Tracing and Evaluating Voice Applications

• How a single OpenAIAgentsInstrumentor().instrument(...) call auto-traces an agents.realtime.RealtimeSession — capturing audio, transcripts, token counts, and tool calls without any manual span code
• The span tree the instrumentor produces (one AUDIO parent per conversational turn, with USER, LLM, and TOOL children)
• The Agents-SDK event stream — RealtimeAudio, RealtimeAudioInterrupted, RealtimeAgentEndEvent, RealtimeToolStart — and how to drive a live microphone + speaker loop on top of it
• How to evaluate the captured audio with an audio-aware OpenAI model and log the results back to Arize

​

Initial setup

​

Create a project directory and virtual environment

​

Install libraries

pip install openai openai-agents openinference-instrumentation-openai-agents \
    arize-otel arize sounddevice numpy pandas

# Linux only:
sudo apt-get install -y libportaudio2   # apt-based distros
sudo dnf install -y portaudio           # dnf-based distros

​

Set environment variables

export ARIZE_SPACE_ID="your-arize-space-id"
export ARIZE_API_KEY="your-arize-api-key"
export OPENAI_API_KEY="your-openai-api-key"

​

What you’ll see in Arize

AUDIO  "conversation.turn"     ← parent; aggregated transcripts, llm.model_name, llm.invocation_parameters
├─ USER "user"                 ← input.audio.url, input.audio.transcript
├─ LLM  "assistant"            ← output.audio.url, output.audio.transcript, token counts, time_to_first_token_ms
│  └─ TOOL "<tool_name>"        ← one per function call within the turn
└─ ...                          ← additional siblings for split input or tool round-trips

• The parent span uses the AUDIO span kind — an instrumentor-introduced extension used for the conversational-turn parent. Arize AX renders it with audio-aware UI: a play button, waveform, and the input/output transcripts inline.
• input.audio.url and output.audio.url carry the captured audio as inline data:audio/wav;base64,… URIs — the bytes ride on the span itself, so no separate audio storage is needed.
• time_to_first_token_ms on the assistant span is the latency from the user’s audio commit to the first byte of the model’s audio response — the metric that matters most for perceived voice-agent responsiveness.
• Token counts include audio-specific breakdowns llm.token_count.prompt_details.audio and llm.token_count.completion_details.audio, so you can see audio vs text token cost separately.

​

Defining the voice agent

​

Tools

from datetime import datetime
from agents import function_tool


@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}."

​

The agent

from agents.realtime import RealtimeAgent

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],
)

​

Audio plumbing

import time

import numpy as np


def make_mic_callback(mic_queue, loop, recording=None):
    """Build a sounddevice InputStream callback that forwards mic chunks into an asyncio queue."""
    def cb(indata, frames, _time, status):
        if status:
            print(f"Mic: {status}")
        # `indata` is a view onto a buffer sounddevice reuses on the next callback,
        # so we copy before handing the bytes to a different thread.
        chunk = indata.copy().tobytes()
        if recording is not None:
            recording.append((time.time_ns(), chunk))
        loop.call_soon_threadsafe(mic_queue.put_nowait, chunk)
    return cb

​

Setting up tracing

import os
from arize.otel import register
from openinference.instrumentation.openai_agents import OpenAIAgentsInstrumentor

# Raise the inline base64 cap to ~30 s of audio (default is ~0.5 s).
# Must be set BEFORE .instrument() — the instrumentor reads it at patch time.
os.environ["OPENINFERENCE_BASE64_AUDIO_MAX_LENGTH"] = "2000000"

PROJECT_NAME = "openai-realtime-voice"

tracer_provider = register(
    space_id=os.environ["ARIZE_SPACE_ID"],
    api_key=os.environ["ARIZE_API_KEY"],
    project_name=PROJECT_NAME,
)

OpenAIAgentsInstrumentor().instrument(tracer_provider=tracer_provider)

​

Running a voice session

