LiveAgent

LiveAgent is a full-duplex voice agent. Unlike the turn-by-turn STT/TTS pipeline, it opens a single session for the entire conversation — audio flows in and out continuously, with voice activity detection and barge-in.

It holds a RealtimeConfig, which comes in two flavors: a provider's native speech-to-speech API (OpenAI, Gemini), or CascadeConfig, which wires a separate STT, LLM, and TTS together behind the same interface. LiveAgent drives both identically, so the choice is one line of config.

Quick start#

A LiveAgent holds a RealtimeConfig and is opened via agent.run(), which yields a ConversationContext. Peers (player, recorder, observers) share that context so they all read from and write to the same event stream.

import asyncio

from ag2.live import (
    LiveAgent,
    SoundDevicePlayer,
    SoundDeviceRecorder,
    openai,
)

agent = LiveAgent(
    name="assistant",
    prompt="You are a helpful voice assistant.",
    config=openai.RealTimeConfig(
        "gpt-realtime-2",
        output=openai.AudioOutput(voice="ballad", speed=1.2),
    ),
)

async def main() -> None:
    async with (
        agent.run() as context,
        SoundDevicePlayer(context=context),
        SoundDeviceRecorder(context=context),
    ):
        print("Starting...")
        await asyncio.Future()  # run until cancelled

if __name__ == "__main__":
    asyncio.run(main())

Note

The three context managers must share the same context so the recorder's RecordedAudioEvents reach the provider session and the provider's SynthesizedAudioEvents reach the player.

Watching the transcript#

The realtime provider streams both audio and a text transcript. Subscribe to ModelMessageChunk to receive the assistant's transcript token-by-token.

import asyncio

from ag2.events import ModelMessageChunk
from ag2.live import (
    LiveAgent,
    OpenAIRealTimeConfig,
    SoundDevicePlayer,
    SoundDeviceRecorder,
)

agent = LiveAgent(
    name="assistant",
    prompt="You are a helpful voice assistant.",
    config=OpenAIRealTimeConfig("gpt-realtime-2"),
)

async def main() -> None:
    async with (
        agent.run() as context,
        SoundDevicePlayer(context=context),
        SoundDeviceRecorder(context=context),
    ):
        print("Starting...")
        with context.stream.where(ModelMessageChunk).join() as events:
            async for event in events:
                print(event)

if __name__ == "__main__":
    asyncio.run(main())

Tip

stream.where(EventType).join() gives you an async iterator that yields filtered events. It's the idiomatic way to consume a single event type from the live session without writing a subscriber.

Text-only output#

To keep the realtime session for its low-latency turn detection but disable audio output entirely, swap AudioOutput for TextOutput. The model returns raw text via ModelMessageChunk and never produces synthesized audio.

import asyncio

from ag2.events import ModelMessageChunk
from ag2.live import (
    LiveAgent,
    SoundDeviceRecorder,
    openai,
)

agent = LiveAgent(
    name="assistant",
    prompt="You are a helpful voice assistant.",
    config=openai.RealTimeConfig(
        "gpt-realtime-2",
        output=openai.TextOutput(),
    ),
)

async def main() -> None:
    async with (
        agent.run() as context,
        SoundDeviceRecorder(context=context),
    ):
        print("Starting...")
        with context.stream.where(ModelMessageChunk).join() as events:
            async for event in events:
                print(event)

if __name__ == "__main__":
    asyncio.run(main())

Keeping the session, swapping the voice#

Because the provider is no longer speaking, you are free to synthesize the reply yourself with any TTS. Attach a TTSObserver: it listens to the ModelMessageChunks the session emits and turns them into SynthesizedAudioEvents, which the player picks up as usual.

import asyncio

from ag2.live import (
    ElevenLabsStreamingTTSConfig,
    LiveAgent,
    SoundDevicePlayer,
    SoundDeviceRecorder,
    TTSObserver,
    openai,
)

agent = LiveAgent(
    name="assistant",
    prompt="You are a helpful voice assistant.",
    config=openai.RealTimeConfig(
        "gpt-realtime-2",
        output=openai.TextOutput(),
    ),
    observers=[
        # any TTS provider — the realtime model is not producing audio
        TTSObserver(ElevenLabsStreamingTTSConfig("eleven_flash_v2_5")),
    ],
)

async def main() -> None:
    async with (
        agent.run() as context,
        SoundDevicePlayer(context=context),
        SoundDeviceRecorder(context=context),
    ):
        await asyncio.Future()

if __name__ == "__main__":
    asyncio.run(main())

This buys you the realtime session's server-side turn detection while sourcing the voice elsewhere — the STT and LLM halves stay with the provider, only the TTS half moves. If you want all three halves to be yours, use CascadeConfig below.

Do not attach a TTSObserver to a session that already speaks

TextOutput() is what stops the provider generating audio. Without it, the session emits both SynthesizedAudioEvent (its own voice) and ModelMessageChunk (the transcript of that voice) — so a TTSObserver would synthesize a second voice from the transcript and the player would play both at once. Two overlapping voices, doubled TTS spend.

