Backends

Backends that run component tasks on open-source models and cloud APIs.

DummyBackend(model=None, *, options=None, max_concurrency=1)

Bases: Backend

Return deterministic placeholder outputs for every task, without network access or model weights.

Outputs depend only on the request, so repeated calls return the same results. The backend is meant for tests and documentation.

Parameters:

Name	Type	Description	Default
model	str | None	Ignored.	None
options	dict | None	Ignored.	None
max_concurrency	int	The maximum number of requests that run at the same time.	1

Examples:

>>> from skai.generation import TextToTextGenerator
>>> generator = TextToTextGenerator(instructions='Summarize.', max_length=3, backend='dummy').fit()
>>> generator.predict(['One two three four five.']).tolist()
['One two three']

Source code in src/skai/core/_backend.py

def __init__(self, model: str | None = None, *, options: dict | None = None, max_concurrency: int = 1) -> None:
    self.model = model
    self.options = options
    self.max_concurrency = max_concurrency

animate_avatar(request)

Return the still avatar repeated for the length of the audio.

Parameters:

Name	Type	Description	Default
request	AvatarRequest	The avatar request.	required

Returns:

Type	Description
Video	A video at 25 frames per second with the audio as its track.

Source code in src/skai/backends/_dummy.py

def animate_avatar(self, request: AvatarRequest) -> Video:
    """Return the still avatar repeated for the length of the audio.

    Args:
        request:
            The avatar request.

    Returns:
        A video at 25 frames per second with the audio as its track.
    """
    avatar = request.avatar.convert('RGB')
    avatar.thumbnail((AVATAR_SIZE, AVATAR_SIZE))
    n_frames = max(1, round(request.audio.duration * FRAMES_PER_SECOND))
    frames = np.repeat(np.asarray(avatar)[np.newaxis], n_frames, axis=0)
    return Video(frames, FRAMES_PER_SECOND, request.audio)

animate_image(request)

Return the still image repeated for the length of the video.

Parameters:

Name	Type	Description	Default
request	AnimationRequest	The animation request.	required

Returns:

Type	Description
Video	A video of the requested duration, at fps or 8 frames per second.

Source code in src/skai/backends/_dummy.py

def animate_image(self, request: AnimationRequest) -> Video:
    """Return the still image repeated for the length of the video.

    Args:
        request:
            The animation request.

    Returns:
        A video of the requested duration, at `fps` or 8 frames per second.
    """
    return _still_video(request.image, request.duration, request.fps or VIDEO_FPS)

classify_audio(request)

Return deterministic scores derived from the audio.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The classification request.	required

Returns:

Type	Description
dict[str, float]	A score per label. The scores sum to one unless the request is multilabel.

Source code in src/skai/backends/_dummy.py

def classify_audio(self, request: ClassificationRequest) -> dict[str, float]:
    """Return deterministic scores derived from the audio.

    Args:
        request:
            The classification request.

    Returns:
        A score per label. The scores sum to one unless the request is multilabel.
    """
    return _score_labels(request)

classify_image(request)

Return deterministic scores derived from the image.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The classification request.	required

Returns:

Type	Description
dict[str, float]	A score per label. The scores sum to one unless the request is multilabel.

Source code in src/skai/backends/_dummy.py

def classify_image(self, request: ClassificationRequest) -> dict[str, float]:
    """Return deterministic scores derived from the image.

    Args:
        request:
            The classification request.

    Returns:
        A score per label. The scores sum to one unless the request is multilabel.
    """
    return _score_labels(request)

classify_multimodal(request)

Return deterministic scores derived from every part of the sample.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The classification request.	required

Returns:

Type	Description
dict[str, float]	A score per label. The scores sum to one unless the request is multilabel.

Source code in src/skai/backends/_dummy.py

def classify_multimodal(self, request: ClassificationRequest) -> dict[str, float]:
    """Return deterministic scores derived from every part of the sample.

    Args:
        request:
            The classification request.

    Returns:
        A score per label. The scores sum to one unless the request is multilabel.
    """
    return _score_labels(request)

classify_text(request)

Return deterministic scores derived from the text.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The classification request.	required

Returns:

Type	Description
dict[str, float]	A score per label. The scores sum to one unless the request is multilabel.

Source code in src/skai/backends/_dummy.py

def classify_text(self, request: ClassificationRequest) -> dict[str, float]:
    """Return deterministic scores derived from the text.

    Args:
        request:
            The classification request.

    Returns:
        A score per label. The scores sum to one unless the request is multilabel.
    """
    return _score_labels(request)

classify_video(request)

Return deterministic scores derived from the sampled frames.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The classification request.	required

Returns:

Type	Description
dict[str, float]	A score per label. The scores sum to one unless the request is multilabel.

Source code in src/skai/backends/_dummy.py

def classify_video(self, request: ClassificationRequest) -> dict[str, float]:
    """Return deterministic scores derived from the sampled frames.

    Args:
        request:
            The classification request.

    Returns:
        A score per label. The scores sum to one unless the request is multilabel.
    """
    return _score_labels(request)

convert_voice(request)

Return a tone as long as the input whose pitch depends on the target voice.

Parameters:

Name	Type	Description	Default
request	AudioOperationRequest	The audio operation request.	required

Returns:

Type	Description
Audio	A mono tone at 16 kHz.

Source code in src/skai/backends/_dummy.py

def convert_voice(self, request: AudioOperationRequest) -> Audio:
    """Return a tone as long as the input whose pitch depends on the target voice.

    Args:
        request:
            The audio operation request.

    Returns:
        A mono tone at 16 kHz.
    """
    return _tone(request.audio.duration, request.voice, request.seed)

detect_objects(request)

Return one box per label, or one box for the prompt.

Parameters:

Name	Type	Description	Default
request	RegionRequest	The region request.	required

Returns:

Type	Description
list[Region]	The regions, each covering a quarter of the image.

Source code in src/skai/backends/_dummy.py

def detect_objects(self, request: RegionRequest) -> list[Region]:
    """Return one box per label, or one box for the prompt.

    Args:
        request:
            The region request.

    Returns:
        The regions, each covering a quarter of the image.
    """
    labels = request.labels or (request.prompt or 'object',)
    return [Region(label, MOMENT_SCORE, _box(request.image.size, label)) for label in labels]

diarize(request)

Split the audio into two halves spoken by two speakers.

Parameters:

Name	Type	Description	Default
request	DiarizationRequest	The diarization request.	required

Returns:

Type	Description
Transcript	The transcript, with two segments.

Source code in src/skai/backends/_dummy.py

def diarize(self, request: DiarizationRequest) -> Transcript:
    """Split the audio into two halves spoken by two speakers.

    Args:
        request:
            The diarization request.

    Returns:
        The transcript, with two segments.
    """
    half = request.audio.duration / 2
    segments = (
        Segment(0.0, half, f'First half of {half:.2f} seconds.', 'SPEAKER_0'),
        Segment(half, 2 * half, f'Second half of {half:.2f} seconds.', 'SPEAKER_1'),
    )
    return Transcript(' '.join(segment.text for segment in segments), segments)

edit_image(request)

Blend the image with a color derived from the instruction, inside the mask when one is given.

Parameters:

Name	Type	Description	Default
request	ImageEditRequest	The image edit request.	required

Returns:

Type	Description
Image	The edited image, of the size of the input.

Source code in src/skai/backends/_dummy.py

def edit_image(self, request: ImageEditRequest) -> Image.Image:
    """Blend the image with a color derived from the instruction, inside the mask when one is given.

    Args:
        request:
            The image edit request.

    Returns:
        The edited image, of the size of the input.
    """
    original = request.image.convert('RGB')
    tint = Image.new('RGB', original.size, _color_of(request.instruction, request.seed))
    edited = Image.blend(original, tint, EDIT_STRENGTH)
    if request.mask is None:
        return edited
    return Image.composite(edited, original, request.mask.convert('L').resize(original.size))

embed_audio(request)

Return a deterministic random vector derived from the audio.

Parameters:

Name	Type	Description	Default
request	EmbeddingRequest	The embedding request.	required

Returns:

Type	Description
NDArray[floating]	A vector with dimensions entries, or 8 when dimensions is None.

Source code in src/skai/backends/_dummy.py

def embed_audio(self, request: EmbeddingRequest) -> NDArray[np.floating]:
    """Return a deterministic random vector derived from the audio.

    Args:
        request:
            The embedding request.

    Returns:
        A vector with `dimensions` entries, or 8 when `dimensions` is `None`.
    """
    return self.embed_text(request)

embed_image(request)

