Recipes

Return configured pipelines that turn inputs into finished results.

Every recipe is a function that returns an ordinary scikit-learn Pipeline built only from the public components, the primitives of skai.pipeline and scikit-learn, so its steps can be inspected, changed with set_params, cloned and tuned. Every recipe takes a backend, passed to every step that uses one, and keyword arguments for its main choices. video_search returns a Retriever, since it answers queries.

Examples:

>>> from skai import recipes
>>> pipeline = recipes.study_pack(backend='dummy')
>>> [name for name, _ in pipeline.steps]
['transcript', 'pack']

audio_description(*, max_length=15, voice=None, backend=None)

Return a pipeline that adds a spoken description of every shot and subtitles of the dialogue to a video.

Every description starts with its shot, and the dialogue is lowered while it plays.

Parameters:

Name	Type	Description	Default
max_length	int	The maximum number of words of every description.	15
voice	str | None	The voice of the descriptions.	None
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, whose predict returns one video per video, as long as the input.

Source code in src/skai/recipes/_recipes.py

def audio_description(*, max_length: int = 15, voice: str | None = None, backend: BackendType = None) -> Pipeline:
    """Return a pipeline that adds a spoken description of every shot and subtitles of the dialogue to a video.

    Every description starts with its shot, and the dialogue is lowered while it plays.

    Args:
        max_length:
            The maximum number of words of every description.

        voice:
            The voice of the descriptions.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, whose `predict` returns one video per video, as long as the input.
    """
    describe = VideoToTextGenerator(
        instructions='Describe what is shown in one short sentence for a blind viewer.',
        max_length=max_length,
        backend=backend,
    )
    shot = MediaUnion(
        [
            ('start', Field('start')),
            ('media', make_pipeline(Field('media'), describe, TextToAudioGenerator(voice=voice, backend=backend))),
        ],
    )
    sources = MediaUnion(
        [
            ('video', 'passthrough'),
            ('dialogue', VideoToAudioGenerator()),
            ('description', make_pipeline(VideoSegmenter(by='shot'), ForEach(shot))),
            ('transcript', make_pipeline(VideoToAudioGenerator(), AudioSegmenter(by='speech', backend=backend))),
        ],
    )
    mixer = MultimodalToAudioGenerator(duck_under=['dialogue'], duck_for='description')
    tracks = MediaUnion(
        [
            ('video', Field('video')),
            ('audio', make_pipeline(Field(['dialogue', 'description']), mixer)),
            ('transcript', Field('transcript')),
        ],
    )
    muxed = MediaUnion(
        [
            ('video', make_pipeline(Field(['video', 'audio']), MultimodalToVideoGenerator(operation='mux'))),
            ('transcript', Field('transcript')),
        ],
    )
    return Pipeline(
        [
            ('sources', sources),
            ('tracks', tracks),
            ('muxed', muxed),
            ('subtitled', MultimodalToVideoGenerator(operation='subtitle')),
        ],
    )

call_topics(*, topics=None, n_clusters=5, random_state=0, backend=None)

Return a pipeline that labels the topic and sentiment of every speaker turn of calls and clusters the turns.

Parameters:

Name	Type	Description	Default
topics	list[str] | None	The topic labels. Billing, technical issue, account, cancellation and other are used when it is None.	None
n_clusters	int	The number of clusters of the turns by meaning.	5
random_state	int | None	Controls the clustering.	0
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, which is fitted on recordings and whose predict returns one list of turn records per call,
Pipeline	each with speaker, text, topic, sentiment and cluster.

Source code in src/skai/recipes/_recipes.py