• send_mic — drains the mic queue and forwards chunks to the session via session.send_audio(...)
• handle_events — consumes the session’s event stream and dispatches on event type
• hard_timer — sets stop after MAX_SESSION_SECONDS as a safety net

import asyncio

import sounddevice as sd
from agents.realtime import RealtimeRunner
from agents.realtime.events import (
    RealtimeAgentEndEvent,
    RealtimeAgentStartEvent,
    RealtimeAudio,
    RealtimeAudioInterrupted,
    RealtimeToolStart,
)

SAMPLE_RATE = 24_000
CHANNELS = 1
MIC_CHUNK_FRAMES = 1_200
MAX_SESSION_SECONDS = 60
EXIT_GRACE_PERIOD = 1.5

# Captured mic chunks with ns-precision timestamps. Populated by the mic
# callback during the session, then sliced by the eval step to extract
# per-USER-span audio.
mic_recording: list[tuple[int, bytes]] = []


async def run_session():
    loop = asyncio.get_running_loop()
    mic_queue: asyncio.Queue = asyncio.Queue(maxsize=100)
    stop = asyncio.Event()
    mic_recording.clear()

    async def hard_timer():
        await asyncio.sleep(MAX_SESSION_SECONDS)
        stop.set()

    runner = RealtimeRunner(agent)
    async with await runner.run() as session:
        print("Connected. Speak now — session will end once the assistant finishes its turn.\n")

        async def send_mic():
            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

        timer_task = asyncio.create_task(hard_timer())
        send_task = asyncio.create_task(send_mic())
        events_task = None

        try:
            with sd.InputStream(
                samplerate=SAMPLE_RATE,
                channels=CHANNELS,
                dtype="int16",
                blocksize=MIC_CHUNK_FRAMES,
                callback=make_mic_callback(mic_queue, loop, recording=mic_recording),
            ), sd.OutputStream(
                samplerate=SAMPLE_RATE,
                channels=CHANNELS,
                dtype="int16",
                blocksize=MIC_CHUNK_FRAMES,
            ) as out_stream:

                async def handle_events():
                    exit_task = None

                    async def maybe_exit():
                        try:
                            await asyncio.sleep(EXIT_GRACE_PERIOD)
                            await loop.run_in_executor(None, out_stream.stop)
                            stop.set()
                        except asyncio.CancelledError:
                            pass

                    def cancel_pending_exit():
                        nonlocal exit_task
                        if exit_task is not None and not exit_task.done():
                            exit_task.cancel()
                        exit_task = None

                    async for event in session:
                        if stop.is_set():
                            break
                        if isinstance(event, RealtimeAudio):
                            arr = np.frombuffer(event.audio.data, dtype=np.int16)
                            await loop.run_in_executor(None, out_stream.write, arr)
                        elif isinstance(event, RealtimeAudioInterrupted):
                            cancel_pending_exit()
                            await loop.run_in_executor(None, out_stream.abort)
                            await loop.run_in_executor(None, out_stream.start)
                            print("[interrupted]")
                        elif isinstance(event, (RealtimeAgentStartEvent, RealtimeToolStart)):
                            cancel_pending_exit()
                        elif isinstance(event, RealtimeAgentEndEvent):
                            cancel_pending_exit()
                            exit_task = asyncio.create_task(maybe_exit())

                events_task = asyncio.create_task(handle_events())
                await stop.wait()
        except KeyboardInterrupt:
            print("\nInterrupted.")
            stop.set()
        finally:
            pending = [t for t in (timer_task, send_task, events_task) if t is not None]
            for t in pending:
                t.cancel()
            await asyncio.gather(*pending, return_exceptions=True)

    tracer_provider.force_flush()
    print("Session ended. Traces flushed to Arize.")


asyncio.run(run_session())

• “What’s the weather in London?” — exercises a tool call (get_weather)
• “What time is it in Tokyo?” — exercises the other tool (get_current_time)
• “Tell me a fun fact about Python” — no tool call; tests the simpler one-response path

​

See your traces in Arize

