Skip to content

Expose TTS fallback controls and activation events (LKINF-495) #7185

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
russellmartin-livekit wants to merge 1 commit into
base: main
Choose a base branch
from
+306 −37
Open

Expose TTS fallback controls and activation events (LKINF-495) #7185

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion livekit-agents/livekit/agents/inference/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
)
from .llm import LLM, LLMModels, LLMStream
from .stt import STT, STTModels
from .tts import TTS, TTSModels
from .tts import TTS, FallbackActivatedEvent, TTSModels
from .vad import VAD, VADModels

if TYPE_CHECKING:
Expand Down Expand Up @@ -39,6 +39,7 @@ def __getattr__(name: str) -> Any:
"LLMStream",
"STTModels",
"TTSModels",
"FallbackActivatedEvent",
"LLMModels",
"VADModels",
"AdaptiveInterruptionDetector",
Expand Down
149 changes: 113 additions & 36 deletions livekit-agents/livekit/agents/inference/tts.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,10 @@
import os
import weakref
from dataclasses import dataclass, replace
from typing import Any, Literal, TypedDict, overload
from typing import Any, Literal, TypedDict, cast, overload

import aiohttp
from pydantic import BaseModel, ValidationError
from typing_extensions import NotRequired

from .. import tts, utils
Expand Down Expand Up @@ -117,6 +118,36 @@ class FallbackModel(TypedDict):

FallbackModelType = FallbackModel | str

FallbackType = Literal["unknown", "fallback", "system_default"]
FallbackCause = Literal["unknown", "timeout", "canceled", "provider_error", "quota_exceeded"]
_FALLBACK_TYPES = frozenset(("unknown", "fallback", "system_default"))
_FALLBACK_CAUSES = frozenset(("unknown", "timeout", "canceled", "provider_error", "quota_exceeded"))


class FallbackActivatedEvent(BaseModel):
"""A fallback model produced content-bearing synthesis output."""

session_id: str
fallback_type: FallbackType
provider: str
model: str
voice: str
cause: FallbackCause


def _normalize_fallback_type(fallback_type: object) -> FallbackType:
if isinstance(fallback_type, str) and fallback_type in _FALLBACK_TYPES:
return cast(FallbackType, fallback_type)
return "unknown"
Comment on lines +138 to +141

@devin-ai-integration devin-ai-integration Bot Sep 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Malformed fallback notices emit events

A missing or non-string fallback_type becomes unknown instead of failing validation. Listeners receive an event when every other field is valid.

Learn more

Unknown string values represent future gateway enum members and can safely map to unknown. Missing values and non-string values are structurally invalid, but this helper maps those values to the same valid enum member. Pydantic therefore cannot reject the notice when its other fields are valid.

Example: A notice containing valid session_id, provider, model, voice, and cause fields but no fallback_type emits FallbackActivatedEvent(fallback_type="unknown"). The notice must instead be ignored while subsequent audio continues.

Recommended fix: Preserve validation failure for missing and non-string values while mapping only unrecognized strings to unknown. Add test cases for an absent value and a non-string value with all other required fields present.

Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.



def _normalize_fallback_cause(
cause: object,
) -> FallbackCause:
if isinstance(cause, str) and cause in _FALLBACK_CAUSES:
return cast(FallbackCause, cause)
return "unknown"


def _has_aligned_transcript(model: str, extra_kwargs: dict[str, Any]) -> bool:
provider = model.split("/")[0]
Expand Down Expand Up @@ -219,16 +250,58 @@ class _TTSOptions:
api_secret: str
extra_kwargs: dict[str, Any]
fallback: NotGivenOr[list[FallbackModel]]
disable_system_default_fallback: bool
conn_options: NotGivenOr[APIConnectOptions]


def _build_session_create_payload(opts: _TTSOptions) -> dict[str, Any]:
params: dict[str, Any] = {
"type": "session.create",
"sample_rate": str(opts.sample_rate),
"encoding": opts.encoding,
"extra": opts.extra_kwargs,
}

if opts.voice:
params["voice"] = opts.voice
if opts.model:
params["model"] = opts.model
if opts.language:
params["language"] = opts.language

if opts.fallback or opts.disable_system_default_fallback:
models = (
[
{
"model": fallback.get("model"),
"voice": fallback.get("voice"),
"extra": fallback.get("extra_kwargs", {}),
}
for fallback in opts.fallback
]
if is_given(opts.fallback)
else []
)
params["fallback"] = {"models": models}
if opts.disable_system_default_fallback:
params["fallback"]["disable_system_default_fallback"] = True

if opts.conn_options:
params["connection"] = {
"timeout": opts.conn_options.timeout,
"retries": opts.conn_options.max_retry,
}

return params


@dataclass(eq=False)
class _TTSConnection:
ws: aiohttp.ClientWebSocketResponse
session_id: str | None