The same applies to CascadeConfig, which already speaks from inside the session. TTSObserver belongs with TextOutput(), or with a plain Agent — never with a config that produces audio of its own.

Tools in a realtime session#

LiveAgent supports the same @agent.tool decorator as a regular Agent. Tool calls are routed through AG2's normal tool executor, and results are sent back to the provider's realtime session automatically.

import asyncio

from ag2.live import (
    LiveAgent,
    OpenAIRealTimeConfig,
    SoundDevicePlayer,
    SoundDeviceRecorder,
)

agent = LiveAgent(
    name="assistant",
    prompt="You are a helpful voice assistant.",
    config=OpenAIRealTimeConfig("gpt-realtime-2"),
)

@agent.tool
async def sum_numbers(a: int, b: int) -> int:
    """You can use this tool to sum two numbers."""
    print(f"Summing {a} and {b}")
    return a + b

async def main() -> None:
    async with (
        agent.run() as context,
        SoundDevicePlayer(context=context),
        SoundDeviceRecorder(context=context),
    ):
        print("Starting...")
        await asyncio.Future()

if __name__ == "__main__":
    asyncio.run(main())

Providers#

LiveAgent is provider-neutral — it accepts any RealtimeConfig. AG2 ships with two native speech-to-speech implementations, plus CascadeConfig for mixing providers.

OpenAIGemini

from ag2.live import openai

config = openai.RealTimeConfig(
    "gpt-realtime-2",
    output=openai.AudioOutput(voice="ballad", speed=1.2),
    input=openai.InputConfig(
        # semantic VAD with interruption is the default
        turn_detection={
            "type": "semantic_vad",
            "create_response": True,
            "interrupt_response": True,
        },
    ),
)

Available voices: alloy, ash, ballad, coral, echo, sage, shimmer, verse, marin, cedar.

from ag2.live import gemini

config = gemini.RealTimeConfig(
    "gemini-3.1-flash-live-preview",
    output=gemini.AudioOutput(voice="Puck", language_code="en-US"),
    input=gemini.InputConfig(transcribe=True),
)

Available voices: Aoede, Charon, Fenrir, Kore, Leda, Orus, Puck, Zephyr.

Warning

Gemini Live's audio I/O is fixed by the API: 16 kHz mono PCM input, 24 kHz mono PCM output. Configure the recorder accordingly:

SoundDeviceRecorder(context=context, sample_rate=16000)

Full Gemini example with a tool

import asyncio

from ag2.events import ModelMessageChunk, TranscriptionChunkEvent
from ag2.live import (
    LiveAgent,
    SoundDevicePlayer,
    SoundDeviceRecorder,
    gemini,
)

agent = LiveAgent(
    name="assistant",
    prompt="You are a helpful voice assistant. Always respond in English.",
    config=gemini.RealTimeConfig(
        "gemini-3.1-flash-live-preview",
        output=gemini.AudioOutput(voice="Puck", language_code="en-US"),
        input=gemini.InputConfig(transcribe=True),
    ),
)

async def main() -> None:
    async with (
        agent.run() as context,
        SoundDevicePlayer(context=context),
        # Gemini Live requires 16 kHz mono PCM input
        SoundDeviceRecorder(context=context, sample_rate=16000),
    ):
        print("Starting...")
        with context.stream.where(ModelMessageChunk | TranscriptionChunkEvent).join() as events:
            async for event in events:
                print(event)

if __name__ == "__main__":
    asyncio.run(main())

CascadeConfig: a cascade that looks realtime#

The configs above are speech-to-speech: one provider owns transcription, the model, and the voice behind a single socket. CascadeConfig is the alternative — three separate providers wired together so that, from LiveAgent's point of view, they behave like one realtime session.

It implements the same RealtimeConfig protocol, so it is a drop-in: same agent.run(), same events, and SoundDevicePlayer / SoundDeviceRecorder need no changes.

import asyncio

from ag2 import config
from ag2.live import (
    CascadeConfig,
    ElevenLabsStreamingTTSConfig,
    LiveAgent,
    OpenAITranscriber,
    SoundDevicePlayer,
    SoundDeviceRecorder,
)

agent = LiveAgent(
    "assistant",
    prompt="You are a voice assistant. Answer in one or two short sentences.",
    config=CascadeConfig(
        stt=OpenAITranscriber("gpt-4o-mini-transcribe"),
        model=config.OpenAIConfig("gpt-5-mini", streaming=True),
        tts=ElevenLabsStreamingTTSConfig("eleven_flash_v2_5"),
    ),
)

async def main() -> None:
    async with (
        agent.run() as context,
        SoundDevicePlayer(context=context),
        SoundDeviceRecorder(context=context),
    ):
        print("Listening — say something. Ctrl+C to stop.")
        await asyncio.Future()

if __name__ == "__main__":
    asyncio.run(main())

Swapping to a native S2S model is a one-line change — config=OpenAIRealTimeConfig("gpt-realtime") — with prompt, tools, player, and recorder untouched. That substitutability is the property CascadeConfig exists to provide.

