Qwen3-TTS 1.7B Base

"""Qwen3-TTS 1.7B Base, generated from a declarative template."""

from __future__ import annotations

import base64
import tempfile

from pydantic import BaseModel, ConfigDict, Field

from runware_serverless import serve

MODEL = "Qwen/Qwen3-TTS-12Hz-1.7B-Base"
REVISION = "fd4b254389122332181a7c3db7f27e918eec64e3"
CACHE = "/weights"
DEFAULT_LANGUAGE = "auto"


class Request(BaseModel):
    # An input this model does not read is a line the caller believed was doing
    # something, so it is refused rather than dropped. It matters more here than
    # elsewhere: the three checkpoint families take different fields, and a
    # speaker sent to a voice-design model would otherwise be silently lost.
    model_config = ConfigDict(extra="forbid")

    text: str = Field(description="What to read aloud.")
    reference_audio: str = Field(description="A recording of the voice to copy, as a base64 audio file. A few seconds of clean speech is enough.")
    reference_text: str | None = Field(default=None, description="What is said in the recording. Given, the copy is closer, because the model is not also transcribing.")
    # Constrained to the model's own list rather than left open. It takes a
    # name and not a code, so "en" is refused where "english" is read, and a
    # pattern says so in the schema instead of in a 400 from the model.
    language: str = Field(
        default=DEFAULT_LANGUAGE,
        pattern="^(auto|chinese|english|french|german|italian|japanese|korean|portuguese|russian|spanish)$",
        description="The language to read in, named rather than coded. Use auto to leave the choice to the model.",
    )


class Spoken(BaseModel):
    audio: str = Field(description="The speech as a base64 WAV.")
    seconds: float
    sample_rate: int


@serve
class Speech:
    def load(self) -> None:
        if hasattr(self, "model"):
            return

        # Imported here rather than at the top of the file. The build imports
        # this file with the platform's own dependencies, so nothing above
        # exists until a worker runs.
        import torch
        from huggingface_hub import snapshot_download
        from qwen_tts import Qwen3TTSModel

        self.torch = torch

        # The whole repository first, then a local path. Handing the repo id
        # straight to from_pretrained fetches only what the root config names,
        # and the speech tokenizer in its own subfolder is then looked for on
        # disk where nothing was written: it fails with "Can't load feature
        # extractor for .../speech_tokenizer". Snapshotting also puts the weights
        # in the volume, which the inner loads do not pass cache_dir down to, so
        # without this every cold start downloads again.
        local = snapshot_download(MODEL, revision=REVISION, cache_dir=CACHE)

        # Straight onto the card. A worker's host RAM is half its card's VRAM,
        # so a model assembled on the host and moved has a ceiling below the
        # card's own.
        self.model = Qwen3TTSModel.from_pretrained(
            local,
            dtype=torch.bfloat16,
            device_map="cuda",
        )

    def predict(self, request: Request) -> Spoken:
        import soundfile

        # Written to a file and passed as a path. The package hands ref_audio
        # straight to soundfile, which opens it as a filename: a base64 string
        # reaches it as one and fails with "Error opening 'UklGRi...'", a
        # LibsndfileError naming the whole payload.
        with tempfile.NamedTemporaryFile(suffix=".wav") as reference:
            reference.write(base64.b64decode(request.reference_audio))
            reference.flush()

            wavs, rate = self.model.generate_voice_clone(
                text=request.text,
                language=request.language,
                ref_audio=reference.name,
                ref_text=request.reference_text,
            )

        # One text in, so one waveform out. The package returns a list because
        # it batches, and the rate beside it rather than on the model.
        track = wavs[0]

        with tempfile.NamedTemporaryFile(suffix=".wav") as handle:
            soundfile.write(handle.name, track, rate, format="WAV")
            handle.seek(0)
            encoded = base64.b64encode(handle.read()).decode("ascii")

        return Spoken(
            audio=encoded,
            seconds=len(track) / rate,
            sample_rate=rate,
        )

configVersion: 1
id: qwen3-tts-17b-base
name: Qwen3-TTS 1.7B Base
source:
  type: code
  entry: model.py
compute:
  gpuType: l40s
build:
  requirements:
    - qwen-tts==0.1.1
    - torch==2.14.1
    - soundfile==0.14.0
    - huggingface-hub==0.36.2
  baseImage: python:3.12
volumes:
  - /weights