Module livekit.plugins.openai
OpenAI plugin for LiveKit Agents
Support for OpenAI Realtime API, LLM, TTS, and STT APIs.
Also includes support for a large number of OpenAI-compatible APIs including Azure OpenAI, Cerebras, Fireworks, Perplexity, Telnyx, xAI, Ollama, DeepSeek, OpenRouter, and OVHcloud AI Endpoints.
See https://docs.livekit.io/agents/integrations/openai/ and https://docs.livekit.io/agents/integrations/llm/ for more information.
Sub-modules
livekit.plugins.openai.realtimelivekit.plugins.openai.responseslivekit.plugins.openai.tools
Functions
async def create_embeddings(*,
input: list[str],
model: models.EmbeddingModels = 'text-embedding-3-small',
dimensions: int | None = None,
api_key: str | None = None,
http_session: aiohttp.ClientSession | None = None) ‑> list[livekit.plugins.openai.embeddings.EmbeddingData]-
Expand source code
async def create_embeddings( *, input: list[str], model: models.EmbeddingModels = "text-embedding-3-small", dimensions: int | None = None, api_key: str | None = None, http_session: aiohttp.ClientSession | None = None, ) -> list[EmbeddingData]: http_session = http_session or utils.http_context.http_session() api_key = api_key or os.environ.get("OPENAI_API_KEY") if not api_key: raise ValueError("OPENAI_API_KEY must be set") async with http_session.post( "https://api.openai.com/v1/embeddings", headers={"Authorization": f"Bearer {api_key}"}, json={ "model": model, "input": input, "encoding_format": "base64", "dimensions": dimensions, }, ) as resp: json = await resp.json() data = json["data"] list_data = [] for d in data: bytes = base64.b64decode(d["embedding"]) num_floats = len(bytes) // 4 floats = list(struct.unpack("f" * num_floats, bytes)) list_data.append(EmbeddingData(index=d["index"], embedding=floats)) return list_data
Classes
class EmbeddingData (index: int, embedding: list[float])-
Expand source code
@dataclass class EmbeddingData: index: int embedding: list[float]EmbeddingData(index: 'int', embedding: 'list[float]')
Instance variables
var embedding : list[float]var index : int
class LLM (*,
model: str | ChatModels = 'gpt-4.1',
api_key: NotGivenOr[str] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
store: NotGivenOr[bool] = NOT_GIVEN,
metadata: NotGivenOr[dict[str, str]] = NOT_GIVEN,
max_completion_tokens: NotGivenOr[int] = NOT_GIVEN,
timeout: httpx.Timeout | None = None,
max_retries: NotGivenOr[int] = NOT_GIVEN,
service_tier: NotGivenOr[str] = NOT_GIVEN,
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
verbosity: NotGivenOr[Verbosity] = NOT_GIVEN,
prompt_cache_retention: NotGivenOr[PromptCacheRetention] = NOT_GIVEN,
extra_body: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
extra_headers: NotGivenOr[dict[str, str]] = NOT_GIVEN,
extra_query: NotGivenOr[dict[str, str]] = NOT_GIVEN)-
Expand source code
class LLM(llm.LLM): def __init__( self, *, model: str | ChatModels = "gpt-4.1", api_key: NotGivenOr[str] = NOT_GIVEN, base_url: NotGivenOr[str] = NOT_GIVEN, client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN, store: NotGivenOr[bool] = NOT_GIVEN, metadata: NotGivenOr[dict[str, str]] = NOT_GIVEN, max_completion_tokens: NotGivenOr[int] = NOT_GIVEN, timeout: httpx.Timeout | None = None, max_retries: NotGivenOr[int] = NOT_GIVEN, service_tier: NotGivenOr[str] = NOT_GIVEN, reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, verbosity: NotGivenOr[Verbosity] = NOT_GIVEN, prompt_cache_retention: NotGivenOr[PromptCacheRetention] = NOT_GIVEN, extra_body: NotGivenOr[dict[str, Any]] = NOT_GIVEN, extra_headers: NotGivenOr[dict[str, str]] = NOT_GIVEN, extra_query: NotGivenOr[dict[str, str]] = NOT_GIVEN, _provider_fmt: NotGivenOr[str] = NOT_GIVEN, _strict_tool_schema: bool = True, ) -> None: """ Create a new instance of OpenAI LLM. ``api_key`` must be set to your OpenAI API key, either using the argument or by setting the ``OPENAI_API_KEY`` environmental variable. """ super().__init__() if not is_given(reasoning_effort) and _supports_reasoning_effort(model): if model in ["gpt-5.1", "gpt-5.2", "gpt-5.4", "gpt-5.4-mini"]: reasoning_effort = "none" else: reasoning_effort = "minimal" self._opts = _LLMOptions( model=model, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, store=store, metadata=metadata, max_completion_tokens=max_completion_tokens, service_tier=service_tier, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, verbosity=verbosity, prompt_cache_retention=prompt_cache_retention, extra_body=extra_body, extra_headers=extra_headers, extra_query=extra_query, ) if is_given(api_key) and not api_key: raise ValueError( "OpenAI API key is required, either as argument or set" " OPENAI_API_KEY environment variable" ) self._provider_fmt = _provider_fmt or "openai" self._strict_tool_schema = _strict_tool_schema self._owns_client = client is None self._client = client or openai.AsyncClient( api_key=api_key if is_given(api_key) else None, base_url=base_url if is_given(base_url) else None, max_retries=max_retries if is_given(max_retries) else 0, http_client=httpx.AsyncClient( timeout=timeout if timeout else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0), follow_redirects=True, limits=httpx.Limits( max_connections=50, max_keepalive_connections=50, keepalive_expiry=120, ), ), ) async def _prewarm_impl(self) -> None: # token-free request supported by openai and openai-compatible servers await self._client.models.list() async def aclose(self) -> None: await super().aclose() if self._owns_client: await self._client.close() @property def model(self) -> str: return self._opts.model @property def provider(self) -> str: return self._client._base_url.netloc.decode("utf-8") @staticmethod def with_azure( *, model: str | ChatModels = "gpt-4o", azure_endpoint: str | None = None, azure_deployment: str | None = None, api_version: str | None = None, api_key: str | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: AsyncAzureADTokenProvider | None = None, organization: str | None = None, project: str | None = None, base_url: str | None = None, user: NotGivenOr[str] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN, timeout: httpx.Timeout | None = None, reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, verbosity: NotGivenOr[Verbosity] = NOT_GIVEN, max_completion_tokens: NotGivenOr[int] = NOT_GIVEN, ) -> LLM: """ This automatically infers the following arguments from their corresponding environment variables if they are not provided: - `api_key` from `AZURE_OPENAI_API_KEY` - `organization` from `OPENAI_ORG_ID` - `project` from `OPENAI_PROJECT_ID` - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN` - `api_version` from `OPENAI_API_VERSION` - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT` """ # noqa: E501 azure_client = openai.AsyncAzureOpenAI( max_retries=0, azure_endpoint=azure_endpoint, azure_deployment=azure_deployment, api_version=api_version, api_key=api_key, azure_ad_token=azure_ad_token, azure_ad_token_provider=azure_ad_token_provider, organization=organization, project=project, base_url=base_url, timeout=timeout if timeout else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0), ) # type: ignore llm = LLM( model=model, client=azure_client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, verbosity=verbosity, max_completion_tokens=max_completion_tokens, ) llm._owns_client = True return llm @staticmethod def with_cerebras( *, model: str | CerebrasChatModels = "gpt-oss-120b", api_key: str | None = None, base_url: str = "https://api.cerebras.ai/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN, reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of Cerebras LLM. ``api_key`` must be set to your Cerebras API key, either using the argument or by setting the ``CEREBRAS_API_KEY`` environment variable. """ api_key = api_key or os.environ.get("CEREBRAS_API_KEY") if api_key is None: raise ValueError( "Cerebras API key is required, either as argument or set CEREBRAS_API_KEY environment variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, _strict_tool_schema=False, ) @staticmethod def with_sambanova( *, model: str | SambaNovaChatModels = "DeepSeek-R1-0528", api_key: str | None = None, base_url: str = "https://api.sambanova.ai/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of SambaNova LLM (OpenAI-compatible). ``api_key`` must be set to your SambaNova API key, either using the argument or by setting the ``SAMBANOVA_API_KEY`` environment variable. """ api_key = api_key or os.environ.get("SAMBANOVA_API_KEY") if api_key is None: raise ValueError( "SambaNova API key is required, either as argument or set SAMBANOVA_API_KEY environment variable" ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, _strict_tool_schema=False, ) @staticmethod def with_fireworks( *, model: str = "accounts/fireworks/models/llama-v3p3-70b-instruct", api_key: str | None = None, base_url: str = "https://api.fireworks.ai/inference/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of Fireworks LLM. ``api_key`` must be set to your Fireworks API key, either using the argument or by setting the ``FIREWORKS_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("FIREWORKS_API_KEY") if api_key is None: raise ValueError( "Fireworks API key is required, either as argument or set FIREWORKS_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, ) @staticmethod def with_x_ai( *, model: str | XAIChatModels = "grok-3-fast", api_key: str | None = None, base_url: str = "https://api.x.ai/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of XAI LLM. ``api_key`` must be set to your XAI API key, either using the argument or by setting the ``XAI_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("XAI_API_KEY") if api_key is None: raise ValueError( "XAI API key is required, either as argument or set XAI_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, # TODO(long): add provider fmt for grok reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, ) @staticmethod def with_openrouter( *, model: str = "auto", api_key: str | None = None, base_url: str = "https://openrouter.ai/api/v1", client: openai.AsyncClient | None = None, site_url: str | None = None, app_name: str | None = None, fallback_models: list[str] | None = None, provider: OpenRouterProviderPreferences | None = None, plugins: list[OpenRouterWebPlugin] | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, timeout: httpx.Timeout | None = None, ) -> LLM: """ Create a new instance of OpenRouter LLM. ``api_key`` must be set to your OpenRouter API key, either using the argument or by setting the ``OPENROUTER_API_KEY`` environment variable. """ api_key = api_key or os.environ.get("OPENROUTER_API_KEY") if api_key is None: raise ValueError( "OpenRouter API key is required, either as argument or set OPENROUTER_API_KEY environment variable" ) # Set up analytics headers for OpenRouter default_headers: dict[str, str] = {} if site_url: default_headers["HTTP-Referer"] = site_url if app_name: default_headers["X-Title"] = app_name # Build OpenRouter-specific request body or_body: dict[str, Any] = {} if provider: or_body["provider"] = provider if fallback_models: # Set fallback models for routing or_body["models"] = [model, *fallback_models] if plugins: or_body["plugins"] = [ {k: v for k, v in asdict(p).items() if v is not None} for p in plugins ] return LLM( model=model, api_key=api_key, client=client, base_url=base_url, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, extra_body=or_body, extra_headers=default_headers, timeout=timeout, ) @staticmethod def with_deepseek( *, model: str | DeepSeekChatModels = "deepseek-chat", api_key: str | None = None, base_url: str = "https://api.deepseek.com/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of DeepSeek LLM. ``api_key`` must be set to your DeepSeek API key, either using the argument or by setting the ``DEEPSEEK_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("DEEPSEEK_API_KEY") if api_key is None: raise ValueError( "DeepSeek API key is required, either as argument or set DEEPSEEK_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, ) @staticmethod def with_cometapi( *, model: str | CometAPIChatModels = "gpt-5-chat-latest", api_key: str | None = None, base_url: str = "https://api.cometapi.com/v1/", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of CometAPI LLM. ``api_key`` must be set to your CometAPI API key, either using the argument or by setting the ``COMETAPI_API_KEY`` environmental variable. CometAPI provides access to 500+ AI models from multiple providers including OpenAI, Anthropic, Google, xAI, DeepSeek, and Qwen through a unified API. Get your API key at: https://api.cometapi.com/console/token Learn more: https://www.cometapi.com/?utm_source=livekit&utm_campaign=integration&utm_medium=integration&utm_content=integration """ api_key = api_key or os.environ.get("COMETAPI_API_KEY") if api_key is None: raise ValueError( "CometAPI API key is required, either as argument or set COMETAPI_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, ) @staticmethod def with_octo( *, model: str | OctoChatModels = "llama-2-13b-chat", api_key: str | None = None, base_url: str = "https://text.octoai.run/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of OctoAI LLM. ``api_key`` must be set to your OctoAI API key, either using the argument or by setting the ``OCTOAI_TOKEN`` environmental variable. """ api_key = api_key or os.environ.get("OCTOAI_TOKEN") if api_key is None: raise ValueError( "OctoAI API key is required, either as argument or set OCTOAI_TOKEN environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, ) @staticmethod def with_ollama( *, model: str = "llama3.1", base_url: str = "http://localhost:11434/v1", client: openai.AsyncClient | None = None, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of Ollama LLM. """ return LLM( model=model, api_key="ollama", base_url=base_url, client=client, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, ) @staticmethod def with_ovhcloud( *, model: str = "gpt-oss-120b", api_key: str | None = None, base_url: str = "https://oai.endpoints.kepler.ai.cloud.ovh.net/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of OVHcloud AI Endpoints LLM. ``api_key`` must be set to your OVHcloud AI Endpoints API key, either using the argument or by setting the ``OVHCLOUD_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("OVHCLOUD_API_KEY") if api_key is None: raise ValueError( "OVHcloud AI Endpoints API key is required, either as argument or set OVHCLOUD_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, ) @staticmethod def with_perplexity( *, model: str | PerplexityChatModels = "llama-3.1-sonar-small-128k-chat", api_key: str | None = None, base_url: str = "https://api.perplexity.ai", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of PerplexityAI LLM. ``api_key`` must be set to your TogetherAI API key, either using the argument or by setting the ``PERPLEXITY_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("PERPLEXITY_API_KEY") if api_key is None: raise ValueError( "Perplexity AI API key is required, either as argument or set PERPLEXITY_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, ) @staticmethod def with_together( *, model: str | TogetherChatModels = "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo", api_key: str | None = None, base_url: str = "https://api.together.xyz/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of TogetherAI LLM. ``api_key`` must be set to your TogetherAI API key, either using the argument or by setting the ``TOGETHER_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("TOGETHER_API_KEY") if api_key is None: raise ValueError( "Together AI API key is required, either as argument or set TOGETHER_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, ) @staticmethod def with_telnyx( *, model: str | TelnyxChatModels = "meta-llama/Meta-Llama-3.1-70B-Instruct", api_key: str | None = None, base_url: str = "https://api.telnyx.com/v2/ai", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of Telnyx LLM. ``api_key`` must be set to your Telnyx API key, either using the argument or by setting the ``TELNYX_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("TELNYX_API_KEY") if api_key is None: raise ValueError( "Telnyx AI API key is required, either as argument or set TELNYX_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, ) @staticmethod def with_nebius( *, model: str | NebiusChatModels = "meta-llama/Meta-Llama-3.1-70B-Instruct", api_key: str | None = None, base_url: str = "https://api.studio.nebius.com/v1/", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of Nebius LLM. ``api_key`` must be set to your Nebius API key, either using the argument or by setting the ``NEBIUS_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("NEBIUS_API_KEY") if api_key is None: raise ValueError( "Nebius API key is required, either as argument or set NEBIUS_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, ) @staticmethod def with_letta( *, agent_id: str, base_url: str = "https://api.letta.com/v1/chat/completions", api_key: str | None = None, ) -> LLM: """ Create a new Letta-backed LLM instance connected to the specified Letta agent. Args: agent_id (str): The Letta agent ID (must be prefixed with 'agent-'). base_url (str): The URL of the Letta server (e.g., http://localhost:8283/v1/chat/completions for local or https://api.letta.com/v1/chat/completions for cloud). api_key (str | None, optional): Optional API key for authentication, required if the Letta server enforces auth. Returns: LLM: A configured LLM instance for interacting with the given Letta agent. """ parsed = urlparse(base_url) if parsed.scheme not in {"http", "https"}: raise ValueError(f"Invalid URL scheme: '{parsed.scheme}'. Must be 'http' or 'https'.") if not parsed.netloc: raise ValueError(f"URL '{base_url}' is missing a network location (e.g., domain name).") api_key = api_key or os.environ.get("LETTA_API_KEY") if api_key is None: raise ValueError( "Letta API key is required, either as argument or set LETTA_API_KEY environmental variable" # noqa: E501 ) return LLM( model=agent_id, api_key=api_key, base_url=base_url, client=None, user=NOT_GIVEN, temperature=NOT_GIVEN, parallel_tool_calls=NOT_GIVEN, tool_choice=NOT_GIVEN, ) def chat( self, *, chat_ctx: ChatContext, tools: list[llm.Tool] | None = None, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN, response_format: NotGivenOr[ completion_create_params.ResponseFormat | type[llm_utils.ResponseFormatT] ] = NOT_GIVEN, extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN, ) -> LLMStream: extra = {} if is_given(extra_kwargs): extra.update(extra_kwargs) if is_given(self._opts.extra_body): extra["extra_body"] = self._opts.extra_body if is_given(self._opts.extra_headers): extra["extra_headers"] = self._opts.extra_headers if is_given(self._opts.extra_query): extra["extra_query"] = self._opts.extra_query if is_given(self._opts.metadata): extra["metadata"] = self._opts.metadata if is_given(self._opts.user): extra["user"] = self._opts.user if is_given(self._opts.max_completion_tokens): extra["max_completion_tokens"] = self._opts.max_completion_tokens if is_given(self._opts.temperature): extra["temperature"] = self._opts.temperature if is_given(self._opts.service_tier): extra["service_tier"] = self._opts.service_tier if is_given(self._opts.reasoning_effort): extra["reasoning_effort"] = self._opts.reasoning_effort if is_given(self._opts.safety_identifier): extra["safety_identifier"] = self._opts.safety_identifier if is_given(self._opts.prompt_cache_key): extra["prompt_cache_key"] = self._opts.prompt_cache_key if is_given(self._opts.top_p): extra["top_p"] = self._opts.top_p if is_given(self._opts.verbosity): extra["verbosity"] = self._opts.verbosity if is_given(self._opts.prompt_cache_retention): extra["prompt_cache_retention"] = self._opts.prompt_cache_retention parallel_tool_calls = ( parallel_tool_calls if is_given(parallel_tool_calls) else self._opts.parallel_tool_calls ) if is_given(parallel_tool_calls): extra["parallel_tool_calls"] = parallel_tool_calls tool_choice = tool_choice if is_given(tool_choice) else self._opts.tool_choice if is_given(tool_choice): oai_tool_choice: ChatCompletionToolChoiceOptionParam if isinstance(tool_choice, dict): oai_tool_choice = { "type": "function", "function": {"name": tool_choice["function"]["name"]}, } extra["tool_choice"] = oai_tool_choice elif tool_choice in ("auto", "required", "none"): oai_tool_choice = tool_choice extra["tool_choice"] = oai_tool_choice if is_given(response_format): extra["response_format"] = llm_utils.to_openai_response_format(response_format) # type: ignore return LLMStream( self, model=self._opts.model, provider_fmt=self._provider_fmt, strict_tool_schema=self._strict_tool_schema, client=self._client, chat_ctx=chat_ctx, tools=tools or [], conn_options=conn_options, extra_kwargs=extra, )Helper class that provides a standard way to create an ABC using inheritance.
Create a new instance of OpenAI LLM.
api_keymust be set to your OpenAI API key, either using the argument or by setting theOPENAI_API_KEYenvironmental variable.Ancestors
- livekit.agents.llm.llm.LLM
- abc.ABC
- EventEmitter
- typing.Generic
Subclasses
- livekit.plugins.baseten.llm.LLM
- livekit.plugins.cerebras.llm.LLM
- livekit.plugins.groq.services.LLM
- livekit.plugins.perplexity.llm.LLM
- livekit.plugins.sarvam.llm.client.LLM
Static methods
def with_azure(*,
model: str | ChatModels = 'gpt-4o',
azure_endpoint: str | None = None,
azure_deployment: str | None = None,
api_version: str | None = None,
api_key: str | None = None,
azure_ad_token: str | None = None,
azure_ad_token_provider: AsyncAzureADTokenProvider | None = None,
organization: str | None = None,
project: str | None = None,
base_url: str | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
timeout: httpx.Timeout | None = None,
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN,
verbosity: NotGivenOr[Verbosity] = NOT_GIVEN,
max_completion_tokens: NotGivenOr[int] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_azure( *, model: str | ChatModels = "gpt-4o", azure_endpoint: str | None = None, azure_deployment: str | None = None, api_version: str | None = None, api_key: str | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: AsyncAzureADTokenProvider | None = None, organization: str | None = None, project: str | None = None, base_url: str | None = None, user: NotGivenOr[str] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN, timeout: httpx.Timeout | None = None, reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, verbosity: NotGivenOr[Verbosity] = NOT_GIVEN, max_completion_tokens: NotGivenOr[int] = NOT_GIVEN, ) -> LLM: """ This automatically infers the following arguments from their corresponding environment variables if they are not provided: - `api_key` from `AZURE_OPENAI_API_KEY` - `organization` from `OPENAI_ORG_ID` - `project` from `OPENAI_PROJECT_ID` - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN` - `api_version` from `OPENAI_API_VERSION` - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT` """ # noqa: E501 azure_client = openai.AsyncAzureOpenAI( max_retries=0, azure_endpoint=azure_endpoint, azure_deployment=azure_deployment, api_version=api_version, api_key=api_key, azure_ad_token=azure_ad_token, azure_ad_token_provider=azure_ad_token_provider, organization=organization, project=project, base_url=base_url, timeout=timeout if timeout else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0), ) # type: ignore llm = LLM( model=model, client=azure_client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, verbosity=verbosity, max_completion_tokens=max_completion_tokens, ) llm._owns_client = True return llmThis automatically infers the following arguments from their corresponding environment variables if they are not provided: -
api_keyfromAZURE_OPENAI_API_KEY-organizationfromOPENAI_ORG_ID-projectfromOPENAI_PROJECT_ID-azure_ad_tokenfromAZURE_OPENAI_AD_TOKEN-api_versionfromOPENAI_API_VERSION-azure_endpointfromAZURE_OPENAI_ENDPOINT def with_cerebras(*,
model: str | CerebrasChatModels = 'gpt-oss-120b',
api_key: str | None = None,
base_url: str = 'https://api.cerebras.ai/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_cerebras( *, model: str | CerebrasChatModels = "gpt-oss-120b", api_key: str | None = None, base_url: str = "https://api.cerebras.ai/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN, reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of Cerebras LLM. ``api_key`` must be set to your Cerebras API key, either using the argument or by setting the ``CEREBRAS_API_KEY`` environment variable. """ api_key = api_key or os.environ.get("CEREBRAS_API_KEY") if api_key is None: raise ValueError( "Cerebras API key is required, either as argument or set CEREBRAS_API_KEY environment variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, _strict_tool_schema=False, )Create a new instance of Cerebras LLM.
api_keymust be set to your Cerebras API key, either using the argument or by setting theCEREBRAS_API_KEYenvironment variable. def with_cometapi(*,
model: str | CometAPIChatModels = 'gpt-5-chat-latest',
api_key: str | None = None,
base_url: str = 'https://api.cometapi.com/v1/',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_cometapi( *, model: str | CometAPIChatModels = "gpt-5-chat-latest", api_key: str | None = None, base_url: str = "https://api.cometapi.com/v1/", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of CometAPI LLM. ``api_key`` must be set to your CometAPI API key, either using the argument or by setting the ``COMETAPI_API_KEY`` environmental variable. CometAPI provides access to 500+ AI models from multiple providers including OpenAI, Anthropic, Google, xAI, DeepSeek, and Qwen through a unified API. Get your API key at: https://api.cometapi.com/console/token Learn more: https://www.cometapi.com/?utm_source=livekit&utm_campaign=integration&utm_medium=integration&utm_content=integration """ api_key = api_key or os.environ.get("COMETAPI_API_KEY") if api_key is None: raise ValueError( "CometAPI API key is required, either as argument or set COMETAPI_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, )Create a new instance of CometAPI LLM.
api_keymust be set to your CometAPI API key, either using the argument or by setting theCOMETAPI_API_KEYenvironmental variable.CometAPI provides access to 500+ AI models from multiple providers including OpenAI, Anthropic, Google, xAI, DeepSeek, and Qwen through a unified API.
Get your API key at: https://api.cometapi.com/console/token Learn more: https://www.cometapi.com/?utm_source=livekit&utm_campaign=integration&utm_medium=integration&utm_content=integration
def with_deepseek(*,
model: str | DeepSeekChatModels = 'deepseek-chat',
api_key: str | None = None,
base_url: str = 'https://api.deepseek.com/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_deepseek( *, model: str | DeepSeekChatModels = "deepseek-chat", api_key: str | None = None, base_url: str = "https://api.deepseek.com/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of DeepSeek LLM. ``api_key`` must be set to your DeepSeek API key, either using the argument or by setting the ``DEEPSEEK_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("DEEPSEEK_API_KEY") if api_key is None: raise ValueError( "DeepSeek API key is required, either as argument or set DEEPSEEK_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, )Create a new instance of DeepSeek LLM.
api_keymust be set to your DeepSeek API key, either using the argument or by setting theDEEPSEEK_API_KEYenvironmental variable. def with_fireworks(*,
model: str = 'accounts/fireworks/models/llama-v3p3-70b-instruct',
api_key: str | None = None,
base_url: str = 'https://api.fireworks.ai/inference/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_fireworks( *, model: str = "accounts/fireworks/models/llama-v3p3-70b-instruct", api_key: str | None = None, base_url: str = "https://api.fireworks.ai/inference/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of Fireworks LLM. ``api_key`` must be set to your Fireworks API key, either using the argument or by setting the ``FIREWORKS_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("FIREWORKS_API_KEY") if api_key is None: raise ValueError( "Fireworks API key is required, either as argument or set FIREWORKS_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, )Create a new instance of Fireworks LLM.
api_keymust be set to your Fireworks API key, either using the argument or by setting theFIREWORKS_API_KEYenvironmental variable. def with_letta(*,
agent_id: str,
base_url: str = 'https://api.letta.com/v1/chat/completions',
api_key: str | None = None) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_letta( *, agent_id: str, base_url: str = "https://api.letta.com/v1/chat/completions", api_key: str | None = None, ) -> LLM: """ Create a new Letta-backed LLM instance connected to the specified Letta agent. Args: agent_id (str): The Letta agent ID (must be prefixed with 'agent-'). base_url (str): The URL of the Letta server (e.g., http://localhost:8283/v1/chat/completions for local or https://api.letta.com/v1/chat/completions for cloud). api_key (str | None, optional): Optional API key for authentication, required if the Letta server enforces auth. Returns: LLM: A configured LLM instance for interacting with the given Letta agent. """ parsed = urlparse(base_url) if parsed.scheme not in {"http", "https"}: raise ValueError(f"Invalid URL scheme: '{parsed.scheme}'. Must be 'http' or 'https'.") if not parsed.netloc: raise ValueError(f"URL '{base_url}' is missing a network location (e.g., domain name).") api_key = api_key or os.environ.get("LETTA_API_KEY") if api_key is None: raise ValueError( "Letta API key is required, either as argument or set LETTA_API_KEY environmental variable" # noqa: E501 ) return LLM( model=agent_id, api_key=api_key, base_url=base_url, client=None, user=NOT_GIVEN, temperature=NOT_GIVEN, parallel_tool_calls=NOT_GIVEN, tool_choice=NOT_GIVEN, )Create a new Letta-backed LLM instance connected to the specified Letta agent.
Args
agent_id:str- The Letta agent ID (must be prefixed with 'agent-').
base_url:str- The URL of the Letta server (e.g., http://localhost:8283/v1/chat/completions for local or https://api.letta.com/v1/chat/completions for cloud).
api_key:str | None, optional- Optional API key for authentication, required if the Letta server enforces auth.
Returns
LLM- A configured LLM instance for interacting with the given Letta agent.
def with_nebius(*,
model: str | NebiusChatModels = 'meta-llama/Meta-Llama-3.1-70B-Instruct',
api_key: str | None = None,
base_url: str = 'https://api.studio.nebius.com/v1/',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_nebius( *, model: str | NebiusChatModels = "meta-llama/Meta-Llama-3.1-70B-Instruct", api_key: str | None = None, base_url: str = "https://api.studio.nebius.com/v1/", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of Nebius LLM. ``api_key`` must be set to your Nebius API key, either using the argument or by setting the ``NEBIUS_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("NEBIUS_API_KEY") if api_key is None: raise ValueError( "Nebius API key is required, either as argument or set NEBIUS_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, )Create a new instance of Nebius LLM.
api_keymust be set to your Nebius API key, either using the argument or by setting theNEBIUS_API_KEYenvironmental variable. def with_octo(*,
model: str | OctoChatModels = 'llama-2-13b-chat',
api_key: str | None = None,
base_url: str = 'https://text.octoai.run/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_octo( *, model: str | OctoChatModels = "llama-2-13b-chat", api_key: str | None = None, base_url: str = "https://text.octoai.run/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of OctoAI LLM. ``api_key`` must be set to your OctoAI API key, either using the argument or by setting the ``OCTOAI_TOKEN`` environmental variable. """ api_key = api_key or os.environ.get("OCTOAI_TOKEN") if api_key is None: raise ValueError( "OctoAI API key is required, either as argument or set OCTOAI_TOKEN environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, )Create a new instance of OctoAI LLM.
api_keymust be set to your OctoAI API key, either using the argument or by setting theOCTOAI_TOKENenvironmental variable. def with_ollama(*,
model: str = 'llama3.1',
base_url: str = 'http://localhost:11434/v1',
client: openai.AsyncClient | None = None,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_ollama( *, model: str = "llama3.1", base_url: str = "http://localhost:11434/v1", client: openai.AsyncClient | None = None, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of Ollama LLM. """ return LLM( model=model, api_key="ollama", base_url=base_url, client=client, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, )Create a new instance of Ollama LLM.
def with_openrouter(*,
model: str = 'auto',
api_key: str | None = None,
base_url: str = 'https://openrouter.ai/api/v1',
client: openai.AsyncClient | None = None,
site_url: str | None = None,
app_name: str | None = None,
fallback_models: list[str] | None = None,
provider: OpenRouterProviderPreferences | None = None,
plugins: list[OpenRouterWebPlugin] | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN,
timeout: httpx.Timeout | None = None) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_openrouter( *, model: str = "auto", api_key: str | None = None, base_url: str = "https://openrouter.ai/api/v1", client: openai.AsyncClient | None = None, site_url: str | None = None, app_name: str | None = None, fallback_models: list[str] | None = None, provider: OpenRouterProviderPreferences | None = None, plugins: list[OpenRouterWebPlugin] | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, timeout: httpx.Timeout | None = None, ) -> LLM: """ Create a new instance of OpenRouter LLM. ``api_key`` must be set to your OpenRouter API key, either using the argument or by setting the ``OPENROUTER_API_KEY`` environment variable. """ api_key = api_key or os.environ.get("OPENROUTER_API_KEY") if api_key is None: raise ValueError( "OpenRouter API key is required, either as argument or set OPENROUTER_API_KEY environment variable" ) # Set up analytics headers for OpenRouter default_headers: dict[str, str] = {} if site_url: default_headers["HTTP-Referer"] = site_url if app_name: default_headers["X-Title"] = app_name # Build OpenRouter-specific request body or_body: dict[str, Any] = {} if provider: or_body["provider"] = provider if fallback_models: # Set fallback models for routing or_body["models"] = [model, *fallback_models] if plugins: or_body["plugins"] = [ {k: v for k, v in asdict(p).items() if v is not None} for p in plugins ] return LLM( model=model, api_key=api_key, client=client, base_url=base_url, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, extra_body=or_body, extra_headers=default_headers, timeout=timeout, )Create a new instance of OpenRouter LLM.
api_keymust be set to your OpenRouter API key, either using the argument or by setting theOPENROUTER_API_KEYenvironment variable. def with_ovhcloud(*,
model: str = 'gpt-oss-120b',
api_key: str | None = None,
base_url: str = 'https://oai.endpoints.kepler.ai.cloud.ovh.net/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_ovhcloud( *, model: str = "gpt-oss-120b", api_key: str | None = None, base_url: str = "https://oai.endpoints.kepler.ai.cloud.ovh.net/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of OVHcloud AI Endpoints LLM. ``api_key`` must be set to your OVHcloud AI Endpoints API key, either using the argument or by setting the ``OVHCLOUD_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("OVHCLOUD_API_KEY") if api_key is None: raise ValueError( "OVHcloud AI Endpoints API key is required, either as argument or set OVHCLOUD_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, )Create a new instance of OVHcloud AI Endpoints LLM.
api_keymust be set to your OVHcloud AI Endpoints API key, either using the argument or by setting theOVHCLOUD_API_KEYenvironmental variable. def with_perplexity(*,
model: str | PerplexityChatModels = 'llama-3.1-sonar-small-128k-chat',
api_key: str | None = None,
base_url: str = 'https://api.perplexity.ai',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_perplexity( *, model: str | PerplexityChatModels = "llama-3.1-sonar-small-128k-chat", api_key: str | None = None, base_url: str = "https://api.perplexity.ai", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of PerplexityAI LLM. ``api_key`` must be set to your TogetherAI API key, either using the argument or by setting the ``PERPLEXITY_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("PERPLEXITY_API_KEY") if api_key is None: raise ValueError( "Perplexity AI API key is required, either as argument or set PERPLEXITY_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, )Create a new instance of PerplexityAI LLM.
api_keymust be set to your TogetherAI API key, either using the argument or by setting thePERPLEXITY_API_KEYenvironmental variable. def with_sambanova(*,
model: str | SambaNovaChatModels = 'DeepSeek-R1-0528',
api_key: str | None = None,
base_url: str = 'https://api.sambanova.ai/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_sambanova( *, model: str | SambaNovaChatModels = "DeepSeek-R1-0528", api_key: str | None = None, base_url: str = "https://api.sambanova.ai/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of SambaNova LLM (OpenAI-compatible). ``api_key`` must be set to your SambaNova API key, either using the argument or by setting the ``SAMBANOVA_API_KEY`` environment variable. """ api_key = api_key or os.environ.get("SAMBANOVA_API_KEY") if api_key is None: raise ValueError( "SambaNova API key is required, either as argument or set SAMBANOVA_API_KEY environment variable" ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, _strict_tool_schema=False, )Create a new instance of SambaNova LLM (OpenAI-compatible).
api_keymust be set to your SambaNova API key, either using the argument or by setting theSAMBANOVA_API_KEYenvironment variable. def with_telnyx(*,
model: str | TelnyxChatModels = 'meta-llama/Meta-Llama-3.1-70B-Instruct',
api_key: str | None = None,
base_url: str = 'https://api.telnyx.com/v2/ai',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_telnyx( *, model: str | TelnyxChatModels = "meta-llama/Meta-Llama-3.1-70B-Instruct", api_key: str | None = None, base_url: str = "https://api.telnyx.com/v2/ai", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of Telnyx LLM. ``api_key`` must be set to your Telnyx API key, either using the argument or by setting the ``TELNYX_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("TELNYX_API_KEY") if api_key is None: raise ValueError( "Telnyx AI API key is required, either as argument or set TELNYX_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, )Create a new instance of Telnyx LLM.
api_keymust be set to your Telnyx API key, either using the argument or by setting theTELNYX_API_KEYenvironmental variable. def with_together(*,
model: str | TogetherChatModels = 'meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo',
api_key: str | None = None,
base_url: str = 'https://api.together.xyz/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_together( *, model: str | TogetherChatModels = "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo", api_key: str | None = None, base_url: str = "https://api.together.xyz/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of TogetherAI LLM. ``api_key`` must be set to your TogetherAI API key, either using the argument or by setting the ``TOGETHER_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("TOGETHER_API_KEY") if api_key is None: raise ValueError( "Together AI API key is required, either as argument or set TOGETHER_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, )Create a new instance of TogetherAI LLM.
api_keymust be set to your TogetherAI API key, either using the argument or by setting theTOGETHER_API_KEYenvironmental variable. def with_x_ai(*,
model: str | XAIChatModels = 'grok-3-fast',
api_key: str | None = None,
base_url: str = 'https://api.x.ai/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM-
Expand source code
@staticmethod def with_x_ai( *, model: str | XAIChatModels = "grok-3-fast", api_key: str | None = None, base_url: str = "https://api.x.ai/v1", client: openai.AsyncClient | None = None, user: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: ToolChoice = "auto", reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN, safety_identifier: NotGivenOr[str] = NOT_GIVEN, prompt_cache_key: NotGivenOr[str] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, ) -> LLM: """ Create a new instance of XAI LLM. ``api_key`` must be set to your XAI API key, either using the argument or by setting the ``XAI_API_KEY`` environmental variable. """ api_key = api_key or os.environ.get("XAI_API_KEY") if api_key is None: raise ValueError( "XAI API key is required, either as argument or set XAI_API_KEY environmental variable" # noqa: E501 ) return LLM( model=model, api_key=api_key, base_url=base_url, client=client, user=user, temperature=temperature, parallel_tool_calls=parallel_tool_calls, tool_choice=tool_choice, # TODO(long): add provider fmt for grok reasoning_effort=reasoning_effort, safety_identifier=safety_identifier, prompt_cache_key=prompt_cache_key, top_p=top_p, )Create a new instance of XAI LLM.
api_keymust be set to your XAI API key, either using the argument or by setting theXAI_API_KEYenvironmental variable.
Instance variables
prop model : str-
Expand source code
@property def model(self) -> str: return self._opts.modelGet the model name/identifier for this LLM instance.
Returns
The model name if available, "unknown" otherwise.
Note
Plugins should override this property to provide their model information.
prop provider : str-
Expand source code
@property def provider(self) -> str: return self._client._base_url.netloc.decode("utf-8")Get the provider name/identifier for this LLM instance.
Returns
The provider name if available, "unknown" otherwise.
Note
Plugins should override this property to provide their provider information.
Methods
async def aclose(self) ‑> None-
Expand source code
async def aclose(self) -> None: await super().aclose() if self._owns_client: await self._client.close() def chat(self,
*,
chat_ctx: ChatContext,
tools: list[llm.Tool] | None = None,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0),
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
response_format: NotGivenOr[completion_create_params.ResponseFormat | type[llm_utils.ResponseFormatT]] = NOT_GIVEN,
extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLMStream-
Expand source code
def chat( self, *, chat_ctx: ChatContext, tools: list[llm.Tool] | None = None, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN, response_format: NotGivenOr[ completion_create_params.ResponseFormat | type[llm_utils.ResponseFormatT] ] = NOT_GIVEN, extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN, ) -> LLMStream: extra = {} if is_given(extra_kwargs): extra.update(extra_kwargs) if is_given(self._opts.extra_body): extra["extra_body"] = self._opts.extra_body if is_given(self._opts.extra_headers): extra["extra_headers"] = self._opts.extra_headers if is_given(self._opts.extra_query): extra["extra_query"] = self._opts.extra_query if is_given(self._opts.metadata): extra["metadata"] = self._opts.metadata if is_given(self._opts.user): extra["user"] = self._opts.user if is_given(self._opts.max_completion_tokens): extra["max_completion_tokens"] = self._opts.max_completion_tokens if is_given(self._opts.temperature): extra["temperature"] = self._opts.temperature if is_given(self._opts.service_tier): extra["service_tier"] = self._opts.service_tier if is_given(self._opts.reasoning_effort): extra["reasoning_effort"] = self._opts.reasoning_effort if is_given(self._opts.safety_identifier): extra["safety_identifier"] = self._opts.safety_identifier if is_given(self._opts.prompt_cache_key): extra["prompt_cache_key"] = self._opts.prompt_cache_key if is_given(self._opts.top_p): extra["top_p"] = self._opts.top_p if is_given(self._opts.verbosity): extra["verbosity"] = self._opts.verbosity if is_given(self._opts.prompt_cache_retention): extra["prompt_cache_retention"] = self._opts.prompt_cache_retention parallel_tool_calls = ( parallel_tool_calls if is_given(parallel_tool_calls) else self._opts.parallel_tool_calls ) if is_given(parallel_tool_calls): extra["parallel_tool_calls"] = parallel_tool_calls tool_choice = tool_choice if is_given(tool_choice) else self._opts.tool_choice if is_given(tool_choice): oai_tool_choice: ChatCompletionToolChoiceOptionParam if isinstance(tool_choice, dict): oai_tool_choice = { "type": "function", "function": {"name": tool_choice["function"]["name"]}, } extra["tool_choice"] = oai_tool_choice elif tool_choice in ("auto", "required", "none"): oai_tool_choice = tool_choice extra["tool_choice"] = oai_tool_choice if is_given(response_format): extra["response_format"] = llm_utils.to_openai_response_format(response_format) # type: ignore return LLMStream( self, model=self._opts.model, provider_fmt=self._provider_fmt, strict_tool_schema=self._strict_tool_schema, client=self._client, chat_ctx=chat_ctx, tools=tools or [], conn_options=conn_options, extra_kwargs=extra, )
Inherited members
class LLMStream (llm: LLM,
*,
model: str | ChatModels,
provider_fmt: str,
strict_tool_schema: bool,
client: openai.AsyncClient,
chat_ctx: llm.ChatContext,
tools: list[llm.Tool],
conn_options: APIConnectOptions,
extra_kwargs: dict[str, Any])-
Expand source code
class LLMStream(_LLMStream): def __init__( self, llm: LLM, *, model: str | ChatModels, provider_fmt: str, strict_tool_schema: bool, client: openai.AsyncClient, chat_ctx: llm.ChatContext, tools: list[llm.Tool], conn_options: APIConnectOptions, extra_kwargs: dict[str, Any], ) -> None: super().__init__( llm, model=model, provider_fmt=provider_fmt, strict_tool_schema=strict_tool_schema, client=client, chat_ctx=chat_ctx, tools=tools, conn_options=conn_options, extra_kwargs=extra_kwargs, )Helper class that provides a standard way to create an ABC using inheritance.
Ancestors
- LLMStream
- livekit.agents.llm.llm.LLMStream
- abc.ABC
class OpenRouterProviderPreferences (*args, **kwargs)-
Expand source code
class OpenRouterProviderPreferences(TypedDict, total=False): """OpenRouter provider routing preferences.""" order: list[str] allow_fallbacks: bool require_parameters: bool data_collection: Literal["allow", "deny"] only: list[str] ignore: list[str] quantizations: list[str] sort: Literal["price", "throughput", "latency"] max_price: dict[str, float]OpenRouter provider routing preferences.
Ancestors
- builtins.dict
Class variables
var allow_fallbacks : boolvar data_collection : Literal['allow', 'deny']var ignore : list[str]var max_price : dict[str, float]var only : list[str]var order : list[str]var quantizations : list[str]var require_parameters : boolvar sort : Literal['price', 'throughput', 'latency']
class OpenRouterWebPlugin (max_results: int = 5, search_prompt: str | None = None, id: str = 'web')-
Expand source code
@dataclass class OpenRouterWebPlugin: """OpenRouter web search plugin configuration""" max_results: int = 5 search_prompt: str | None = None id: str = "web"OpenRouter web search plugin configuration
Instance variables
var id : strvar max_results : intvar search_prompt : str | None
class STT (*,
language: str | list[str] = 'en',
detect_language: bool = False,
model: STTModels | str = 'gpt-4o-mini-transcribe',
prompt: NotGivenOr[str] = NOT_GIVEN,
keywords: NotGivenOr[list[str]] = NOT_GIVEN,
turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN,
noise_reduction_type: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
api_key: NotGivenOr[str] = NOT_GIVEN,
client: openai.AsyncClient | None = None,
use_realtime: NotGivenOr[bool] = NOT_GIVEN,
vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN)-
Expand source code
class STT(stt.STT): def __init__( self, *, language: str | list[str] = "en", detect_language: bool = False, model: STTModels | str = "gpt-4o-mini-transcribe", prompt: NotGivenOr[str] = NOT_GIVEN, keywords: NotGivenOr[list[str]] = NOT_GIVEN, turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN, noise_reduction_type: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, base_url: NotGivenOr[str] = NOT_GIVEN, api_key: NotGivenOr[str] = NOT_GIVEN, client: openai.AsyncClient | None = None, use_realtime: NotGivenOr[bool] = NOT_GIVEN, vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN, ): """ Create a new instance of OpenAI STT. Args: language: The language code to use for transcription (e.g., "en" for English). gpt-transcribe and gpt-live-transcribe accept a list for code-switched audio. detect_language: Whether to automatically detect the language. model: The OpenAI model to use for transcription. prompt: Optional free-form description of the audio, such as its topic or setting. keywords: Literal terms to expect, such as product names or acronyms. Only for gpt-transcribe and gpt-live-transcribe, and only a hint. turn_detection: When using realtime transcription, this controls how model detects the user is done speaking. Final transcripts are generated only after the turn is over. See: https://platform.openai.com/docs/guides/realtime-vad Ignored for `gpt-realtime-whisper` and `gpt-live-transcribe`, which do not support server-side turn detection. noise_reduction_type: Type of noise reduction to apply. "near_field" or "far_field" This isn't needed when using LiveKit's noise cancellation. temperature: Sampling temperature between 0 and 1. Lower values make the transcription more deterministic. Not supported for realtime transcription. base_url: Custom base URL for OpenAI API. api_key: Your OpenAI API key. If not provided, will use the OPENAI_API_KEY environment variable. client: Optional pre-configured OpenAI AsyncClient instance. use_realtime: Whether to use the realtime transcription API. Defaults to True for `gpt-realtime-whisper` and `gpt-live-transcribe`, which are served only there, and to False otherwise. vad: Optional Voice Activity Detector used to commit the audio buffer when the model does not support server-side turn detection (`gpt-realtime-whisper`, `gpt-live-transcribe`). When not provided and the model requires it, the bundled Silero VAD is used with default settings. Pass `vad=None` to opt out and drive `input_audio_buffer.commit` yourself. """ # noqa: E501 if not is_given(use_realtime): use_realtime = _is_realtime_only(model) if use_realtime and is_given(temperature): logger.warning( "temperature is not supported for realtime transcription; " "ignoring the provided value" ) temperature = NOT_GIVEN if use_realtime and _is_realtime_only(model): if is_given(turn_detection): logger.warning( "turn_detection is not supported for %s; ignoring the provided value", model ) turn_detection = NOT_GIVEN if not is_given(vad): vad = inference.VAD(model="silero") super().__init__( capabilities=stt.STTCapabilities( streaming=use_realtime, interim_results=use_realtime, aligned_transcript=False, keyterms=_supports_context_hints(model), ) ) # the last language asked for, kept while detection is on so it can be restored self._specified_languages = _as_languages(language) languages = [] if detect_language else self._specified_languages resolved_keywords = list(keywords) if is_given(keywords) else [] _validate_context(model, languages, resolved_keywords) if not is_given(turn_detection): turn_detection = { "type": "server_vad", "threshold": 0.5, "prefix_padding_ms": 600, "silence_duration_ms": 350, } self._opts = _STTOptions( languages=languages, detect_language=detect_language, model=model, prompt=prompt, keywords=resolved_keywords, turn_detection=_as_turn_detection(turn_detection), temperature=temperature, ) if is_given(noise_reduction_type): self._opts.noise_reduction_type = noise_reduction_type # user keywords; _opts.keywords holds the effective set (user + session) self._user_keywords: list[str] = list(self._opts.keywords) self._session_keyterms: list[str] = [] self._vad = vad if is_given(vad) else None # an explicit `vad=None` means the caller commits the audio buffer itself self._vad_opted_out = vad is None if is_given(api_key) and not api_key: raise ValueError( "OpenAI API key is required, either as argument or set" " OPENAI_API_KEY environment variable" ) self._client = client or openai.AsyncClient( max_retries=0, api_key=api_key if is_given(api_key) else None, base_url=base_url if is_given(base_url) else None, http_client=httpx.AsyncClient( timeout=httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0), follow_redirects=True, limits=httpx.Limits( max_connections=50, max_keepalive_connections=50, keepalive_expiry=120, ), ), ) self._streams = weakref.WeakSet[SpeechStream]() self._session: aiohttp.ClientSession | None = None self._pool = utils.ConnectionPool[aiohttp.ClientWebSocketResponse]( max_session_duration=_max_session_duration, connect_cb=self._connect_ws, close_cb=self._close_ws, ) @property def model(self) -> str: return self._opts.model @property def provider(self) -> str: return self._client._base_url.netloc.decode("utf-8") @staticmethod def with_azure( *, language: str | list[str] = "en", detect_language: bool = False, model: STTModels | str = "gpt-4o-mini-transcribe", prompt: NotGivenOr[str] = NOT_GIVEN, keywords: NotGivenOr[list[str]] = NOT_GIVEN, turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN, noise_reduction_type: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, azure_endpoint: str | None = None, azure_deployment: str | None = None, api_version: str | None = None, api_key: str | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: AsyncAzureADTokenProvider | None = None, organization: str | None = None, project: str | None = None, base_url: str | None = None, use_realtime: NotGivenOr[bool] = NOT_GIVEN, timeout: httpx.Timeout | None = None, vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN, ) -> STT: """ Create a new instance of Azure OpenAI STT. This automatically infers the following arguments from their corresponding environment variables if they are not provided: - `api_key` from `AZURE_OPENAI_API_KEY` - `organization` from `OPENAI_ORG_ID` - `project` from `OPENAI_PROJECT_ID` - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN` - `api_version` from `OPENAI_API_VERSION` - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT` """ # noqa: E501 azure_client = openai.AsyncAzureOpenAI( max_retries=0, azure_endpoint=azure_endpoint, azure_deployment=azure_deployment, api_version=api_version, api_key=api_key, azure_ad_token=azure_ad_token, azure_ad_token_provider=azure_ad_token_provider, organization=organization, project=project, base_url=base_url, timeout=timeout if timeout else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0), ) # type: ignore return STT( language=language, detect_language=detect_language, model=model, prompt=prompt, keywords=keywords, turn_detection=turn_detection, noise_reduction_type=noise_reduction_type, temperature=temperature, client=azure_client, use_realtime=use_realtime, vad=vad, ) @staticmethod def with_ovhcloud( *, model: str = "whisper-large-v3-turbo", api_key: NotGivenOr[str] = NOT_GIVEN, base_url: str = "https://oai.endpoints.kepler.ai.cloud.ovh.net/v1", client: openai.AsyncClient | None = None, language: str | list[str] = "en", detect_language: bool = False, prompt: NotGivenOr[str] = NOT_GIVEN, ) -> STT: """ Create a new instance of OVHcloud AI Endpoints STT. ``api_key`` must be set to your OVHcloud AI Endpoints API key, either using the argument or by setting the ``OVHCLOUD_API_KEY`` environmental variable. """ ovhcloud_api_key = api_key if is_given(api_key) else os.environ.get("OVHCLOUD_API_KEY") if not ovhcloud_api_key: raise ValueError("OVHcloud AI Endpoints API key is required") return STT( model=model, api_key=ovhcloud_api_key, base_url=base_url, client=client, language=language, detect_language=detect_language, prompt=prompt, use_realtime=False, ) def stream( self, *, language: NotGivenOr[str | list[str]] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> SpeechStream: opts = dataclasses.replace(self._opts) if is_given(language): opts.languages = _as_languages(language) _validate_context(opts.model, opts.languages, opts.keywords) stream = SpeechStream( stt=self, pool=self._pool, conn_options=conn_options, opts=opts, vad_instance=self._vad, ) self._streams.add(stream) return stream def update_options( self, *, model: NotGivenOr[STTModels | str] = NOT_GIVEN, language: NotGivenOr[str | list[str]] = NOT_GIVEN, detect_language: NotGivenOr[bool] = NOT_GIVEN, prompt: NotGivenOr[str] = NOT_GIVEN, keywords: NotGivenOr[list[str]] = NOT_GIVEN, turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN, noise_reduction_type: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, ) -> None: """ Update the options for the speech stream. Open streams apply the change in place; only a new `model` reconnects them. Args: language: The language to transcribe in, or a list for gpt-transcribe and gpt-live-transcribe. An empty list detects the language. detect_language: Whether to detect the language. Turning it off falls back to the last language asked for. model: The model to use for transcription. prompt: Optional free-form description of the audio. keywords: Literal terms to expect. Only for gpt-transcribe and gpt-live-transcribe. turn_detection: When using realtime, this controls how model detects the user is done speaking. noise_reduction_type: Type of noise reduction to apply. "near_field" or "far_field" temperature: Sampling temperature between 0 and 1. Not supported for realtime transcription. """ # noqa: E501 # resolve first: an unsupported combination must raise before anything is applied resolved_model = model if is_given(model) else self._opts.model if is_given(language): languages = _as_languages(language) elif detect_language: languages = [] elif detect_language is False and not self._opts.languages: logger.warning( "detect_language=False names no language, falling back to %s; " "pass `language` to transcribe in another", self._specified_languages, ) languages = self._specified_languages else: languages = self._opts.languages user_keywords = list(keywords) if is_given(keywords) else self._user_keywords _validate_context(resolved_model, languages, user_keywords) if not is_given(language): # a stream keeps its own language, which the new model may not take for stream in self._streams: _validate_context(resolved_model, stream._opts.languages, user_keywords) # the transport is fixed at construction: AgentSession wraps a non-streaming STT once if is_given(model) and _is_realtime_only(model): if not self.capabilities.streaming: raise ValueError( f"{resolved_model} is served only over the realtime API, and this STT was " "created for the transcriptions endpoint; pass `use_realtime=True` to the " "constructor to reach it" ) if self._vad is None and not self._vad_opted_out: raise ValueError( f"{resolved_model} has no server-side endpointing, so it needs a `vad` to " "commit the audio buffer; pass one to the constructor, or pass `vad=None` " "to drive `input_audio_buffer.commit` yourself" ) languages_changed = languages != self._opts.languages model_changed = resolved_model != self._opts.model # a stream keeps a language of its own unless this call names or moves the language languages_given = is_given(language) or languages_changed self._opts.model = resolved_model self._capabilities.keyterms = _supports_context_hints(resolved_model) self._opts.languages = languages if languages: self._specified_languages = languages self._user_keywords = user_keywords # detected keyterms must not survive a switch to a model that rejects keywords self._opts.keywords = ( list(dict.fromkeys([*user_keywords, *self._session_keyterms])) if self.capabilities.keyterms else [] ) if is_given(detect_language): self._opts.detect_language = detect_language if is_given(prompt): self._opts.prompt = prompt if is_given(turn_detection): self._opts.turn_detection = _as_turn_detection(turn_detection) if is_given(noise_reduction_type): self._opts.noise_reduction_type = noise_reduction_type if is_given(temperature): if self.capabilities.streaming: logger.warning( "temperature is not supported for realtime transcription; " "ignoring the provided value" ) else: self._opts.temperature = temperature for stream in self._streams: stream.update_options(language=languages if languages_given else NOT_GIVEN) if model_changed: # every stream has dropped its own socket by now, so this reaches the idle ones the # pool holds between two speech sessions self._pool.invalidate() def _update_session_keyterms(self, keyterms: list[str]) -> None: if not self.capabilities.keyterms: super()._update_session_keyterms(keyterms) return if keyterms == self._session_keyterms: return self._session_keyterms = list(keyterms) self._opts.keywords = list(dict.fromkeys([*self._user_keywords, *keyterms])) for stream in self._streams: stream.update_options() async def _connect_ws(self, timeout: float) -> aiohttp.ClientWebSocketResponse: query_params: dict[str, str] = { "intent": "transcription", } # OpenAI's native realtime endpoint treats ?model= as selecting a # conversation session and rejects the transcription-mode # session.update with invalid_model — the model is conveyed via # audio.input.transcription.model instead. Gateways need the model # on the upgrade URL to route the connection before the first frame. if urlparse(str(self._client.base_url)).hostname != "api.openai.com": query_params["model"] = self._opts.model headers = { "User-Agent": "LiveKit Agents", "Authorization": f"Bearer {self._client.api_key}", } url = f"{str(self._client.base_url).rstrip('/')}/realtime?{urlencode(query_params)}" if url.startswith("http"): url = url.replace("http", "ws", 1) session = self._ensure_session() # the config is sent once the stream acquires the connection, since the pool also # hands back sockets it opened earlier return await asyncio.wait_for(session.ws_connect(url, headers=headers), timeout) async def _close_ws(self, ws: aiohttp.ClientWebSocketResponse) -> None: await ws.close() def _ensure_session(self) -> aiohttp.ClientSession: if not self._session: self._session = utils.http_context.http_session() return self._session async def _recognize_impl( self, buffer: AudioBuffer, *, language: NotGivenOr[str | list[str]] = NOT_GIVEN, conn_options: APIConnectOptions, ) -> stt.SpeechEvent: if is_given(language): languages = _as_languages(language) _validate_context(self._opts.model, languages, self._opts.keywords) self._opts.languages = languages try: data = rtc.combine_audio_frames(buffer).to_wav_bytes() format = "json" if self._opts.model == "whisper-1": # verbose_json returns language and other details, only supported for whisper-1 format = "verbose_json" transcription = _transcription(self._opts) resp = await self._client.audio.transcriptions.create( file=( "file.wav", data, "audio/wav", ), model=self._opts.model, # type: ignore language=transcription.language or openai.omit, languages=transcription.languages or openai.omit, keywords=transcription.keywords or openai.omit, prompt=transcription.prompt or openai.omit, response_format=format, temperature=self._opts.temperature if is_given(self._opts.temperature) else openai.omit, timeout=httpx.Timeout(30, connect=conn_options.timeout), ) # the detected language beats the hint, dominant code first; an empty list keeps it sd = stt.SpeechData(text=resp.text, language=_transcript_language(self._opts.languages)) if isinstance(resp, TranscriptionVerbose) and resp.language: sd.language = LanguageCode(resp.language) elif isinstance(resp, Transcription) and resp.languages: sd.language = LanguageCode(resp.languages[0].code) return stt.SpeechEvent( type=stt.SpeechEventType.FINAL_TRANSCRIPT, alternatives=[sd], ) except openai.APITimeoutError: raise APITimeoutError() from None except openai.APIStatusError as e: raise APIStatusError( e.message, status_code=e.status_code, request_id=e.request_id, body=e.body ) from None except Exception as e: raise APIConnectionError() from eHelper class that provides a standard way to create an ABC using inheritance.
Create a new instance of OpenAI STT.
Args
language- The language code to use for transcription (e.g., "en" for English). gpt-transcribe and gpt-live-transcribe accept a list for code-switched audio.
detect_language- Whether to automatically detect the language.
model- The OpenAI model to use for transcription.
prompt- Optional free-form description of the audio, such as its topic or setting.
keywords- Literal terms to expect, such as product names or acronyms. Only for gpt-transcribe and gpt-live-transcribe, and only a hint.
turn_detection- When using realtime transcription, this controls how model detects the user is done speaking.
Final transcripts are generated only after the turn is over. See: https://platform.openai.com/docs/guides/realtime-vad
Ignored for
gpt-realtime-whisperandgpt-live-transcribe, which do not support server-side turn detection. noise_reduction_type- Type of noise reduction to apply. "near_field" or "far_field" This isn't needed when using LiveKit's noise cancellation.
temperature- Sampling temperature between 0 and 1. Lower values make the transcription more deterministic. Not supported for realtime transcription.
base_url- Custom base URL for OpenAI API.
api_key- Your OpenAI API key. If not provided, will use the OPENAI_API_KEY environment variable.
client- Optional pre-configured OpenAI AsyncClient instance.
use_realtime- Whether to use the realtime transcription API. Defaults to True for
gpt-realtime-whisperandgpt-live-transcribe, which are served only there, and to False otherwise. vad- Optional Voice Activity Detector used to commit the audio buffer when the model
does not support server-side turn detection (
gpt-realtime-whisper,gpt-live-transcribe). When not provided and the model requires it, the bundled Silero VAD is used with default settings. Passvad=Noneto opt out and driveinput_audio_buffer.commityourself.
Ancestors
- livekit.agents.stt.stt.STT
- abc.ABC
- EventEmitter
- typing.Generic
Subclasses
- livekit.plugins.groq.services.STT
Static methods
def with_azure(*,
language: str | list[str] = 'en',
detect_language: bool = False,
model: STTModels | str = 'gpt-4o-mini-transcribe',
prompt: NotGivenOr[str] = NOT_GIVEN,
keywords: NotGivenOr[list[str]] = NOT_GIVEN,
turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN,
noise_reduction_type: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
azure_endpoint: str | None = None,
azure_deployment: str | None = None,
api_version: str | None = None,
api_key: str | None = None,
azure_ad_token: str | None = None,
azure_ad_token_provider: AsyncAzureADTokenProvider | None = None,
organization: str | None = None,
project: str | None = None,
base_url: str | None = None,
use_realtime: NotGivenOr[bool] = NOT_GIVEN,
timeout: httpx.Timeout | None = None,
vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN) ‑> livekit.plugins.openai.stt.STT-
Expand source code
@staticmethod def with_azure( *, language: str | list[str] = "en", detect_language: bool = False, model: STTModels | str = "gpt-4o-mini-transcribe", prompt: NotGivenOr[str] = NOT_GIVEN, keywords: NotGivenOr[list[str]] = NOT_GIVEN, turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN, noise_reduction_type: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, azure_endpoint: str | None = None, azure_deployment: str | None = None, api_version: str | None = None, api_key: str | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: AsyncAzureADTokenProvider | None = None, organization: str | None = None, project: str | None = None, base_url: str | None = None, use_realtime: NotGivenOr[bool] = NOT_GIVEN, timeout: httpx.Timeout | None = None, vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN, ) -> STT: """ Create a new instance of Azure OpenAI STT. This automatically infers the following arguments from their corresponding environment variables if they are not provided: - `api_key` from `AZURE_OPENAI_API_KEY` - `organization` from `OPENAI_ORG_ID` - `project` from `OPENAI_PROJECT_ID` - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN` - `api_version` from `OPENAI_API_VERSION` - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT` """ # noqa: E501 azure_client = openai.AsyncAzureOpenAI( max_retries=0, azure_endpoint=azure_endpoint, azure_deployment=azure_deployment, api_version=api_version, api_key=api_key, azure_ad_token=azure_ad_token, azure_ad_token_provider=azure_ad_token_provider, organization=organization, project=project, base_url=base_url, timeout=timeout if timeout else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0), ) # type: ignore return STT( language=language, detect_language=detect_language, model=model, prompt=prompt, keywords=keywords, turn_detection=turn_detection, noise_reduction_type=noise_reduction_type, temperature=temperature, client=azure_client, use_realtime=use_realtime, vad=vad, )Create a new instance of Azure OpenAI STT.
This automatically infers the following arguments from their corresponding environment variables if they are not provided: -
api_keyfromAZURE_OPENAI_API_KEY-organizationfromOPENAI_ORG_ID-projectfromOPENAI_PROJECT_ID-azure_ad_tokenfromAZURE_OPENAI_AD_TOKEN-api_versionfromOPENAI_API_VERSION-azure_endpointfromAZURE_OPENAI_ENDPOINT def with_ovhcloud(*,
model: str = 'whisper-large-v3-turbo',
api_key: NotGivenOr[str] = NOT_GIVEN,
base_url: str = 'https://oai.endpoints.kepler.ai.cloud.ovh.net/v1',
client: openai.AsyncClient | None = None,
language: str | list[str] = 'en',
detect_language: bool = False,
prompt: NotGivenOr[str] = NOT_GIVEN) ‑> livekit.plugins.openai.stt.STT-
Expand source code
@staticmethod def with_ovhcloud( *, model: str = "whisper-large-v3-turbo", api_key: NotGivenOr[str] = NOT_GIVEN, base_url: str = "https://oai.endpoints.kepler.ai.cloud.ovh.net/v1", client: openai.AsyncClient | None = None, language: str | list[str] = "en", detect_language: bool = False, prompt: NotGivenOr[str] = NOT_GIVEN, ) -> STT: """ Create a new instance of OVHcloud AI Endpoints STT. ``api_key`` must be set to your OVHcloud AI Endpoints API key, either using the argument or by setting the ``OVHCLOUD_API_KEY`` environmental variable. """ ovhcloud_api_key = api_key if is_given(api_key) else os.environ.get("OVHCLOUD_API_KEY") if not ovhcloud_api_key: raise ValueError("OVHcloud AI Endpoints API key is required") return STT( model=model, api_key=ovhcloud_api_key, base_url=base_url, client=client, language=language, detect_language=detect_language, prompt=prompt, use_realtime=False, )Create a new instance of OVHcloud AI Endpoints STT.
api_keymust be set to your OVHcloud AI Endpoints API key, either using the argument or by setting theOVHCLOUD_API_KEYenvironmental variable.
Instance variables
prop model : str-
Expand source code
@property def model(self) -> str: return self._opts.modelGet the model name/identifier for this STT instance.
Returns
The model name if available, "unknown" otherwise.
Note
Plugins should override this property to provide their model information.
prop provider : str-
Expand source code
@property def provider(self) -> str: return self._client._base_url.netloc.decode("utf-8")Get the provider name/identifier for this STT instance.
Returns
The provider name if available, "unknown" otherwise.
Note
Plugins should override this property to provide their provider information.
Methods
def stream(self,
*,
language: NotGivenOr[str | list[str]] = NOT_GIVEN,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.openai.stt.SpeechStream-
Expand source code
def stream( self, *, language: NotGivenOr[str | list[str]] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> SpeechStream: opts = dataclasses.replace(self._opts) if is_given(language): opts.languages = _as_languages(language) _validate_context(opts.model, opts.languages, opts.keywords) stream = SpeechStream( stt=self, pool=self._pool, conn_options=conn_options, opts=opts, vad_instance=self._vad, ) self._streams.add(stream) return stream def update_options(self,
*,
model: NotGivenOr[STTModels | str] = NOT_GIVEN,
language: NotGivenOr[str | list[str]] = NOT_GIVEN,
detect_language: NotGivenOr[bool] = NOT_GIVEN,
prompt: NotGivenOr[str] = NOT_GIVEN,
keywords: NotGivenOr[list[str]] = NOT_GIVEN,
turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN,
noise_reduction_type: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, model: NotGivenOr[STTModels | str] = NOT_GIVEN, language: NotGivenOr[str | list[str]] = NOT_GIVEN, detect_language: NotGivenOr[bool] = NOT_GIVEN, prompt: NotGivenOr[str] = NOT_GIVEN, keywords: NotGivenOr[list[str]] = NOT_GIVEN, turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN, noise_reduction_type: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, ) -> None: """ Update the options for the speech stream. Open streams apply the change in place; only a new `model` reconnects them. Args: language: The language to transcribe in, or a list for gpt-transcribe and gpt-live-transcribe. An empty list detects the language. detect_language: Whether to detect the language. Turning it off falls back to the last language asked for. model: The model to use for transcription. prompt: Optional free-form description of the audio. keywords: Literal terms to expect. Only for gpt-transcribe and gpt-live-transcribe. turn_detection: When using realtime, this controls how model detects the user is done speaking. noise_reduction_type: Type of noise reduction to apply. "near_field" or "far_field" temperature: Sampling temperature between 0 and 1. Not supported for realtime transcription. """ # noqa: E501 # resolve first: an unsupported combination must raise before anything is applied resolved_model = model if is_given(model) else self._opts.model if is_given(language): languages = _as_languages(language) elif detect_language: languages = [] elif detect_language is False and not self._opts.languages: logger.warning( "detect_language=False names no language, falling back to %s; " "pass `language` to transcribe in another", self._specified_languages, ) languages = self._specified_languages else: languages = self._opts.languages user_keywords = list(keywords) if is_given(keywords) else self._user_keywords _validate_context(resolved_model, languages, user_keywords) if not is_given(language): # a stream keeps its own language, which the new model may not take for stream in self._streams: _validate_context(resolved_model, stream._opts.languages, user_keywords) # the transport is fixed at construction: AgentSession wraps a non-streaming STT once if is_given(model) and _is_realtime_only(model): if not self.capabilities.streaming: raise ValueError( f"{resolved_model} is served only over the realtime API, and this STT was " "created for the transcriptions endpoint; pass `use_realtime=True` to the " "constructor to reach it" ) if self._vad is None and not self._vad_opted_out: raise ValueError( f"{resolved_model} has no server-side endpointing, so it needs a `vad` to " "commit the audio buffer; pass one to the constructor, or pass `vad=None` " "to drive `input_audio_buffer.commit` yourself" ) languages_changed = languages != self._opts.languages model_changed = resolved_model != self._opts.model # a stream keeps a language of its own unless this call names or moves the language languages_given = is_given(language) or languages_changed self._opts.model = resolved_model self._capabilities.keyterms = _supports_context_hints(resolved_model) self._opts.languages = languages if languages: self._specified_languages = languages self._user_keywords = user_keywords # detected keyterms must not survive a switch to a model that rejects keywords self._opts.keywords = ( list(dict.fromkeys([*user_keywords, *self._session_keyterms])) if self.capabilities.keyterms else [] ) if is_given(detect_language): self._opts.detect_language = detect_language if is_given(prompt): self._opts.prompt = prompt if is_given(turn_detection): self._opts.turn_detection = _as_turn_detection(turn_detection) if is_given(noise_reduction_type): self._opts.noise_reduction_type = noise_reduction_type if is_given(temperature): if self.capabilities.streaming: logger.warning( "temperature is not supported for realtime transcription; " "ignoring the provided value" ) else: self._opts.temperature = temperature for stream in self._streams: stream.update_options(language=languages if languages_given else NOT_GIVEN) if model_changed: # every stream has dropped its own socket by now, so this reaches the idle ones the # pool holds between two speech sessions self._pool.invalidate()Update the options for the speech stream. Open streams apply the change in place; only a new
modelreconnects them.Args
language- The language to transcribe in, or a list for gpt-transcribe and gpt-live-transcribe. An empty list detects the language.
detect_language- Whether to detect the language. Turning it off falls back to the last language asked for.
model- The model to use for transcription.
prompt- Optional free-form description of the audio.
keywords- Literal terms to expect. Only for gpt-transcribe and gpt-live-transcribe.
turn_detection- When using realtime, this controls how model detects the user is done speaking.
noise_reduction_type- Type of noise reduction to apply. "near_field" or "far_field"
temperature- Sampling temperature between 0 and 1. Not supported for realtime transcription.
Inherited members
class TTS (*,
model: TTSModels | str = 'gpt-4o-mini-tts',
voice: TTSVoices | str = 'ash',
speed: float = 1.0,
instructions: NotGivenOr[str] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
api_key: NotGivenOr[str] = NOT_GIVEN,
client: openai.AsyncClient | None = None,
response_format: NotGivenOr[RESPONSE_FORMATS] = NOT_GIVEN)-
Expand source code
class TTS(tts.TTS): def __init__( self, *, model: TTSModels | str = DEFAULT_MODEL, voice: TTSVoices | str = DEFAULT_VOICE, speed: float = 1.0, instructions: NotGivenOr[str] = NOT_GIVEN, base_url: NotGivenOr[str] = NOT_GIVEN, api_key: NotGivenOr[str] = NOT_GIVEN, client: openai.AsyncClient | None = None, response_format: NotGivenOr[RESPONSE_FORMATS] = NOT_GIVEN, ) -> None: """ Create a new instance of OpenAI TTS. ``api_key`` must be set to your OpenAI API key, either using the argument or by setting the ``OPENAI_API_KEY`` environmental variable. """ super().__init__( capabilities=tts.TTSCapabilities(streaming=False), sample_rate=SAMPLE_RATE, num_channels=NUM_CHANNELS, ) self._opts = _TTSOptions( model=model, voice=voice, speed=speed, instructions=instructions if is_given(instructions) else None, response_format=response_format if is_given(response_format) else "mp3", ) if is_given(api_key) and not api_key: raise ValueError( "OpenAI API key is required, either as argument or set" " OPENAI_API_KEY environment variable" ) self._owns_client = client is None self._client = client or openai.AsyncClient( max_retries=0, api_key=api_key if is_given(api_key) else None, base_url=base_url if is_given(base_url) else None, http_client=httpx.AsyncClient( timeout=httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0), follow_redirects=True, limits=httpx.Limits( max_connections=50, max_keepalive_connections=50, keepalive_expiry=120 ), ), ) self._prewarm_task: asyncio.Task | None = None @property def model(self) -> str: return self._opts.model @property def provider(self) -> str: return self._client._base_url.netloc.decode("utf-8") def update_options( self, *, model: NotGivenOr[TTSModels | str] = NOT_GIVEN, voice: NotGivenOr[TTSVoices | str] = NOT_GIVEN, speed: NotGivenOr[float] = NOT_GIVEN, instructions: NotGivenOr[str] = NOT_GIVEN, ) -> None: if is_given(model): self._opts.model = model if is_given(voice): self._opts.voice = voice if is_given(speed): self._opts.speed = speed if is_given(instructions): self._opts.instructions = instructions @staticmethod def with_azure( *, model: TTSModels | str = DEFAULT_MODEL, voice: TTSVoices | str = DEFAULT_VOICE, speed: float = 1.0, instructions: NotGivenOr[str] = NOT_GIVEN, azure_endpoint: str | None = None, azure_deployment: str | None = None, api_version: str | None = None, api_key: str | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: AsyncAzureADTokenProvider | None = None, organization: str | None = None, project: str | None = None, base_url: str | None = None, response_format: NotGivenOr[RESPONSE_FORMATS] = NOT_GIVEN, timeout: httpx.Timeout | None = None, ) -> TTS: """ Create a new instance of Azure OpenAI TTS. This automatically infers the following arguments from their corresponding environment variables if they are not provided: - `api_key` from `AZURE_OPENAI_API_KEY` - `organization` from `OPENAI_ORG_ID` - `project` from `OPENAI_PROJECT_ID` - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN` - `api_version` from `OPENAI_API_VERSION` - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT` """ azure_client = openai.AsyncAzureOpenAI( max_retries=0, azure_endpoint=azure_endpoint, azure_deployment=azure_deployment, api_version=api_version, api_key=api_key, azure_ad_token=azure_ad_token, azure_ad_token_provider=azure_ad_token_provider, organization=organization, project=project, base_url=base_url, timeout=timeout if timeout else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0), ) # type: ignore tts = TTS( model=model, voice=voice, speed=speed, instructions=instructions, client=azure_client, response_format=response_format, ) tts._owns_client = True return tts def synthesize( self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> tts.ChunkedStream: return ChunkedStream(tts=self, input_text=text, conn_options=conn_options) def prewarm(self) -> None: async def _prewarm() -> None: try: await self._client.get("/", cast_to=str) except Exception: pass self._prewarm_task = asyncio.create_task(_prewarm()) async def aclose(self) -> None: if self._prewarm_task: await aio.cancel_and_wait(self._prewarm_task) if self._owns_client: await self._client.close()Helper class that provides a standard way to create an ABC using inheritance.
Create a new instance of OpenAI TTS.
api_keymust be set to your OpenAI API key, either using the argument or by setting theOPENAI_API_KEYenvironmental variable.Ancestors
- livekit.agents.tts.tts.TTS
- abc.ABC
- EventEmitter
- typing.Generic
Static methods
def with_azure(*,
model: TTSModels | str = 'gpt-4o-mini-tts',
voice: TTSVoices | str = 'ash',
speed: float = 1.0,
instructions: NotGivenOr[str] = NOT_GIVEN,
azure_endpoint: str | None = None,
azure_deployment: str | None = None,
api_version: str | None = None,
api_key: str | None = None,
azure_ad_token: str | None = None,
azure_ad_token_provider: AsyncAzureADTokenProvider | None = None,
organization: str | None = None,
project: str | None = None,
base_url: str | None = None,
response_format: NotGivenOr[RESPONSE_FORMATS] = NOT_GIVEN,
timeout: httpx.Timeout | None = None) ‑> livekit.plugins.openai.tts.TTS-
Expand source code
@staticmethod def with_azure( *, model: TTSModels | str = DEFAULT_MODEL, voice: TTSVoices | str = DEFAULT_VOICE, speed: float = 1.0, instructions: NotGivenOr[str] = NOT_GIVEN, azure_endpoint: str | None = None, azure_deployment: str | None = None, api_version: str | None = None, api_key: str | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: AsyncAzureADTokenProvider | None = None, organization: str | None = None, project: str | None = None, base_url: str | None = None, response_format: NotGivenOr[RESPONSE_FORMATS] = NOT_GIVEN, timeout: httpx.Timeout | None = None, ) -> TTS: """ Create a new instance of Azure OpenAI TTS. This automatically infers the following arguments from their corresponding environment variables if they are not provided: - `api_key` from `AZURE_OPENAI_API_KEY` - `organization` from `OPENAI_ORG_ID` - `project` from `OPENAI_PROJECT_ID` - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN` - `api_version` from `OPENAI_API_VERSION` - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT` """ azure_client = openai.AsyncAzureOpenAI( max_retries=0, azure_endpoint=azure_endpoint, azure_deployment=azure_deployment, api_version=api_version, api_key=api_key, azure_ad_token=azure_ad_token, azure_ad_token_provider=azure_ad_token_provider, organization=organization, project=project, base_url=base_url, timeout=timeout if timeout else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0), ) # type: ignore tts = TTS( model=model, voice=voice, speed=speed, instructions=instructions, client=azure_client, response_format=response_format, ) tts._owns_client = True return ttsCreate a new instance of Azure OpenAI TTS.
This automatically infers the following arguments from their corresponding environment variables if they are not provided: -
api_keyfromAZURE_OPENAI_API_KEY-organizationfromOPENAI_ORG_ID-projectfromOPENAI_PROJECT_ID-azure_ad_tokenfromAZURE_OPENAI_AD_TOKEN-api_versionfromOPENAI_API_VERSION-azure_endpointfromAZURE_OPENAI_ENDPOINT
Instance variables
prop model : str-
Expand source code
@property def model(self) -> str: return self._opts.modelGet the model name/identifier for this TTS instance.
Returns
The model name if available, "unknown" otherwise.
Note
Plugins should override this property to provide their model information.
prop provider : str-
Expand source code
@property def provider(self) -> str: return self._client._base_url.netloc.decode("utf-8")Get the provider name/identifier for this TTS instance.
Returns
The provider name if available, "unknown" otherwise.
Note
Plugins should override this property to provide their provider information.
Methods
async def aclose(self) ‑> None-
Expand source code
async def aclose(self) -> None: if self._prewarm_task: await aio.cancel_and_wait(self._prewarm_task) if self._owns_client: await self._client.close() def prewarm(self) ‑> None-
Expand source code
def prewarm(self) -> None: async def _prewarm() -> None: try: await self._client.get("/", cast_to=str) except Exception: pass self._prewarm_task = asyncio.create_task(_prewarm())Pre-warm connection to the TTS service
def synthesize(self,
text: str,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.agents.tts.tts.ChunkedStream-
Expand source code
def synthesize( self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> tts.ChunkedStream: return ChunkedStream(tts=self, input_text=text, conn_options=conn_options) def update_options(self,
*,
model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
voice: NotGivenOr[TTSVoices | str] = NOT_GIVEN,
speed: NotGivenOr[float] = NOT_GIVEN,
instructions: NotGivenOr[str] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, model: NotGivenOr[TTSModels | str] = NOT_GIVEN, voice: NotGivenOr[TTSVoices | str] = NOT_GIVEN, speed: NotGivenOr[float] = NOT_GIVEN, instructions: NotGivenOr[str] = NOT_GIVEN, ) -> None: if is_given(model): self._opts.model = model if is_given(voice): self._opts.voice = voice if is_given(speed): self._opts.speed = speed if is_given(instructions): self._opts.instructions = instructions
Inherited members