Module livekit.agents.beta.translation.translation_model

Classes

class InputSpeechStarted

Expand source code

@dataclass
class InputSpeechStarted:
    """The speaker began a turn, as the model heard it."""

The speaker began a turn, as the model heard it.

class InputSpeechStopped

Expand source code

@dataclass
class InputSpeechStopped:
    """The speaker ended a turn, as the model heard it."""

The speaker ended a turn, as the model heard it.

class InputTranscript (text: str,
is_final: bool,
language: str | None = None,
speaker_id: str | None = None)

Expand source code

@dataclass
class InputTranscript:
    """What the model heard, in the language it was spoken."""

    text: str
    is_final: bool
    language: str | None = None
    speaker_id: str | None = None

What the model heard, in the language it was spoken.

Instance variables

var is_final : bool

var language : str | None

var speaker_id : str | None

var text : str

class OutputSpeechStopped

Expand source code

@dataclass
class OutputSpeechStopped:
    """The model finished speaking a translation; its audio up to here is one segment.

    Yielded by :attr:`TranslationModelSession.audio_stream` rather than emitted as an event,
    so it cannot overtake audio the model has already queued ahead of it.
    """

The model finished speaking a translation; its audio up to here is one segment.

Yielded by :attr:TranslationModelSession.audio_stream rather than emitted as an event, so it cannot overtake audio the model has already queued ahead of it.

class OutputTranscript (text: str, is_final: bool, language: str | None = None)

Expand source code

@dataclass
class OutputTranscript:
    """What the model said, in the language it translates into."""

    text: str
    is_final: bool
    language: str | None = None

What the model said, in the language it translates into.

Instance variables

var is_final : bool

var language : str | None

var text : str

class TranslatedAudioFrame (frame: rtc.AudioFrame, language: str | None = None)

Expand source code

@dataclass
class TranslatedAudioFrame:
    """One frame of the translation the model speaks."""

    frame: rtc.AudioFrame
    language: str | None = None
    """The language spoken in this frame, for a provider that stamps its audio."""

One frame of the translation the model speaks.

Instance variables

var frame : AudioFrame

var language : str | None

The language spoken in this frame, for a provider that stamps its audio.

class TranslationCapabilities (input_transcription: bool = False,
output_transcription: bool = False,
turn_reporting: bool = False,
output_segmentation: bool = False,
retargetable: bool = False)

Expand source code

@dataclass
class TranslationCapabilities:
    """What varies between translation providers."""

    input_transcription: bool = False
    """Whether the model transcribes the speech it hears, in the language it was spoken."""
    output_transcription: bool = False
    """Whether the model transcribes the translation it speaks."""
    turn_reporting: bool = False
    """Whether the model reports where the speaker's turns start and end.

    About the speech it hears. For the speech it speaks, see ``output_segmentation``.
    """
    output_segmentation: bool = False
    """Whether the model marks where each translation it speaks ends.

    When it does, the session ends a playout segment on that mark alone. When it does not,
    the session falls back to a silence timer, which cannot tell a pause inside a sentence
    from the end of one.
    """
    retargetable: bool = False
    """Whether the target language can be changed on a running session."""

What varies between translation providers.

Instance variables

var input_transcription : bool

Whether the model transcribes the speech it hears, in the language it was spoken.

var output_segmentation : bool

Whether the model marks where each translation it speaks ends.

When it does, the session ends a playout segment on that mark alone. When it does not, the session falls back to a silence timer, which cannot tell a pause inside a sentence from the end of one.

var output_transcription : bool

Whether the model transcribes the translation it speaks.

var retargetable : bool

Whether the target language can be changed on a running session.

var turn_reporting : bool

Whether the model reports where the speaker's turns start and end.

About the speech it hears. For the speech it speaks, see output_segmentation.

class TranslationModel (*,
capabilities: TranslationCapabilities,
target_language: str,
source_language: str | None = None)

Expand source code

class TranslationModel(ABC):
    """A model that hears speech in one language and speaks it in another.

    Its own primitive, beside :class:`~livekit.agents.llm.LLM`,
    :class:`~livekit.agents.llm.RealtimeModel` and :class:`~livekit.agents.llm.DuplexModel`: a
    translator holds no conversation, so it takes no instructions, carries no chat context,
    calls no tools and never decides to reply. What it has instead is a language it translates
    into. Run one with
    :class:`~livekit.agents.beta.translation.TranslationSession`.
    """

    def __init__(
        self,
        *,
        capabilities: TranslationCapabilities,
        target_language: str,
        source_language: str | None = None,
    ) -> None:
        """
        Args:
            capabilities: What this provider can do.
            target_language: The language the model speaks its translation in.
            source_language: The language the model expects to hear, or ``None`` when it
                detects the language itself.
        """
        self._capabilities = capabilities
        self._target_language = target_language
        self._source_language = source_language
        self._label = f"{type(self).__module__}.{type(self).__name__}"

    @property
    def model(self) -> str:
        return "unknown"

    @property
    def provider(self) -> str:
        return "unknown"

    @property
    def capabilities(self) -> TranslationCapabilities:
        return self._capabilities

    @property
    def label(self) -> str:
        return self._label

    @property
    def target_language(self) -> str:
        """The language the model speaks its translation in."""
        return self._target_language

    @property
    def source_language(self) -> str | None:
        """The language the model expects to hear, or ``None`` when it detects it itself."""
        return self._source_language

    def update_options(self, *, target_language: str) -> None:
        """Point the model at another language.

        Takes effect on the sessions opened after it. To retarget a running session, call
        :meth:`TranslationModelSession.update_options` on the session itself, which a
        provider only serves when ``capabilities.retargetable``.
        """
        self._target_language = target_language

    @abstractmethod
    def session(self) -> TranslationModelSession:
        """Open a session; the caller pushes audio into it and reads its translation out."""

    @abstractmethod
    async def aclose(self) -> None: ...

    async def __aenter__(self) -> TranslationModel:
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        await self.aclose()

