Realtime Voice Agent with Gemini Live in 10xGraph (Python)

"""Realtime audio-to-audio from a WAV file, using AudioAgent + Gemini Live.

No microphone or speaker required. This streams a 16 kHz mono PCM16 WAV into the live
session, writes the model's 24 kHz audio reply to ``out.wav``, and prints the input and
output transcripts plus any tool calls. It is the headless counterpart to
``audio_agent_mic.py`` and is the easiest way to sanity-check your setup.

Setup:  pip install "10xgraph[realtime]" and set GEMINI_API_KEY
Run:    python examples/realtime_audio/audio_agent_file.py path/to/input.wav

``input.wav`` must be mono, 16-bit PCM, 16 kHz (the format Gemini Live expects for input).
"""

import asyncio
import os
import sys
import wave

from dotenv import load_dotenv

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


load_dotenv()

# Gemini Live is the only realtime provider today. Check Google's Live API docs for the
# model names available to your key.
MODEL = "gemini-3.8-live"


def get_weather(location: str) -> str:
    """Return the current weather for a city. Called by the model during the conversation."""
    return f"It is 22 degrees Celsius and sunny in {location}."


def build_app():
    """Compile a single realtime audio agent with one tool and a voice."""
    config = RealtimeConfig(
        model=MODEL,
        voice="Puck",
        system_instruction="You are a concise voice assistant. Answer in one or two sentences.",
    )
    return AudioAgent(MODEL, realtime_config=config, tools=[get_weather]).compile()


def read_pcm16(path: str) -> tuple[int, bytes]:
    """Read a mono PCM16 WAV file, returning (sample_rate, raw_pcm_bytes)."""
    with wave.open(path, "rb") as wf:
        if wf.getsampwidth() != 2 or wf.getnchannels() != 1:
            raise ValueError("input must be mono, 16-bit PCM (sampwidth=2, channels=1)")
        return wf.getframerate(), wf.readframes(wf.getnframes())


