Ag2 Live

Realtime voice / live audio agents

When to use

• The user wants a talking agent: speak into a mic, the agent replies with voice, hands-free, low latency (Gemini Live or OpenAI Realtime).
• They want live transcription of the user's speech while the agent talks.
• They want a voice pipeline over an existing text Agent: transcribe a recorded clip → run the agent → (optionally) speak the reply.
• They want to add spoken output (TTS) to an otherwise text-only streaming Agent.

If the user only needs to send a recorded audio file into an agent as one input (not a live session), use ag2-multimodal-input (AudioInput) instead.

> Hardware / keys caveat. A real session needs (a) a provider API key and (b) a working microphone + speaker. Neither is available headless. Everything in this skill constructs without hardware; the parts that actually open a socket or audio device are marked [needs keys + audio] below.

Installation

The live module splits across optional extras — install the ones you need:

# OpenAI Realtime + OpenAI TTS/STT
pip install "ag2[openai-realtime]"

# Gemini Live
pip install "ag2[gemini-realtime]"

# Local microphone / speaker I/O (SoundDeviceRecorder / SoundDevicePlayer)
pip install "sounddevice[numpy]"

A typical OpenAI voice app on the local sound card: pip install "ag2[openai-realtime]" "sounddevice[numpy]".

> Required. Run the relevant install before delivering code. Without the matching extra, the public symbol resolves to a placeholder that raises ImportError (openai/gemini) or the additional-dependency error (sounddevice[numpy]) the moment you use it.

Public API

All exported from ag2.live:

| Symbol | Role | |---|---| | LiveAgent | Wraps a prompt + realtime config; agent.run() opens the bidirectional session | | GeminiRealTimeConfig | Gemini Live realtime provider config | | OpenAIRealTimeConfig | OpenAI Realtime provider config | | OpenAITTSConfig | Text → speech (PCM bytes) | | OpenAITranscriber | One-shot speech → text (transcription) | | OpenAITranslationTranscriber | One-shot speech → English text (translation) | | SoundDeviceRecorder | Mic capture → RecordedAudioEvent on the stream | | SoundDevicePlayer | Plays SynthesizedAudioEvent PCM out the speaker | | TTSObserver | Observer that speaks a text Agent's streamed tokens via a TTS config |

How a live session works

LiveAgent is built around an event stream (a ConversationContext). The recorder, the provider session, and the player all share that one context:

mic ──SoundDeviceRecorder──▶ RecordedAudioEvent ──▶ provider session (Gemini/OpenAI)
                                                       │
                                       SynthesizedAudioEvent (assistant audio)
                                                       ▼
                                              SoundDevicePlayer ──▶ speaker

Along the way the provider also emits TranscriptionChunkEvent / TranscriptionCompletedEvent (your speech), ModelMessageChunk (assistant text), ToolCallEvent/ToolResultEvent, and UsageEvent. agent.run() is an async context manager that yields the shared ConversationContext so you can attach the recorder and player to it.

Minimal recipe — full duplex voice loop [needs keys + audio]

```python title="voice_loop.py" import asyncio

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

agent = LiveAgent( "voice_bot", "You are a friendly realtime voice assistant. Keep replies short and conversational.", config=OpenAIRealTimeConfig( "gpt-realtime", output=AudioOutput(voice="marin"), # Transcribe the user's speech so we can see it too. input=InputConfig(transcription={"model": "gpt-4o-mini-transcribe"}), ), )

async def main() -> None: # run() yields the shared ConversationContext. async with agent.run() as ctx: # Recorder and player bind to the SAME context so audio flows on one bus. async with ( SoundDeviceRecorder(context=ctx), # mic -> RecordedAudioEvent SoundDevicePlayer(context=ctx), # SynthesizedAudioEvent -> speaker ): print("Talk to the agent. Ctrl-C to stop.") await asyncio.Event().wait() # keep the session open

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

Set `OPENAI_API_KEY` (or pass `client=AsyncOpenAI(...)`). For Gemini, set `GEMINI_API_KEY` / `GOOGLE_API_KEY` (or pass `client=`).

Swap to Gemini by changing only the config:

```python
from ag2.live import GeminiRealTimeConfig
from ag2.live.gemini import AudioOutput, InputConfig

agent = LiveAgent(
    "voice_bot",
    "You are a friendly realtime voice assistant.",
    config=GeminiRealTimeConfig(
        "gemini-2.5-flash-native-audio-preview-12-2025",
        output=AudioOutput(voice="Kore"),
        input=InputConfig(transcribe=True),   # user-speech transcription
    ),
)

Provider matrix — Gemini Live vs OpenAI Realtime