A model that hears speech in one language and speaks it in another.

Its own primitive, beside :class:~livekit.agents.llm.LLM, :class:~livekit.agents.llm.RealtimeModel and :class:~livekit.agents.llm.DuplexModel: a translator holds no conversation, so it takes no instructions, carries no chat context, calls no tools and never decides to reply. What it has instead is a language it translates into. Run one with :class:~livekit.agents.beta.translation.TranslationSession.

Args

capabilities

What this provider can do.

target_language

The language the model speaks its translation in.

source_language

The language the model expects to hear, or None when it detects the language itself.

Ancestors

• abc.ABC

Instance variables

prop capabilities : TranslationCapabilities

Expand source code

@property
def capabilities(self) -> TranslationCapabilities:
    return self._capabilities

prop label : str

Expand source code

@property
def label(self) -> str:
    return self._label

prop model : str

Expand source code

@property
def model(self) -> str:
    return "unknown"

prop provider : str

Expand source code

@property
def provider(self) -> str:
    return "unknown"

prop source_language : str | None

Expand source code

@property
def source_language(self) -> str | None:
    """The language the model expects to hear, or ``None`` when it detects it itself."""
    return self._source_language

The language the model expects to hear, or None when it detects it itself.

prop target_language : str

Expand source code

@property
def target_language(self) -> str:
    """The language the model speaks its translation in."""
    return self._target_language

The language the model speaks its translation in.

Methods

async def aclose(self) ‑> None

Expand source code

@abstractmethod
async def aclose(self) -> None: ...

def session(self) ‑> TranslationModelSession

Expand source code

@abstractmethod
def session(self) -> TranslationModelSession:
    """Open a session; the caller pushes audio into it and reads its translation out."""

Open a session; the caller pushes audio into it and reads its translation out.

def update_options(self, *, target_language: str) ‑> None

Expand source code

def update_options(self, *, target_language: str) -> None:
    """Point the model at another language.

    Takes effect on the sessions opened after it. To retarget a running session, call
    :meth:`TranslationModelSession.update_options` on the session itself, which a
    provider only serves when ``capabilities.retargetable``.
    """
    self._target_language = target_language

Point the model at another language.

Takes effect on the sessions opened after it. To retarget a running session, call :meth:TranslationModelSession.update_options() on the session itself, which a provider only serves when capabilities.retargetable.

class TranslationModelError (**data: Any)

Expand source code