Return a deterministic random vector derived from the image.

Parameters:

Name	Type	Description	Default
request	EmbeddingRequest	The embedding request.	required

Returns:

Type	Description
NDArray[floating]	A vector with dimensions entries, or 8 when dimensions is None.

Source code in src/skai/backends/_dummy.py

def embed_image(self, request: EmbeddingRequest) -> NDArray[np.floating]:
    """Return a deterministic random vector derived from the image.

    Args:
        request:
            The embedding request.

    Returns:
        A vector with `dimensions` entries, or 8 when `dimensions` is `None`.
    """
    return self.embed_text(request)

embed_text(request)

Return a deterministic random vector derived from the text.

Parameters:

Name	Type	Description	Default
request	EmbeddingRequest	The embedding request.	required

Returns:

Type	Description
NDArray[floating]	A vector with dimensions entries, or 8 when dimensions is None.

Source code in src/skai/backends/_dummy.py

def embed_text(self, request: EmbeddingRequest) -> NDArray[np.floating]:
    """Return a deterministic random vector derived from the text.

    Args:
        request:
            The embedding request.

    Returns:
        A vector with `dimensions` entries, or 8 when `dimensions` is `None`.
    """
    rng = np.random.default_rng(_seed_of(request.sample))
    return rng.standard_normal(request.dimensions or EMBEDDING_DIMENSIONS)

enhance_audio(request)

Return the audio clipped to [-1, 1] as the denoised copy.

Parameters:

Name	Type	Description	Default
request	AudioOperationRequest	The audio operation request.	required

Returns:

Type	Description
Audio	Audio as long as the input.

Source code in src/skai/backends/_dummy.py

def enhance_audio(self, request: AudioOperationRequest) -> Audio:
    """Return the audio clipped to `[-1, 1]` as the denoised copy.

    Args:
        request:
            The audio operation request.

    Returns:
        Audio as long as the input.
    """
    mono = request.audio.to_mono()
    return Audio(np.clip(mono.samples, -1.0, 1.0).astype(np.float32), mono.sample_rate)

extract_fields(request)

Fill every field with words of the sample, following the nesting of the declarations.

Parameters:

Name	Type	Description	Default
request	ExtractionRequest	The extraction request.	required

Returns:

Type	Description
dict[str, Any]	The value of every field. A list field holds two items.

Source code in src/skai/backends/_dummy.py

def extract_fields(self, request: ExtractionRequest) -> dict[str, Any]:
    """Fill every field with words of the sample, following the nesting of the declarations.

    Args:
        request:
            The extraction request.

    Returns:
        The value of every field. A list field holds two items.
    """
    words = _describe(request.sample).split()
    return {name: _fill(declaration, words, _seed_of(name, request.sample)) for name, declaration in request.fields}

find_moments(request)

Return the middle third of the audio or video as the one matching moment.

Parameters:

Name	Type	Description	Default
request	MomentRequest	The moment request.	required

Returns:

Type	Description
list[Moment]	One moment.

Source code in src/skai/backends/_dummy.py

def find_moments(self, request: MomentRequest) -> list[Moment]:
    """Return the middle third of the audio or video as the one matching moment.

    Args:
        request:
            The moment request.

    Returns:
        One moment.
    """
    third = request.sample.duration / 3
    return [Moment(third, 2 * third, MOMENT_SCORE)]

fine_tune(request)

Return a model id derived from the training prompts and answers, without training anything.

Parameters:

Name	Type	Description	Default
request	FineTuneRequest	The fine-tuning request.	required

Returns:

Type	Description
str	A deterministic model id.

Source code in src/skai/backends/_dummy.py

def fine_tune(self, request: FineTuneRequest) -> str:
    """Return a model id derived from the training prompts and answers, without training anything.

    Args:
        request:
            The fine-tuning request.

    Returns:
        A deterministic model id.
    """
    parts = [part for prompt, answer in request.examples for part in (prompt.user, answer)]
    return f'dummy-finetuned-{_seed_of(request.task, *parts)}'

generate_image(request)

Return an image of one color derived from the request.

Parameters:

Name	Type	Description	Default
request	ImageGenerationRequest	The image generation request.	required

Returns:

Type	Description
Image	An RGB image of the requested size.

Source code in src/skai/backends/_dummy.py

def generate_image(self, request: ImageGenerationRequest) -> Image.Image:
    """Return an image of one color derived from the request.

    Args:
        request:
            The image generation request.

    Returns:
        An RGB image of the requested size.
    """
    color = _color_of(
        request.prompt,
        request.style,
        request.negative_prompt,
        request.seed,
        request.reference_images,
    )
    return Image.new('RGB', request.size, color)

generate_music(request)

Return a sine tone of the requested length.

Parameters:

Name	Type	Description	Default
request	AudioGenerationRequest	The audio generation request.	required

Returns:

Type	Description
Audio	A mono tone at 16 kHz.

Source code in src/skai/backends/_dummy.py

def generate_music(self, request: AudioGenerationRequest) -> Audio:
    """Return a sine tone of the requested length.

    Args:
        request:
            The audio generation request.

    Returns:
        A mono tone at 16 kHz.
    """
    return _tone(request.duration, 'music', request.prompt, request.seed)

generate_sound(request)

Return a sine tone of the requested length.

Parameters:

Name	Type	Description	Default
request	AudioGenerationRequest	The audio generation request.	required

Returns:

Type	Description
Audio	A mono tone at 16 kHz.

Source code in src/skai/backends/_dummy.py

def generate_sound(self, request: AudioGenerationRequest) -> Audio:
    """Return a sine tone of the requested length.

    Args:
        request:
            The audio generation request.

    Returns:
        A mono tone at 16 kHz.
    """
    return _tone(request.duration, 'sound', request.prompt, request.seed)

generate_text(request)

Return a sentence about the sample, with the glossary applied and cut to the maximum length.

Parameters:

Name	Type	Description	Default
request	TextGenerationRequest	The text generation request.	required

Returns:

Type	Description
str	The text itself for a text sample, or a sentence that states the size of other samples.

Source code in src/skai/backends/_dummy.py

def generate_text(self, request: TextGenerationRequest) -> str:
    """Return a sentence about the sample, with the glossary applied and cut to the maximum length.

    Args:
        request:
            The text generation request.

    Returns:
        The text itself for a text sample, or a sentence that states the size of other samples.
    """
    text = _describe(request.sample)
    for term, rendering in request.glossary:
        text = text.replace(term, rendering)
    return ' '.join(text.split()[: request.max_length])

generate_video(request)

Return a video of one color derived from the request.

Parameters:

Name	Type	Description	Default
request	VideoGenerationRequest	The video generation request.	required

Returns:

Type	Description
Video	A video of the requested duration and size, at fps or 8 frames per second.

Source code in src/skai/backends/_dummy.py

def generate_video(self, request: VideoGenerationRequest) -> Video:
    """Return a video of one color derived from the request.

    Args:
        request:
            The video generation request.

    Returns:
        A video of the requested duration and size, at `fps` or 8 frames per second.
    """
    color = _color_of(request.prompt, request.style, request.negative_prompt, request.seed)
    return _still_video(Image.new('RGB', request.size, color), request.duration, request.fps or VIDEO_FPS)

lip_sync(request)

Loop or cut the frames of the video to the length of the audio, with the audio as its track.

Parameters:

Name	Type	Description	Default
request	LipSyncRequest	The lip sync request.	required

Returns:

Type	Description
Video	A video as long as the audio.

Source code in src/skai/backends/_dummy.py

def lip_sync(self, request: LipSyncRequest) -> Video:
    """Loop or cut the frames of the video to the length of the audio, with the audio as its track.

    Args:
        request:
            The lip sync request.

    Returns:
        A video as long as the audio.
    """
    frames = request.video.frames
    count = max(1, round(request.audio.duration * request.video.fps))
    return Video(frames[np.arange(count) % frames.shape[0]], request.video.fps, request.audio)

pick_thumbnail(request)

Pick a frame derived from the frames.

Parameters:

Name	Type	Description	Default
request	ThumbnailRequest	The thumbnail request.	required

Returns:

Type	Description
int	The index of the frame.

Source code in src/skai/backends/_dummy.py

def pick_thumbnail(self, request: ThumbnailRequest) -> int:
    """Pick a frame derived from the frames.

    Args:
        request:
            The thumbnail request.

    Returns:
        The index of the frame.
    """
    return _seed_of(request.frames) % len(request.frames.frames)

remove_background(request)

Keep a centered ellipse of the image and make the rest transparent.

Parameters:

