Skip to content

Overview

ag2.live is the AG2 module for building voice-enabled agents. It covers three patterns:

  1. Turn-by-turn / Pipeline — an STT → Agent → TTS round-trip built on top of a regular Agent. One utterance in, one reply out.

  2. Realtime / Speech-to-Speech (S2S) — a LiveAgent streaming audio to and from a provider's realtime API, which owns all three stages.

  3. Cascade — a LiveAgent running CascadeConfig: separate STT, LLM, and TTS providers wired to behave like a realtime session.

When to use which#

Pattern Class Latency Use when…
Turn-by-turn voice Agent + a transcriber + a TTS observer ~1–3 s per turn You already have a text agent and want to add a voice front-end; you need structured output, tasks, or any other text-agent feature.
Realtime full-duplex LiveAgent + OpenAIRealTimeConfig / GeminiRealTimeConfig <500 ms You want barge-in, interruption, semantic VAD, or a phone-call-like UX.
Cascade LiveAgent + CascadeConfig in between You want the continuous session, but need a specific voice, a specific model, or a provider mix no single S2S vendor sells.

Patterns 2 and 3 are interchangeable at the config line — LiveAgent drives both through the same RealtimeConfig protocol, so switching costs nothing but the config.

See Provider support for the matrix of currently supported providers.

Note

All three patterns share the same audio I/O primitives — SoundDevicePlayer and SoundDeviceRecorder — and the same event stream. You can mix observers (TTS, logging, persistence) across both.

Installation#

The audio I/O classes depend on sounddevice and numpy; the OpenAI, Gemini, and ElevenLabs integrations need their respective SDKs.

pip install "ag2[openai,gemini,elevenlabs] sounddevice[numpy]"

Warning

SoundDevicePlayer and SoundDeviceRecorder require sounddevice[numpy] as an additional dependency (not optional)

Imports and credentials#

Every provider config is importable from ag2.live under a provider-prefixed name: OpenAIRealTimeConfig, OpenAITranscriber, OpenAITranslationTranscriber, OpenAITTSConfig, GeminiRealTimeConfig, ElevenLabsTranscriber, ElevenLabsTTSConfig, ElevenLabsStreamingTTSConfig. The same classes also live in the provider modules ag2.live.openai, ag2.live.gemini, and ag2.live.elevenlabs under their short names (openai.RealTimeConfig, openai.STTConfig, …).

The provider-specific option types — AudioOutput, TextOutput, InputConfig — are only available from the provider module, so a realtime example that tunes the voice imports the module (from ag2.live import openai) and uses openai.RealTimeConfig alongside openai.AudioOutput. Both spellings are equivalent; the pages in this section use whichever keeps the example shortest.

Each provider reads its key from the environment unless you pass one:

Provider Environment variable Explicit key
OpenAI OPENAI_API_KEY api_key= on the four OpenAI configs, see OpenAI credentials
Gemini GOOGLE_API_KEY or GEMINI_API_KEY pass a preconfigured client=
ElevenLabs ELEVENLABS_API_KEY pass a preconfigured client=

The flows at a glance#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import asyncio

from ag2 import Agent, config
from ag2.live import (
    OpenAITTSConfig,
    OpenAITranscriber,
    SoundDevicePlayer,
    SoundDeviceRecorder,
    TTSObserver,
)

agent = Agent(
    name="assistant",
    prompt="You are a helpful voice assistant.",
    config=config.OpenAIResponsesConfig(model="gpt-5", streaming=True),
    observers=[TTSObserver(config=OpenAITTSConfig(model="gpt-4o-mini-tts"))],
)

async def main() -> None:
    pipeline = OpenAITranscriber("gpt-4o-mini-transcribe").pipe(agent)

    async with SoundDevicePlayer() as player:
        voice = SoundDeviceRecorder().record(duration=3)
        reply = await pipeline.ask(voice, stream=player.stream)
        print(reply.body)

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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()

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import asyncio

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

agent = LiveAgent(
    name="assistant",
    prompt="You are a helpful voice assistant.",
    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("Starting...")
        await asyncio.Future()

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

What's next#

  • STT & TTS — wrap any Agent with speech input and output.
  • LiveAgent — realtime voice agents: native speech-to-speech with OpenAI or Gemini, or a CascadeConfig mixing providers.