STT & TTS

The STT → Agent → TTS flow turns any existing Agent into a voice agent without changing the agent itself. Speech-to-text is added as a pipeline wrapper; text-to-speech is added as an observer that listens to the model's streamed message chunks.

Provider support#

Provider	STT	Streaming STT	TTS	Streaming TTS	Audio to Audio
OpenAI	✅	✅	✅	—	✅
ElevenLabs	✅	—	✅	✅	—
Gemini	—	—	—	—	✅

Mixing providers across the two halves is supported and often the best setup: transcribe with OpenAI for live captions, speak with ElevenLabs for low time-to-first-byte.

Note

The Audio to Audio column means a native speech-to-speech endpoint. To get continuous, full-duplex audio out of providers that have none — ElevenLabs voices on any LLM, for instance — use CascadeConfig, which presents an STT + LLM + TTS cascade to LiveAgent as if it were one realtime session.

ElevenLabs classes need the optional dependency and ELEVENLABS_API_KEY in the environment:

pip install "ag2[elevenlabs] sounddevice[numpy]"

Audio I/O primitives#

SoundDeviceRecorder captures microphone input and SoundDevicePlayer plays synthesized speech. Both are thin wrappers around the sounddevice library and share the same event stream.

from ag2.live import SoundDevicePlayer, SoundDeviceRecorder

recorder = SoundDeviceRecorder()
voice = recorder.record(duration=5)  # blocks for 5s, returns VoiceInput

The recorder produces a VoiceInput containing 16-bit PCM bytes plus the sample rate and channel count. The player subscribes to SynthesizedAudioEvent on its context's stream and plays each chunk on a background thread.

Note

Recorder.record(duration=...) is a one-shot, blocking helper for the turn-by-turn flow. For continuous streaming (used by LiveAgent), use the recorder as an async context manager — see LiveAgent.

Speech-to-Text#

OpenAITranscriber implements the STTConfig protocol and exposes a .pipe(agent) method that wraps an Agent in a VoicePipeline. Calling pipeline.ask(voice) transcribes the audio and forwards the text to the agent's normal ask() flow.

import asyncio

from ag2 import Agent, config
from ag2.live import OpenAITranscriber, SoundDeviceRecorder

agent = Agent(
    "assistant",
    config=config.OpenAIConfig("gpt-5", streaming=True),
)

async def main():
    # pipe STT model to agent input
    pipeline = OpenAITranscriber("gpt-4o-mini-transcribe").pipe(agent)
    recorder = SoundDeviceRecorder()

    print("Say something...")
    voice_input = recorder.record(duration=5)
    reply = await pipeline.ask(voice_input)
    print(reply.body)

    print("Say something...")
    voice_input = recorder.record(duration=5)
    # continue the same conversation
    reply = await reply.ask(voice_input)
    print(reply.body)

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

pipeline.ask(...) returns a VoiceReply that exposes the same surface as AgentReply (.body, .response, .history, .ask(...)) plus .ask(voice_input) for the next voice turn. The agent's history is preserved across turns.

Tip

The transcriber emits TranscriptionChunkEvent and TranscriptionCompletedEvent on the agent's stream as soon as the transcription server starts producing tokens. Subscribe to them to display live captions.

Translation#

If you want the user's speech transcribed into English regardless of input language, swap in OpenAITranslationTranscriber. It has the same API as OpenAITranscriber but uses OpenAI's translation endpoint.

from ag2.live import OpenAITranslationTranscriber

pipeline = OpenAITranslationTranscriber("whisper-1").pipe(agent)

ElevenLabs (Scribe)#

ElevenLabsTranscriber implements the same STTConfig protocol, so it drops into the same .pipe(agent) call:

from ag2.live import ElevenLabsTranscriber

pipeline = ElevenLabsTranscriber("scribe_v2").pipe(agent)

language_code is only sent when you set it — leave it unset and Scribe auto-detects. diarize=True asks for speaker labels.

Text-to-Speech#

TTSObserver is an observer that listens to ModelMessageChunk events as the agent streams its response, batches them into sentence-sized chunks, calls a TTS provider, and emits SynthesizedAudioEvents onto the stream. A SoundDevicePlayer attached to the same stream then plays them.

import asyncio