def call_topics(
    *,
    topics: list[str] | None = None,
    n_clusters: int = 5,
    random_state: int | None = 0,
    backend: BackendType = None,
) -> Pipeline:
    """Return a pipeline that labels the topic and sentiment of every speaker turn of calls and clusters the turns.

    Args:
        topics:
            The topic labels. Billing, technical issue, account, cancellation and other are used when it is `None`.

        n_clusters:
            The number of clusters of the turns by meaning.

        random_state:
            Controls the clustering.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, which is fitted on recordings and whose `predict` returns one list of turn records per call,
        each with `speaker`, `text`, `topic`, `sentiment` and `cluster`.
    """
    topics = topics or ['billing', 'technical issue', 'account', 'cancellation', 'other']
    cluster = make_pipeline(
        Field('text'),
        TextEmbedder(backend=backend),
        KMeans(n_clusters=n_clusters, n_init='auto', random_state=random_state),
        FunctionTransformer(np.argmin, kw_args={'axis': 1}),
    )
    turn = MediaUnion(
        [
            ('speaker', Field('speaker')),
            ('text', Field('text')),
            ('topic', make_pipeline(Field('text'), TextClassifier(labels=topics, backend=backend))),
            (
                'sentiment',
                make_pipeline(
                    Field('text'),
                    TextClassifier(labels=['positive', 'neutral', 'negative'], backend=backend),
                ),
            ),
            ('cluster', cluster),
        ],
    )
    return Pipeline([('turns', AudioSegmenter(by='speaker', backend=backend)), ('label', ForEach(turn))])

dubbing(*, target_language, backend=None)

Return a pipeline that dubs a video into another language in the original voice.

The speech is transcribed and translated, the vocals and the accompaniment are separated, the translation is spoken in the voice of the vocals, the video is lip-synced to it, the accompaniment is mixed back, and subtitles in the new language are burned in.

Parameters:

Name	Type	Description	Default
target_language	str	The language of the dubbed video.	required
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, whose predict returns one video per video, as long as the input.

Source code in src/skai/recipes/_recipes.py

def dubbing(*, target_language: str, backend: BackendType = None) -> Pipeline:
    """Return a pipeline that dubs a video into another language in the original voice.

    The speech is transcribed and translated, the vocals and the accompaniment are separated, the translation is spoken
    in the voice of the vocals, the video is lip-synced to it, the accompaniment is mixed back, and subtitles in the
    new language are burned in.

    Args:
        target_language:
            The language of the dubbed video.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, whose `predict` returns one video per video, as long as the input.
    """
    translation = INSTRUCTIONS['translator'].format(target_language=target_language)
    translate = TextToTextGenerator(instructions=translation, backend=backend)
    sources = MediaUnion(
        [
            ('video', 'passthrough'),
            ('text', make_pipeline(VideoToTextGenerator(operation='transcribe', backend=backend), translate)),
            (
                'voice',
                make_pipeline(
                    VideoToAudioGenerator(),
                    AudioToAudioGenerator(operation='separate', stem='vocals', backend=backend),
                ),
            ),
            (
                'accompaniment',
                make_pipeline(
                    VideoToAudioGenerator(),
                    AudioToAudioGenerator(operation='separate', stem='accompaniment', backend=backend),
                ),
            ),
        ],
    )
    speech = MediaUnion(
        [
            ('video', Field('video')),
            (
                'audio',
                make_pipeline(
                    Field(['text', 'voice']),
                    MultimodalToAudioGenerator(operation='speech', backend=backend),
                ),
            ),
            ('accompaniment', Field('accompaniment')),
        ],
    )
    fitted = MediaUnion(
        [
            ('speaking', make_pipeline(Field(['video', 'audio']), MultimodalToVideoGenerator(operation='mux'))),
            ('accompaniment', Field('accompaniment')),
            ('transcript', make_pipeline(Field('audio'), AudioSegmenter(by='speech', backend=backend))),
        ],
    )
    synced = MediaUnion(
        [
            ('video', Field('speaking')),
            (
                'mix',
                make_pipeline(Field(['speaking', 'accompaniment']), MultimodalToAudioGenerator(duck_for='speaking')),
            ),
            ('transcript', Field('transcript')),
        ],
    )
    lipsynced = MediaUnion(
        [
            (
                'video',
                make_pipeline(
                    MediaUnion([('video', Field('video')), ('audio', Field('video'))]),
                    MultimodalToVideoGenerator(operation='lip_sync', backend=backend),
                ),
            ),
            ('audio', Field('mix')),
            ('transcript', Field('transcript')),
        ],
    )
    muxed = MediaUnion(
        [
            ('video', make_pipeline(Field(['video', 'audio']), MultimodalToVideoGenerator(operation='mux'))),
            ('transcript', Field('transcript')),
        ],
    )
    return Pipeline(
        [
            ('sources', sources),
            ('speech', speech),
            ('fitted', fitted),
            ('synced', synced),
            ('lipsynced', lipsynced),
            ('muxed', muxed),
            ('subtitled', MultimodalToVideoGenerator(operation='subtitle')),
        ],
    )