| | Gemini (GeminiRealTimeConfig) | OpenAI (OpenAIRealTimeConfig) | |---|---|---| | Import the knob types from | ag2.live.gemini | ag2.live.openai | | Output modes | AudioOutput (default) / TextOutput | AudioOutput (default) / TextOutput | | Voices (AudioOutput(voice=...)) | Aoede, Charon, Fenrir, Kore (default), Leda, Orus, Puck, Zephyr | alloy (default), ash, ballad, coral, echo, sage, shimmer, verse, marin, cedar | | Example models (first positional arg) | gemini-2.5-flash-native-audio-preview-12-2025, gemini-3.1-flash-live-preview, gemini-live-2.5-flash-preview, gemini-2.0-flash-live-001 | gpt-realtime, gpt-realtime-mini, gpt-audio-1.5, gpt-4o-realtime-preview-2024-10-01 | | User-speech transcription | InputConfig(transcribe=True, transcription_languages=[...]) | InputConfig(transcription={"model": "gpt-4o-mini-transcribe"}) | | Turn detection / VAD | InputConfig(automatic_activity_detection=..., activity_handling=..., turn_coverage=...) | InputConfig(turn_detection={"type": "semantic_vad", ...}) (default on) | | Audio I/O sample rate | in 16 kHz / out 24 kHz (fixed by API) | configurable via AudioOutput.format / InputConfig.format (default PCM 24 kHz) | | Generation knobs | temperature=, max_output_tokens= | max_output_tokens= (int or "inf"), tool_choice=, tracing= | | Escape hatch for raw provider config | config= | session= | | Bring your own client | client=google.genai.Client(...) | client=openai.AsyncOpenAI(...) |

The model name is a positional arg; everything else is keyword-only. The Literal model/voice lists above are accepted, but any string is allowed too (forward-compatible with new models).

AudioOutput adds language_code= (Gemini) and speed= + format= (OpenAI). TextOutput() takes no args — it returns text only (ModelMessageChunk), no audio playback.

Tools, HITL, observers on a LiveAgent

LiveAgent(...) accepts the same surface as a normal Agent: tools=, hitl_hook=, middleware=, observers=, dependencies=, variables=, plugins=. Tool calls from the model arrive as ToolCallEvent and results are forwarded back into the session automatically. run() can override config=, prompt=, tools=, observers=, hitl_hook= per session.

> Note: LiveAgent only supports function tools over realtime — provider server-side tool types raise NotImplementedError in both backends.

Audio I/O — SoundDeviceRecorder / SoundDevicePlayer

Both are async context managers (async with) and bind to a ConversationContext via context=. The device opens on __aenter__ [needs audio]; construction alone touches no hardware.

from ag2.live import SoundDeviceRecorder, SoundDevicePlayer

# Recorder: mic -> RecordedAudioEvent on the stream.
recorder = SoundDeviceRecorder(context=ctx, sample_rate=16000, channels=1)  # block_size optional

# Player: subscribes to SynthesizedAudioEvent, writes PCM to the speaker (24 kHz mono).
player = SoundDevicePlayer(context=ctx)

SoundDeviceRecorder also has a one-shot blocking record(duration) that returns a VoiceInput (16-bit PCM) — handy to feed the STT pipeline below without running a live session:

voice = recorder.record(duration=4.0)   # blocks 4s, returns VoiceInput  [needs audio]

One-shot speech-to-text + voice pipeline [needs keys; audio only if recording live]

OpenAITranscriber (transcription) and OpenAITranslationTranscriber (translate to English) take a VoiceInput and return text. .pipe(agent) builds a VoicePipeline that transcribes then runs a normal text Agent:

from ag2 import Agent
from ag2.config import OpenAIConfig
from ag2.live import OpenAITranscriber, SoundDeviceRecorder
from ag2.live.stt import VoiceInput

agent = Agent("assistant", "You answer questions.", config=OpenAIConfig(model="gpt-4o-mini"))
pipeline = OpenAITranscriber("gpt-4o-transcribe").pipe(agent)

# Record a clip, transcribe it, and ask the agent — in one call.  [needs keys + audio]
voice: VoiceInput = SoundDeviceRecorder().record(4.0)
reply = await pipeline.ask(voice)
print(await reply.content())

# Continue the conversation with another clip:
followup = await reply.ask(SoundDeviceRecorder().record(4.0))

VoiceInput(content: bytes, frame_rate: int, channels: int) wraps 16-bit PCM; build it directly from any PCM source if you aren't using the recorder.

OpenAITranslationTranscriber is identical but always outputs English (useful for non-English speech in → English text out).

Text-to-speech — OpenAITTSConfig and TTSObserver

OpenAITTSConfig synthesizes text into PCM bytes:

from ag2.live import OpenAITTSConfig