from ag2 import Agent, config
from ag2.live import OpenAITTSConfig, SoundDevicePlayer, 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:
    async with SoundDevicePlayer() as player:
        # pass the player's stream so synthesized audio reaches the speakers
        await agent.ask("Hello, agent!", stream=player.stream)

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

Warning

The agent's config must be set up for streaming output (e.g., streaming=True). TTSObserver works at the ModelMessageChunk granularity — if the model emits a single non-streaming ModelMessage, the observer will still synthesize it, but you lose the sentence-level pipelining that keeps latency low.

Never attach it to something that already speaks

TTSObserver synthesizes from the model's text. A LiveAgent running a speech-to-speech config or a CascadeConfig already produces its own audio and emits the matching transcript — so the observer would speak a second copy over it. Two voices, doubled TTS spend. See LiveAgent for the one realtime setup where attaching it is correct.

Voice and speed#

Any config implementing the TTSConfig protocol works with TTSObserver:

OpenAIElevenLabs

from ag2.live import OpenAITTSConfig

config = OpenAITTSConfig(
    model="gpt-4o-mini-tts",
    voice="ballad",  # alloy, ash, ballad, coral, echo, sage, shimmer, verse...
    speed=1.1,
)

from ag2.live import ElevenLabsTTSConfig

config = ElevenLabsTTSConfig(
    "eleven_v3",
    voice_id="21m00Tcm4TlvDq8ikWAM",  # Rachel
    output_format="pcm_24000",  # matches SoundDevicePlayer
)

ElevenLabsTTSConfig is the buffered mode — required by eleven_v3, which cannot stream. For the flash and turbo families, prefer streaming (below).

Note

output_format defaults to pcm_24000, matching what SoundDevicePlayer and SoundDeviceRecorder expect. If you change it, change the player to match.

Streaming TTS#

With a buffered config, TTSObserver waits for each sentence to be fully synthesized before any of it plays. A streaming config removes that wait: audio chunks are forwarded as they are generated, so playback starts at time-to-first-byte.

Note

ElevenLabs is currently the only provider with a streaming TTS config. OpenAITTSConfig performs a single buffered read and has no streaming equivalent.

Only the config changes — TTSObserver detects that the config can stream and uses that path automatically:

import asyncio

from ag2 import Agent, config
from ag2.live import ElevenLabsStreamingTTSConfig, SoundDevicePlayer, TTSObserver

tts = ElevenLabsStreamingTTSConfig("eleven_flash_v2_5", voice_id="21m00Tcm4TlvDq8ikWAM")

agent = Agent(
    "assistant",
    prompt="You are a helpful voice assistant. Keep answers to a few sentences.",
    config=config.OpenAIConfig("gpt-5-mini", streaming=True),
    observers=[TTSObserver(tts)],
)

async def main() -> None:
    async with SoundDevicePlayer() as player:
        await agent.ask("Hello, agent!", stream=player.stream)

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

Streaming directly to the speaker#

ElevenLabsStreamingTTSConfig.stream(...) is usable on its own, outside any agent, when you just want to speak a known string:

import asyncio

from ag2.live import ElevenLabsStreamingTTSConfig, SoundDevicePlayer

async def main() -> None:
    tts = ElevenLabsStreamingTTSConfig("eleven_flash_v2_5")
    async with SoundDevicePlayer() as player:
        async for chunk in tts.stream("Audio arrives as it is generated."):
            await player.play(chunk)

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

player.play(...) only enqueues — a background worker thread writes to the output device, so this loop is never blocked by playback.

Sentence batching#

TTSObserver batches streamed text before synthesizing, since a per-token request would be slow and expensive. min_chars (default 60) is the smallest amount of text worth a request; below it the buffer keeps accumulating even past a sentence boundary.

observers=[TTSObserver(tts, min_chars=40)]

Lowering it starts the first sentence sooner, at the cost of more requests and more granular sentences.

Combining STT and TTS#

The full round-trip — voice in, voice out — is just both halves wired up at once: pipe the agent through the transcriber, attach a TTSObserver, and share a stream with the player.

import asyncio

from ag2 import Agent, config
from ag2.context import ConversationContext
from ag2.live import (
    OpenAITTSConfig,
    OpenAITranscriber,
    SoundDevicePlayer,
    SoundDeviceRecorder,
    TTSObserver,
)
from ag2.stream import MemoryStream

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