explainer_video(*, n_scenes='a few', size=(1280, 720), scene_seconds=4.0, voice=None, backend=None)

Return a pipeline that turns a brief into one narrated, subtitled video with music.

A storyboard splits the brief into scenes, every scene gets a generated image animated into a clip and a spoken narration, the clips and narrations are joined, music is mixed under the narration, and the narration is burned in as subtitles.

Parameters:

Name	Type	Description	Default
n_scenes	int | str	The number of scenes, or words such as 'a few'.	'a few'
size	tuple[int, int]	The width and height of the video.	(1280, 720)
scene_seconds	float	The length of every scene clip in seconds.	4.0
voice	str | None	The voice of the narration.	None
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, whose predict returns one video per brief.

Source code in src/skai/recipes/_recipes.py

def explainer_video(
    *,
    n_scenes: int | str = 'a few',
    size: tuple[int, int] = (1280, 720),
    scene_seconds: float = 4.0,
    voice: str | None = None,
    backend: BackendType = None,
) -> Pipeline:
    """Return a pipeline that turns a brief into one narrated, subtitled video with music.

    A storyboard splits the brief into scenes, every scene gets a generated image animated into a clip and a spoken
    narration, the clips and narrations are joined, music is mixed under the narration, and the narration is burned
    in as subtitles.

    Args:
        n_scenes:
            The number of scenes, or words such as `'a few'`.

        size:
            The width and height of the video.

        scene_seconds:
            The length of every scene clip in seconds.

        voice:
            The voice of the narration.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, whose `predict` returns one video per brief.
    """
    steps = [('plan', _plan(backend, n_scenes, music=True))]
    return Pipeline(steps + _narrated_video(backend, size=size, scene_seconds=scene_seconds, voice=voice, music=True))

highlights(*, k=5, music_seconds=30.0, backend=None)

Return a pipeline that turns a long video into a reel of its most exciting shots with music.

Parameters:

Name	Type	Description	Default
k	int	The number of shots of the reel, kept in their original order.	5
music_seconds	float	The length of the music in seconds.	30.0
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, whose predict returns one reel per video.

Source code in src/skai/recipes/_recipes.py

def highlights(*, k: int = 5, music_seconds: float = 30.0, backend: BackendType = None) -> Pipeline:
    """Return a pipeline that turns a long video into a reel of its most exciting shots with music.

    Args:
        k:
            The number of shots of the reel, kept in their original order.

        music_seconds:
            The length of the music in seconds.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, whose `predict` returns one reel per video.
    """
    reel = make_pipeline(
        VideoSegmenter(by='shot'),
        ForEach(Field('media')),
        TopK(VideoClassifier(labels=['exciting', 'calm'], backend=backend), label='exciting', k=k),
        VideoToVideoGenerator(operation='concatenate'),
    )
    music = make_pipeline(
        VideoToTextGenerator(
            instructions='Describe energetic music that fits the video, in one sentence.',
            backend=backend,
        ),
        TextToAudioGenerator(operation='music', duration=music_seconds, backend=backend),
    )
    mixer = MultimodalToAudioGenerator(duck_under=['music'], duck_for='video', volumes={'music': 0.6})
    tracks = MediaUnion([('video', Field('video')), ('audio', make_pipeline(Field(['video', 'music']), mixer))])
    return Pipeline(
        [
            ('parts', MediaUnion([('video', reel), ('music', music)])),
            ('tracks', tracks),
            ('muxed', MultimodalToVideoGenerator(operation='mux')),
        ],
    )