Name	Type	Description	Default
request	ImageOperationRequest	The image operation request.	required

Returns:

Type	Description
Image	An RGBA image of the input size.

Source code in src/skai/backends/_dummy.py

def remove_background(self, request: ImageOperationRequest) -> Image.Image:
    """Keep a centered ellipse of the image and make the rest transparent.

    Args:
        request:
            The image operation request.

    Returns:
        An RGBA image of the input size.
    """
    image = request.image.convert('RGBA')
    alpha = Image.new('L', image.size, 0)
    ImageDraw.Draw(alpha).ellipse((0, 0, image.width - 1, image.height - 1), fill=255)
    image.putalpha(alpha)
    return image

segment_image(request)

Return one box and the mask of the box per label, or for the prompt.

Parameters:

Name	Type	Description	Default
request	RegionRequest	The region request.	required

Returns:

Type	Description
list[Region]	The regions, with masks of the image size.

Source code in src/skai/backends/_dummy.py

def segment_image(self, request: RegionRequest) -> list[Region]:
    """Return one box and the mask of the box per label, or for the prompt.

    Args:
        request:
            The region request.

    Returns:
        The regions, with masks of the image size.
    """
    regions = []
    for region in self.detect_objects(request):
        mask = Image.new('L', request.image.size, 0)
        mask.paste(255, region.box)
        regions.append(Region(region.label, region.score, region.box, mask))
    return regions

separate_audio(request)

Return the audio at half its volume as the stem.

Parameters:

Name	Type	Description	Default
request	AudioOperationRequest	The audio operation request.	required

Returns:

Type	Description
Audio	Audio as long as the input.

Source code in src/skai/backends/_dummy.py

def separate_audio(self, request: AudioOperationRequest) -> Audio:
    """Return the audio at half its volume as the stem.

    Args:
        request:
            The audio operation request.

    Returns:
        Audio as long as the input.
    """
    mono = request.audio.to_mono()
    return Audio((mono.samples * STEM_GAIN).astype(np.float32), mono.sample_rate)

synthesize_speech(request)

Return a sine tone whose length grows with the number of words.

Parameters:

Name	Type	Description	Default
request	SpeechRequest	The speech request.	required

Returns:

Type	Description
Audio	A mono tone at 16 kHz.

Source code in src/skai/backends/_dummy.py

def synthesize_speech(self, request: SpeechRequest) -> Audio:
    """Return a sine tone whose length grows with the number of words.

    Args:
        request:
            The speech request.

    Returns:
        A mono tone at 16 kHz.
    """
    duration = SPEECH_PADDING_SECONDS + SPEECH_SECONDS_PER_WORD * len(request.text.split())
    return _tone(duration, request.text, request.voice, request.seed)

transcribe(request)

Return a sentence that states the length of the speech.

Parameters:

Name	Type	Description	Default
request	TranscriptionRequest	The transcription request.	required

Returns:

Type	Description
Transcript	The transcript, with one segment or with evenly spaced words when timestamps are requested.

Source code in src/skai/backends/_dummy.py

def transcribe(self, request: TranscriptionRequest) -> Transcript:
    """Return a sentence that states the length of the speech.

    Args:
        request:
            The transcription request.

    Returns:
        The transcript, with one segment or with evenly spaced words when timestamps are requested.
    """
    duration = request.audio.duration
    text = f'Speech of {duration:.2f} seconds.'
    text = f'{text} Terms: {", ".join(request.vocabulary)}.' if request.vocabulary else text
    if request.timestamps == 'segment':
        return Transcript(text, (Segment(0.0, duration, text),))
    if request.timestamps == 'word':
        words = text.split()
        step = duration / len(words)
        return Transcript(text, tuple(Segment(i * step, (i + 1) * step, word) for i, word in enumerate(words)))
    return Transcript(text)

transcribe_image(request)

Return a sentence that states the size of the image.

Parameters:

Name	Type	Description	Default
request	ImageTranscriptionRequest	The image transcription request.	required

Returns:

Type	Description
str	The text, as a Markdown heading when the request asks for Markdown.

Source code in src/skai/backends/_dummy.py

def transcribe_image(self, request: ImageTranscriptionRequest) -> str:
    """Return a sentence that states the size of the image.

    Args:
        request:
            The image transcription request.

    Returns:
        The text, as a Markdown heading when the request asks for Markdown.
    """
    text = f'Text of a {request.image.width}x{request.image.height} image.'
    return f'# {text}' if request.markdown else text

upscale_image(request)

Resize the image by the scale.

Parameters:

Name	Type	Description	Default
request	ImageOperationRequest	The image operation request.	required

Returns:

Type	Description
Image	The upscaled image.

Source code in src/skai/backends/_dummy.py

def upscale_image(self, request: ImageOperationRequest) -> Image.Image:
    """Resize the image by the scale.

    Args:
        request:
            The image operation request.

    Returns:
        The upscaled image.
    """
    width, height = request.image.size
    return request.image.convert('RGB').resize((width * request.scale, height * request.scale))

HFBackend(model=None, *, device=None, max_new_tokens=512, token_env='HF_TOKEN', options=None, max_concurrency=1)

Bases: PromptedBackend

Run component tasks on open-source Hugging Face models on the local machine.

Text classification uses a zero-shot natural language inference model and image classification uses CLIP, so both return real scores but ignore few-shot examples. Text generation and extraction prompt small instruction and vision models, and audio text generation uses an audio-language model. Object detection uses OWLv2, and segmentation finds boxes with OWLv2 and masks with SAM. Diarization uses pyannote, which needs the diarize extra and a Hugging Face token that has accepted the model conditions, and transcribes every speaker turn with Whisper. Audio separation uses the Hybrid Demucs bundle of torchaudio, and the model parameter names another bundle. Audio enhancement is spectral-gating noise reduction with no model. Models are downloaded on first use. The backend needs the hf extra.

Background removal runs the code of briaai/RMBG-1.4 that is downloaded from the Hugging Face Hub, pinned to the commit BACKGROUND_MODEL_REVISION. Another model set with model loads without remote code.

Parameters:

Name	Type	Description	Default
model	str | None	The Hugging Face model id to use. The default model of the task is used when it is None.	None
device	str | None	The device that runs the models, for example 'cpu', 'cuda' or 'mps'. It is chosen automatically when it is None.	None
max_new_tokens	int	The maximum number of tokens that a prompted model generates.	512
token_env	str	The name of the environment variable that holds the Hugging Face token of gated models.	'HF_TOKEN'
options	dict | None	Extra keyword arguments passed to every pipeline call.	None
max_concurrency	int	The maximum number of requests that run at the same time.	1

Source code in src/skai/backends/_hf.py

def __init__(
    self,
    model: str | None = None,
    *,
    device: str | None = None,
    max_new_tokens: int = 512,
    token_env: str = 'HF_TOKEN',  # noqa: S107 - the default names an environment variable
    options: dict | None = None,
    max_concurrency: int = 1,
) -> None:
    super().__init__(model, options=options, max_concurrency=max_concurrency)
    self.device = device
    self.max_new_tokens = max_new_tokens
    self.token_env = token_env

animate_image(request)

Animate an image with an image-to-video diffusion model.

The motion prompt is not used.

Parameters:

Name	Type	Description	Default
request	AnimationRequest	The animation request.	required

Returns:

Type	Description
Video	The video, resized to the size of the image.

Source code in src/skai/backends/_hf.py

def animate_image(self, request: AnimationRequest) -> Video:
    """Animate an image with an image-to-video diffusion model.

    The motion prompt is not used.

    Args:
        request:
            The animation request.

    Returns:
        The video, resized to the size of the image.
    """
    pipeline = self._load_diffusion('animate_image', 'StableVideoDiffusionPipeline')
    fps = request.fps or SVD_FPS
    output = pipeline(
        request.image.convert('RGB').resize(SVD_SIZE),
        num_frames=max(1, round(request.duration * fps)),
        generator=self._generator(request.seed),
        **self.call_options,
    )
    return Video(_to_frames(output.frames[0]), fps).resize(request.image.size)

classify_audio(request)

Score audio with a zero-shot audio classification model.

Label descriptions are used as the candidate labels when they are given.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The classification request.	required

Returns:

Type	Description
dict[str, float]	A score per label.

Source code in src/skai/backends/_hf.py

