Module livekit.plugins.rime.realtime.realtime_model
Prism speech-to-speech through the Rime SDK.
LiveKit history, generations and playback are adapted here. The SDK owns all wire protocol and admission rules. Adapted from sglang-omni's reference plugin at bc8a500249536f369d20bc094baa9169e5986758.
Classes
class PrismError (fault: r.RealtimeFault)-
Expand source code
class PrismError(llm.RealtimeError): """A Prism refusal with its original fault and retry classification.""" def __init__(self, fault: r.RealtimeFault): super().__init__(f"{fault.code}: {fault.message}", code=fault.code) self.fault = fault self.retryable = fault.retryable self.request_id = fault.request_id self.response_id = fault.response_id self.scope = fault.scopeA Prism refusal with its original fault and retry classification.
Ancestors
- livekit.agents.llm.realtime.RealtimeError
- builtins.Exception
- builtins.BaseException
class PrismModelError (**data: Any)-
Expand source code
class PrismModelError(llm.RealtimeModelError): """A LiveKit model error with Prism's separate retry classification.""" retryable: bool | None = None """Whether the unchanged request may succeed again; does not initiate a retry."""A LiveKit model error with Prism's separate retry classification.
Create a new model by parsing and validating input data from keyword arguments.
Raises [
ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.selfis explicitly positional-only to allowselfas a field name.Ancestors
- livekit.agents.llm.realtime.RealtimeModelError
- pydantic.main.BaseModel
Class variables
var model_configvar retryable : bool | None-
Whether the unchanged request may succeed again; does not initiate a retry.
class RealtimeModel (*,
playback: "Literal['auto', 'estimated', 'discard']" = 'auto',
context_mode: "Literal['strict', 'shared']" = 'shared',
endpoint: str | None = None,
client: Rime | None = None,
voice: str | None = None,
turn_detection: r.TurnDetection | None = None,
truncation: r.Truncation | None = None,
timeouts: r.RealtimeTimeouts | None = None,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0, liveness_timeout=6.0),
startup_timeout: float = 30.0,
request_id: str | None = None,
response_metadata: Mapping[str, str] | None = None,
tool_fillers: Mapping[str, r.PrismFiller] | None = None)-
Expand source code
class RealtimeModel(llm.RealtimeModel): """Prism for one AgentSession. Requires the Rime SDK realtime extra and Python 3.11. Omit endpoint to use public Prism. An injected Rime client stays caller-owned. Otherwise Rime reads RIME_API_KEY. update_options() queues live settings updates and returns immediately. Stored settings change after provider acknowledgment; refusals are reported through error events. Context updates preserve omitted history by default. Use context_mode="strict" for replacement semantics through Agent.update_chat_ctx(). A closed session may be replaced with a fresh conversation, such as an explicit fallback restart. Concurrent sessions are unsupported. """ def __init__( self, *, playback: Literal["auto", "estimated", "discard"] = "auto", context_mode: Literal["strict", "shared"] = "shared", endpoint: str | None = None, client: Rime | None = None, voice: str | None = None, turn_detection: r.TurnDetection | None = None, truncation: r.Truncation | None = None, timeouts: r.RealtimeTimeouts | None = None, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, startup_timeout: float = 30.0, request_id: str | None = None, response_metadata: Mapping[str, str] | None = None, tool_fillers: Mapping[str, r.PrismFiller] | None = None, ): super().__init__( capabilities=llm.RealtimeCapabilities( message_truncation=True, turn_detection=True, user_transcription=True, auto_tool_reply_generation=False, audio_output=True, manual_function_calls=False, mutable_chat_context=True, mutable_instructions=True, mutable_tools=True, ) ) if not math.isfinite(startup_timeout) or startup_timeout <= 0: raise ValueError("startup_timeout must be finite and greater than zero") if not math.isfinite(conn_options.timeout) or not math.isfinite( conn_options.retry_interval ): raise ValueError("Connection timeout and retry interval must be finite") if playback not in ("auto", "estimated", "discard"): raise ValueError("playback must be auto, estimated, or discard") if context_mode not in ("strict", "shared"): raise ValueError("context_mode must be strict or shared") self._playback_mode = playback self._context_mode = context_mode self._conn_options = conn_options self._startup_timeout = startup_timeout self._endpoint = endpoint self._voice = voice self._turn_detection = turn_detection self._truncation = truncation self._timeouts = timeouts or r.RealtimeTimeouts() self._request_id = request_id self._response_metadata = dict(response_metadata) if response_metadata is not None else None self._tool_fillers = dict(tool_fillers or {}) self._client = client if client is not None else Rime() self._owns_client = client is None self._session: RealtimeSession | None = None self._wrapped_output = False def _wrap_audio_output(self, output: AudioOutput) -> AudioOutput: """Install the returned sink through AgentSession.output.audio before playback.""" self._wrapped_output = True return PrismAudioOutput(output) async def _replace_chat_ctx(self, agent: Agent, chat_ctx: llm.ChatContext) -> None: """Apply an exact edit. Pause other local edits and handoffs during this call.""" await self._edit_context(agent, chat_ctx, "replace") async def _clear_chat_ctx(self, agent: Agent) -> None: await self._replace_chat_ctx(agent, llm.ChatContext.empty()) async def _reconcile_chat_ctx(self, agent: Agent) -> None: """Restore confirmed provider history locally, without provider writes.""" await self._edit_context(agent, None, "reconcile") async def _edit_context( self, agent: Agent, target: llm.ChatContext | None, operation: Literal["replace", "reconcile"], ) -> None: try: owner = agent.session except RuntimeError: raise HistoryUpdateError("The agent must be active on this Rime model") from None model = agent.llm if is_given(agent.llm) else owner.llm session = self._session if owner.current_agent is not agent or model is not self or session is None: raise HistoryUpdateError("The active agent must use this Rime model directly") await asyncio.shield(session._initialized) session._require() history = session._history assert history is not None if operation == "reconcile": target = history.projection(agent.chat_ctx) else: assert target is not None target = target.copy() # Preserve local handoff/configuration records by stable identity. # Keep live tool references; a deep copy can reach the running agent. ids = {item.id for item in target.items} target.items.extend( item.model_copy() for item in agent.chat_ctx.items if not isinstance(item, (llm.ChatMessage, llm.FunctionCall, llm.FunctionCallOutput)) and item.id not in ids ) token = _Edit(session, operation, target.copy(), asyncio.current_task(), history.revision) scope = _edit.set(token) failure: HistoryUpdateError | None = None try: await agent.update_chat_ctx(target, exclude_invalid_function_calls=False) if not token.consumed or self._session is not session: raise HistoryUpdateError("The live provider did not consume the history operation") except HistoryUpdateError as error: failure = error finally: token.active = False _edit.reset(scope) if failure is not None: if operation == "replace" and failure.recoverable and token.consumed: try: await self._reconcile_chat_ctx(agent) failure.needs_reconciliation = False except HistoryUpdateError: pass raise failure def update_options( self, *, voice: NotGivenOr[str] = NOT_GIVEN, response_metadata: NotGivenOr[Mapping[str, str]] = NOT_GIVEN, tool_fillers: NotGivenOr[Mapping[str, r.PrismFiller]] = NOT_GIVEN, truncation: NotGivenOr[r.Truncation] = NOT_GIVEN, ) -> None: """Queue settings on the live session, or configure the next initial session. Live updates return immediately. Refusals are reported through error events; stored settings change only after the provider acknowledges the update. """ if self._session is None: if is_given(voice): self._voice = voice if is_given(response_metadata): self._response_metadata = dict(response_metadata) if is_given(tool_fillers): self._tool_fillers = dict(tool_fillers) if is_given(truncation): self._truncation = cast(r.Truncation, truncation) return self._session.update_options( voice=voice, response_metadata=response_metadata, tool_fillers=tool_fillers, truncation=truncation, ) async def _get_history(self) -> tuple[r.HistoryItem, ...]: """Read completed server history. Running replies are not included.""" if self._session is None: raise llm.RealtimeError("Prism session is not open") return await self._session._get_history() async def _delete_instruction_item(self, item_id: str) -> None: """Remove a saved nudge instruction after its reply ends.""" if self._session is None: raise llm.RealtimeError("Prism session is not open") await self._session._delete_instruction_item(item_id) @property def model(self) -> str: return "prism" @property def provider(self) -> str: return "rime" def session(self, *, turn_detection_disabled: bool = False) -> RealtimeSession: if turn_detection_disabled: raise llm.RealtimeError("Prism requires server turn detection") if self._session is not None and ( self._session._shutdown_task is None or not self._session._shutdown_task.done() ): raise llm.RealtimeError( "Close the previous Prism session before opening a fresh conversation" ) self._session = RealtimeSession(self) return self._session async def aclose(self) -> None: if self._session is not None: await self._session.aclose() if self._owns_client: await self._client.close()Prism for one AgentSession. Requires the Rime SDK realtime extra and Python 3.11.
Omit endpoint to use public Prism. An injected Rime client stays caller-owned. Otherwise Rime reads RIME_API_KEY. update_options() queues live settings updates and returns immediately. Stored settings change after provider acknowledgment; refusals are reported through error events. Context updates preserve omitted history by default. Use context_mode="strict" for replacement semantics through Agent.update_chat_ctx(). A closed session may be replaced with a fresh conversation, such as an explicit fallback restart. Concurrent sessions are unsupported.
Ancestors
- livekit.agents.llm.realtime.RealtimeModel
Instance variables
prop model : str-
Expand source code
@property def model(self) -> str: return "prism" prop provider : str-
Expand source code
@property def provider(self) -> str: return "rime"
Methods
async def aclose(self) ‑> None-
Expand source code
async def aclose(self) -> None: if self._session is not None: await self._session.aclose() if self._owns_client: await self._client.close() def session(self, *, turn_detection_disabled: bool = False) ‑> RealtimeSession-
Expand source code
def session(self, *, turn_detection_disabled: bool = False) -> RealtimeSession: if turn_detection_disabled: raise llm.RealtimeError("Prism requires server turn detection") if self._session is not None and ( self._session._shutdown_task is None or not self._session._shutdown_task.done() ): raise llm.RealtimeError( "Close the previous Prism session before opening a fresh conversation" ) self._session = RealtimeSession(self) return self._sessionCreate a new session, optionally with server-side turn detection disabled.
turn_detection_disabledis honored only by plugins reportingcan_disable_turn_detection; the model itself is left unchanged and reusable. def update_options(self,
*,
voice: NotGivenOr[str] = NOT_GIVEN,
response_metadata: NotGivenOr[Mapping[str, str]] = NOT_GIVEN,
tool_fillers: NotGivenOr[Mapping[str, r.PrismFiller]] = NOT_GIVEN,
truncation: NotGivenOr[r.Truncation] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, voice: NotGivenOr[str] = NOT_GIVEN, response_metadata: NotGivenOr[Mapping[str, str]] = NOT_GIVEN, tool_fillers: NotGivenOr[Mapping[str, r.PrismFiller]] = NOT_GIVEN, truncation: NotGivenOr[r.Truncation] = NOT_GIVEN, ) -> None: """Queue settings on the live session, or configure the next initial session. Live updates return immediately. Refusals are reported through error events; stored settings change only after the provider acknowledges the update. """ if self._session is None: if is_given(voice): self._voice = voice if is_given(response_metadata): self._response_metadata = dict(response_metadata) if is_given(tool_fillers): self._tool_fillers = dict(tool_fillers) if is_given(truncation): self._truncation = cast(r.Truncation, truncation) return self._session.update_options( voice=voice, response_metadata=response_metadata, tool_fillers=tool_fillers, truncation=truncation, )Queue settings on the live session, or configure the next initial session.
Live updates return immediately. Refusals are reported through error events; stored settings change only after the provider acknowledges the update.
class RealtimeSession (model: RealtimeModel)-
Expand source code
class RealtimeSession(llm.RealtimeSession[Literal["rime_server_event_received"]]): """A Prism session that emits SDK SessionEvent objects before processing them.""" def __init__(self, model: RealtimeModel): super().__init__(model) self._model = model self._sdk: r.RealtimeSession | None = None self._connection: AbstractAsyncContextManager[r.RealtimeSession] | None = None self._opening_task: asyncio.Task[None] | None = None self._initialized: asyncio.Future[None] = asyncio.get_running_loop().create_future() self._initialized.add_done_callback(lambda f: None if f.cancelled() else f.exception()) self._closed = False self._failure: Exception | None = None self._shutdown_task: asyncio.Task[None] | None = None self._fatal_emitted = False self._tool_list: list[llm.Tool] = [] self._generations: OrderedDict[str, _Generation] = OrderedDict() self._generation_ready = asyncio.Condition() self._turn_started: OrderedDict[str, float] = OrderedDict() self._latest: r.ResponseRef | None = None self._active: r.ResponseRef | None = None self._parent: r.ResponseRef | None = None self._report: r.ToolCallRef | None = None self._audio: asyncio.Queue[r.AudioChunk] = asyncio.Queue(100) self._audio_format: tuple[int, int] | None = None self._audio_resampler: rtc.AudioResampler | None = None self._tasks: set[asyncio.Task[Any]] = set() self._settings_lock = asyncio.Lock() self._options_task: asyncio.Task[None] | None = None self._tool_fillers = dict(model._tool_fillers) self._tool_choice: Literal["auto", "none"] = "auto" self._suppress_tools = False self._suppressed_revisions: set[int] = set() self._tool_policy_floor = 0 self._transcript_deltas: dict[str, str] = {} self._history: PrismHistory | None = None self._playback: Playback | None = None self._instruction_lock = asyncio.Lock() self._instruction_items: OrderedDict[str, tuple[str, bool]] = OrderedDict() self._spawn(self._connect()) def _sdk_tools( self, tools: list[llm.Tool], fillers: Mapping[str, r.PrismFiller] ) -> list[r.ToolDefinition]: return [_tool(tool, fillers) for tool in tools] @property def chat_ctx(self) -> llm.ChatContext: return self._history.chat_ctx if self._history is not None else llm.ChatContext.empty() @property def tools(self) -> llm.ToolContext: return llm.ToolContext(self._tool_list) def _require(self) -> r.RealtimeSession: if self._failure: raise self._failure if self._sdk is None or self._closed: raise llm.RealtimeError("Prism session is not open") return self._sdk async def _connect(self) -> None: try: async with asyncio.timeout(self._model._startup_timeout): await self._open_connection() if self._failure is not None: # A startup failure can occur while the handshake is pending, # before _fatal has an SDK connection to close. self._shutdown_task = asyncio.create_task(self._shutdown()) return assert self._sdk is not None self._history = PrismHistory(self._sdk) self._playback = Playback( self._sdk, self._model._playback_mode, self._model._timeouts.request_s, self._report_playback, self._spawn, lambda: self._model._wrapped_output, ) self._spawn(self._read_events()) self._initialized.set_result(None) self._spawn(self._send_audio()) except asyncio.CancelledError: raise except TimeoutError: self._fatal(RimeTimeoutError("Prism startup timed out")) except Exception as error: self._fatal(error) async def _open_connection(self) -> None: """Open the SDK session with bounded retries for retryable startup failures. Try at most ``max_retry + 1`` times, with ``conn_options.timeout`` per attempt. Retry delays honor the server delay and must fit the startup budget; the caller enforces the overall startup timeout. Cancellation or an attempt timeout cancels the opening task without retrying. Shutdown waits for that task and exits any successfully entered context. """ options = self._model._conn_options deadline = asyncio.get_running_loop().time() + self._model._startup_timeout for attempt in range(options.max_retry + 1): if self._closed: raise asyncio.CancelledError if self._failure: raise self._failure self._connection = self._model._client.realtime.connect( endpoint=self._model._endpoint, voice=self._model._voice, turn_detection=self._model._turn_detection, truncation=self._model._truncation, timeouts=self._model._timeouts, interrupt_on_speech=True, request_id=self._model._request_id, response_metadata=self._model._response_metadata, ) self._opening_task = asyncio.create_task(self._enter_connection(self._connection)) try: async with asyncio.timeout(options.timeout): await asyncio.shield(self._opening_task) return except asyncio.CancelledError: self._opening_task.cancel() raise except TimeoutError: self._opening_task.cancel() raise RimeTimeoutError("Prism connection initialization timed out") from None except Exception as error: if ( self._failure or self._closed or attempt == options.max_retry or not _startup_retryable(error) ): raise delay = options._interval_for_retry(attempt) retry_after = getattr(error, "retry_after_s", None) if isinstance(retry_after, (int, float)) and math.isfinite(retry_after): delay = max(delay, retry_after) # A server delay that cannot fit the startup budget ends with its original refusal. if asyncio.get_running_loop().time() + delay >= deadline: raise self._emit_error(error, recoverable=True) await asyncio.sleep(delay) async def _enter_connection( self, connection: AbstractAsyncContextManager[r.RealtimeSession] ) -> None: """Enter the context and retain its session for shutdown cleanup. Store success before returning to a waiter that may already be cancelled. Shutdown waits for this task before exiting a successfully entered context. Args: connection: SDK connection context to enter; shutdown owns its exit. """ self._sdk = await connection.__aenter__() async def update_instructions(self, instructions: str) -> None: await self._apply_settings(instructions=instructions) async def update_tools(self, tools: list[llm.Tool]) -> None: await self._apply_settings(tools=tools) async def _update_model_options( self, previous: asyncio.Task[None] | None, *, voice: NotGivenOr[str], response_metadata: NotGivenOr[Mapping[str, str]], tool_fillers: NotGivenOr[Mapping[str, r.PrismFiller]], truncation: NotGivenOr[r.Truncation], ) -> None: """Apply queued options in order and report recoverable refusals as events.""" if previous is not None: await asyncio.shield(previous) await asyncio.shield(self._initialized) self._require() try: await self._settle_update( self._send_settings( voice=voice, response_metadata=response_metadata, tool_fillers=tool_fillers, truncation=truncation, ) ) except (PrismError, RimeInputError, RimeResourceLimitError) as error: if self._failure is not None: raise self._failure from None if isinstance(error, PrismError) and error.scope != "event": raise # Local SDK admission limits leave the session usable. A terminal SDK # failure is also delivered by the event reader, which closes the session. self._emit_error(error, recoverable=True) async def _wait_for_options(self) -> None: """Wait for options queued before this operation without canceling them.""" if self._options_task is not None: await asyncio.shield(self._options_task) async def _apply_settings( self, *, instructions: NotGivenOr[str] = NOT_GIVEN, tools: NotGivenOr[list[llm.Tool]] = NOT_GIVEN, voice: NotGivenOr[str] = NOT_GIVEN, response_metadata: NotGivenOr[Mapping[str, str]] = NOT_GIVEN, tool_fillers: NotGivenOr[Mapping[str, r.PrismFiller]] = NOT_GIVEN, truncation: NotGivenOr[r.Truncation] = NOT_GIVEN, ) -> None: # A cancelled caller still waits for acknowledgment so local settings cannot # contradict an update the server accepted. No update is replayed. await self._wait_for_options() await asyncio.shield(self._initialized) self._require() await self._settle_update( self._send_settings( instructions=instructions, tools=tools, voice=voice, response_metadata=response_metadata, tool_fillers=tool_fillers, truncation=truncation, ) ) async def _send_settings( self, *, instructions: NotGivenOr[str] = NOT_GIVEN, tools: NotGivenOr[list[llm.Tool]] = NOT_GIVEN, voice: NotGivenOr[str] = NOT_GIVEN, response_metadata: NotGivenOr[Mapping[str, str]] = NOT_GIVEN, tool_fillers: NotGivenOr[Mapping[str, r.PrismFiller]] = NOT_GIVEN, truncation: NotGivenOr[r.Truncation] = NOT_GIVEN, ) -> None: """Send settings and retain local state only after acknowledgment.""" async with self._settings_lock: sdk = self._require() fillers = dict(tool_fillers) if is_given(tool_fillers) else self._tool_fillers new_tools = list(tools) if is_given(tools) else self._tool_list definitions = ( self._sdk_tools(new_tools, fillers) if is_given(tools) or is_given(tool_fillers) else None ) try: if any( is_given(v) for v in ( instructions, tools, voice, tool_fillers, truncation, ) ): await sdk.update( truncation=cast(r.Truncation, truncation) if is_given(truncation) else None, instructions=instructions if is_given(instructions) else None, tools=definitions if is_given(tools) or is_given(tool_fillers) else None, voice=voice if is_given(voice) else None, ) self._tool_list = new_tools self._tool_fillers = fillers self._model._tool_fillers = dict(fillers) if is_given(voice): self._model._voice = voice if is_given(truncation): self._model._truncation = cast(r.Truncation, truncation) if is_given(response_metadata): await sdk.set_response_metadata(dict(response_metadata)) self._model._response_metadata = dict(response_metadata) except r.RimeRealtimeError as error: raise PrismError(error.fault) from error def update_options( self, *, tool_choice: NotGivenOr[llm.ToolChoice | None] = NOT_GIVEN, voice: NotGivenOr[str] = NOT_GIVEN, response_metadata: NotGivenOr[Mapping[str, str]] = NOT_GIVEN, tool_fillers: NotGivenOr[Mapping[str, r.PrismFiller]] = NOT_GIVEN, truncation: NotGivenOr[r.Truncation] = NOT_GIVEN, ) -> None: """Set tool choice and queue provider options without waiting for acknowledgment.""" if self._failure is not None or self._closed: self._require() if is_given(tool_choice): if tool_choice not in (None, "auto", "none"): raise llm.RealtimeError("Prism supports only tool_choice='auto' or 'none'") self._tool_choice = "none" if tool_choice == "none" else "auto" if any(is_given(value) for value in (voice, response_metadata, tool_fillers, truncation)): # Copy caller-owned maps before scheduling an asynchronous operation. self._options_task = self._spawn( self._update_model_options( self._options_task, voice=voice, response_metadata=dict(response_metadata) if is_given(response_metadata) else NOT_GIVEN, tool_fillers=dict(tool_fillers) if is_given(tool_fillers) else NOT_GIVEN, truncation=truncation, ) ) async def _get_history(self) -> tuple[r.HistoryItem, ...]: """Read Prism's completed history without changing LiveKit's conversation record.""" await asyncio.shield(self._initialized) self._require() assert self._history is not None try: return await self._history.get_history() except r.RimeRealtimeError as error: raise PrismError(error.fault) from error async def _delete_instruction_item(self, item_id: str) -> None: """Delete an instruction observed in this session without changing LiveKit history.""" await asyncio.shield(self._initialized) sdk = self._require() assert self._history is not None async with self._instruction_lock: state = self._instruction_items.get(item_id) if state is None: raise RimeInputError("Unknown saved Prism instruction item") if not state[1]: raise RimeInputError("Cannot delete a saved instruction before its reply ends") await self._history.wait_for_updates() try: await sdk.delete_item(item_id) except r.RimeRealtimeError as error: raise PrismError(error.fault) from error if self._instruction_items.get(item_id) == state: del self._instruction_items[item_id] async def update_chat_ctx(self, chat_ctx: llm.ChatContext) -> None: await asyncio.shield(self._initialized) token = _edit.get() if token is not None: if ( not token.active or token.consumed or token.session is not self or token.task is not asyncio.current_task() or token.target.copy(exclude_handoff=True, exclude_config_update=True).items != chat_ctx.copy(exclude_handoff=True, exclude_config_update=True).items ): raise HistoryUpdateError("Expired or mismatched history operation") token.consumed = True await self._settle_update(self._sync_chat_ctx(chat_ctx, token)) async def _sync_chat_ctx(self, chat_ctx: llm.ChatContext, token: _Edit | None) -> None: """Sync history and preserve continuation ownership for accepted tool results.""" self._require() assert self._history is not None if token is not None and token.operation == "reconcile": await self._history.reconcile(chat_ctx, token.revision) return result = await self._history.sync( chat_ctx, shared=token is None and self._model._context_mode == "shared", progress=token is None, ) for ref, output in result.submitted: if output.reply_required: if ( self._latest and ref.response_id == self._latest.response_id and not self._history.has_pending_calls(self._latest) ): self._set_tool_reply(self._latest, self._report) else: self._set_tool_reply(self._parent, ref) else: self._release_tool_response(ref) if result.failure is not None: error = result.failure logger.warning("Prism history update failed", extra={"status": error.status}) if not error.recoverable: self._fatal(error) if ( token is None and error.partial and result.tool_results_only and any(output.reply_required for _, output in result.submitted) ): raise _PartialToolResultError( str(error), status=error.status, confirmed_chat_ctx=error.confirmed_chat_ctx, needs_reconciliation=error.needs_reconciliation, ) from error raise error async def _settle_update(self, operation: Coroutine[Any, Any, None]) -> None: """Finish acknowledged state changes before propagating caller cancellation. Session shutdown still cancels the operation. An operation failure takes precedence over caller cancellation so uncertain state is not hidden. """ task = asyncio.create_task(operation) self._tasks.add(task) task.add_done_callback(self._tasks.discard) cancelled = False while not task.done(): try: await asyncio.shield(task) except asyncio.CancelledError: if self._closed: raise cancelled = True task.result() if cancelled: raise asyncio.CancelledError def push_audio(self, frame: rtc.AudioFrame) -> None: if self._failure is not None or self._closed: self._require() try: source_format = (frame.sample_rate, frame.num_channels) if source_format != self._audio_format: self._audio_format = source_format self._audio_resampler = None # Leave SDK-supported input unchanged. Convert only formats that its # PCMFormat cannot represent, such as a 44.1 kHz custom audio input. if frame.num_channels > 2: samples = np.frombuffer(frame.data, dtype=np.int16).reshape(-1, frame.num_channels) mono = (samples.sum(axis=1, dtype=np.int32) // frame.num_channels).astype(np.int16) frame = rtc.AudioFrame( data=mono.tobytes(), sample_rate=frame.sample_rate, num_channels=1, samples_per_channel=frame.samples_per_channel, ) if frame.sample_rate not in (8000, 16000, 24000, 48000): if self._audio_resampler is None: self._audio_resampler = rtc.AudioResampler( input_rate=frame.sample_rate, output_rate=16000, num_channels=frame.num_channels, ) frames = self._audio_resampler.push(frame) else: frames = [frame] for converted in frames: self._audio.put_nowait( r.AudioChunk( data=bytes(converted.data), format=r.PCMFormat( sample_rate=cast( Literal[8000, 16000, 24000, 48000], converted.sample_rate ), channels=cast(Literal[1, 2], converted.num_channels), ), ) ) except Exception as error: self._fatal(error) async def _send_audio(self) -> None: sdk = self._require() while True: chunk = await self._audio.get() await sdk.send_audio(chunk) def push_video(self, frame: rtc.VideoFrame) -> None: logger.warning("Prism does not accept video; ignoring frame") def commit_audio(self) -> None: logger.warning("Prism commits speech through server turn detection; ignoring commit_audio") def clear_audio(self) -> None: if self._failure is not None or self._closed: self._require() self._audio_resampler = None self._audio_format = None while not self._audio.empty(): self._audio.get_nowait() # Before connection, audio exists only in the local queue. The sender # starts after the SDK handshake, so no server clear is needed yet. if self._sdk is None: return # The SDK holds its audio lock across sends and clears, so an in-flight # chunk finishes before this clear reaches the server. self._spawn(self._require().clear_audio()) def _release_tool_response(self, ref: r.ResponseRef | r.ToolCallRef) -> None: assert self._history is not None if any( staged is not None and staged.response_id == ref.response_id for staged in (self._parent, self._report) ): return if not self._history.has_pending_calls( r.ResponseRef(session_id=ref.session_id, response_id=ref.response_id) ): self._history.release_response(ref.response_id) def _set_tool_reply(self, parent: r.ResponseRef | None, report: r.ToolCallRef | None) -> None: previous = (self._parent, self._report) self._parent, self._report = parent, report for ref in previous: if ref is not None: self._release_tool_response(ref) def generate_reply( self, *, instructions: NotGivenOr[str] = NOT_GIVEN, tool_choice: NotGivenOr[llm.ToolChoice] = NOT_GIVEN, tools: NotGivenOr[list[llm.Tool]] = NOT_GIVEN, metadata: Mapping[str, str] | None = None, ) -> asyncio.Future[llm.GenerationCreatedEvent]: if is_given(tools): logger.warning("Prism does not support per-reply tools; ignoring tools override") self.update_options(tool_choice=tool_choice) parent, report = self._parent, self._report holds: list[r.ResponseHold] = [] try: # A request owns its references independently of later staged replies. # The callback also releases them if cancelled before its coroutine runs. for ref in (parent, report): if ref is not None: holds.append(self._require().hold_response(ref)) task = self._spawn( self._reply( instructions if is_given(instructions) and instructions else None, parent, report, self._tool_choice == "none", dict(metadata) if metadata is not None else None, ), request=True, ) except BaseException: for hold in holds: hold.release() raise self._set_tool_reply(None, None) def release_holds(_: asyncio.Task[Any]) -> None: for hold in holds: hold.release() task.add_done_callback(release_holds) return task async def _reply( self, instruction: str | None, parent: r.ResponseRef | None, report: r.ToolCallRef | None, disable_tools: bool, metadata: dict[str, str] | None, ) -> llm.GenerationCreatedEvent: await self._wait_for_options() await asyncio.shield(self._initialized) # Do not hold the history lock during admission: a pending tool result # may need that lock before Prism can admit another reply. self._require() assert self._history is not None await self._history.wait_for_updates() if self._playback is not None: await self._playback.settle() try: if not disable_tools: return await self._request_reply(instruction, parent, report, metadata) async with self._settings_lock: sdk = self._require() restore = self._sdk_tools(self._tool_list, self._tool_fillers) self._suppress_tools = True try: info = await sdk.update(tools=[]) if info.settings_revision is not None: self._suppressed_revisions.add(info.settings_revision) if len(self._suppressed_revisions) > 128: oldest = min(self._suppressed_revisions) self._suppressed_revisions.remove(oldest) self._tool_policy_floor = oldest + 1 # Shield admission through response creation. Restoring the tools # before an accepted late response starts changes its settings. task = self._spawn( self._request_reply(instruction, parent, report, metadata), request=True ) try: return await asyncio.shield(task) except asyncio.CancelledError: try: created = await asyncio.shield(task) if created.response_id is not None: await self._abandon_reply( self._generations[created.response_id].ref ) finally: raise finally: if not self._closed and self._failure is None: try: await sdk.update(tools=restore) except BaseException: self._fatal(RuntimeError("Prism tool restoration did not complete")) raise self._suppress_tools = False except r.RimeRealtimeError as error: raise PrismError(error.fault) from error except r.RealtimeAdmissionTimeout as error: raise llm.RealtimeError(str(error), code="admission_timeout") from error async def _request_reply( self, instruction: str | None, parent: r.ResponseRef | None, report: r.ToolCallRef | None, metadata: dict[str, str] | None, ) -> llm.GenerationCreatedEvent: sdk = self._require() assert self._history is not None ref: r.ResponseRef | None = None try: if instruction is not None: ref = await sdk.request_reply(instruction=instruction, metadata=metadata) elif parent is not None: try: ref = await sdk.continue_reply(parent, metadata=metadata) except (RimeInputError, r.RimeRealtimeError) as error: if isinstance(error, r.RimeRealtimeError) and ( error.fault.scope != "event" or error.fault.code != "tool_continuation_unavailable" ): raise assert self._history is not None call = self._history.last_call(parent) if call is None: raise ref = await sdk.request_reply( tool_call=call, instruction=_TOOL_RESULT_INSTRUCTION, metadata=metadata ) else: ref = await sdk.request_reply( tool_call=report, instruction=_TOOL_RESULT_INSTRUCTION if report is not None else None, metadata=metadata, ) async with self._generation_ready: await self._generation_ready.wait_for( lambda: ref.response_id in self._generations or self._failure is not None ) self._require() return self._generations[ref.response_id].event except r.RimeRealtimeError as error: raise PrismError(error.fault) from error except r.RealtimeAdmissionTimeout as error: raise llm.RealtimeError(str(error), code="admission_timeout") from error except asyncio.CancelledError: if ref is not None and not self._closed: self._spawn(self._abandon_reply(ref)) raise async def _abandon_reply(self, response: r.ResponseRef) -> None: generation = self._generations.get(response.response_id) if generation is not None: generation.abandoned = True self._end(generation) await self._cancel(response) def interrupt(self) -> None: if not self._closed and self._failure is None and self._active is not None: generation = self._generations[self._active.response_id] generation.abandoned = True self._spawn(self._cancel(self._active)) self._end(generation) def truncate( self, *, message_id: str, modalities: list[Literal["text", "audio"]], audio_end_ms: int, audio_transcript: NotGivenOr[str] = NOT_GIVEN, ) -> None: if self._closed or self._failure is not None: return generation = next((g for g in self._generations.values() if message_id in g.outputs), None) if generation is None: return message = generation.outputs[message_id] message.discarded = True message.text.close() message.audio.close() if is_given(audio_transcript) or audio_end_ms == 0: message.played_transcript = audio_transcript if is_given(audio_transcript) else "" assert self._history is not None self._history.record_assistant(message_id, message.played_transcript, played=True) assert self._playback is not None self._playback.interrupt( r.PlaybackInterrupted(output=message.output, audio_end_ms=audio_end_ms) ) async def _cancel(self, response: r.ResponseRef) -> None: if self._closed or self._failure is not None: return try: await self._require().cancel(response) except r.RimeRealtimeError as error: if error.fault.scope != "event" or error.fault.code != "response_cancel_not_active": raise # The response ended before cancellation reached the server. Playback # the server retains its playback estimate. self._emit_error(error, recoverable=True) async def _report_playback(self, report: r.PlaybackReport) -> None: if self._closed or self._failure is not None: return try: await self._require().report_playback(report) except r.RimeRealtimeError as error: if error.fault.scope != "event": raise # A newer response owns the playback window; this report changes nothing. self._emit_error(error, recoverable=True) async def _read_events(self) -> None: async for event in self._require().events: self._on_event(event) async with self._generation_ready: self._generation_ready.notify_all() if not self._closed: raise ConnectionError("Prism session ended") def _on_event(self, event: r.SessionEvent) -> None: self.emit("rime_server_event_received", event) payload = event.payload assert self._history is not None if isinstance(payload, r.InputCommitted): self._turn_started.setdefault(payload.item_id, time.time()) self._history.observe( payload, turn_started_at=self._turn_started.get(payload.item_id) if isinstance(payload, (r.InputCommitted, r.TranscriptFinal)) else None, ) if isinstance(payload, r.SpeechStarted): self._latest = None self._set_tool_reply(None, self._report) self._turn_started[payload.item_id] = time.time() while len(self._turn_started) > 128: self._turn_started.popitem(last=False) self.emit( "input_speech_started", llm.InputSpeechStartedEvent(), ) elif isinstance(payload, r.SpeechStopped): stopped = llm.InputSpeechStoppedEvent(user_transcription_enabled=True) self.emit("input_speech_stopped", stopped) elif isinstance(payload, r.TranscriptDelta): text = self._transcript_deltas.get(payload.item_id, "") + payload.delta if len(text) > 65536 or ( payload.item_id not in self._transcript_deltas and len(self._transcript_deltas) >= 128 ): raise RuntimeError("Prism input transcript exceeds the adapter limit") self._transcript_deltas[payload.item_id] = text self.emit( "input_audio_transcription_completed", llm.InputTranscriptionCompleted( item_id=payload.item_id, transcript=text, is_final=False, turn_started_at=self._turn_started.get(payload.item_id), ), ) elif isinstance(payload, r.TranscriptFinal): self._transcript_deltas.pop(payload.item_id, None) started = self._turn_started.pop(payload.item_id, None) self.emit( "input_audio_transcription_completed", llm.InputTranscriptionCompleted( item_id=payload.item_id, transcript=payload.text, is_final=True, turn_started_at=started, ), ) elif isinstance(payload, r.TranscriptFailed): self._transcript_deltas.pop(payload.item_id, None) self._turn_started.pop(payload.item_id, None) elif isinstance(payload, r.ResponseStarted): assert self._playback is not None self._playback.start(payload.response) if ( payload.settings_revision is not None and payload.settings_revision < self._tool_policy_floor ): raise RimeResourceLimitError("Prism response tool policy is no longer retained") if len(self._generations) >= _GENERATION_LIMIT: # Local interruption closes streams before the provider's terminal event. oldest = next( (key for key, value in self._generations.items() if value.provider_ended), None, ) if oldest is None: raise RimeResourceLimitError( "Prism generation limit reached with an active reply" ) del self._generations[oldest] messages = utils.aio.Chan[llm.MessageGeneration](maxsize=128) functions = utils.aio.Chan[llm.FunctionCall](maxsize=128) created = llm.GenerationCreatedEvent( message_stream=messages, function_stream=functions, user_initiated=event.request_id is not None, response_id=payload.response.response_id, ) self._generations[payload.response.response_id] = _Generation( payload.response, created, messages, functions, created_timestamp=time.time(), created_monotonic=time.monotonic(), instruction_item_id=payload.instruction_item_id, # A speculative speech reply can arrive after a settings update # while still using the older revision's registered tools. tools_disabled=( payload.settings_revision in self._suppressed_revisions if payload.settings_revision is not None else self._suppress_tools ), ) if payload.instruction_item_id is not None: self._instruction_items.pop(payload.instruction_item_id, None) if len(self._instruction_items) >= 128: self._instruction_items.popitem(last=False) self._instruction_items[payload.instruction_item_id] = ( payload.response.response_id, False, ) self._latest = self._active = payload.response self._set_tool_reply(None, None) if event.request_id is None: self.emit("generation_created", created) elif isinstance(payload, r.ResponseAbandoned): self._spawn(self._abandon_reply(payload.response)) elif isinstance(payload, r.MessageStarted): generation = self._generations[payload.output.response.response_id] if generation.abandoned: return assert self._playback is not None message = _Message(payload.output, token=self._playback.output(payload.output)) if self._model._playback_mode == "discard": message.audio.close() generation.outputs[payload.output.item_id] = message modalities: asyncio.Future[list[Literal["text", "audio"]]] = ( asyncio.get_running_loop().create_future() ) modalities.set_result(["text", "audio"]) generation.messages.send_nowait( llm.MessageGeneration( message_id=payload.output.item_id, text_stream=message.text, audio_stream=message.audio, modalities=modalities, ) ) elif isinstance(payload, (r.TextDelta, r.TextDone, r.AudioDelta, r.AudioDone)): generation = self._generations[payload.output.response.response_id] if ( isinstance(payload, r.AudioDelta) and payload.audio.data and generation.first_audio_monotonic is None ): generation.first_audio_monotonic = time.monotonic() if generation.abandoned: return message = generation.outputs[payload.output.item_id] if message.discarded: return if isinstance(payload, r.TextDelta): message.transcript += payload.delta if len(message.transcript) > 65536: raise RuntimeError("Prism response text exceeds the adapter limit") message.text.send_nowait(payload.delta) elif isinstance(payload, r.TextDone): message.transcript = payload.text message.text.close() elif isinstance(payload, r.AudioDelta): if self._model._playback_mode == "discard": return data = payload.audio.data if data: frame = rtc.AudioFrame( data=data, sample_rate=24000, num_channels=1, samples_per_channel=len(data) // 2, ) assert self._playback is not None self._playback.tag(message.token, frame) message.audio.send_nowait(frame) else: message.audio.close() elif isinstance(payload, r.ToolCall): generation = self._generations[payload.call.response_id] if generation.tools_disabled: raise llm.RealtimeError("Prism emitted a tool call during a tool-disabled reply") if generation.abandoned: return call = self._history.record_tool_call(payload) self._generations[payload.call.response_id].functions.send_nowait(call) elif isinstance(payload, r.ResponseEnded): assert self._playback is not None self._playback.end( payload.response, interrupted=( self._generations[payload.response.response_id].abandoned or payload.status != "completed" ), ) instruction_item_id = self._generations[ payload.response.response_id ].instruction_item_id if instruction_item_id is not None: state = self._instruction_items.get(instruction_item_id) if state is not None and state[0] == payload.response.response_id: self._instruction_items[instruction_item_id] = (state[0], True) self._end(self._generations[payload.response.response_id]) if self._active == payload.response: self._active = None self._emit_response_metrics(payload) elif isinstance(payload, r.ItemTruncated): if payload.kept_text is not None and payload.content_index == 0: for generation in self._generations.values(): if not generation.ended and ( truncated_message := generation.outputs.get(payload.item_id) ): truncated_message.retained_text = payload.kept_text break elif isinstance(payload, r.FaultEvent) and payload.error.request_id is None: self._emit_error( PrismError(payload.error), recoverable=payload.error.scope != "session" ) def _emit_response_metrics(self, ended: r.ResponseEnded) -> None: generation = self._generations[ended.response.response_id] if generation.provider_ended: return generation.provider_ended = True duration = time.monotonic() - generation.created_monotonic usage = ended.usage output_tokens = usage.output_tokens if usage is not None else 0 self.emit( "metrics_collected", RealtimeModelMetrics( request_id=ended.response.response_id, timestamp=generation.created_timestamp, duration=duration, ttft=( generation.first_audio_monotonic - generation.created_monotonic if generation.first_audio_monotonic is not None else -1 ), cancelled=ended.status == "cancelled", label=self._model.label, input_tokens=usage.input_tokens if usage is not None else 0, output_tokens=output_tokens, total_tokens=usage.total_tokens if usage is not None else 0, tokens_per_second=output_tokens / duration if duration > 0 else 0, input_token_details=RealtimeModelMetrics.InputTokenDetails(), output_token_details=RealtimeModelMetrics.OutputTokenDetails(), metadata=Metadata( model_name=self._model.model, model_provider=self._model.provider ), ), ) def _end(self, generation: _Generation) -> None: if generation.ended: return generation.ended = True for item_id, message in generation.outputs.items(): transcript = ( message.transcript if message.played_transcript is None else message.played_transcript ) if message.retained_text is not None: transcript = message.retained_text if transcript or message.retained_text is not None: assert self._history is not None self._history.record_assistant(item_id, message.transcript) if message.played_transcript is not None: self._history.record_assistant(item_id, message.played_transcript, played=True) if message.retained_text is not None: self._history.record_assistant(item_id, message.retained_text) message.retained_text = None message.text.close() message.audio.close() generation.messages.close() generation.functions.close() def _emit_error(self, error: Exception, *, recoverable: bool) -> None: self.emit( "error", PrismModelError( timestamp=time.time(), label=self._model.label, error=error, recoverable=recoverable, retryable=error.retryable if isinstance(error, PrismError) else ( error.fault.retryable if isinstance(error, r.RimeRealtimeError) else ( _startup_retryable(error) if getattr(error, "status_code", None) is not None else None ) ), ), ) def _fatal(self, error: Exception) -> None: if isinstance(error, r.RimeRealtimeError): error = PrismError(error.fault) if self._failure is None: self._failure = error self._active = None if self._fatal_emitted: return self._fatal_emitted = True self._emit_error(error, recoverable=False) if not self._initialized.done(): self._initialized.set_exception(error) if (self._sdk is not None or self._opening_task is not None) and not self._closed: self._shutdown_task = asyncio.create_task(self._shutdown()) for generation in self._generations.values(): self._end(generation) def _spawn(self, coro: Coroutine[Any, Any, Any], *, request: bool = False) -> asyncio.Task[Any]: task = asyncio.create_task(coro) self._tasks.add(task) task.add_done_callback(lambda finished: self._done(finished, request=request)) return task def _done(self, task: asyncio.Task[Any], *, request: bool) -> None: self._tasks.discard(task) if not task.cancelled() and (error := task.exception()) is not None: # The caller receives reply refusals and admission timeouts. An error # in an unobserved background operation must also stop the session. if isinstance(error, Exception) and not ( request and isinstance(error, llm.RealtimeError) ): self._fatal(error) async def aclose(self) -> None: if self._shutdown_task is None: self._shutdown_task = asyncio.create_task(self._shutdown()) await asyncio.shield(self._shutdown_task) async def _shutdown(self) -> None: self._closed = True if self._playback is not None: self._playback.close() if self._history is not None: self._history.close() if not self._initialized.done(): self._initialized.cancel() tasks = list(self._tasks) if self._opening_task is not None: tasks.append(self._opening_task) for task in tasks: task.cancel() await asyncio.gather(*tasks, return_exceptions=True) if self._connection is not None and self._sdk is not None: await self._connection.__aexit__(None, None, None) for generation in self._generations.values(): self._end(generation)A Prism session that emits SDK SessionEvent objects before processing them.
Ancestors
- livekit.agents.llm.realtime.RealtimeSession
- abc.ABC
- EventEmitter
- typing.Generic
Instance variables
prop chat_ctx : llm.ChatContext-
Expand source code
@property def chat_ctx(self) -> llm.ChatContext: return self._history.chat_ctx if self._history is not None else llm.ChatContext.empty() prop tools : llm.ToolContext-
Expand source code
@property def tools(self) -> llm.ToolContext: return llm.ToolContext(self._tool_list)
Methods
async def aclose(self) ‑> None-
Expand source code
async def aclose(self) -> None: if self._shutdown_task is None: self._shutdown_task = asyncio.create_task(self._shutdown()) await asyncio.shield(self._shutdown_task) def clear_audio(self) ‑> None-
Expand source code
def clear_audio(self) -> None: if self._failure is not None or self._closed: self._require() self._audio_resampler = None self._audio_format = None while not self._audio.empty(): self._audio.get_nowait() # Before connection, audio exists only in the local queue. The sender # starts after the SDK handshake, so no server clear is needed yet. if self._sdk is None: return # The SDK holds its audio lock across sends and clears, so an in-flight # chunk finishes before this clear reaches the server. self._spawn(self._require().clear_audio()) def commit_audio(self) ‑> None-
Expand source code
def commit_audio(self) -> None: logger.warning("Prism commits speech through server turn detection; ignoring commit_audio") def generate_reply(self,
*,
instructions: NotGivenOr[str] = NOT_GIVEN,
tool_choice: NotGivenOr[llm.ToolChoice] = NOT_GIVEN,
tools: NotGivenOr[list[llm.Tool]] = NOT_GIVEN,
metadata: Mapping[str, str] | None = None) ‑> _asyncio.Future[livekit.agents.llm.realtime.GenerationCreatedEvent]-
Expand source code
def generate_reply( self, *, instructions: NotGivenOr[str] = NOT_GIVEN, tool_choice: NotGivenOr[llm.ToolChoice] = NOT_GIVEN, tools: NotGivenOr[list[llm.Tool]] = NOT_GIVEN, metadata: Mapping[str, str] | None = None, ) -> asyncio.Future[llm.GenerationCreatedEvent]: if is_given(tools): logger.warning("Prism does not support per-reply tools; ignoring tools override") self.update_options(tool_choice=tool_choice) parent, report = self._parent, self._report holds: list[r.ResponseHold] = [] try: # A request owns its references independently of later staged replies. # The callback also releases them if cancelled before its coroutine runs. for ref in (parent, report): if ref is not None: holds.append(self._require().hold_response(ref)) task = self._spawn( self._reply( instructions if is_given(instructions) and instructions else None, parent, report, self._tool_choice == "none", dict(metadata) if metadata is not None else None, ), request=True, ) except BaseException: for hold in holds: hold.release() raise self._set_tool_reply(None, None) def release_holds(_: asyncio.Task[Any]) -> None: for hold in holds: hold.release() task.add_done_callback(release_holds) return task def interrupt(self) ‑> None-
Expand source code
def interrupt(self) -> None: if not self._closed and self._failure is None and self._active is not None: generation = self._generations[self._active.response_id] generation.abandoned = True self._spawn(self._cancel(self._active)) self._end(generation) def push_audio(self, frame: rtc.AudioFrame) ‑> None-
Expand source code
def push_audio(self, frame: rtc.AudioFrame) -> None: if self._failure is not None or self._closed: self._require() try: source_format = (frame.sample_rate, frame.num_channels) if source_format != self._audio_format: self._audio_format = source_format self._audio_resampler = None # Leave SDK-supported input unchanged. Convert only formats that its # PCMFormat cannot represent, such as a 44.1 kHz custom audio input. if frame.num_channels > 2: samples = np.frombuffer(frame.data, dtype=np.int16).reshape(-1, frame.num_channels) mono = (samples.sum(axis=1, dtype=np.int32) // frame.num_channels).astype(np.int16) frame = rtc.AudioFrame( data=mono.tobytes(), sample_rate=frame.sample_rate, num_channels=1, samples_per_channel=frame.samples_per_channel, ) if frame.sample_rate not in (8000, 16000, 24000, 48000): if self._audio_resampler is None: self._audio_resampler = rtc.AudioResampler( input_rate=frame.sample_rate, output_rate=16000, num_channels=frame.num_channels, ) frames = self._audio_resampler.push(frame) else: frames = [frame] for converted in frames: self._audio.put_nowait( r.AudioChunk( data=bytes(converted.data), format=r.PCMFormat( sample_rate=cast( Literal[8000, 16000, 24000, 48000], converted.sample_rate ), channels=cast(Literal[1, 2], converted.num_channels), ), ) ) except Exception as error: self._fatal(error) def push_video(self, frame: rtc.VideoFrame) ‑> None-
Expand source code
def push_video(self, frame: rtc.VideoFrame) -> None: logger.warning("Prism does not accept video; ignoring frame") def truncate(self,
*,
message_id: str,
modalities: "list[Literal['text', 'audio']]",
audio_end_ms: int,
audio_transcript: NotGivenOr[str] = NOT_GIVEN) ‑> None-
Expand source code
def truncate( self, *, message_id: str, modalities: list[Literal["text", "audio"]], audio_end_ms: int, audio_transcript: NotGivenOr[str] = NOT_GIVEN, ) -> None: if self._closed or self._failure is not None: return generation = next((g for g in self._generations.values() if message_id in g.outputs), None) if generation is None: return message = generation.outputs[message_id] message.discarded = True message.text.close() message.audio.close() if is_given(audio_transcript) or audio_end_ms == 0: message.played_transcript = audio_transcript if is_given(audio_transcript) else "" assert self._history is not None self._history.record_assistant(message_id, message.played_transcript, played=True) assert self._playback is not None self._playback.interrupt( r.PlaybackInterrupted(output=message.output, audio_end_ms=audio_end_ms) ) async def update_chat_ctx(self, chat_ctx: llm.ChatContext) ‑> None-
Expand source code
async def update_chat_ctx(self, chat_ctx: llm.ChatContext) -> None: await asyncio.shield(self._initialized) token = _edit.get() if token is not None: if ( not token.active or token.consumed or token.session is not self or token.task is not asyncio.current_task() or token.target.copy(exclude_handoff=True, exclude_config_update=True).items != chat_ctx.copy(exclude_handoff=True, exclude_config_update=True).items ): raise HistoryUpdateError("Expired or mismatched history operation") token.consumed = True await self._settle_update(self._sync_chat_ctx(chat_ctx, token)) async def update_instructions(self, instructions: str) ‑> None-
Expand source code
async def update_instructions(self, instructions: str) -> None: await self._apply_settings(instructions=instructions) def update_options(self,
*,
tool_choice: NotGivenOr[llm.ToolChoice | None] = NOT_GIVEN,
voice: NotGivenOr[str] = NOT_GIVEN,
response_metadata: NotGivenOr[Mapping[str, str]] = NOT_GIVEN,
tool_fillers: NotGivenOr[Mapping[str, r.PrismFiller]] = NOT_GIVEN,
truncation: NotGivenOr[r.Truncation] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, tool_choice: NotGivenOr[llm.ToolChoice | None] = NOT_GIVEN, voice: NotGivenOr[str] = NOT_GIVEN, response_metadata: NotGivenOr[Mapping[str, str]] = NOT_GIVEN, tool_fillers: NotGivenOr[Mapping[str, r.PrismFiller]] = NOT_GIVEN, truncation: NotGivenOr[r.Truncation] = NOT_GIVEN, ) -> None: """Set tool choice and queue provider options without waiting for acknowledgment.""" if self._failure is not None or self._closed: self._require() if is_given(tool_choice): if tool_choice not in (None, "auto", "none"): raise llm.RealtimeError("Prism supports only tool_choice='auto' or 'none'") self._tool_choice = "none" if tool_choice == "none" else "auto" if any(is_given(value) for value in (voice, response_metadata, tool_fillers, truncation)): # Copy caller-owned maps before scheduling an asynchronous operation. self._options_task = self._spawn( self._update_model_options( self._options_task, voice=voice, response_metadata=dict(response_metadata) if is_given(response_metadata) else NOT_GIVEN, tool_fillers=dict(tool_fillers) if is_given(tool_fillers) else NOT_GIVEN, truncation=truncation, ) )Set tool choice and queue provider options without waiting for acknowledgment.
async def update_tools(self, tools: list[llm.Tool]) ‑> None-
Expand source code
async def update_tools(self, tools: list[llm.Tool]) -> None: await self._apply_settings(tools=tools)
Inherited members