image_translation(*, target_language, backend=None)

Return a pipeline that redraws the text of images, such as comics or screenshots, in another language.

The text of every image is transcribed and translated into an edit instruction, the text regions are masked, and the image is edited inside the mask.

Parameters:

Name	Type	Description	Default
target_language	str	The language of the redrawn text.	required
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, whose predict returns one image per image, of the input size.

Source code in src/skai/recipes/_recipes.py

def image_translation(*, target_language: str, backend: BackendType = None) -> Pipeline:
    """Return a pipeline that redraws the text of images, such as comics or screenshots, in another language.

    The text of every image is transcribed and translated into an edit instruction, the text regions are masked, and
    the image is edited inside the mask.

    Args:
        target_language:
            The language of the redrawn text.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, whose `predict` returns one image per image, of the input size.
    """
    instructions = (
        f'Translate the text into {target_language}. Answer with an image edit instruction of the form: Replace the '
        'text with "<translation>", keeping the font, size, color and layout.'
    )
    parts = MediaUnion(
        [
            ('image', 'passthrough'),
            (
                'instructions',
                make_pipeline(
                    ImageToTextGenerator(operation='transcribe', backend=backend),
                    TextToTextGenerator(instructions=instructions, backend=backend),
                ),
            ),
            (
                'mask',
                make_pipeline(
                    ImageSegmenter(by='mask', prompt='written text', backend=backend),
                    FunctionTransformer(_merge_masks),
                ),
            ),
        ],
    )
    return Pipeline([('parts', parts), ('redraw', MultimodalToImageGenerator(backend=backend))])

invoice_anomalies(*, contamination='auto', backend=None)

Return a pipeline that scores invoice scans as normal or anomalous from their vendor, amount and date.

Parameters:

Name	Type	Description	Default
contamination	float | str	The expected share of anomalies, passed to IsolationForest.	'auto'
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, which is fitted on scans and whose predict returns 1 for normal and -1 for anomalous scans.

Source code in src/skai/recipes/_recipes.py

def invoice_anomalies(*, contamination: float | str = 'auto', backend: BackendType = None) -> Pipeline:
    """Return a pipeline that scores invoice scans as normal or anomalous from their vendor, amount and date.

    Args:
        contamination:
            The expected share of anomalies, passed to `IsolationForest`.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, which is fitted on scans and whose `predict` returns `1` for normal and `-1` for anomalous scans.
    """
    fields = {'vendor': 'vendor name', 'amount': 'total amount', 'date': 'invoice date as YYYY-MM-DD'}
    features = ColumnTransformer(
        [
            ('vendor', OrdinalEncoder(handle_unknown='use_encoded_value', unknown_value=-1), [0]),
            ('amount', FunctionTransformer(_to_number), [1]),
            ('date', FunctionTransformer(_to_ordinal), [2]),
        ],
    )
    return Pipeline(
        [
            ('fields', ImageExtractor(fields=fields, backend=backend)),
            ('features', features),
            ('score', IsolationForest(contamination=contamination, random_state=0)),
        ],
    )

meeting_briefing(*, avatar, max_length=80, backend=None)

Return a pipeline that turns a meeting recording into minutes, action items and a talking-head briefing.

Parameters:

Name	Type	Description	Default
avatar	Any	The portrait that reads the briefing.	required
max_length	int	The maximum number of words of the briefing.	80
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, whose predict returns one dict per recording with minutes, action_items and video.

Source code in src/skai/recipes/_recipes.py