def classify_audio(self, request: ClassificationRequest) -> dict[str, float]:
    """Score audio with a zero-shot audio classification model.

    Label descriptions are used as the candidate labels when they are given.

    Args:
        request:
            The classification request.

    Returns:
        A score per label.
    """
    assert isinstance(request.sample, Audio)
    candidates = [request.descriptions.get(label, label) for label in request.labels]
    pipeline = self._pipeline('zero-shot-audio-classification', self.resolve_model('classify_audio'))
    samples = request.sample.resample(CLAP_SAMPLE_RATE).samples
    scores = {item['label']: item['score'] for item in pipeline(samples, candidate_labels=candidates)}
    return {label: float(scores[candidate]) for label, candidate in zip(request.labels, candidates, strict=True)}

classify_image(request)

Score an image with a zero-shot image classification model.

Label descriptions are used as the candidate labels when they are given.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The classification request.	required

Returns:

Type	Description
dict[str, float]	A score per label.

Source code in src/skai/backends/_hf.py

def classify_image(self, request: ClassificationRequest) -> dict[str, float]:
    """Score an image with a zero-shot image classification model.

    Label descriptions are used as the candidate labels when they are given.

    Args:
        request:
            The classification request.

    Returns:
        A score per label.
    """
    candidates = [request.descriptions.get(label, label) for label in request.labels]
    pipeline = self._pipeline('zero-shot-image-classification', self.resolve_model('classify_image'))
    scores = {item['label']: item['score'] for item in pipeline(request.sample, candidate_labels=candidates)}
    return {label: float(scores[candidate]) for label, candidate in zip(request.labels, candidates, strict=True)}

classify_text(request)

Score a text with a zero-shot classification model.

Label descriptions are used as the candidate labels when they are given.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The classification request.	required

Returns:

Type	Description
dict[str, float]	A score per label.

Source code in src/skai/backends/_hf.py

def classify_text(self, request: ClassificationRequest) -> dict[str, float]:
    """Score a text with a zero-shot classification model.

    Label descriptions are used as the candidate labels when they are given.

    Args:
        request:
            The classification request.

    Returns:
        A score per label.
    """
    candidates = [request.descriptions.get(label, label) for label in request.labels]
    pipeline = self._pipeline('zero-shot-classification', self.resolve_model('classify_text'))
    output = pipeline(request.sample, candidate_labels=candidates, multi_label=request.multi_label)
    scores = dict(zip(output['labels'], output['scores'], strict=True))
    return {label: float(scores[candidate]) for label, candidate in zip(request.labels, candidates, strict=True)}

complete(prompt, model)

Return the answer of a local chat, vision or audio-language model to a prompt.

Parameters:

Name	Type	Description	Default
prompt	Prompt	The prompt.	required
model	str	The model to prompt.	required

Returns:

Type	Description
str	The answer of the model.

Source code in src/skai/backends/_hf.py

def complete(self, prompt: Prompt, model: str) -> str:
    """Return the answer of a local chat, vision or audio-language model to a prompt.

    Args:
        prompt:
            The prompt.

        model:
            The model to prompt.

    Returns:
        The answer of the model.
    """
    if prompt.audio:
        return self._complete_audio(prompt, model)
    options = {'max_new_tokens': self.max_new_tokens, 'return_full_text': False, **self.call_options}
    if not prompt.images:
        messages = [{'role': 'system', 'content': prompt.system}, {'role': 'user', 'content': prompt.user}]
        return self._pipeline('text-generation', model)(messages, **options)[0]['generated_text']
    content: list[dict[str, Any]] = [{'type': 'image', 'image': image} for image in prompt.images]
    content.append({'type': 'text', 'text': f'{prompt.system}\n\n{prompt.user}'})
    output = self._pipeline('image-text-to-text', model)(text=[{'role': 'user', 'content': content}], **options)
    return output[0]['generated_text']

connect()

Check that the hf extra is installed.

Returns:

Type	Description
HFBackend	The backend.

Source code in src/skai/backends/_hf.py

def connect(self) -> 'HFBackend':
    """Check that the `hf` extra is installed.

    Returns:
        The backend.
    """
    import_optional('transformers', 'hf')
    return self

detect_objects(request)

Find the boxes of regions with a zero-shot object detection model.

The labels are the candidate labels, or the prompt when there are no labels.

Parameters:

Name	Type	Description	Default
request	RegionRequest	The region request.	required

Returns:

Type	Description
list[Region]	The regions, without masks.

Source code in src/skai/backends/_hf.py

def detect_objects(self, request: RegionRequest) -> list[Region]:
    """Find the boxes of regions with a zero-shot object detection model.

    The labels are the candidate labels, or the prompt when there are no labels.

    Args:
        request:
            The region request.

    Returns:
        The regions, without masks.
    """
    return self._find_regions(request, self.resolve_model('detect_objects'))

diarize(request)

Find speaker turns with pyannote and transcribe every turn with the default Whisper model.

Parameters:

Name	Type	Description	Default
request	DiarizationRequest	The diarization request.	required

Returns:

Type	Description
Transcript	The transcript, with a speaker per segment.

Source code in src/skai/backends/_hf.py

def diarize(self, request: DiarizationRequest) -> Transcript:
    """Find speaker turns with pyannote and transcribe every turn with the default Whisper model.

    Args:
        request:
            The diarization request.

    Returns:
        The transcript, with a speaker per segment.
    """
    pipeline = self._load_diarization(self.resolve_model('diarize'))
    torch = import_optional('torch', 'hf')
    audio = request.audio.to_mono()
    waveform = torch.from_numpy(audio.samples.astype(np.float32)[None, :])
    annotation = pipeline({'waveform': waveform, 'sample_rate': audio.sample_rate})
    segments = []
    for turn, _, speaker in annotation.itertracks(yield_label=True):
        part = cut_audio(audio, turn.start, turn.end)
        turn_request = TranscriptionRequest(part, request.language, None, request.vocabulary)
        text = self._transcribe(turn_request, TRANSCRIPTION_MODEL).text if part.samples.size else ''
        segments.append(Segment(float(turn.start), float(turn.end), text, str(speaker)))
    return Transcript(' '.join(segment.text for segment in segments if segment.text), tuple(segments))

edit_image(request)

Edit an image with an instruction-following diffusion model.

Parameters:

Name	Type	Description	Default
request	ImageEditRequest	The image edit request.	required

Returns:

Type	Description
Image	The edited image, of the size of the input.

Source code in src/skai/backends/_hf.py

def edit_image(self, request: ImageEditRequest) -> Image.Image:
    """Edit an image with an instruction-following diffusion model.

    Args:
        request:
            The image edit request.

    Returns:
        The edited image, of the size of the input.
    """
    pipeline = self._load_diffusion('edit_image', 'StableDiffusionInstructPix2PixPipeline')
    output = pipeline(
        prompt=request.instruction,
        image=request.image.convert('RGB'),
        generator=self._generator(request.seed),
        **self.call_options,
    )
    edited = output.images[0]
    return edited if edited.size == request.image.size else edited.resize(request.image.size)

embed_audio(request)

Embed audio with a CLAP model.

Parameters:

Name	Type	Description	Default
request	EmbeddingRequest	The embedding request.	required

Returns:

Type	Description
NDArray[floating]	The embedding vector.

Source code in src/skai/backends/_hf.py

def embed_audio(self, request: EmbeddingRequest) -> NDArray[np.floating]:
    """Embed audio with a CLAP model.

    Args:
        request:
            The embedding request.

    Returns:
        The embedding vector.
    """
    assert isinstance(request.sample, Audio)
    return np.asarray(self._encode_audio(request.sample))

embed_image(request)

Embed an image with a CLIP model of sentence-transformers.

Parameters:

Name	Type	Description	Default
request	EmbeddingRequest	The embedding request.	required

Returns:

Type	Description
NDArray[floating]	The embedding vector.

Source code in src/skai/backends/_hf.py

def embed_image(self, request: EmbeddingRequest) -> NDArray[np.floating]:
    """Embed an image with a CLIP model of sentence-transformers.

    Args:
        request:
            The embedding request.

    Returns:
        The embedding vector.
    """
    sentence_transformers = import_optional('sentence_transformers', 'hf')
    model = self.resolve_model('embed_image')
    encoder = self._load(
        ('embed_image', model),
        lambda: sentence_transformers.SentenceTransformer(model, device=self.device),
    )
    return np.asarray(encoder.encode([request.sample], **self.call_options)[0])

embed_text(request)

Embed a text with a sentence-transformers model.

Parameters:

Name	Type	Description	Default
request	EmbeddingRequest	The embedding request.	required

Returns:

Type	Description
NDArray[floating]	The embedding vector.

Source code in src/skai/backends/_hf.py

