"""realtime_with_tools.py — Realtime voice agent with tool calls. Local mic/speaker setup with two function tools (get_weather, get_current_time). Tool calls produce TOOL child spans nested under the LLM span in the trace. Traces are sent to Arize AX. Set OPENAI_API_KEY, ARIZE_SPACE_ID, and ARIZE_API_KEY in the environment (or replace the placeholders below) before running. Requirements: see examples/requirements.txt Usage: python realtime_with_tools.py # Try: "What's the weather in London?" or "What time is it in Tokyo?" # With verbose instrumentation logging: python realtime_with_tools.py --debug """ from __future__ import annotations import argparse import asyncio import collections import logging import os import signal import sys from datetime import datetime import numpy as np import sounddevice as sd from agents import function_tool # type: ignore[import] from agents.realtime import RealtimeAgent, RealtimeRunner # type: ignore[import] from agents.realtime.events import ( # type: ignore[import] RealtimeAudio, RealtimeAudioInterrupted, ) from arize.otel import register as register_arize from openinference.instrumentation.openai_agents import OpenAIAgentsInstrumentor # Set these before running (export them or replace the placeholders). os.environ.setdefault("OPENAI_API_KEY", "YOUR_OPENAI_API_KEY") os.environ.setdefault("ARIZE_SPACE_ID", "YOUR_ARIZE_SPACE_ID") os.environ.setdefault("ARIZE_API_KEY", "YOUR_ARIZE_API_KEY") logging.basicConfig(level=logging.WARNING) logger = logging.getLogger(__name__) SAMPLE_RATE = 24_000 CHANNELS = 1 MIC_CHUNK_FRAMES = 1_200 def setup_tracer_provider(): # type: ignore[no-untyped-def] """Register an Arize AX tracer provider.""" project_name = os.getenv("PROJECT_NAME", "realtime-with-tools") kwargs: dict = { "space_id": os.environ["ARIZE_SPACE_ID"], "api_key": os.environ["ARIZE_API_KEY"], "project_name": project_name, } endpoint = os.getenv("ARIZE_ENDPOINT") if endpoint: kwargs["endpoint"] = endpoint print(f"Traces → Arize AX (project={project_name})") return register_arize(**kwargs) # --------------------------------------------------------------------------- # Tools # --------------------------------------------------------------------------- @function_tool def get_weather(location: str, unit: str = "fahrenheit") -> str: """Get the current weather for a location.""" if unit == "celsius": return f"The weather in {location} is 22 °C and sunny." return f"The weather in {location} is 72 °F and sunny." @function_tool def get_current_time(timezone: str = "UTC") -> str: """Get the current time for a timezone.""" now = datetime.now().strftime("%I:%M %p") return f"The current time in {timezone} is {now}." # --------------------------------------------------------------------------- # Audio plumbing # --------------------------------------------------------------------------- def _make_mic_callback( mic_queue: asyncio.Queue[bytes], loop: asyncio.AbstractEventLoop, ) -> object: def _cb(indata: np.ndarray, frames: int, _time, status) -> None: if status: logger.warning("Mic: %s", status) loop.call_soon_threadsafe(mic_queue.put_nowait, indata.copy().tobytes()) return _cb def _make_speaker_callback(speaker_buf: collections.deque[bytes]) -> object: def _cb(outdata: np.ndarray, frames: int, _time, status) -> None: needed = frames * CHANNELS * 2 buf = bytearray(needed) pos = 0 while pos < needed and speaker_buf: chunk = speaker_buf.popleft() take = min(len(chunk), needed - pos) buf[pos : pos + take] = chunk[:take] if take < len(chunk): speaker_buf.appendleft(chunk[take:]) pos += take outdata[:] = np.frombuffer(bytes(buf), dtype=np.int16).reshape(-1, CHANNELS) return _cb # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- async def run(args: argparse.Namespace) -> None: if args.debug: logging.getLogger().setLevel(logging.DEBUG) logging.getLogger("openinference.instrumentation.openai_agents").setLevel(logging.DEBUG) logger.setLevel(logging.DEBUG) print("Debug logging enabled (instrumentation + exporter)") provider = setup_tracer_provider() OpenAIAgentsInstrumentor().instrument(tracer_provider=provider) loop = asyncio.get_running_loop() mic_queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=100) speaker_buf: collections.deque[bytes] = collections.deque() agent = RealtimeAgent( name="Assistant", instructions=( "You are a helpful voice assistant. " "You have tools to look up weather and the current time — use them when asked. " "Keep responses short and conversational." ), tools=[get_weather, get_current_time], ) stop = asyncio.Event() loop.add_signal_handler(signal.SIGINT, stop.set) print("Connecting to OpenAI Realtime… (Ctrl-C to stop)") runner = RealtimeRunner(agent) async with await runner.run() as session: print("Connected. Speak now.\n") print(" Try: 'What's the weather in London?'") print(" Try: 'What time is it in Tokyo?'\n") print(" (Type a message and press Enter to send text input.)\n") async def _send_mic() -> None: while not stop.is_set(): try: chunk = await asyncio.wait_for(mic_queue.get(), timeout=0.1) await session.send_audio(chunk) except asyncio.TimeoutError: continue async def _handle_events() -> None: async for event in session: if stop.is_set(): break if isinstance(event, RealtimeAudio): speaker_buf.append(event.audio.data) elif isinstance(event, RealtimeAudioInterrupted): speaker_buf.clear() print("[interrupted]") async def _stdin_reader() -> None: """Read typed lines from stdin and inject them as user text messages.""" while not stop.is_set(): try: line = await loop.run_in_executor(None, sys.stdin.readline) except Exception: return if not line: # EOF return text = line.strip() if not text: continue print(f"[text→] {text}") try: await session.send_message(text) # type: ignore[attr-defined] except Exception as exc: logger.warning("send_message failed: %s", exc) send_task = asyncio.create_task(_send_mic()) events_task = asyncio.create_task(_handle_events()) stdin_task = asyncio.create_task(_stdin_reader()) async def _wait_and_teardown() -> None: await stop.wait() print("\nStopping…") send_task.cancel() events_task.cancel() stdin_task.cancel() await asyncio.gather(send_task, events_task, stdin_task, return_exceptions=True) with sd.InputStream( samplerate=SAMPLE_RATE, channels=CHANNELS, dtype="int16", blocksize=MIC_CHUNK_FRAMES, callback=_make_mic_callback(mic_queue, loop), ), sd.OutputStream( samplerate=SAMPLE_RATE, channels=CHANNELS, dtype="int16", blocksize=MIC_CHUNK_FRAMES, callback=_make_speaker_callback(speaker_buf), ): await _wait_and_teardown() provider.force_flush() provider.shutdown() print("Done. Traces flushed.") def main() -> None: parser = argparse.ArgumentParser( description="Realtime voice agent with tool calls and OpenInference tracing.", ) parser.add_argument( "--debug", action="store_true", help="Verbose debug logging for instrumentation and span export.", ) asyncio.run(run(parser.parse_args())) if __name__ == "__main__": try: main() except KeyboardInterrupt: sys.exit(0)