def meeting_briefing(*, avatar: Any, max_length: int = 80, backend: BackendType = None) -> Pipeline:
    """Return a pipeline that turns a meeting recording into minutes, action items and a talking-head briefing.

    Args:
        avatar:
            The portrait that reads the briefing.

        max_length:
            The maximum number of words of the briefing.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, whose `predict` returns one dict per recording with `minutes`, `action_items` and `video`.
    """
    minutes = TextToTextGenerator(
        instructions='Write the minutes of the meeting: decisions, open questions and key points.',
        backend=backend,
    )
    action_items = TextExtractor(
        fields={'action_items': [{'owner': 'who does it', 'task': 'what to do', 'due': 'when, if stated'}]},
        backend=backend,
    )
    briefing = TextToTextGenerator(
        instructions='Write a short spoken briefing of the meeting for people who missed it.',
        max_length=max_length,
        backend=backend,
    )
    video = make_pipeline(
        briefing,
        TextToAudioGenerator(backend=backend),
        AudioToVideoGenerator(avatar=avatar, backend=backend),
    )
    return Pipeline(
        [
            ('transcript', AudioToTextGenerator(backend=backend)),
            ('briefing', MediaUnion([('minutes', minutes), ('action_items', action_items), ('video', video)])),
        ],
    )

moderation(*, labels=None, cv=3, backend=None)

Return a pipeline that classifies posts of an image and a caption, calibrated on labeled posts.

Every post is classified with its image and caption in one request, with labeled posts as few-shot examples, and the scores are calibrated with cross-validation into probabilities.

Parameters:

Name	Type	Description	Default
labels	list[str] | None	The labels, ['safe', 'unsafe'] when it is None.	None
cv	int	The number of cross-validation folds of the calibration.	3
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, which is fitted on posts, as dicts with image and caption, and their labels.

Source code in src/skai/recipes/_recipes.py

def moderation(*, labels: list[str] | None = None, cv: int = 3, backend: BackendType = None) -> Pipeline:
    """Return a pipeline that classifies posts of an image and a caption, calibrated on labeled posts.

    Every post is classified with its image and caption in one request, with labeled posts as few-shot examples, and
    the scores are calibrated with cross-validation into probabilities.

    Args:
        labels:
            The labels, `['safe', 'unsafe']` when it is `None`.

        cv:
            The number of cross-validation folds of the calibration.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, which is fitted on posts, as dicts with `image` and `caption`, and their labels.
    """
    labels = labels or ['safe', 'unsafe']
    instructions = INSTRUCTIONS['moderator'].format(kind='post, judging its image and caption together,')
    classifier = MultimodalClassifier(labels=labels, instructions=instructions, backend=backend)
    return Pipeline([('moderate', CalibratedClassifierCV(classifier, cv=cv))])

podcast(*, voices=('alloy', 'echo'), music_seconds=8.0, backend=None)

Return a pipeline that turns an article into a two-host podcast episode with intro music.

The article is summarized and written as a conversation of host A and host B, every line is spoken in the voice of its host, the lines are joined in order, and generated music is mixed under them.

Parameters:

Name	Type	Description	Default
voices	tuple[str, str]	The voices of host A and host B.	('alloy', 'echo')
music_seconds	float	The length of the music in seconds.	8.0
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, whose predict returns one audio episode per article.

Source code in src/skai/recipes/_recipes.py

def podcast(
    *,
    voices: tuple[str, str] = ('alloy', 'echo'),
    music_seconds: float = 8.0,
    backend: BackendType = None,
) -> Pipeline:
    """Return a pipeline that turns an article into a two-host podcast episode with intro music.

    The article is summarized and written as a conversation of host A and host B, every line is spoken in the voice
    of its host, the lines are joined in order, and generated music is mixed under them.

    Args:
        voices:
            The voices of host A and host B.

        music_seconds:
            The length of the music in seconds.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, whose `predict` returns one audio episode per article.
    """
    script = TextExtractor(
        fields={'lines': [{'host': 'A or B', 'text': 'what the host says'}]},
        instructions='Write a lively podcast conversation between host A and host B that covers the text.',
        backend=backend,
    )
    voice = make_pipeline(Field('host'), FunctionTransformer(_pick_voice, kw_args={'voices': voices}))
    line = make_pipeline(
        MediaUnion([('text', Field('text')), ('voice', voice)]),
        MultimodalToAudioGenerator(operation='speech', backend=backend),
    )
    speech = make_pipeline(script, ForEach(line), AudioToAudioGenerator(operation='concatenate', gap=0.3))
    music_prompt = TextToTextGenerator(
        instructions='Describe short intro music for a podcast about the text, in one sentence.',
        backend=backend,
    )
    music = make_pipeline(
        music_prompt,
        TextToAudioGenerator(operation='music', duration=music_seconds, backend=backend),
    )
    return Pipeline(
        [
            ('summary', summarizer(backend=backend)),
            ('parts', MediaUnion([('speech', speech), ('music', music)])),
            ('mix', MultimodalToAudioGenerator(duck_under=['music'], duck_for='speech', volumes={'music': 0.4})),
        ],
    )