class TranslationModelError(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    type: Literal["translation_model_error"] = "translation_model_error"
    timestamp: float
    label: str
    error: Exception = Field(..., exclude=True)
    recoverable: bool

Usage Documentation

Models

A base class for creating Pydantic models.

Attributes

__class_vars__

The names of the class variables defined on the model.

__private_attributes__

Metadata about the private attributes of the model.

__signature__

The synthesized __init__ [Signature][inspect.Signature] of the model.

__pydantic_complete__

Whether model building is completed, or if there are still undefined fields.

__pydantic_core_schema__

The core schema of the model.

__pydantic_custom_init__

Whether the model has a custom __init__ function.

__pydantic_decorators__

Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.

__pydantic_generic_metadata__

A dictionary containing metadata about generic Pydantic models. The origin and args items map to the [__origin__][genericalias.origin] and [__args__][genericalias.args] attributes of [generic aliases][types-genericalias], and the parameter item maps to the __parameter__ attribute of generic classes.

__pydantic_parent_namespace__

Parent namespace of the model, used for automatic rebuilding of models.

__pydantic_post_init__

The name of the post-init method for the model, if defined.

__pydantic_root_model__

Whether the model is a [RootModel][pydantic.root_model.RootModel].

__pydantic_serializer__

The pydantic-core SchemaSerializer used to dump instances of the model.

__pydantic_validator__

The pydantic-core SchemaValidator used to validate instances of the model.

__pydantic_fields__

A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.

__pydantic_computed_fields__

A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.

__pydantic_extra__

A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.

__pydantic_fields_set__

The names of fields explicitly set during instantiation.

__pydantic_private__

Values of private attributes set on the model instance.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var error : Exception

var label : str

var model_config

var recoverable : bool

var timestamp : float

var type : Literal['translation_model_error']

class TranslationModelSession (translation_model: TranslationModel)

Expand source code

class TranslationModelSession(
    ABC, rtc.EventEmitter[TranslationEventTypes | TEvent], Generic[TEvent]
):
    """One run of a :class:`TranslationModel`: audio in, translated audio out.

    The speech the model hears is pushed in frame by frame with :meth:`push_audio`; the
    translation it speaks is read out of :attr:`audio_stream`, which also carries an
    :class:`OutputSpeechStopped` where each spoken translation ends. What it heard and what it
    said arrive as ``input_transcript`` and ``output_transcript`` events, and the speaker's
    turns as ``input_speech_started`` and ``input_speech_stopped``.
    """

    def __init__(self, translation_model: TranslationModel) -> None:
        super().__init__()
        self._translation_model = translation_model
        self._closing = False

    @property
    def translation_model(self) -> TranslationModel:
        return self._translation_model

    @property
    def capabilities(self) -> TranslationCapabilities:
        return self._translation_model.capabilities

    @property
    @abstractmethod
    def audio_stream(self) -> AsyncIterable[TranslatedAudioFrame | OutputSpeechStopped]:
        """The translation the model speaks, and where each one ends.

        A model whose ``capabilities.output_segmentation`` is set yields an
        :class:`OutputSpeechStopped` after the last frame of each translation. Because it
        travels in the stream, it is always read after the audio it closes.
        """

    @abstractmethod
    def push_audio(self, frame: rtc.AudioFrame) -> None:
        """Hand the model one frame of the speech to translate."""

    def clear_audio(self) -> None:
        """Drop the translation the model has not spoken yet.

        A no-op for a provider that cannot discard what it has already produced.
        """

    def update_options(self, *, target_language: str) -> None:
        """Point this running session at another language.

        Raises:
            RuntimeError: If the provider fixes the target language for the session's life.
        """
        raise RuntimeError(
            f"{self._translation_model.label} translates into a language fixed for the "
            "session's life; start a new session to translate into another"
        )

    @abstractmethod
    async def aclose(self) -> None: ...

One run of a :class:TranslationModel: audio in, translated audio out.

The speech the model hears is pushed in frame by frame with :meth:push_audio; the translation it speaks is read out of :attr:audio_stream, which also carries an :class:OutputSpeechStopped where each spoken translation ends. What it heard and what it said arrive as input_transcript and output_transcript events, and the speaker's turns as input_speech_started and input_speech_stopped.

Ancestors

• abc.ABC
• EventEmitter
• typing.Generic

Instance variables

prop audio_stream : AsyncIterable[TranslatedAudioFrame | OutputSpeechStopped]

Expand source code

@property
@abstractmethod
def audio_stream(self) -> AsyncIterable[TranslatedAudioFrame | OutputSpeechStopped]:
    """The translation the model speaks, and where each one ends.

    A model whose ``capabilities.output_segmentation`` is set yields an
    :class:`OutputSpeechStopped` after the last frame of each translation. Because it
    travels in the stream, it is always read after the audio it closes.
    """

The translation the model speaks, and where each one ends.

A model whose capabilities.output_segmentation is set yields an :class:OutputSpeechStopped after the last frame of each translation. Because it travels in the stream, it is always read after the audio it closes.

prop capabilities : TranslationCapabilities

Expand source code

@property
def capabilities(self) -> TranslationCapabilities:
    return self._translation_model.capabilities

prop translation_model : TranslationModel

Expand source code

@property
def translation_model(self) -> TranslationModel:
    return self._translation_model

Methods

async def aclose(self) ‑> None

Expand source code

@abstractmethod
async def aclose(self) -> None: ...

def clear_audio(self) ‑> None

Expand source code

def clear_audio(self) -> None:
    """Drop the translation the model has not spoken yet.

    A no-op for a provider that cannot discard what it has already produced.
    """

Drop the translation the model has not spoken yet.

A no-op for a provider that cannot discard what it has already produced.

def push_audio(self, frame: rtc.AudioFrame) ‑> None

Expand source code

@abstractmethod
def push_audio(self, frame: rtc.AudioFrame) -> None:
    """Hand the model one frame of the speech to translate."""

Hand the model one frame of the speech to translate.

def update_options(self, *, target_language: str) ‑> None

Expand source code

def update_options(self, *, target_language: str) -> None:
    """Point this running session at another language.

    Raises:
        RuntimeError: If the provider fixes the target language for the session's life.
    """
    raise RuntimeError(
        f"{self._translation_model.label} translates into a language fixed for the "
        "session's life; start a new session to translate into another"
    )

Point this running session at another language.

Raises

RuntimeError

If the provider fixes the target language for the session's life.

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once