• Audio playback — click the play button on the user and assistant spans. The audio is served from the inline data:audio/wav;base64,... URI on the span attribute.
• Transcripts — the input.audio.transcript and output.audio.transcript attributes show what the model heard and what it said. Useful for debugging mishearings.
• Tool calls — child TOOL spans under the assistant span show the function name, arguments JSON, and the value the function returned.
• Latency — time_to_first_token_ms on the assistant span gives the user-perceived “how fast did the agent start talking” latency.
• Audio token cost — llm.token_count.prompt_details.audio and llm.token_count.completion_details.audio break out audio tokens separately from text tokens.

​

Audio redaction

• OPENINFERENCE_HIDE_INPUT_AUDIO=true — drop input.audio.* from USER spans
• OPENINFERENCE_HIDE_OUTPUT_AUDIO=true — drop output.audio.* from LLM spans
• OPENINFERENCE_BASE64_AUDIO_MAX_LENGTH=<n> — cap the inline base64 payload length (default 32000)

​

Evaluating the voice session

1. Export the USER spans from Arize — we only need them for span IDs and time bounds, not the audio bytes
2. Slice mic_recording between each USER span’s start_time and end_time, wrap the raw PCM in a WAV header
3. Classify the resulting audio with gpt-audio-mini (OpenAI’s audio-input chat model)
4. Log the evaluations back to Arize via client.spans.update_evaluations(...)

​

Export the USER spans from Arize

import os
from datetime import datetime, timedelta, timezone

import pandas as pd
from arize import ArizeClient

PROJECT_NAME = "openai-realtime-voice"

arize_client = ArizeClient(api_key=os.environ["ARIZE_API_KEY"])

# 1-hour lookback covers the session you just ran plus a generous margin for
# Arize's ingestion latency. Narrow the window if you have a noisy project.
end_time = datetime.now(timezone.utc)
start_time = end_time - timedelta(hours=1)

traces_df = arize_client.spans.export_to_df(
    space_id=os.environ["ARIZE_SPACE_ID"],
    project_name=PROJECT_NAME,
    start_time=start_time,
    end_time=end_time,
)
print(f"Exported {len(traces_df)} spans")

# USER spans are the per-utterance children of each AUDIO turn. Their
# start_time / end_time bound the audio we want to evaluate.
user_spans = traces_df[
    traces_df.get("attributes.openinference.span.kind") == "USER"
].copy()
print(f"{len(user_spans)} user spans to evaluate")

​

Slice the user audio out of the local recording

import base64
import io
import wave


def _ts_to_ns(ts) -> int:
    """Convert a pandas/datetime/string timestamp to ns since the Unix epoch."""
    return int(pd.Timestamp(ts).value)


def _wrap_pcm_as_wav_b64(pcm_bytes: bytes) -> str:
    """Wrap raw PCM16 mono 24 kHz bytes in a WAV header and return base64."""
    buf = io.BytesIO()
    with wave.open(buf, "wb") as wav:
        wav.setnchannels(CHANNELS)
        wav.setsampwidth(2)  # int16
        wav.setframerate(SAMPLE_RATE)
        wav.writeframes(pcm_bytes)
    return base64.b64encode(buf.getvalue()).decode("utf-8")


def extract_audio_b64(span, recording=mic_recording) -> str | None:
    """Slice the recording by the span's [start_time, end_time] and return WAV base64."""
    start_ns = _ts_to_ns(span["start_time"])
    end_ns = _ts_to_ns(span["end_time"])
    pcm = b"".join(chunk for ts, chunk in recording if start_ns <= ts <= end_ns)
    if not pcm:
        return None
    return _wrap_pcm_as_wav_b64(pcm)

​

Define the tone classifier

1. Task description — a system prompt that tells the model what to listen for and how to reply
2. Rails — the fixed set of valid labels (positive / neutral / negative); the prompt forbids anything else
3. Output format — a small JSON object with label and explanation, parsed back into the score columns Arize expects

import json
import re

from openai import OpenAI

openai_client = OpenAI()