product_ad(*, scene='Place the product in an attractive lifestyle scene with soft natural light.', motion='a slow camera orbit around the product', duration=4.0, backend=None)

Return a pipeline that turns product photos into short promotional clips.

Every photo is edited into a scene, upscaled and animated.

Parameters:

Name	Type	Description	Default
scene	str	The edit that places the product in a scene.	'Place the product in an attractive lifestyle scene with soft natural light.'
motion	str	The motion of the clip.	'a slow camera orbit around the product'
duration	float	The length of every clip in seconds.	4.0
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, whose predict returns one video per photo.

Source code in src/skai/recipes/_recipes.py

def product_ad(
    *,
    scene: str = 'Place the product in an attractive lifestyle scene with soft natural light.',
    motion: str = 'a slow camera orbit around the product',
    duration: float = 4.0,
    backend: BackendType = None,
) -> Pipeline:
    """Return a pipeline that turns product photos into short promotional clips.

    Every photo is edited into a scene, upscaled and animated.

    Args:
        scene:
            The edit that places the product in a scene.

        motion:
            The motion of the clip.

        duration:
            The length of every clip in seconds.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, whose `predict` returns one video per photo.
    """
    return Pipeline(
        [
            ('scene', ImageToImageGenerator(operation='edit', instructions=scene, backend=backend)),
            ('upscale', ImageToImageGenerator(operation='upscale', backend=backend)),
            ('animate', ImageToVideoGenerator(instructions=motion, duration=duration, backend=backend)),
        ],
    )

product_catalog(*, attributes=None, backend=None)

Return a pipeline that turns product photos into catalog records with marketing copy.

The background of every photo is removed, and the attributes and a short marketing text are extracted from it.

Parameters:

Name	Type	Description	Default
attributes	dict[str, str] | None	The attribute fields, as names and descriptions. Name, category, color and material are used when it is None.	None
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, whose predict returns one record per photo.

Source code in src/skai/recipes/_recipes.py

def product_catalog(*, attributes: dict[str, str] | None = None, backend: BackendType = None) -> Pipeline:
    """Return a pipeline that turns product photos into catalog records with marketing copy.

    The background of every photo is removed, and the attributes and a short marketing text are extracted from it.

    Args:
        attributes:
            The attribute fields, as names and descriptions. Name, category, color and material are used when it is
            `None`.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, whose `predict` returns one record per photo.
    """
    fields: dict[str, Any] = attributes or {
        'name': 'product name',
        'category': 'product category',
        'color': 'main color',
        'material': 'main material',
    }
    fields = {**fields, 'copy': 'two sentences of marketing copy for the product'}
    return Pipeline(
        [
            ('cutout', ImageToImageGenerator(operation='remove_background', backend=backend)),
            ('record', ImageExtractor(fields=fields, backend=backend)),
        ],
    )

quality_control(*, defects=None, backend=None)

Return a pipeline that classifies product photos as defective or not, trained on labeled photos.

A visual question answer about defects and the detected defect regions of every photo are combined into features of a logistic regression.

Parameters:

Name	Type	Description	Default
defects	list[str] | None	The kinds of defects to detect. Scratches, dents, cracks and stains are used when it is None.	None
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, which is fitted on photos and their labels.

Source code in src/skai/recipes/_recipes.py