async def feed_audio(queue: LiveInputQueue, pcm: bytes, sample_rate: int) -> None:
    """Stream the file into the session in ~100 ms chunks (the audio hot path)."""
    chunk = (sample_rate // 10) * 2  # 100 ms * 2 bytes/sample
    for offset in range(0, len(pcm), chunk):
        queue.send_audio(pcm[offset : offset + chunk], sample_rate=sample_rate)
        await asyncio.sleep(0.0)  # yield so the pump task can flush to the socket
    # Automatic VAD detects end-of-speech; we leave the queue open to receive the reply
    # and close it from the main loop once the model finishes its turn.


async def main() -> None:
    in_path = sys.argv[1] if len(sys.argv) > 1 else "input.wav"
    if not os.path.exists(in_path):
        sys.exit(f"Provide a 16 kHz mono PCM16 WAV path. Not found: {in_path}")

    sample_rate, pcm = read_pcm16(in_path)
    app = build_app()
    queue = LiveInputQueue()

    out = wave.open("out.wav", "wb")
    out.setnchannels(1)
    out.setsampwidth(2)
    out.setframerate(OUTPUT_SAMPLE_RATE)

    feeder = asyncio.create_task(feed_audio(queue, pcm, sample_rate))
    try:
        async for event in app.arealtime(queue, {"thread_id": "audio-file-demo"}):
            if event.type == "audio_delta":
                out.writeframes(event.data)
            elif event.type == "input_transcript" and event.finished:
                print(f"you:   {event.text}")
            elif event.type == "output_transcript" and event.finished:
                print(f"agent: {event.text}")
            elif event.type == "tool_call":
                print(f"[tool] {event.name}({event.args})")
            elif event.type == "turn_complete":
                queue.close()  # one turn for this demo: end the session
    finally:
        feeder.cancel()
        out.close()
        await app.aclose()

    print("Wrote model audio reply to out.wav")


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

"""Live voice weather assistant -- AudioAgent + Gemini Live, React-style tool calling.

Talk to it through your microphone and it talks back in real time. Ask "what's the
weather in Tokyo?" and the model calls the ``get_weather`` tool, then speaks the answer
it gets back -- the realtime analog of a ReactAgent's reason -> tool -> respond loop.

Features
    - Voice playback: the model's reply is played out loud on your speakers.
    - Tool calling: ``get_weather`` is advertised to the model and invoked on demand.
    - Echo-safe by default: the mic is muted while the agent is speaking, so it doesn't
      hear (and reply to) its own voice through your speakers.

Echo / feedback
    Without echo cancellation, your speaker audio leaks into the mic and the model
    transcribes its own replies as your input. This demo avoids that by muting the mic
    while the agent talks (half-duplex). Use headphones and set MIC_FULL_DUPLEX=1 for
    true full duplex with barge-in (speak over the agent to interrupt it).

Setup:  pip install "10xgraph[realtime]" sounddevice, and set GEMINI_API_KEY
        (or the Vertex AI environment variables, see README)
Run:    python examples/realtime_audio/audio_agent_mic.py
        then say e.g. "What's the weather in Paris?" and press Ctrl+C to stop.
"""

import asyncio
import contextlib
import os
import sys

from dotenv import load_dotenv

from tenxgraph.core.realtime.base import INPUT_SAMPLE_RATE, OUTPUT_SAMPLE_RATE, RealtimeConfig
from tenxgraph.core.realtime.queue import LiveInputQueue
from tenxgraph.prebuilt.agent import AudioAgent


load_dotenv()

# Use Vertex AI (service account / ADC) when GOOGLE_GENAI_USE_VERTEXAI is set; otherwise
# use a Gemini API key. Live model names can differ between the two, so check Google's
# Live API docs for what your project can use.
USE_VERTEX = os.getenv("GOOGLE_GENAI_USE_VERTEXAI", "").strip().lower() in ("1", "true", "yes")
MODEL = "gemini-3.8-live"
MIC_BLOCK = INPUT_SAMPLE_RATE // 10  # 100 ms frames
# Full duplex (no mic muting) -- only sensible with headphones, enables barge-in.
FULL_DUPLEX = os.getenv("MIC_FULL_DUPLEX", "").strip().lower() in ("1", "true", "yes")
# How long to keep the mic muted after the agent's turn ends, to let the speaker drain.
MUTE_TAIL_SEC = 0.4

# A tiny canned forecast table so different cities give different answers. Swap the body
# for a real HTTP call to a weather API and nothing else here changes.
_FORECASTS = {
    "tokyo": "18 degrees Celsius, light rain",
    "paris": "24 degrees Celsius and sunny",
    "london": "15 degrees Celsius, overcast",
    "new york": "21 degrees Celsius, partly cloudy",
    "san francisco": "17 degrees Celsius, foggy",
}


def get_weather(location: str) -> str:
    """Get the current weather for a city. Call this whenever the user asks about weather.

    Args:
        location: The city name, e.g. "Tokyo" or "Paris".
    """
    forecast = _FORECASTS.get(location.strip().lower(), "22 degrees Celsius and clear")
    print(f"  [tool] get_weather(location={location!r}) -> {forecast}")
    return f"The weather in {location} is {forecast}."


def build_app():
    config = RealtimeConfig(
        model=MODEL,
        voice="Puck",
        system_instruction=(
            "You are a friendly, concise voice assistant. When the user asks about the "
            "weather, always call the get_weather tool and answer using its result. "
            "Keep replies to one or two sentences."
        ),
    )
    return AudioAgent(
        MODEL, realtime_config=config, tools=[get_weather], use_vertex_ai=USE_VERTEX
    ).compile()


async def main() -> None:
    try:
        import sounddevice as sd
    except ImportError:
        sys.exit("This example needs sounddevice:  pip install sounddevice")

    app = build_app()
    queue = LiveInputQueue()
    loop = asyncio.get_running_loop()

    # Mic gate: while the agent is speaking we drop mic frames so its own voice (played on
    # the speaker and picked up by the mic) is never sent back as user input. Disabled in
    # full-duplex mode (headphones), where barge-in is wanted instead.
    agent_speaking = {"on": False}

    def on_mic(indata, _frames, _time, _status) -> None:
        # PortAudio calls this on its own thread; marshal onto the event loop so the
        # asyncio-backed queue is touched only from the loop thread.
        if agent_speaking["on"] and not FULL_DUPLEX:
            return  # muted while the agent talks (echo guard)
        loop.call_soon_threadsafe(queue.send_audio, bytes(indata))

    def unmute() -> None:
        agent_speaking["on"] = False

    speaker = sd.RawOutputStream(samplerate=OUTPUT_SAMPLE_RATE, channels=1, dtype="int16")
    mic = sd.RawInputStream(
        samplerate=INPUT_SAMPLE_RATE,
        channels=1,
        dtype="int16",
        blocksize=MIC_BLOCK,
        callback=on_mic,
    )
    speaker.start()
    mic.start()
    mode = "full-duplex (barge-in)" if FULL_DUPLEX else "echo-safe (mic muted while agent talks)"
    print(f"Listening [{mode}]. Try: 'What's the weather in Tokyo?'   (Ctrl+C to stop)")

    try:
        async for event in app.arealtime(queue, {"thread_id": "audio-mic-demo"}):
            if event.type == "audio_delta":
                agent_speaking["on"] = True  # mute the mic for the duration of the reply
                speaker.write(event.data)  # play the model's voice
            elif event.type == "turn_complete":
                # Reopen the mic after the speaker has drained the buffered tail.
                loop.call_later(MUTE_TAIL_SEC, unmute)
            elif event.type == "interrupted":
                # Barge-in (full-duplex only): discard audio already queued for playback.
                speaker.stop()
                speaker.start()
                agent_speaking["on"] = False
            elif event.type == "input_transcript" and event.finished:
                print(f"you:   {event.text}")
            elif event.type == "output_transcript" and event.finished:
                print(f"agent: {event.text}")
            elif event.type == "tool_call":
                print(f"  [tool-call requested] {event.name}({event.args})")
            elif event.type == "error":
                print(f"  [error] {event.message}")
    except (KeyboardInterrupt, asyncio.CancelledError):
        print("\nStopping...")
    finally:
        queue.close()
        mic.stop()
        mic.close()
        speaker.stop()
        speaker.close()
        await app.aclose()


if __name__ == "__main__":
    with contextlib.suppress(KeyboardInterrupt):
        asyncio.run(main())

"""Realtime AudioAgent exposed through the 10xGraph API server.

`10xgraph.json` points the server at `app` below. The server then offers a WebSocket at
/v1/graph/live that bridges client audio to this agent: binary PCM16 frames upstream,
model audio back as binary frames, transcripts and tool calls as JSON.

Run:  cd examples/realtime_audio && 10xgraph dev --no-open
"""

from dotenv import load_dotenv

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


load_dotenv()

# Gemini Live is the only realtime provider today. Model names change often, so check
# Google's Live API docs for what your key can use.
MODEL = "gemini-3.8-live"


def get_weather(location: str) -> str:
    """Return the current weather for a city. Called by the model during the conversation."""
    return f"It is 22 degrees Celsius and sunny in {location}."


checkpointer = InMemoryCheckpointer()

app = AudioAgent(
    MODEL,
    realtime_config=RealtimeConfig(
        model=MODEL,
        voice="Puck",
        system_instruction="You are a concise, friendly voice assistant.",
    ),
    tools=[get_weather],
).compile(checkpointer=checkpointer)

{
  "agent": "graph:app",
  "checkpointer": "graph:checkpointer",
  "env": ".env"
}