# Three-way tone classification. The model is constrained to these labels by
# the prompt; the score mapping turns each label into a numeric value Arize
# can sort and chart on (any monotonic mapping works — 0 / 0.5 / 1 is just
# convenient for filtering "tone went negative" in the trace list).
TONE_RAILS = ["positive", "neutral", "negative"]
TONE_SCORES = {"positive": 1.0, "neutral": 0.5, "negative": 0.0}

TONE_SYSTEM_PROMPT = (
    "You are an evaluator that listens to a user's voice clip and classifies the tone of voice. "
    f"Reply with one of exactly these labels: {', '.join(TONE_RAILS)}. "
    "Respond with ONLY a JSON object on a single line, with two keys: "
    '"label" (one of the labels above) and "explanation" (a one-sentence reason). '
    "Do not wrap the JSON in markdown, do not include any preamble, do not add any other text."
)


def _extract_json_object(text: str) -> dict:
    """Pull the first {...} object out of the model's response and parse it.

    Audio models don't support `response_format={"type": "json_object"}`, so we
    rely on prompt instructions and a lenient extractor that survives minor
    deviations like the model wrapping its reply in markdown fences.
    """
    match = re.search(r"\{.*?\}", text, re.DOTALL)
    if not match:
        raise ValueError(f"No JSON object in classifier response: {text!r}")
    return json.loads(match.group(0))


def classify_tone(audio_b64: str) -> dict:
    """Classify the tone of one user utterance. Returns {label, score, explanation}."""
    response = openai_client.chat.completions.create(
        model="gpt-audio-mini",
        # We want text out (the JSON label), not synthesized audio.
        modalities=["text"],
        messages=[
            {"role": "system", "content": TONE_SYSTEM_PROMPT},
            {"role": "user", "content": [
                {"type": "text", "text": "Classify the tone of this audio clip:"},
                # `input_audio` content blocks accept base64-encoded WAV directly.
                {"type": "input_audio", "input_audio": {"data": audio_b64, "format": "wav"}},
            ]},
        ],
    )
    parsed = _extract_json_object(response.choices[0].message.content)
    label = parsed.get("label", "neutral").lower()
    if label not in TONE_RAILS:
        # Model went off-rails — snap to neutral so the eval row still lands.
        label = "neutral"
    return {
        "label": label,
        "score": TONE_SCORES[label],
        "explanation": parsed.get("explanation", ""),
    }

​

Run the evaluation and log results back to Arize

• context.span_id — the span the eval attaches to
• eval.<name>.label — string label (one of the rails)
• eval.<name>.score — numeric score
• eval.<name>.explanation — free-text justification from the model

# Build one row per USER span in the shape Arize's `update_evaluations`
# requires: `context.span_id` plus `eval.<name>.{label,score,explanation}` columns.
rows = []
for _, span in user_spans.iterrows():
    audio_b64 = extract_audio_b64(span)
    if audio_b64 is None:
        print(f"  skipping {span['context.span_id']}: no mic audio in span window")
        continue
    result = classify_tone(audio_b64)
    rows.append({
        "context.span_id": span["context.span_id"],
        "eval.tone.label": result["label"],
        "eval.tone.score": result["score"],
        "eval.tone.explanation": result["explanation"],
    })

evals_df = pd.DataFrame(rows)

arize_client.spans.update_evaluations(
    space_id=os.environ["ARIZE_SPACE_ID"],
    project_name=PROJECT_NAME,
    dataframe=evals_df,
)
print(f"Logged {len(evals_df)} tone evaluations to Arize.")

​

Review the evaluation in Arize

​

Reference

​

Key Realtime API events the instrumentor listens for

1. Session events
   • session.created — the session was opened
   • session.updated — session parameters changed (model, instructions, tools, VAD config)
2. Audio input events
   • input_audio_buffer.speech_started — server-side VAD detected the user beginning to speak
   • input_audio_buffer.speech_stopped — server-side VAD detected the user finishing
   • input_audio_buffer.committed — audio buffer committed for processing