def quality_control(*, defects: list[str] | None = None, backend: BackendType = None) -> Pipeline:
    """Return a pipeline that classifies product photos as defective or not, trained on labeled photos.

    A visual question answer about defects and the detected defect regions of every photo are combined into features
    of a logistic regression.

    Args:
        defects:
            The kinds of defects to detect. Scratches, dents, cracks and stains are used when it is `None`.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, which is fitted on photos and their labels.
    """
    defects = defects or ['scratch', 'dent', 'crack', 'stain']
    question = ImageToTextGenerator(
        instructions=(
            f'Does the product show a defect such as a {", ".join(defects)}? Answer yes or no, then describe it.'
        ),
        backend=backend,
    )
    features = FeatureUnion(
        [
            ('answer', make_pipeline(question, TextEmbedder(backend=backend))),
            (
                'regions',
                make_pipeline(
                    ImageSegmenter(labels=defects, backend=backend),
                    FunctionTransformer(_region_features),
                ),
            ),
        ],
    )
    return Pipeline([('features', features), ('classify', LogisticRegression(max_iter=1000))])

read_along(*, characters='the main characters of the story, in a friendly picture-book style', size=(1024, 768), page_seconds=6.0, voice=None, backend=None)

Return a pipeline that turns a story into a read-along video with consistent characters and word subtitles.

fit generates one reference image of the characters, which every page image keeps consistent, and the narration is shown word by word.

Parameters:

Name	Type	Description	Default
characters	str	A description of the characters, from which the reference image is generated.	'the main characters of the story, in a friendly picture-book style'
size	tuple[int, int]	The width and height of the pages.	(1024, 768)
page_seconds	float	The length of every page in seconds.	6.0
voice	str | None	The voice of the narration.	None
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, whose predict returns one video per story.

Source code in src/skai/recipes/_recipes.py

def read_along(
    *,
    characters: str = 'the main characters of the story, in a friendly picture-book style',
    size: tuple[int, int] = (1024, 768),
    page_seconds: float = 6.0,
    voice: str | None = None,
    backend: BackendType = None,
) -> Pipeline:
    """Return a pipeline that turns a story into a read-along video with consistent characters and word subtitles.

    `fit` generates one reference image of the characters, which every page image keeps consistent, and the narration
    is shown word by word.

    Args:
        characters:
            A description of the characters, from which the reference image is generated.

        size:
            The width and height of the pages.

        page_seconds:
            The length of every page in seconds.

        voice:
            The voice of the narration.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, whose `predict` returns one video per story.
    """
    steps = [('plan', _plan(backend, 'one per page', music=False))]
    pages = _narrated_video(
        backend,
        size=size,
        scene_seconds=page_seconds,
        voice=voice,
        music=False,
        reference_prompt=characters,
        subtitles_by='word',
    )
    return Pipeline(steps + pages)

study_pack(*, n_questions=5, voice=None, backend=None)

Return a pipeline that turns a lecture video into notes, key terms, a quiz and a narrated recap.

Parameters:

Name	Type	Description	Default
n_questions	int	The number of quiz questions.	5
voice	str | None	The voice of the recap.	None
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, whose predict returns one dict per lecture with notes, key_terms, quiz and recap.

Source code in src/skai/recipes/_recipes.py

def study_pack(*, n_questions: int = 5, voice: str | None = None, backend: BackendType = None) -> Pipeline:
    """Return a pipeline that turns a lecture video into notes, key terms, a quiz and a narrated recap.

    Args:
        n_questions:
            The number of quiz questions.

        voice:
            The voice of the recap.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, whose `predict` returns one dict per lecture with `notes`, `key_terms`, `quiz` and `recap`.
    """
    notes = TextToTextGenerator(
        instructions='Write structured study notes of the lecture, with headings and bullet points.',
        backend=backend,
    )
    key_terms = TextExtractor(
        fields={'key_terms': [{'term': 'a key term', 'definition': 'its meaning'}]},
        backend=backend,
    )
    recap = make_pipeline(
        summarizer(max_length=60, backend=backend),
        TextToAudioGenerator(voice=voice, backend=backend),
    )
    return Pipeline(
        [
            ('transcript', VideoToTextGenerator(operation='transcribe', backend=backend)),
            (
                'pack',
                MediaUnion(
                    [
                        ('notes', notes),
                        ('key_terms', key_terms),
                        ('quiz', quiz_writer(n_questions=n_questions, backend=backend)),
                        ('recap', recap),
                    ],
                ),
            ),
        ],
    )

