|
| 1 | +# Copyright (c) 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors |
| 2 | +# |
| 3 | +# SPDX-License-Identifier: Apache-2.0 |
| 4 | + |
| 5 | +import asyncio |
| 6 | +from collections.abc import AsyncIterator, Callable, Iterable, Sequence |
| 7 | +from contextlib import asynccontextmanager, suppress |
| 8 | + |
| 9 | +from fast_depends.library.serializer import SerializerProto |
| 10 | + |
| 11 | +from ag2.config.config import ModelConfig |
| 12 | +from ag2.context import ConversationContext |
| 13 | +from ag2.events import ( |
| 14 | + AudioInterruptedEvent, |
| 15 | + AudioPlaybackCompletedEvent, |
| 16 | + AudioPlaybackStartedEvent, |
| 17 | + BaseEvent, |
| 18 | + ModelMessageChunk, |
| 19 | + ModelResponse, |
| 20 | + RecordedAudioEvent, |
| 21 | + SynthesizedAudioEvent, |
| 22 | + TextInput, |
| 23 | + ToolResultsEvent, |
| 24 | + UsageEvent, |
| 25 | +) |
| 26 | +from ag2.events.input_events import ModelRequest |
| 27 | +from ag2.tools.schemas import ToolSchema |
| 28 | + |
| 29 | +from ..config import LLMClient |
| 30 | +from .observer import Speech |
| 31 | +from .protocols import TTSConfig |
| 32 | +from .realtime import RealtimeConfig |
| 33 | +from .stt import STTConfig, VoiceInput |
| 34 | +from .turn import SilenceTurnDetector, TurnDetector |
| 35 | + |
| 36 | +# Cap on tool round-trips within one turn. A cascade turn is driven by a live |
| 37 | +# microphone; a model stuck in a tool loop would hold the floor indefinitely. |
| 38 | +MAX_TOOL_ITERATIONS = 10 |
| 39 | + |
| 40 | + |
| 41 | +class CascadeConfig(RealtimeConfig): |
| 42 | + """A separate STT, LLM, and TTS wired to look like one realtime session. |
| 43 | +
|
| 44 | + Implements `RealtimeConfig`, so `LiveAgent` drives it exactly as it drives |
| 45 | + `OpenAIRealTimeConfig` or `GeminiRealTimeConfig` — same events in, same |
| 46 | + events out, and `SoundDeviceRecorder` / `SoundDevicePlayer` need no changes: |
| 47 | +
|
| 48 | + agent = LiveAgent( |
| 49 | + "assistant", |
| 50 | + config=CascadeConfig( |
| 51 | + stt=ElevenLabsTranscriber("scribe_v2"), |
| 52 | + model=config.OpenAIConfig("gpt-5-mini", streaming=True), |
| 53 | + tts=ElevenLabsStreamingTTSConfig("eleven_flash_v2_5"), |
| 54 | + ), |
| 55 | + ) |
| 56 | +
|
| 57 | + The session is half-duplex by default: while the reply is playing, the |
| 58 | + microphone is ignored. On speakers the mic hears the reply, and a cascade |
| 59 | + that listens to itself interrupts its own sentence and then answers its own |
| 60 | + words — see `barge_in` to trade that safety for interruptibility. |
| 61 | +
|
| 62 | + barge_in=True allows for interruptions, but if you are on speakers, AI will detect |
| 63 | + its own speech as yours. With headphones, interruptions work nicely, no VAD logic currently. |
| 64 | + """ |
| 65 | + |
| 66 | + def __init__( |
| 67 | + self, |
| 68 | + *, |
| 69 | + stt: STTConfig, |
| 70 | + model: ModelConfig, |
| 71 | + tts: TTSConfig[bytes], |
| 72 | + turn_detector: Callable[[], TurnDetector] = SilenceTurnDetector, |
| 73 | + barge_in: bool = False, |
| 74 | + min_chars: int = 60, |
| 75 | + ) -> None: |
| 76 | + self._stt = stt |
| 77 | + self._model = model |
| 78 | + self._tts = tts |
| 79 | + self._turn_detector = turn_detector |
| 80 | + # Off by default. Shouldn't be on speakers to be on, otherwise AI's own speech gets detected as user's input. |
| 81 | + self._barge_in = barge_in |
| 82 | + self._min_chars = min_chars |
| 83 | + |
| 84 | + @asynccontextmanager |
| 85 | + async def session( |
| 86 | + self, |
| 87 | + context: ConversationContext, |
| 88 | + *, |
| 89 | + instructions: Iterable[str] = (), |
| 90 | + tools: Iterable[ToolSchema] = (), |
| 91 | + serializer: SerializerProto, |
| 92 | + ) -> AsyncIterator[None]: |
| 93 | + client: LLMClient = self._model.create() |
| 94 | + detector = self._turn_detector() |
| 95 | + schemas = list(tools) |
| 96 | + |
| 97 | + # `LiveAgent` builds the context without a prompt and hands the agent's |
| 98 | + # instructions to the session instead, so the system prompt has to be |
| 99 | + # installed here for the client to pick it up. |
| 100 | + prompt_mark = len(context.prompt) |
| 101 | + context.prompt.extend(instructions) |
| 102 | + |
| 103 | + turn: asyncio.Task[None] | None = None |
| 104 | + # Whether our own voice is currently coming out of the speaker |
| 105 | + playing = False |
| 106 | + |
| 107 | + async def _on_playback_started(event: AudioPlaybackStartedEvent) -> None: |
| 108 | + nonlocal playing |
| 109 | + playing = True |
| 110 | + |
| 111 | + async def _on_playback_completed(event: AudioPlaybackCompletedEvent) -> None: |
| 112 | + nonlocal playing |
| 113 | + playing = False |
| 114 | + # The tail of the reply is still in the detector's prefix buffer; |
| 115 | + # without this the first "user" turn after every answer opens on an |
| 116 | + # echo of the answer. |
| 117 | + detector.reset() |
| 118 | + |
| 119 | + async def _on_audio(event: RecordedAudioEvent) -> None: |
| 120 | + nonlocal turn |
| 121 | + |
| 122 | + if playing and not self._barge_in: |
| 123 | + return |
| 124 | + |
| 125 | + was_speaking = detector.speaking |
| 126 | + voice = detector.push(event.content) |
| 127 | + |
| 128 | + # Barge-in: the user started talking over the reply. |
| 129 | + if self._barge_in and not was_speaking and detector.speaking and _running(turn): |
| 130 | + assert turn is not None |
| 131 | + turn.cancel() |
| 132 | + await context.send(AudioInterruptedEvent()) |
| 133 | + |
| 134 | + if voice is None: |
| 135 | + return |
| 136 | + |
| 137 | + if _running(turn): |
| 138 | + assert turn is not None |
| 139 | + await turn |
| 140 | + |
| 141 | + # `spawn_background` (not a bare `create_task`) so a turn that |
| 142 | + # raises is logged rather than vanishing into an unretrieved task. |
| 143 | + turn = context.spawn_background(self._turn(voice, context, client, schemas, serializer)) |
| 144 | + |
| 145 | + with ( |
| 146 | + context.stream.where(RecordedAudioEvent).sub_scope(_on_audio), |
| 147 | + context.stream.where(AudioPlaybackStartedEvent).sub_scope(_on_playback_started), |
| 148 | + context.stream.where(AudioPlaybackCompletedEvent).sub_scope(_on_playback_completed), |
| 149 | + ): |
| 150 | + try: |
| 151 | + yield |
| 152 | + |
| 153 | + finally: |
| 154 | + if _running(turn): |
| 155 | + assert turn is not None |
| 156 | + turn.cancel() |
| 157 | + with suppress(asyncio.CancelledError): |
| 158 | + await turn |
| 159 | + del context.prompt[prompt_mark:] |
| 160 | + |
| 161 | + async def _turn( |
| 162 | + self, |
| 163 | + voice: VoiceInput, |
| 164 | + context: ConversationContext, |
| 165 | + client: LLMClient, |
| 166 | + schemas: list[ToolSchema], |
| 167 | + serializer: SerializerProto, |
| 168 | + ) -> None: |
| 169 | + """One user utterance: transcribe, answer, speak.""" |
| 170 | + text = await self._stt.transcribe(voice, context) |
| 171 | + if not text.strip(): |
| 172 | + return |
| 173 | + |
| 174 | + await context.send(ModelRequest([TextInput(text)])) |
| 175 | + await self._respond(context, client, schemas, serializer) |
| 176 | + |
| 177 | + async def _respond( |
| 178 | + self, |
| 179 | + context: ConversationContext, |
| 180 | + client: LLMClient, |
| 181 | + schemas: list[ToolSchema], |
| 182 | + serializer: SerializerProto, |
| 183 | + ) -> None: |
| 184 | + speech = Speech(self._tts, self._min_chars) |
| 185 | + |
| 186 | + with context.stream.where(ModelMessageChunk).sub_scope(speech.on_chunk): |
| 187 | + for _ in range(MAX_TOOL_ITERATIONS): |
| 188 | + messages = _conversation(await context.stream.history.get_events()) |
| 189 | + response = await client( # type: ignore[operator] |
| 190 | + messages, |
| 191 | + context, |
| 192 | + tools=schemas, |
| 193 | + response_schema=None, |
| 194 | + serializer=serializer, |
| 195 | + ) |
| 196 | + |
| 197 | + if response.usage: |
| 198 | + await context.send( |
| 199 | + UsageEvent( |
| 200 | + response.usage, |
| 201 | + kind="model_call", |
| 202 | + model=response.model, |
| 203 | + provider=response.provider, |
| 204 | + finish_reason=response.finish_reason, |
| 205 | + ) |
| 206 | + ) |
| 207 | + await context.send(response) |
| 208 | + |
| 209 | + if not response.tool_calls: |
| 210 | + await speech.finish(response.content, context) |
| 211 | + return |
| 212 | + |
| 213 | + async with context.stream.get(ToolResultsEvent | ModelResponse) as results: |
| 214 | + await context.send(response.tool_calls) |
| 215 | + settled = await results |
| 216 | + |
| 217 | + if isinstance(settled, ModelResponse): |
| 218 | + await speech.finish(settled.content, context) |
| 219 | + return |
| 220 | + |
| 221 | + raise RuntimeError(f"Tool loop exceeded {MAX_TOOL_ITERATIONS} iterations in one voice turn") |
| 222 | + |
| 223 | + |
| 224 | +def _conversation(events: "Sequence[BaseEvent]") -> "list[BaseEvent]": |
| 225 | + """Drop the raw audio before handing history to the model. |
| 226 | +
|
| 227 | + A live session logs every microphone chunk and every synthesized chunk — |
| 228 | + ten-plus events per second, each carrying PCM. The provider mappers ignore |
| 229 | + them, but they would still be walked (and held) on every model call, so a |
| 230 | + long conversation pays for the whole recording on each turn. Transcripts |
| 231 | + carry the meaning; the bytes are for the recorder and the speaker. |
| 232 | + """ |
| 233 | + return [e for e in events if not isinstance(e, (RecordedAudioEvent, SynthesizedAudioEvent))] |
| 234 | + |
| 235 | + |
| 236 | +def _running(task: "asyncio.Task[None] | None") -> bool: |
| 237 | + return task is not None and not task.done() |
0 commit comments