Skip to content
Docs Pydantic Docs

Voice Assistant

Example of a voice assistant built on a realtime speech-to-speech model: it streams your microphone to OpenAI’s gpt-realtime model and plays the model’s spoken replies back through your speakers. Talk to it — and try interrupting while it’s speaking: the model stops and listens (barge-in).

Demonstrates:

The agent exposes a single get_weather tool the model can call mid-conversation, and the terminal shows a running transcript of both sides of the conversation plus any tool calls.

Audio I/O runs on listentome, whose microphone is an async iterator that send_audio() consumes directly, and whose speaker write() suspends until the device has played each chunk from stream_audio(). Both audio directions stay bounded rather than growing without limit: the microphone stream and the session’s audio buffer each drop their oldest blocks if their consumer falls behind, so a machine that stutters glitches instead of ending the call.

Barge-in costs the example no code at all: because playback is a single device-paced stream_audio() loop, the session can track the playback position itself, so handle_barge_in=True does the local half of it — dropping the buffered audio the user will never hear, truncating the provider’s transcript to what was really heard, and staying out of the way on an ordinary turn where the previous reply was heard in full. The one thing it can’t reach is the block already inside the speaker, so up to a chunk of stale audio finishes playing. Playback loops the session can’t follow, and triggers you’d rather own yourself, take the manual paths in the barge-in guide instead.

Running the Example

The example’s dependencies include listentome for microphone and speaker access. It also requires the PortAudio system library: brew install portaudio on macOS, apt install libportaudio2 on Debian/Ubuntu.

The realtime model runs on gpt-realtime, so you’ll need an OpenAI API key set via OPENAI_API_KEY.

With dependencies installed and environment variables set, run:

Terminal
python -m pydantic_ai_examples.realtime_voice

Example Code

realtime_voice.py
from __future__ import annotations

import anyio
import listentome
import logfire

from pydantic_ai import (
    Agent,
    FunctionToolCallEvent,
    FunctionToolResultEvent,
    PartEndEvent,
    SpeechPart,
)
from pydantic_ai.realtime import RealtimeSession

# 'if-token-present' means nothing will be sent (and the example will work) if you don't have logfire configured
logfire.configure(send_to_logfire='if-token-present')
logfire.instrument_pydantic_ai()

agent = Agent(
    instructions='You are a friendly voice assistant. Keep your replies short and conversational.'
)


@agent.tool_plain
def get_weather(city: str) -> str:
    """Look up the current weather in a city."""
    return f'It is currently 21 degrees and sunny in {city}.'


async def conversation(session: RealtimeSession) -> None:
    """Wire the microphone and speaker to the session and run the conversation."""
    # Capture and play at the rates this model expects; they can differ per direction.
    mic = listentome.InputStream(
        samplerate=session.audio_input_sample_rate,
        channels=1,
        dtype='int16',
        blocksize=session.audio_input_sample_rate // 10,  # 100 ms per block
    )
    speaker = listentome.OutputStream(
        samplerate=session.audio_output_sample_rate, channels=1, dtype='int16'
    )

    async with mic, speaker, anyio.create_task_group() as tg:
        # The microphone is an async iterator of PCM blocks; `send_audio` forwards them
        # all. If the network falls behind, the stream drops its oldest blocks rather
        # than letting latency grow without bound.
        tg.start_soon(session.send_audio, mic)

        # `write()` returns once the device has consumed a chunk, so playback advances at
        # speaker pace while the model runs ahead; `stream_audio()`'s own buffer bounds
        # the backlog, dropping its oldest chunks if playback falls too far behind, so a
        # machine that stutters glitches instead of ending the call. Pulling the next
        # chunk only after the device consumed the previous one also lets the session
        # track the playback position itself, which is what `handle_barge_in=True` uses
        # to handle interruptions without any code here.
        async def play_audio() -> None:
            async for chunk in session.stream_audio():
                await speaker.write(chunk)

        tg.start_soon(play_audio)

        print('Listening — start talking (Ctrl-C to quit).')
        async for event in session:
            match event:
                case PartEndEvent(part=SpeechPart() as part) if part.transcript:
                    print(f'{part.speaker}: {part.transcript}')
                case FunctionToolCallEvent(part=call):
                    print(f'[calling {call.tool_name}]')
                case FunctionToolResultEvent(part=result):
                    print(f'[{result.tool_name} returned: {result.content}]')
                case _:
                    pass
        tg.cancel_scope.cancel()


async def main():
    # The session opens before the microphone starts capturing, so no audio from before
    # the conversation began is queued up and sent to the model as stale input. With
    # `handle_barge_in=True`, interrupting the model mid-sentence is handled by the
    # session itself: it stops playback of the rest of the reply and truncates the
    # provider's transcript to what was actually heard.
    realtime = agent.realtime('openai:gpt-realtime')
    async with realtime.session(handle_barge_in=True) as session:
        await conversation(session)


if __name__ == '__main__':
    try:
        anyio.run(main)
    except KeyboardInterrupt:
        pass