def embed_text(self, request: EmbeddingRequest) -> NDArray[np.floating]:
    """Embed a text with a sentence-transformers model.

    Args:
        request:
            The embedding request.

    Returns:
        The embedding vector.
    """
    sentence_transformers = import_optional('sentence_transformers', 'hf')
    model = self.resolve_model('embed_text')
    encoder = self._load(
        ('embed_text', model),
        lambda: sentence_transformers.SentenceTransformer(model, device=self.device),
    )
    return np.asarray(encoder.encode([request.sample], **self.call_options)[0])

enhance_audio(request)

Remove stationary noise with spectral gating, learning the noise from the quietest frames.

Parameters:

Name	Type	Description	Default
request	AudioOperationRequest	The audio operation request.	required

Returns:

Type	Description
Audio	The denoised mono audio, of the length and rate of the input.

Source code in src/skai/backends/_hf.py

def enhance_audio(self, request: AudioOperationRequest) -> Audio:
    """Remove stationary noise with spectral gating, learning the noise from the quietest frames.

    Args:
        request:
            The audio operation request.

    Returns:
        The denoised mono audio, of the length and rate of the input.
    """
    mono = request.audio.to_mono()
    return Audio(_reduce_noise(mono.samples), mono.sample_rate)

generate_image(request)

Generate an image with a diffusion model.

Parameters:

Name	Type	Description	Default
request	ImageGenerationRequest	The image generation request.	required

Returns:

Type	Description
Image	The image.

Source code in src/skai/backends/_hf.py

def generate_image(self, request: ImageGenerationRequest) -> Image.Image:
    """Generate an image with a diffusion model.

    Args:
        request:
            The image generation request.

    Returns:
        The image.
    """
    model = self.resolve_model('generate_image')
    pipeline = self._load_diffusion('generate_image', 'AutoPipelineForText2Image')
    generator = self._generator(request.seed)
    width, height = request.size
    output = pipeline(
        prompt=build_image_prompt(request),
        negative_prompt=request.negative_prompt,
        width=width,
        height=height,
        generator=generator,
        **(TURBO_OPTIONS if model == IMAGE_MODEL else {}),
        **self.call_options,
    )
    return output.images[0]

generate_music(request)

Generate music with MusicGen, about 50 tokens per second, trimmed or padded to the duration.

Parameters:

Name	Type	Description	Default
request	AudioGenerationRequest	The audio generation request.	required

Returns:

Type	Description
Audio	The mono music.

Source code in src/skai/backends/_hf.py

def generate_music(self, request: AudioGenerationRequest) -> Audio:
    """Generate music with MusicGen, about 50 tokens per second, trimmed or padded to the duration.

    Args:
        request:
            The audio generation request.

    Returns:
        The mono music.
    """
    transformers = import_optional('transformers', 'hf')
    if request.seed is not None:
        transformers.set_seed(request.seed)
    pipeline = self._pipeline('text-to-audio', self.resolve_model('generate_music'))
    tokens = max(1, math.ceil(request.duration * MUSICGEN_TOKENS_PER_SECOND))
    output = pipeline(
        request.prompt,
        forward_params={'do_sample': True, 'max_new_tokens': tokens},
        **self.call_options,
    )
    rate = int(output['sampling_rate'])
    return Audio(_to_mono_samples(output['audio'], round(request.duration * rate)), rate)

generate_sound(request)

Generate a sound with AudioLDM2, trimmed or padded to the duration.

Parameters:

Name	Type	Description	Default
request	AudioGenerationRequest	The audio generation request.	required

Returns:

Type	Description
Audio	The mono sound at 16 kHz.

Source code in src/skai/backends/_hf.py

def generate_sound(self, request: AudioGenerationRequest) -> Audio:
    """Generate a sound with AudioLDM2, trimmed or padded to the duration.

    Args:
        request:
            The audio generation request.

    Returns:
        The mono sound at 16 kHz.
    """
    pipeline = self._load_diffusion('generate_sound', 'AudioLDM2Pipeline')
    output = pipeline(
        request.prompt,
        audio_length_in_s=request.duration,
        generator=self._generator(request.seed),
        **self.call_options,
    )
    length = round(request.duration * AUDIOLDM_SAMPLE_RATE)
    return Audio(_to_mono_samples(output.audios[0], length), AUDIOLDM_SAMPLE_RATE)

generate_video(request)

Generate a video with a text-to-video diffusion model.

Parameters:

Name	Type	Description	Default
request	VideoGenerationRequest	The video generation request.	required

Returns:

Type	Description
Video	The video, at fps or 16 frames per second.

Source code in src/skai/backends/_hf.py

def generate_video(self, request: VideoGenerationRequest) -> Video:
    """Generate a video with a text-to-video diffusion model.

    Args:
        request:
            The video generation request.

    Returns:
        The video, at `fps` or 16 frames per second.
    """
    pipeline = self._load_diffusion('generate_video', 'DiffusionPipeline')
    fps = request.fps or WAN_FPS
    width, height = request.size
    output = pipeline(
        prompt=build_image_prompt(ImageGenerationRequest(request.prompt, request.size, request.style)),
        negative_prompt=request.negative_prompt,
        width=width,
        height=height,
        num_frames=4 * max(1, round(request.duration * fps / 4)) + 1,
        generator=self._generator(request.seed),
        **self.call_options,
    )
    return Video(_to_frames(output.frames[0]), fps)

remove_background(request)

Remove the background with the mask of an image segmentation model, by default RMBG at a pinned commit.

Parameters:

Name	Type	Description	Default
request	ImageOperationRequest	The image operation request.	required

Returns:

Type	Description
Image	An RGBA image of the input size whose alpha is the predicted mask.

Source code in src/skai/backends/_hf.py

def remove_background(self, request: ImageOperationRequest) -> Image.Image:
    """Remove the background with the mask of an image segmentation model, by default RMBG at a pinned commit.

    Args:
        request:
            The image operation request.

    Returns:
        An RGBA image of the input size whose alpha is the predicted mask.
    """
    model = self.resolve_model('remove_background')
    image = request.image.convert('RGB')
    if model == BACKGROUND_MODEL:
        pipeline = self._pipeline(
            'image-segmentation',
            model,
            trust_remote_code=True,
            revision=BACKGROUND_MODEL_REVISION,
        )
        mask = pipeline(image, return_mask=True, **self.call_options).convert('L')
    else:
        output = self._pipeline('image-segmentation', model)(image, **self.call_options)
        masks = [np.asarray(item['mask'].convert('L')) for item in output]
        mask = Image.fromarray(np.max(masks, axis=0).astype(np.uint8)) if masks else Image.new('L', image.size)
    cutout = image.convert('RGBA')
    cutout.putalpha(mask if mask.size == image.size else mask.resize(image.size))
    return cutout

segment_image(request)

Find the masks of regions with SAM, prompted by the boxes of the default object detection model.

Parameters:

Name	Type	Description	Default
request	RegionRequest	The region request.	required

Returns:

Type	Description
list[Region]	The regions, with masks of the image size.

Source code in src/skai/backends/_hf.py

def segment_image(self, request: RegionRequest) -> list[Region]:
    """Find the masks of regions with SAM, prompted by the boxes of the default object detection model.

    Args:
        request:
            The region request.

    Returns:
        The regions, with masks of the image size.
    """
    regions = self._find_regions(request, DETECTION_MODEL)
    if not regions:
        return []
    torch = import_optional('torch', 'hf')
    transformers = import_optional('transformers', 'hf')
    model = self.resolve_model('segment_image')
    device = self.device or 'cpu'
    processor = self._load(('sam-processor', model), lambda: transformers.SamProcessor.from_pretrained(model))
    sam = self._load(('sam-model', model), lambda: transformers.SamModel.from_pretrained(model).to(device))
    boxes = [[list(region.box) for region in regions]]
    inputs = processor(request.image.convert('RGB'), input_boxes=boxes, return_tensors='pt').to(device)
    with torch.no_grad():
        outputs = sam(**inputs, multimask_output=False)
    masks = processor.image_processor.post_process_masks(
        outputs.pred_masks.cpu(),
        inputs['original_sizes'],
        inputs['reshaped_input_sizes'],
    )[0]
    return [
        Region(region.label, region.score, region.box, Image.fromarray(np.asarray(mask[0]).astype(np.uint8) * 255))
        for region, mask in zip(regions, masks, strict=True)
    ]

separate_audio(request)

Separate the vocals, or the sum of drums, bass and other as the accompaniment, with Hybrid Demucs.

Parameters:

Name	Type	Description	Default
request	AudioOperationRequest	The audio operation request.	required

Returns:

