AudioAgent

pip install "10xgraph[realtime]"

export GEMINI_API_KEY=your-api-key

export GOOGLE_GENAI_USE_VERTEXAI=1
# Use standard Application Default Credentials (gcloud auth application-default login)

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, helpful voice assistant."}],
).compile()

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

    async for event in app.arealtime(queue, {"thread_id": "demo-1"}):
        # Print model output as transcription completes each sentence
        if event.type == "output_transcript" and event.finished:
            print(f"Agent: {event.text}")
        elif event.type == "turn_complete":
            # Session finished; close the queue
            queue.close()

    await app.aclose()

asyncio.run(main())

Agent: Hello! I'm here to help with any questions or tasks you need assistance with. What would you like to know or do?

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:
    """Get the current weather for a city."""
    # In practice, call a real weather API
    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 the weather tool to answer questions about the weather.",
    }],
    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"}):
        # Observe tool calls for debugging
        if event.type == "tool_call":
            print(f"[Tool] Calling {event.name}({event.args})")
        elif event.type == "tool_result":
            print(f"[Tool] Result: {event.result}")
        # Listen for the agent's final answer
        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())

[Tool] Calling get_weather({'city': 'Tokyo'})
[Tool] Result: Sunny, 24°C in Tokyo
Agent: The weather in Tokyo is sunny with a temperature of 24 degrees Celsius. It's a beautiful day there!

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"

# For production, use PgCheckpointer (needs postgres_dsn, redis_url and a user_id in config)
checkpointer = InMemoryCheckpointer()

app = AudioAgent(
    MODEL,
    realtime_config=RealtimeConfig(model=MODEL, voice="Puck"),
    system_prompt=[{"role": "system", "content": "Remember important details the user tells you."}],
).compile(checkpointer=checkpointer)

async def main():
    # First session: establish context
    print("=== Session 1 ===")
    queue = LiveInputQueue()
    queue.send_text("My name is Alex and I work in Berlin.")

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

    await app.aclose()

    # Second session: resume with the saved context
    print("\n=== Session 2 (reconnect) ===")
    app2 = AudioAgent(
        MODEL,
        realtime_config=RealtimeConfig(model=MODEL, voice="Puck"),
        system_prompt=[{"role": "system", "content": "Remember important details the user tells you."}],
    ).compile(checkpointer=checkpointer)

    queue = LiveInputQueue()
    queue.send_text("Remind me where I work.")

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

    await app2.aclose()

asyncio.run(main())

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).compile()

async def main():
    playback_buffer = bytearray()
    queue = LiveInputQueue()
    queue.send_text("Tell me a long story.")

    async for event in app.arealtime(queue, {"thread_id": "barge-1"}):
        if event.type == "audio_delta":
            # Queue audio for playback
            playback_buffer.extend(event.data)
        elif event.type == "interrupted":
            # User spoke over the model; discard pending audio
            print("[Interrupt] Flushing audio buffer")
            playback_buffer.clear()
        elif event.type == "turn_complete":
            # Play remaining buffer and close
            print(f"[Done] Playback {len(playback_buffer)} bytes")
            queue.close()

    await app.aclose()

asyncio.run(main())

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

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

config = RealtimeConfig(
    model=MODEL,
    vad=VADConfig(enabled=False),  # Disable automatic detection
)
app = AudioAgent(MODEL, realtime_config=config).compile()

async def main():
    queue = LiveInputQueue()

    # Simulate push-to-talk: signal start of user activity
    queue.send_activity_start()

    # Send audio (e.g., from microphone or WAV file)
    # pcm_bytes = read_audio_samples()  # 16-bit PCM, 16 kHz mono
    pcm_bytes = b"\x00" * 16000  # 1 second of silence for demo
    queue.send_audio(pcm_bytes, sample_rate=16000)

    # Signal end of user activity
    queue.send_activity_end()

    async for event in app.arealtime(queue, {"thread_id": "ptt-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())