Why use it#

	Native S2S	`CascadeConfig`
Latency	Lowest — one socket, no hops	Higher — three sequential API calls per turn
Provider choice	Whatever the S2S vendor offers	Any STT × any LLM × any TTS
Voice	The vendor's voices	Any TTS voice, including cloned
Model	The vendor's realtime model	Any `ModelConfig` AG2 supports
Turn detection	Server-side VAD	Local, energy-gated (see below)

Reach for it when the voice or the model matters more than the last few hundred milliseconds — a specific ElevenLabs voice, a model with no realtime endpoint, or a provider mix that no single vendor sells.

Turn detection#

An S2S provider segments the microphone stream server-side. A cascade has no server doing that, so the decision "the user stopped talking, transcribe now" is made locally by a TurnDetector.

The default SilenceTurnDetector is energy-gated: root-mean-square amplitude against a fixed threshold, no model. That is enough for a close-talking mic in a quiet room.

from ag2.live import CascadeConfig, SilenceTurnDetector

config = CascadeConfig(
    stt=...,
    model=...,
    tts=...,
    turn_detector=lambda: SilenceTurnDetector(
        threshold=500.0,     # RMS above this counts as speech
        silence=0.7,         # seconds of quiet that close a turn
        min_speech=0.2,      # shorter bursts are discarded, not transcribed
        prefix_padding=0.3,  # audio kept from just before the trigger
        max_duration=30.0,   # hard cap on a single turn
    ),
)

turn_detector takes any callable returning a TurnDetector, so a real VAD model can be substituted for noisy input.

Tip

prefix_padding exists because the chunk that crosses the threshold is normally already a few milliseconds into the first word. Without it, turns start clipped. min_speech is the opposite guard — a cough or a door slam should not cost an STT round-trip.

Barge-in and echo#

The session is half-duplex by default: while the reply is playing, the microphone is ignored.

That default is about echo, not simplicity. On speakers, the mic hears the reply; a cascade that listens to itself will interrupt its own sentence and then answer its own words. There is no acoustic echo cancellation in ag2.live.

config = CascadeConfig(stt=..., model=..., tts=..., barge_in=True)

With barge_in=True, speech detected during playback cancels the in-flight turn and emits AudioInterruptedEvent, which drops any audio still queued in the player.

Warning

Use headphones with barge_in=True. On speakers the assistant's own voice trips the detector and interrupts itself. The turn detector is energy-gated with no speaker-voice model, so it cannot tell your voice from its own.

Streaming TTS#

If the TTS config exposes a stream method — like ElevenLabsStreamingTTSConfig — the session detects it and forwards audio chunk by chunk, so playback starts at time-to-first-byte instead of after the last sample of the sentence. Plain configs are synthesized whole. No flag to set; it is chosen from what the config offers.

Either way the reply is spoken sentence by sentence as the model produces it. min_chars (default 60) sets the smallest amount of text worth a TTS request:

config = CascadeConfig(stt=..., model=..., tts=..., min_chars=40)

Lowering it starts the first sentence sooner, at the cost of more requests and choppier prosody.

Tools#

Tools work exactly as they do on any other RealtimeConfig — declare them on the LiveAgent and the cascade runs the loop internally, feeding results back to the model before it speaks.

A cascade turn holds the microphone, so the tool loop is capped at 10 round-trips per turn; a model stuck in a loop would otherwise hold the floor indefinitely. Exceeding the cap raises rather than hanging.

Note

Raw audio is stripped from the conversation history before each model call. A live session logs every microphone and synthesized chunk — ten-plus events per second, each carrying PCM — and replaying that on every turn would make a long conversation pay for the whole recording each time. Transcripts carry the meaning; the bytes are for the recorder and the player.

LiveAgent vs Agent#

LiveAgent mirrors Agent's constructor surface — name, prompt, tools, middleware, observers, dependencies, variables, plugins, hitl_hook — so most agent-level concepts carry over. The differences:

Feature	`Agent`	`LiveAgent`
Entry point	`await agent.ask(input)`	`async with agent.run() as context`
History	Returned via `AgentReply`	Lives on the session's stream
Turn detection	Application-driven (you call `ask`)	Provider-driven (VAD)
Structured output	Supported	Not supported
`tasks` / `run_subtask`	Supported	Not supported

If you need both — for example, a realtime voice front-end that hands off to a tasking agent — drive the handoff through a tool on the LiveAgent that delegates to a separate Agent using Agent.as_tool().

Examples#

examples/live_playground/live_to_text_and_voice.py — native S2S, audio in and out
examples/live_playground/live_to_text_only.py — TextOutput(), transcript only
examples/live_playground/cascade_live_agent.py — CascadeConfig with a tool, and the one-line swap to S2S
examples/live_playground/live_tools_execution.py — tool calls in a realtime session

What's next#

STT & TTS — the turn-by-turn alternative, and the TTSObserver used above.
Tools — tool authoring, middleware, and approval flows that all work inside a LiveAgent.