tts = OpenAITTSConfig("gpt-4o-mini-tts", voice="alloy", speed=1.0)
pcm: bytes = await tts.synthesize("Hello there!")   # [needs keys]

TTSObserver turns any text streaming Agent into a talking one: attach it as an observer and it accumulates ModelMessageChunk tokens, synthesizes complete sentences, and emits SynthesizedAudioEvent — which a SoundDevicePlayer on the same stream plays aloud. This is the bridge between a normal text agent and live audio output (no realtime provider needed).

from ag2 import Agent, MemoryStream
from ag2.config import OpenAIConfig
from ag2.context import ConversationContext
from ag2.live import OpenAITTSConfig, SoundDevicePlayer, TTSObserver

agent = Agent(
    "narrator",
    "You are a helpful assistant.",
    config=OpenAIConfig(model="gpt-4o-mini"),
    observers=[TTSObserver(OpenAITTSConfig("gpt-4o-mini-tts", voice="nova"))],
)

# Agent and Player must share ONE stream: agent.ask(stream=...) and
# SoundDevicePlayer(context=...) whose .stream is that same stream.
stream = MemoryStream()
ctx = ConversationContext(stream=stream)

async with SoundDevicePlayer(context=ctx):     # plays the synthesized speech  [needs audio]
    reply = await agent.ask("Tell me a one-line joke.", stream=stream)
    print(await reply.content())

TTSObserver(config) returns a CompositeObserver. It flushes any remaining buffered text on the final ModelMessage, so trailing partial sentences are still spoken.

Usage reporting

await LiveAgent.usage_report(ctx) aggregates token usage over the live session's event log into a UsageReport.

What is construction-tested vs needs live keys + hardware

Run by this skill's references/test_samples.py (all green) — constructed / exercised offline:

• All 9 public symbols import as real classes (not missing-dependency placeholders).
• GeminiRealTimeConfig / OpenAIRealTimeConfig construction with AudioOutput / TextOutput / InputConfig, voices, temperature/max-tokens, and _build_session(instructions=...) merge.
• OpenAITTSConfig, OpenAITranscriber, OpenAITranslationTranscriber construction.
• LiveAgent construction (string prompt, list prompt, with tools=/observers=).
• SoundDeviceRecorder / SoundDevicePlayer construction (no device opened).
• TTSObserver returns a CompositeObserver.
• OpenAITranscriber(...).pipe(agent) → VoicePipeline; VoiceInput dataclass.

Requires real API keys + microphone/speaker (NOT runnable headless):

• agent.run() opening a live websocket to Gemini/OpenAI.
• async with SoundDeviceRecorder(...) / SoundDevicePlayer(...) (opens the sound card on __aenter__).
• SoundDeviceRecorder.record(duration) (blocks on the mic).
• OpenAITTSConfig.synthesize(...), OpenAITranscriber.transcribe(...), and VoicePipeline.ask(...) (hit the OpenAI API).

Common pitfalls

• Wrong extra installed. pip install "ag2[openai-realtime]" for OpenAI, "ag2[gemini-realtime]" for Gemini, "sounddevice[numpy]" for local audio. The symbol imports fine but raises on first use otherwise.
• numpy missing. SoundDeviceRecorder / SoundDevicePlayer need numpy — sounddevice[numpy] pulls it in. Without it you get an additional-dependency ImportError.
• Importing knob types from the wrong module. AudioOutput / TextOutput / InputConfig are provider-specific: ag2.live.gemini vs ag2.live.openai. They are not re-exported from ag2.live.
• Recorder, player, and run() not sharing one context. Pass the ctx yielded by agent.run() into SoundDeviceRecorder(context=ctx) / SoundDevicePlayer(context=ctx), or audio events won't reach the session/speaker.
• Voice not in the provider's list. Gemini and OpenAI have different voice names (see matrix). A bare string is accepted but an unknown voice is rejected by the provider at session open.
• Expecting server-side / provider tools over realtime. Only function tools are supported; other tool types raise NotImplementedError.
• Forgetting to keep the loop alive. agent.run() is a context manager — exit closes the session. Keep it open (await asyncio.Event().wait() or your own loop) for a continuous conversation.

Going deeper

• Source: ag2/live/{realtime.py,gemini.py,openai.py,protocols.py,observer.py,stt.py,sound_device.py}.
• realtime.py — LiveAgent + the RealtimeConfig protocol.
• For sending a recorded audio file as a one-off input to a text agent, see ag2-multimodal-input (AudioInput).
• For observers in general (token monitors, loop detection), see ag2-observers-and-alerts.

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: ag2ai
• Source: ag2ai/ag2-skills
• License: Apache-2.0

Install and usage instructions live in the source repository linked above.