async def main():
    pipeline = OpenAITranscriber("gpt-4o-mini-transcribe").pipe(agent)
    recorder = SoundDeviceRecorder()
    # one stream for the whole conversation, but a player per turn
    stream = MemoryStream()
    reply = None

    for _ in range(3):
        print("Say something...")
        # safe to open the mic: the previous turn's player has fully drained
        voice_input = recorder.record(duration=5)
        # exiting this block waits for playback to finish
        async with SoundDevicePlayer(context=ConversationContext(stream)):
            if reply is None:
                reply = await pipeline.ask(voice_input, stream=stream)
            else:
                reply = await reply.ask(voice_input)
        print(reply.body)

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

The player is scoped to a single turn and attached to a shared MemoryStream, so history carries across turns while each turn's audio is guaranteed to have finished before the microphone opens again.

player.join() does not wait for playback to finish

join() returns as soon as the queue is empty — but the worker thread has already dequeued the final chunk and is still writing it to the device, so audio is very much still coming out of the speaker. With streaming TTS it is worse: the queue legitimately hits zero between chunks arriving from the provider, so join() can return mid-sentence.

If you open the microphone at that point, the recorder captures the assistant's own voice, it gets transcribed as the user's next turn, and the conversation talks to itself:

you: "Hello, are you there?"
bot: Yes — I'm here and ready to help. What can I do for you?
you: "Yes, I'm here and ready to help. What can I do for you?"

Use one player per speaking phase instead. Leaving the async with block calls close(), which pushes a sentinel, lets the worker finish every queued write, and joins the thread — that genuinely waits for the audio to finish.

async def main() -> None:
    # a player scoped to this turn — exiting drains it completely
    async with SoundDevicePlayer() as player:
        reply = await pipeline.ask(voice_input, stream=player.stream)
        print(reply.body)

    # only now is the speaker silent, so it is safe to record
    voice_input = recorder.record(duration=5)

This is echo suppression by turn-taking, not true echo cancellation

Even with the microphone opened only after playback ends, a loud speaker plus a sensitive mic can still bleed in. There is no AEC in ag2.live — use headphones when testing a speaker-and-mic loop. For genuine overlap handling, LiveAgent has server-side VAD and barge-in.

Mixing providers#

This is an example of how one can use openai STT and elevenlabs TTS together.

import asyncio

from ag2 import Agent, config
from ag2.events import TranscriptionChunkEvent
from ag2.live import (
    ElevenLabsStreamingTTSConfig,
    OpenAITranscriber,
    SoundDevicePlayer,
    SoundDeviceRecorder,
    TTSObserver,
)

SAMPLE_RATE = 24000  # pcm_24000, what the player and recorder use

agent = Agent(
    "assistant",
    prompt="You are a helpful voice assistant. Keep answers to a few sentences.",
    config=config.OpenAIConfig("gpt-5-mini", streaming=True),
    observers=[TTSObserver(ElevenLabsStreamingTTSConfig("eleven_flash_v2_5"))],
)

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

    # OpenAI streams the transcript back as deltas — print them as live captions.
    async def on_caption(event: TranscriptionChunkEvent) -> None:
        print(event.content, end="", flush=True)

    print("Say something...")
    voice_input = recorder.record(duration=5)

    async with SoundDevicePlayer() as player:
        player.stream.where(TranscriptionChunkEvent).subscribe(on_caption)
        reply = await pipeline.ask(voice_input, stream=player.stream)
        print(f"\n{reply.body}")

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

Both halves share one stream, so TranscriptionChunkEvent (from the transcriber) and SynthesizedAudioEvent (from the observer) land in the same place — the player only subscribes to the latter, leaving the captions to your own subscriber.

Examples#

examples/stt_tts.py/stt_tts.py — the OpenAI round-trip
examples/stt_tts.py/elevenlabs_stt_tts_streaming.py — streaming TTS, prints time-to-first-byte
examples/stt_tts.py/elevenlabs_stt_tts_sync.py — buffered TTS with eleven_v3

Run the last two back to back to hear the difference on the first sentence.

What's next#

LiveAgent — drop the turn-by-turn round-trip in favor of a streaming, full-duplex realtime session. CascadeConfig keeps the mix-and-match providers from this page while giving you continuous audio and barge-in.
Observers — TTSObserver is one of many observer patterns; see the harness docs for logging, persistence, and custom observers.