Type	Description
Audio	The mono stem, of the length and rate of the input.

Source code in src/skai/backends/_hf.py

def separate_audio(self, request: AudioOperationRequest) -> Audio:
    """Separate the vocals, or the sum of drums, bass and other as the accompaniment, with Hybrid Demucs.

    Args:
        request:
            The audio operation request.

    Returns:
        The mono stem, of the length and rate of the input.
    """
    mono = request.audio.to_mono()
    rate = getattr(import_optional('torchaudio', 'hf').pipelines, self.resolve_model('separate_audio')).sample_rate
    stem = Audio(self._separate(mono, request.stem), rate).resample(mono.sample_rate)
    return Audio(_to_mono_samples(stem.samples, mono.samples.shape[0]), mono.sample_rate)

synthesize_speech(request)

Speak a text with a text-to-speech model.

The default model has a single voice, so the voice of the request is ignored.

Parameters:

Name	Type	Description	Default
request	SpeechRequest	The speech request.	required

Returns:

Type	Description
Audio	The speech.

Source code in src/skai/backends/_hf.py

def synthesize_speech(self, request: SpeechRequest) -> Audio:
    """Speak a text with a text-to-speech model.

    The default model has a single voice, so the voice of the request is ignored.

    Args:
        request:
            The speech request.

    Returns:
        The speech.
    """
    output = self._pipeline('text-to-speech', self.resolve_model('synthesize_speech'))(
        request.text,
        **self.call_options,
    )
    return Audio(np.asarray(output['audio'], dtype=np.float32).squeeze(), int(output['sampling_rate']))

transcribe(request)

Transcribe speech with a speech recognition model.

Parameters:

Name	Type	Description	Default
request	TranscriptionRequest	The transcription request.	required

Returns:

Type	Description
Transcript	The transcript.

Source code in src/skai/backends/_hf.py

def transcribe(self, request: TranscriptionRequest) -> Transcript:
    """Transcribe speech with a speech recognition model.

    Args:
        request:
            The transcription request.

    Returns:
        The transcript.
    """
    return self._transcribe(request, self.resolve_model('transcribe'))

upscale_image(request)

Upscale an image with Swin2SR, using the x2 model and a resize for scales other than 2 and 4.

Parameters:

Name	Type	Description	Default
request	ImageOperationRequest	The image operation request.	required

Returns:

Type	Description
Image	The upscaled image, exactly scale times the size of the input.

Source code in src/skai/backends/_hf.py

def upscale_image(self, request: ImageOperationRequest) -> Image.Image:
    """Upscale an image with Swin2SR, using the x2 model and a resize for scales other than 2 and 4.

    Args:
        request:
            The image operation request.

    Returns:
        The upscaled image, exactly `scale` times the size of the input.
    """
    image = request.image.convert('RGB')
    size = (image.width * request.scale, image.height * request.scale)
    if request.scale == 1:
        return image
    model = self.model or UPSCALE_MODELS.get(request.scale, UPSCALE_MODELS[2])
    upscaled = self._pipeline('image-to-image', model)(image, **self.call_options)
    return upscaled if upscaled.size == size else upscaled.resize(size)

LiteLLMBackend(model=None, *, api_key_env=None, api_base=None, video_audio=True, options=None, max_concurrency=8)

Bases: PromptedBackend

Run prompted tasks and embeddings on any provider that LiteLLM supports.

Models are named as LiteLLM names them, for example anthropic/claude-haiku-5-5, gemini/gemini-2.5-flash or ollama/llama3.2. LiteLLM reads the standard credential variable of each provider. The backend needs the litellm extra.

Parameters:

Name	Type	Description	Default
model	str | None	The model to use. The default model of the task is used when it is None.	None
api_key_env	str | None	The name of an environment variable that holds the API key, used instead of the provider's standard one.	None
api_base	str | None	The base URL of the provider's API.	None
video_audio	bool	Whether video requests include the audio track. It suits models that take audio, such as Gemini, and must be False for models that take text and images only.	True
options	dict | None	Extra keyword arguments passed to every LiteLLM call.	None
max_concurrency	int	The maximum number of requests that run at the same time.	8

Source code in src/skai/backends/_litellm.py

def __init__(
    self,
    model: str | None = None,
    *,
    api_key_env: str | None = None,
    api_base: str | None = None,
    video_audio: bool = True,
    options: dict | None = None,
    max_concurrency: int = 8,
) -> None:
    super().__init__(model, options=options, max_concurrency=max_concurrency)
    self.api_key_env = api_key_env
    self.api_base = api_base
    self.video_audio = video_audio

complete(prompt, model)

Return the answer of a chat model to a prompt.

Parameters:

Name	Type	Description	Default
prompt	Prompt	The prompt.	required
model	str	The model to prompt.	required

Returns:

Type	Description
str	The answer of the model.

Source code in src/skai/backends/_litellm.py

def complete(self, prompt: Prompt, model: str) -> str:
    """Return the answer of a chat model to a prompt.

    Args:
        prompt:
            The prompt.

        model:
            The model to prompt.

    Returns:
        The answer of the model.
    """
    options = {'response_format': {'type': 'json_object'}} if prompt.json_output else {}
    response = self._litellm.completion(
        model=model,
        messages=to_chat_messages(prompt),
        **options,
        **self._credentials(),
        **self.call_options,
    )
    return response.choices[0].message.content or ''

connect()

Import LiteLLM and check that the credential variable is set.

Returns:

Type	Description
LiteLLMBackend	The backend.

Source code in src/skai/backends/_litellm.py

def connect(self) -> 'LiteLLMBackend':
    """Import LiteLLM and check that the credential variable is set.

    Returns:
        The backend.
    """
    self._module = import_optional('litellm', 'litellm')
    self._credentials()
    return self

embed_image(request)

Embed an image, sent as a data URL, with a multimodal embedding model.

Parameters:

Name	Type	Description	Default
request	EmbeddingRequest	The embedding request.	required

Returns:

Type	Description
NDArray[floating]	The embedding vector.

Source code in src/skai/backends/_litellm.py

def embed_image(self, request: EmbeddingRequest) -> NDArray[np.floating]:
    """Embed an image, sent as a data URL, with a multimodal embedding model.

    Args:
        request:
            The embedding request.

    Returns:
        The embedding vector.
    """
    assert isinstance(request.sample, Image.Image)
    return self._embed('embed_image', to_data_url(request.sample), request.dimensions)

embed_text(request)

Embed a text with an embedding model.

Parameters:

Name	Type	Description	Default
request	EmbeddingRequest	The embedding request.	required

Returns:

Type	Description
NDArray[floating]	The embedding vector.

Source code in src/skai/backends/_litellm.py

def embed_text(self, request: EmbeddingRequest) -> NDArray[np.floating]:
    """Embed a text with an embedding model.

    Args:
        request:
            The embedding request.

    Returns:
        The embedding vector.
    """
    return self._embed('embed_text', str(request.sample), request.dimensions)

sends_video_audio_now()

Return the video_audio parameter.

Returns:

Type	Description
bool	Whether video requests include the audio track.

Source code in src/skai/backends/_litellm.py

def sends_video_audio_now(self) -> bool:
    """Return the `video_audio` parameter.

    Returns:
        Whether video requests include the audio track.
    """
    return self.video_audio

OpenAIBackend(model=None, *, api_key_env='OPENAI_API_KEY', base_url=None, client=None, voice=DEFAULT_VOICE, finetune_model=FINETUNE_MODEL, finetune_timeout=3600.0, video_timeout=900.0, poll_interval=15.0, options=None, max_concurrency=8)

Bases: PromptedBackend

Run component tasks on the OpenAI API.

The backend prompts chat, vision and audio models for classification, text generation, extraction, image transcription, object detection, moments and thumbnails, and runs embeddings, speech, transcription, image generation and editing, and Sora videos. It needs the openai extra.

Parameters:

Name	Type	Description	Default
model	str | None	The model to use. The default model of the task is used when it is None.	None
api_key_env	str	The name of the environment variable that holds the API key.	'OPENAI_API_KEY'
base_url	str | None	The base URL of an OpenAI-compatible API.	None
client	Any	A preconfigured openai.OpenAI client, used instead of creating one.	None
voice	str	The voice used for speech synthesis when the request names none.	DEFAULT_VOICE
finetune_model	str	The base model that fine-tuning starts from.	FINETUNE_MODEL
finetune_timeout	float	The number of seconds to wait for a fine-tuning job before giving up.	3600.0
video_timeout	float	The number of seconds to wait for a video job before giving up.	900.0
poll_interval	float	The number of seconds between checks of a fine-tuning or video job.	15.0
options	dict | None	Extra keyword arguments passed to every API call.	None
max_concurrency	int	The maximum number of requests that run at the same time.	8