3. Conversation events
   • conversation.item.created — a new conversation item (user message, function call, etc.) was added
4. Response events
   • response.created — the model has started generating a response (becomes RealtimeAgentStartEvent at the SDK layer)
   • response.audio_transcript.delta — incremental transcript of the audio response
   • response.audio_transcript.done — transcript complete
   • response.audio.delta — output audio bytes (become RealtimeAudio events)
   • response.done — response finished (becomes RealtimeAgentEndEvent); may include function_call output items
5. Error events
   • error — any error encountered during processing

​

Semantic conventions

1. Session attributes
   • session.id — unique identifier for the session
2. Audio attributes
   • input.audio.url — URL of the input audio (a data:audio/wav;base64,... URI by default, or any HTTPS URL if you’ve wired a custom storage backend)
   • input.audio.mime_type — MIME type of the input audio (e.g. audio/wav)
   • input.audio.transcript — transcript of the input audio
   • output.audio.url — URL of the output audio
   • output.audio.mime_type — MIME type of the output audio
   • output.audio.transcript — transcript of the output audio
3. Span kind
   • openinference.span.kind — AUDIO for turn parents, USER for user inputs, LLM for assistant responses, TOOL for tool calls. See Voice Extensions for AUDIO and USER (instrumentor-introduced extensions) and the canonical span-kinds reference for LLM and TOOL.
4. LLM attributes
   • llm.model_name — Realtime model used (e.g. gpt-4o-realtime-preview)
   • llm.invocation_parameters — session config as JSON
   • llm.token_count.prompt_details.audio — audio input tokens consumed
   • llm.token_count.completion_details.audio — audio output tokens generated
   • time_to_first_token_ms — latency from input commit to first response byte
5. Tool attributes
   • tool.name — name of the called function
   • tool.arguments — JSON of the call arguments
   • Output value populated on the TOOL span when the function returns
6. Error attributes
   • error.type — class of error
   • error.message — error detail

​

Implementation considerations

• Audio length: raise OPENINFERENCE_BASE64_AUDIO_MAX_LENGTH to whatever your longest expected turn needs. Truncated base64 produces a broken WAV that the trace UI cannot play and the evaluator cannot decode. If your turns are long enough to bump the cap regularly, swap the inline data URI for an HTTPS URL pointing at your storage of choice (S3, GCS, your own CDN) using a custom SpanProcessor or wrapping SpanExporter.
• Cost of audio tokens: audio input and output tokens are billed at a higher rate than text tokens. Track llm.token_count.prompt_details.audio and llm.token_count.completion_details.audio on the assistant span as you would any cost metric.
• Redaction: if your traces will leave a controlled environment, set OPENINFERENCE_HIDE_INPUT_AUDIO=true and OPENINFERENCE_HIDE_OUTPUT_AUDIO=true. Transcripts and durations stay; the raw audio attributes are dropped before the span ships.
• Voice evaluations are still primarily an OpenAI capability: gpt-audio-mini is the audio-capable model used in the eval step. Other providers’ audio models can be substituted with the same dataframe-shaping pattern; the Arize update_evaluations half doesn’t care which LLM produced the labels.
• Replay-friendly storage: if you swap to external storage, key your audio objects by <timestamp>_<trace_id>_<span_id>_<input|output>.wav (or a similar scheme involving the trace + span IDs). That makes each clip traceable back to the span that produced it — useful when you want to spot-check a particular trace months later.

​

Read more

• Companion notebook on GitHub — the full runnable version of this guide
• OpenInference OpenAI Agents instrumentor — source, changelog, and the canonical realtime_with_tools.py example
• OpenAI Agents SDK realtime docs — RealtimeAgent, RealtimeRunner, and the full event reference
• OpenAI Realtime API reference — the underlying WebSocket protocol the SDK speaks
• OpenAI audio input guide — the gpt-audio / gpt-audio-mini API used by the evaluator
• OpenInference semantic conventions — the formal definitions for span kinds and audio attributes