AudioAgent

A prebuilt realtime audio-to-audio agent backed by Gemini Live — mirrors ReactAgent’s construction surface while wrapping a LiveAgent as the graph root.

Import path: tenxgraph.prebuilt.agent

Concept

AudioAgent opens a persistent WebSocket to Gemini Live and runs a duplex audio session. Unlike ReactAgent, the provider owns the turn loop. 10xGraph wraps it in a graph node (LiveAgent) and exposes a separate execution path — arealtime() — that drives the session.

Session graph

flowchart LR
    START([START]) --> LIVE["LIVE\n(LiveAgent)"]
    LIVE --> END_NODE([END])

The graph is a single node. The LIVE -> END edge exists only to satisfy graph validation; it is never traversed during a realtime session because LiveAgent holds the WebSocket for the full session lifetime.

Tool loop

Tools are registered through a ToolNode at build time and advertised to the Gemini Live session at connect time. When the model requests a tool call, LiveAgent dispatches it through the same ToolNode path used by ReactAgent (parallel execution, callbacks, publisher events), then feeds the result back over the socket without interrupting the audio stream. Barge-in works normally — an interrupted event discards any in-flight tool result accumulation.

System prompt and skills

system_prompt, skills, and memory are all supported. At connect time, the agent flattens them — including any {field} placeholder interpolation from AgentState — into a single system_instruction string sent to Gemini Live. This snapshot is fixed for the session; the model cannot receive a new system prompt mid-session. Dynamic behavior after connect goes through activate_skill or memory tools.

Execution path

A graph containing a LiveAgent must be driven with arealtime() (async generator) or realtime() (sync wrapper). Calling invoke, ainvoke, stream, or astream raises a RuntimeError.

Transcripts and storage

Audio is never stored at rest. Finished speech turns are persisted as Message objects with metadata={"modality": "audio"} — both user (input transcription) and model (output transcription).

Provider requirement

LiveAgent v1 resolves the provider from the model string and requires "google". Passing any other provider string raises a ValueError at construction time.

Installation

pip install "10xgraph[realtime]"

Set your credentials:

export GEMINI_API_KEY=your-api-key

For Vertex AI, set GOOGLE_GENAI_USE_VERTEXAI=1 with standard ADC credentials.

Constructor Parameters

compile() Parameters

compile() does not accept media_store, interrupt_before, or interrupt_after. Realtime media (images, video frames) is sent directly to the model via LiveInputQueue.send_image() — it is never routed through a media store. Interrupt hooks are not applicable to the realtime execution path.

Full Code

Minimal example

import asyncio
from tenxgraph.core.realtime.base import RealtimeConfig
from tenxgraph.core.realtime.queue import LiveInputQueue
from tenxgraph.prebuilt.agent import AudioAgent

MODEL = "gemini-live-2.5-flash-preview"

app = AudioAgent(
    MODEL,
    realtime_config=RealtimeConfig(model=MODEL, voice="Puck"),
    system_prompt=[{"role": "system", "content": "You are a concise voice assistant."}],
).compile()

async def main():
    queue = LiveInputQueue()
    queue.send_text("Hello, what can you do?")

    async for event in app.arealtime(queue, {"thread_id": "demo-1"}):
        if event.type == "output_transcript" and event.finished:
            print(f"agent: {event.text}")
        elif event.type == "turn_complete":
            queue.close()

    await app.aclose()

asyncio.run(main())

With tools

import asyncio
from tenxgraph.core.realtime.base import RealtimeConfig
from tenxgraph.core.realtime.queue import LiveInputQueue
from tenxgraph.prebuilt.agent import AudioAgent

MODEL = "gemini-live-2.5-flash-preview"

def get_weather(city: str) -> str:
    """Return the current weather for a city."""
    return f"Sunny, 24°C in {city}"

app = AudioAgent(
    MODEL,
    realtime_config=RealtimeConfig(model=MODEL, voice="Aoede"),
    system_prompt=[{
        "role": "system",
        "content": "You are a helpful voice assistant. Use tools when they help you answer.",
    }],
    tools=[get_weather],
).compile()

async def main():
    queue = LiveInputQueue()
    queue.send_text("What's the weather in Tokyo?")

    async for event in app.arealtime(queue, {"thread_id": "tools-1"}):
        if event.type == "tool_call":
            print(f"calling tool: {event.name}({event.args})")
        elif event.type == "tool_result":
            print(f"tool result: {event.result}")
        elif event.type == "output_transcript" and event.finished:
            print(f"agent: {event.text}")
        elif event.type == "turn_complete":
            queue.close()

    await app.aclose()

asyncio.run(main())

With a checkpointer (persistent sessions)

A checkpointer stores both the transcript Message history and the session resumption handle. On reconnect, 10xGraph either resumes via the provider handle (no reseed overhead) or falls back to replaying the full transcript into the new session.

import asyncio
from tenxgraph.core.realtime.base import RealtimeConfig
from tenxgraph.core.realtime.queue import LiveInputQueue
from tenxgraph.prebuilt.agent import AudioAgent
from tenxgraph.storage.checkpointer import InMemoryCheckpointer

MODEL = "gemini-live-2.5-flash-preview"

checkpointer = InMemoryCheckpointer()

app = AudioAgent(
    MODEL,
    realtime_config=RealtimeConfig(model=MODEL, voice="Puck"),
).compile(checkpointer=checkpointer)

async def main():
    queue = LiveInputQueue()
    queue.send_text("Remember that my name is Alex.")

    async for event in app.arealtime(queue, {"thread_id": "persist-1"}):
        if event.type == "output_transcript" and event.finished:
            print(f"agent: {event.text}")
        elif event.type == "turn_complete":
            queue.close()

    await app.aclose()

asyncio.run(main())

Handling barge-in

When the user speaks over the model, the provider emits an interrupted event. Discard any audio playback buffer on your side and resume listening.

async for event in app.arealtime(queue, {"thread_id": "barge-1"}):
    if event.type == "audio_delta":
        playback_buffer.extend(event.data)
    elif event.type == "interrupted":
        playback_buffer.clear()   # flush in-flight model audio
    elif event.type == "turn_complete":
        pass  # flush and play remaining buffer

VAD and push-to-talk

Voice activity detection (VAD) is enabled by default. To switch to push-to-talk (manual activity), disable VAD and signal boundaries explicitly:

from tenxgraph.core.realtime.base import RealtimeConfig, VADConfig
from tenxgraph.core.realtime.queue import LiveInputQueue

config = RealtimeConfig(
    model="gemini-live-2.5-flash-preview",
    vad=VADConfig(enabled=False),
)
app = AudioAgent("gemini-live-2.5-flash-preview", realtime_config=config).compile()

queue = LiveInputQueue()
queue.send_activity_start()
queue.send_audio(pcm_bytes, sample_rate=16000)
queue.send_activity_end()

RealtimeConfig key fields

ReconnectConfig defaults: base_delay=0.5, max_delay=10.0, max_attempts=5. Provider-initiated go_away rotations always reconnect immediately (no backoff).

RealtimeEvent types

Further reading

• How to build a realtime audio agent — end-to-end guide covering WAV file I/O, microphone streaming, image input, and the API WebSocket bridge.
• Realtime reference — full API surface for RealtimeConfig, VADConfig, ReconnectConfig, LiveInputQueue, RealtimeClient, and all event types.