Source code in src/skai/backends/_openai.py

def __init__(
    self,
    model: str | None = None,
    *,
    api_key_env: str = 'OPENAI_API_KEY',
    base_url: str | None = None,
    client: Any = None,
    voice: str = DEFAULT_VOICE,
    finetune_model: str = FINETUNE_MODEL,
    finetune_timeout: float = 3600.0,
    video_timeout: float = 900.0,
    poll_interval: float = 15.0,
    options: dict | None = None,
    max_concurrency: int = 8,
) -> None:
    super().__init__(model, options=options, max_concurrency=max_concurrency)
    self.api_key_env = api_key_env
    self.base_url = base_url
    self.client = client
    self.voice = voice
    self.finetune_model = finetune_model
    self.finetune_timeout = finetune_timeout
    self.video_timeout = video_timeout
    self.poll_interval = poll_interval

animate_image(request)

Animate an image with Sora, using it as the first frame.

Parameters:

Name	Type	Description	Default
request	AnimationRequest	The animation request.	required

Returns:

Type	Description
Video	The video, trimmed to the requested duration and resized to the size of the image.

Source code in src/skai/backends/_openai.py

def animate_image(self, request: AnimationRequest) -> Video:
    """Animate an image with Sora, using it as the first frame.

    Args:
        request:
            The animation request.

    Returns:
        The video, trimmed to the requested duration and resized to the size of the image.
    """
    size = _closest_video_size(request.image.size)
    reference = _to_png_file(request.image.convert('RGB').resize(size), 'image.png')
    video = self._create_video(
        self.resolve_model('animate_image'),
        request.motion_prompt or DEFAULT_MOTION_PROMPT,
        _closest_video_seconds(request.duration),
        size,
        input_reference=reference,
    )
    return video.trim(request.duration).resize(request.image.size)

complete(prompt, model)

Return the answer of a chat model to a prompt.

Parameters:

Name	Type	Description	Default
prompt	Prompt	The prompt.	required
model	str	The model to prompt.	required

Returns:

Type	Description
str	The answer of the model.

Source code in src/skai/backends/_openai.py

def complete(self, prompt: Prompt, model: str) -> str:
    """Return the answer of a chat model to a prompt.

    Args:
        prompt:
            The prompt.

        model:
            The model to prompt.

    Returns:
        The answer of the model.
    """
    options = {'response_format': {'type': 'json_object'}} if prompt.json_output else {}
    response = self._api.chat.completions.create(
        model=model,
        messages=to_chat_messages(prompt),
        **options,
        **self.call_options,
    )
    return response.choices[0].message.content or ''

connect()

Use the given client, or create one with the API key read from api_key_env.

Returns:

Type	Description
OpenAIBackend	The backend.

Source code in src/skai/backends/_openai.py

def connect(self) -> 'OpenAIBackend':
    """Use the given client, or create one with the API key read from `api_key_env`.

    Returns:
        The backend.
    """
    if self.client is not None:
        self._client = self.client
        return self
    openai = import_optional('openai', 'openai')
    self._client = openai.OpenAI(api_key=read_env(self.api_key_env), base_url=self.base_url)
    return self

edit_image(request)

Edit an image with an image model, inside the mask when one is given.

Parameters:

Name	Type	Description	Default
request	ImageEditRequest	The image edit request.	required

Returns:

Type	Description
Image	The edited image, resized to the size of the input.

Source code in src/skai/backends/_openai.py

def edit_image(self, request: ImageEditRequest) -> Image.Image:
    """Edit an image with an image model, inside the mask when one is given.

    Args:
        request:
            The image edit request.

    Returns:
        The edited image, resized to the size of the input.
    """
    image = request.image.convert('RGBA')
    mask = {'mask': _to_png_file(_to_alpha_mask(request.mask, image.size), 'mask.png')} if request.mask else {}
    response = self._api.images.edit(
        model=self.resolve_model('edit_image'),
        image=_to_png_file(image, 'image.png'),
        prompt=request.instruction,
        **mask,
        **self.call_options,
    )
    edited = Image.open(io.BytesIO(base64.b64decode(response.data[0].b64_json))).convert('RGB')
    return edited if edited.size == request.image.size else edited.resize(request.image.size)

embed_text(request)

Embed a text with an embedding model.

Parameters:

Name	Type	Description	Default
request	EmbeddingRequest	The embedding request.	required

Returns:

Type	Description
NDArray[floating]	The embedding vector.

Source code in src/skai/backends/_openai.py

def embed_text(self, request: EmbeddingRequest) -> NDArray[np.floating]:
    """Embed a text with an embedding model.

    Args:
        request:
            The embedding request.

    Returns:
        The embedding vector.
    """
    dimensions = {'dimensions': request.dimensions} if request.dimensions else {}
    response = self._api.embeddings.create(
        model=self.resolve_model('embed_text'),
        input=[request.sample],
        **dimensions,
        **self.call_options,
    )
    return np.asarray(response.data[0].embedding)

fine_tune(request)

Train a chat model on the prompts and answers, and wait until the job finishes.

A job that does not finish within finetune_timeout seconds raises TimeoutError naming the job, whose model can be used later through the model parameter.

Parameters:

Name	Type	Description	Default
request	FineTuneRequest	The fine-tuning request.	required

Returns:

Type	Description
str	The id of the fine-tuned model.

Raises:

Type	Description
FineTuneError	If OpenAI refuses the job, or the job fails or is cancelled.

Source code in src/skai/backends/_openai.py

def fine_tune(self, request: FineTuneRequest) -> str:
    """Train a chat model on the prompts and answers, and wait until the job finishes.

    A job that does not finish within `finetune_timeout` seconds raises `TimeoutError` naming the job, whose model
    can be used later through the `model` parameter.

    Args:
        request:
            The fine-tuning request.

    Returns:
        The id of the fine-tuned model.

    Raises:
        FineTuneError:
            If OpenAI refuses the job, or the job fails or is cancelled.
    """
    lines = [
        json.dumps({'messages': [*to_chat_messages(prompt), {'role': 'assistant', 'content': answer}]})
        for prompt, answer in request.examples
    ]
    job = self._start_fine_tuning('\n'.join(lines).encode())
    retrieve = self._api.fine_tuning.jobs.retrieve
    job = self._wait_for(job, retrieve, FINISHED_STATUSES, self.finetune_timeout, 'Fine-tuning job')
    if job.status != 'succeeded':
        msg = f'Fine-tuning job {job.id} ended with status `{job.status}` ({job.error}).'
        raise FineTuneError(msg)
    return job.fine_tuned_model

generate_image(request)

Generate an image with an image model, from the reference images of a subject when the request has them.

Reference images go to the image edit endpoint, which keeps their subject in the new image.

Parameters:

Name	Type	Description	Default
request	ImageGenerationRequest	The image generation request.	required

Returns:

Type	Description
Image	The image.

Source code in src/skai/backends/_openai.py

def generate_image(self, request: ImageGenerationRequest) -> Image.Image:
    """Generate an image with an image model, from the reference images of a subject when the request has them.

    Reference images go to the image edit endpoint, which keeps their subject in the new image.

    Args:
        request:
            The image generation request.

    Returns:
        The image.
    """
    width, height = request.size
    options = {'model': self.resolve_model('generate_image'), 'prompt': build_image_prompt(request), 'n': 1}
    options |= {'size': f'{width}x{height}', **self.call_options}
    if request.reference_images:
        references = [
            _to_png_file(image.convert('RGB'), f'reference-{index}.png')
            for index, image in enumerate(request.reference_images)
        ]
        response = self._api.images.edit(image=references, **options)
    else:
        response = self._api.images.generate(**options)
    image = Image.open(io.BytesIO(base64.b64decode(response.data[0].b64_json)))
    image.load()
    return image

generate_video(request)

Generate a video with Sora, then fit it to the requested duration and size.

Sora makes videos of 4, 8 or 12 seconds at a few sizes, so the closest are requested, and the video is trimmed and resized. Trimming and resizing need the video extra.

Parameters:

Name	Type	Description	Default
request	VideoGenerationRequest	The video generation request.	required

Returns:

Type	Description
Video	The video.

Source code in src/skai/backends/_openai.py