vertical_clip(*, max_length=60, size=(1280, 720), scene_seconds=3.0, voice=None, backend=None)

Return a pipeline that turns a post into a short vertical video with burned subtitles.

The post is shortened, turned into an explainer video, and reframed to 9:16 following its main subject.

Parameters:

Name	Type	Description	Default
max_length	int	The maximum number of words of the shortened post.	60
size	tuple[int, int]	The width and height of the generated scenes, before reframing.	(1280, 720)
scene_seconds	float	The length of every scene clip in seconds.	3.0
voice	str | None	The voice of the narration.	None
backend	BackendType	The backend of every step that uses one.	None

Returns:

Type	Description
Pipeline	The pipeline, whose predict returns one vertical video per post.

Source code in src/skai/recipes/_recipes.py

def vertical_clip(
    *,
    max_length: int = 60,
    size: tuple[int, int] = (1280, 720),
    scene_seconds: float = 3.0,
    voice: str | None = None,
    backend: BackendType = None,
) -> Pipeline:
    """Return a pipeline that turns a post into a short vertical video with burned subtitles.

    The post is shortened, turned into an explainer video, and reframed to 9:16 following its main subject.

    Args:
        max_length:
            The maximum number of words of the shortened post.

        size:
            The width and height of the generated scenes, before reframing.

        scene_seconds:
            The length of every scene clip in seconds.

        voice:
            The voice of the narration.

        backend:
            The backend of every step that uses one.

    Returns:
        The pipeline, whose `predict` returns one vertical video per post.
    """
    instructions = 'Rewrite the post as a short, punchy script for a vertical video.'
    steps: list[tuple[str, Any]] = [
        ('shorten', TextToTextGenerator(instructions=instructions, max_length=max_length, backend=backend)),
        ('plan', _plan(backend, 'three', music=True)),
    ]
    video = _narrated_video(
        backend,
        size=size,
        scene_seconds=scene_seconds,
        voice=voice,
        music=True,
        aspect_ratio=9 / 16,
    )
    return Pipeline(steps + video)

video_search(*, threshold=0.5, n_frames=4, backend=None)

Return a retriever that indexes the shots of a video library and finds the closest shots of text queries.

Shots and texts are embedded with one joint text and image model, CLIP on Hugging Face by default.

Parameters:

Name	Type	Description	Default
threshold	float	The histogram distance between consecutive frames above which a cut is found.	0.5
n_frames	int	The number of frames embedded per shot.	4
backend	BackendType	The backend of the embedders. It must embed texts and images in one space.	None

Returns:

Type	Description
Retriever	The retriever, whose kneighbors returns shot records with sample, start, end and distance.

Source code in src/skai/recipes/_recipes.py

def video_search(*, threshold: float = 0.5, n_frames: int = 4, backend: BackendType = None) -> Retriever:
    """Return a retriever that indexes the shots of a video library and finds the closest shots of text queries.

    Shots and texts are embedded with one joint text and image model, CLIP on Hugging Face by default.

    Args:
        threshold:
            The histogram distance between consecutive frames above which a cut is found.

        n_frames:
            The number of frames embedded per shot.

        backend:
            The backend of the embedders. It must embed texts and images in one space.

    Returns:
        The retriever, whose `kneighbors` returns shot records with `sample`, `start`, `end` and `distance`.
    """
    backend = JOINT_MODEL if backend is None else backend
    shot = MediaUnion(
        [
            ('start', Field('start')),
            ('end', Field('end')),
            ('vector', make_pipeline(Field('media'), VideoEmbedder(n_frames=n_frames, center=False, backend=backend))),
        ],
    )
    items = make_pipeline(VideoSegmenter(by='shot', threshold=threshold), ForEach(shot))
    return Retriever(items, TextEmbedder(center=False, backend=backend))