class TTS(tts.TTS):
class TTS(tts.TTS[Literal["fallback_activated"]]):
@overload
def __init__(
self,
Expand All @@ -244,6 +317,7 @@ def __init__(
http_session: aiohttp.ClientSession | None = None,
extra_kwargs: NotGivenOr[CartesiaOptions] = NOT_GIVEN,
fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
disable_system_default_fallback: bool = False,
conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
) -> None:
pass
Expand All @@ -263,6 +337,7 @@ def __init__(
http_session: aiohttp.ClientSession | None = None,
extra_kwargs: NotGivenOr[DeepgramOptions] = NOT_GIVEN,
fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
disable_system_default_fallback: bool = False,
conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
) -> None:
pass
Expand All @@ -282,6 +357,7 @@ def __init__(
http_session: aiohttp.ClientSession | None = None,
extra_kwargs: NotGivenOr[RimeOptions] = NOT_GIVEN,
fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
disable_system_default_fallback: bool = False,
conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
) -> None:
pass
Expand All @@ -301,6 +377,7 @@ def __init__(
http_session: aiohttp.ClientSession | None = None,
extra_kwargs: NotGivenOr[InworldOptions] = NOT_GIVEN,
fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
disable_system_default_fallback: bool = False,
conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
) -> None:
pass
Expand All @@ -320,6 +397,7 @@ def __init__(
http_session: aiohttp.ClientSession | None = None,
extra_kwargs: NotGivenOr[XaiOptions] = NOT_GIVEN,
fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
disable_system_default_fallback: bool = False,
conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
) -> None:
pass
Expand All @@ -339,6 +417,7 @@ def __init__(
http_session: aiohttp.ClientSession | None = None,
extra_kwargs: NotGivenOr[FishAudioOptions] = NOT_GIVEN,
fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
disable_system_default_fallback: bool = False,
conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
) -> None:
pass
Expand All @@ -358,6 +437,7 @@ def __init__(
http_session: aiohttp.ClientSession | None = None,
extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
disable_system_default_fallback: bool = False,
conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
) -> None:
pass
Expand All @@ -384,6 +464,7 @@ def __init__(
| FishAudioOptions
] = NOT_GIVEN,
fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
disable_system_default_fallback: bool = False,
conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
) -> None:
"""Livekit Cloud Inference TTS
Expand All @@ -401,6 +482,8 @@ def __init__(
extra_kwargs (dict, optional): Extra kwargs to pass to the TTS model.
fallback (FallbackModelType, optional): Fallback models - either a list of model names,
a list of FallbackModel instances.
disable_system_default_fallback (bool, optional): Disable system-configured fallback
models.
conn_options (APIConnectOptions, optional): Connection options for request attempts.
"""
sample_rate = sample_rate if is_given(sample_rate) else DEFAULT_SAMPLE_RATE
Expand Down Expand Up @@ -459,6 +542,7 @@ def __init__(
api_secret=lk_api_secret,
extra_kwargs=resolved_extra_kwargs,
fallback=fallback_models,
disable_system_default_fallback=disable_system_default_fallback,
conn_options=conn_options if is_given(conn_options) else DEFAULT_API_CONNECT_OPTIONS,
)
self._session = http_session
Expand Down Expand Up @@ -532,35 +616,7 @@ async def _connect_ws(self, timeout: float) -> _TTSConnection:
except aiohttp.ClientConnectorError as e:
raise APIConnectionError("failed to connect to LiveKit Inference TTS") from e

params: dict[str, Any] = {
"type": "session.create",
"sample_rate": str(self._opts.sample_rate),
"encoding": self._opts.encoding,
"extra": self._opts.extra_kwargs,
}

if self._opts.voice:
params["voice"] = self._opts.voice
if self._opts.model:
params["model"] = self._opts.model
if self._opts.language:
params["language"] = self._opts.language
if self._opts.fallback:
models = [
{
"model": m.get("model"),
"voice": m.get("voice"),
"extra": m.get("extra_kwargs", {}),
}
for m in self._opts.fallback
]
params["fallback"] = {"models": models}

if self._opts.conn_options:
params["connection"] = {
"timeout": self._opts.conn_options.timeout,
"retries": self._opts.conn_options.max_retry,
}
params = _build_session_create_payload(self._opts)

try:
payload = json.dumps(params)
Expand Down Expand Up @@ -725,12 +781,33 @@ async def _recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
current_session_id = session_id
output_emitter.start_segment(segment_id=session_id)

if data.get("type") == "session.created":
msg_type = data.get("type")
if msg_type == "session.created":
pass
elif data.get("type") == "output_audio":
elif msg_type == "fallback_activated":
try:
event = FallbackActivatedEvent(
session_id=data.get("session_id"),
fallback_type=_normalize_fallback_type(data.get("fallback_type")),
provider=data.get("provider"),
model=data.get("model"),
voice=data.get("voice"),
cause=_normalize_fallback_cause(data.get("cause")),
)
except ValidationError as e:
logger.warning(
"ignoring invalid fallback activation notice",
extra={"session_id": data.get("session_id"), "error": str(e)},
Comment on lines +797 to +800

@devin-ai-integration devin-ai-integration Bot Sep 11, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟨 Validation errors bypass PII redaction

A malformed fallback notice logs str(e) under the unmarked error attribute. Rejected gateway values can reach logs without collector redaction.

Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

)
continue
self._tts.emit(
"fallback_activated",
event,
)
elif msg_type == "output_audio":
b64data = base64.b64decode(data["audio"])
output_emitter.push(b64data)
elif data.get("type") == "output_alignment":
elif msg_type == "output_alignment":
aligned: list[TimedString] = []
if words := data.get("words"):
aligned = [
Expand All @@ -754,14 +831,14 @@ async def _recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
if self._expressive
else aligned
)
elif data.get("type") == "done":
elif msg_type == "done":
if self._held_tokens: # release an unclosed span, cue unresolved
output_emitter.push_timed_transcript(
drop_bracket_cues([], self._held_tokens, final=True)
)
output_emitter.end_input()
break
elif data.get("type") == "error":
elif msg_type == "error":
raise APIError(f"LiveKit Inference TTS returned error: {msg.data}")

try:
Expand Down
Loading