def generate_video(self, request: VideoGenerationRequest) -> Video:
    """Generate a video with Sora, then fit it to the requested duration and size.

    Sora makes videos of 4, 8 or 12 seconds at a few sizes, so the closest are requested, and the video is trimmed
    and resized. Trimming and resizing need the `video` extra.

    Args:
        request:
            The video generation request.

    Returns:
        The video.
    """
    prompt = build_image_prompt(ImageGenerationRequest(request.prompt, request.size, request.style))
    prompt = f'{prompt}. Avoid: {request.negative_prompt}' if request.negative_prompt else prompt
    size = _closest_video_size(request.size)
    seconds = _closest_video_seconds(request.duration)
    video = self._create_video(self.resolve_model('generate_video'), prompt, seconds, size)
    return video.trim(request.duration).resize(request.size)

synthesize_speech(request)

Speak a text with a text-to-speech model.

Parameters:

Name	Type	Description	Default
request	SpeechRequest	The speech request.	required

Returns:

Type	Description
Audio	The speech.

Source code in src/skai/backends/_openai.py

def synthesize_speech(self, request: SpeechRequest) -> Audio:
    """Speak a text with a text-to-speech model.

    Args:
        request:
            The speech request.

    Returns:
        The speech.
    """
    response = self._api.audio.speech.create(
        model=self.resolve_model('synthesize_speech'),
        voice=request.voice if isinstance(request.voice, str) else self.voice,
        input=request.text,
        response_format='wav',
        **self.call_options,
    )
    return Audio.from_bytes(response.content)

transcribe(request)

Transcribe speech with a transcription model.

Parameters:

Name	Type	Description	Default
request	TranscriptionRequest	The transcription request.	required

Returns:

Type	Description
Transcript	The transcript.

Source code in src/skai/backends/_openai.py

def transcribe(self, request: TranscriptionRequest) -> Transcript:
    """Transcribe speech with a transcription model.

    Args:
        request:
            The transcription request.

    Returns:
        The transcript.
    """
    options: dict[str, Any] = {'language': request.language} if request.language else {}
    options |= {'prompt': ', '.join(request.vocabulary)} if request.vocabulary else {}
    if request.timestamps:
        options |= {'response_format': 'verbose_json', 'timestamp_granularities': [request.timestamps]}
    response = self._api.audio.transcriptions.create(
        model=self.resolve_model('transcribe'),
        file=('audio.wav', request.audio.to_bytes()),
        **options,
        **self.call_options,
    )
    return Transcript(response.text.strip(), _to_segments(response, request.timestamps))

ReplicateBackend(model=None, *, api_token_env='REPLICATE_API_TOKEN', client=None, options=None, max_concurrency=4)

Bases: Backend

Run talking avatars, lip sync, image generation, speech synthesis and voice conversion on Replicate models.

The default models are open-source. A model is named owner/name or owner/name:version, and a version should be pinned for reproducible results. Inputs are sent under the names the default models use, and options adds or overrides inputs for other models. The backend needs the replicate extra.

Parameters:

Name	Type	Description	Default
model	str | None	The model to use. The default model of the task is used when it is None.	None
api_token_env	str	The name of the environment variable that holds the API token.	'REPLICATE_API_TOKEN'
client	Any	A preconfigured replicate.Client, used instead of creating one.	None
options	dict | None	Extra inputs sent to every model run.	None
max_concurrency	int	The maximum number of requests that run at the same time.	4

Source code in src/skai/backends/_replicate.py

def __init__(
    self,
    model: str | None = None,
    *,
    api_token_env: str = 'REPLICATE_API_TOKEN',  # noqa: S107 - the default names an environment variable
    client: Any = None,
    options: dict | None = None,
    max_concurrency: int = 4,
) -> None:
    super().__init__(model, options=options, max_concurrency=max_concurrency)
    self.api_token_env = api_token_env
    self.client = client

animate_avatar(request)

Animate a portrait with a talking-head model.

Parameters:

Name	Type	Description	Default
request	AvatarRequest	The avatar request.	required

Returns:

Type	Description
Video	The video.

Source code in src/skai/backends/_replicate.py

def animate_avatar(self, request: AvatarRequest) -> Video:
    """Animate a portrait with a talking-head model.

    Args:
        request:
            The avatar request.

    Returns:
        The video.
    """
    inputs = {
        'source_image': _to_file(encode_png(request.avatar.convert('RGB')), 'avatar.png'),
        'driven_audio': _to_file(request.audio.to_bytes(), 'audio.wav'),
    }
    return Video(data=self._run_model(self.resolve_model('animate_avatar'), inputs))

connect()

Use the given client, or create one with the API token read from api_token_env.

Returns:

Type	Description
ReplicateBackend	The backend.

Source code in src/skai/backends/_replicate.py

def connect(self) -> 'ReplicateBackend':
    """Use the given client, or create one with the API token read from `api_token_env`.

    Returns:
        The backend.
    """
    if self.client is not None:
        self._client = self.client
        return self
    replicate = import_optional('replicate', 'replicate')
    self._client = replicate.Client(api_token=read_env(self.api_token_env))
    return self

convert_voice(request)

Say the speech of audio in the voice of the reference audio with a voice conversion model.

Parameters:

Name	Type	Description	Default
request	AudioOperationRequest	The audio operation request, whose voice holds the reference audio.	required

Returns:

Type	Description
Audio	The converted speech.

Source code in src/skai/backends/_replicate.py

def convert_voice(self, request: AudioOperationRequest) -> Audio:
    """Say the speech of audio in the voice of the reference audio with a voice conversion model.

    Args:
        request:
            The audio operation request, whose `voice` holds the reference audio.

    Returns:
        The converted speech.
    """
    assert request.voice is not None
    inputs = {
        'source_audio': _to_file(request.audio.to_bytes(), 'source.wav'),
        'reference_audio': _to_file(request.voice.to_bytes(), 'reference.wav'),
    }
    return Audio.from_bytes(self._run_model(self.resolve_model('convert_voice'), inputs))

generate_image(request)

Generate an image and resize it to the requested size.

Parameters:

Name	Type	Description	Default
request	ImageGenerationRequest	The image generation request.	required

Returns:

Type	Description
Image	The image.

Source code in src/skai/backends/_replicate.py

def generate_image(self, request: ImageGenerationRequest) -> Image.Image:
    """Generate an image and resize it to the requested size.

    Args:
        request:
            The image generation request.

    Returns:
        The image.
    """
    inputs: dict[str, Any] = {
        'prompt': build_image_prompt(request),
        'aspect_ratio': _closest_aspect_ratio(request.size),
        'output_format': 'png',
    }
    inputs |= {'seed': request.seed} if request.seed is not None else {}
    inputs |= {'negative_prompt': request.negative_prompt} if request.negative_prompt else {}
    image = Image.open(io.BytesIO(self._run_model(self.resolve_model('generate_image'), inputs))).convert('RGB')
    return image if image.size == request.size else image.resize(request.size)

lip_sync(request)

Make a video speak new audio with a lip sync model.

Parameters:

Name	Type	Description	Default
request	LipSyncRequest	The lip sync request.	required

Returns:

Type	Description
Video	The video.

Source code in src/skai/backends/_replicate.py

def lip_sync(self, request: LipSyncRequest) -> Video:
    """Make a video speak new audio with a lip sync model.

    Args:
        request:
            The lip sync request.

    Returns:
        The video.
    """
    inputs: dict[str, Any] = {
        'video': _to_file(request.video.to_bytes(), 'video.mp4'),
        'audio': _to_file(request.audio.to_bytes(), 'audio.wav'),
    }
    inputs |= {'seed': request.seed} if request.seed is not None else {}
    return Video(data=self._run_model(self.resolve_model('lip_sync'), inputs))

synthesize_speech(request)

Speak a text, cloning the voice when the request holds reference audio.

Parameters:

Name	Type	Description	Default
request	SpeechRequest	The speech request.	required

Returns:

Type	Description
Audio	The speech.

Source code in src/skai/backends/_replicate.py

def synthesize_speech(self, request: SpeechRequest) -> Audio:
    """Speak a text, cloning the voice when the request holds reference audio.

    Args:
        request:
            The speech request.

    Returns:
        The speech.
    """
    if isinstance(request.voice, Audio):
        model = self.resolve_model('clone_voice')
        speaker = _to_file(request.voice.to_bytes(), 'speaker.wav')
        inputs: dict[str, Any] = {'text': request.text, 'speaker': speaker, 'language': request.language or 'en'}
    else:
        model = self.resolve_model('synthesize_speech')
        inputs = {'text': request.text, 'voice': request.voice or DEFAULT_VOICE}
    return Audio.from_bytes(self._run_model(model, inputs))