AI Engineering No. 260

Realtime Voice Agents skill for Claude Code

Answer fast, stop when interrupted, never talk over callers. Claude builds voice agents around a measured voice-to-voice budget rather than vendor demo numbers. It picks a cascaded or speech-to-speech design, tunes VAD or semantic turn detection, cancels and truncates speech on barge-in so history matches what the caller heard, covers slow tools with spoken filler, and replays a recorded-audio suite in real time to catch regressions.

  • namerealtime-voice-agents
  • size221 lines · ≈2,800 tokens
  • files2
  • licenseMIT
  • updated

What Claude sees before loading it

This description is the only part of the skill that sits in context all the time. Claude reads it to decide when to load the full playbook below.

description: Builds low-latency conversational voice agents - choosing a streaming STT to LLM to TTS pipeline or a speech-to-speech realtime API, a per-stage latency budget with voice-to-voice targets, VAD and semantic turn detection, barge-in that truncates what the caller never heard, echo cancellation, WebRTC and SIP telephony basics, spoken filler around mid-conversation tool calls, and regression testing with recorded audio. Use when building a phone or in-app voice assistant, when an agent feels laggy, cuts callers off, talks over them, or ignores interruptions, or when connecting an agent to Twilio, SIP or WebRTC.

Try asking

  • Our phone agent takes two seconds to answer and callers keep repeating themselves. Fix it
  • Build a voice assistant for appointment booking over Twilio with a tool to check availability
  • The agent keeps talking after the user interrupts. Make barge-in work properly

Use it when

  • Building a phone, kiosk or in-app conversational voice assistant
  • Callers report lag, dead air, or being cut off mid-sentence
  • Interruptions are ignored or the agent hears its own voice
  • Connecting an agent to SIP trunks, a CPaaS media stream or WebRTC

Skip it when

How to check the result

The recorded-audio suite passes its outcome assertions and scripts/latency_report.py shows voice-to-voice p50 and p95 and barge-in stop time within budget, confirmed against stereo call recordings.

The common mistake it prevents

Stopping the generator on barge-in but not flushing the playback buffer, so the agent keeps talking for seconds and its history includes words the caller never heard.

~/.claude/skills/realtime-voice-agents/SKILL.md

--- name: realtime-voice-agents description: Builds low-latency conversational voice agents - choosing a streaming STT to LLM to TTS pipeline or a speech-to-speech realtime API, a… license: MIT ---

A voice agent is ready when voice-to-voice latency, measured from the caller’s side, is within budget at p50 and p95; it never answers before the caller has finished but never leaves dead air; it stops talking within a few hundred milliseconds when interrupted and forgets the words the caller never heard; slow tools are covered by speech; and a suite of recorded calls proves all of that on every change. A demo over office Wi-Fi proves none of it.

Ground rules

  • Stream every stage. Nothing waits for a complete transcript, a complete LLM answer, or a complete audio file. Sequential batch stages cannot meet a conversational budget.
  • Measure voice-to-voice: from the caller’s last voiced frame to the first agent audio the caller hears. Per-vendor TTFT numbers do not add up to that.
  • Conversation history holds what was actually spoken, not what was generated.
  • Write for the ear: one or two short sentences per turn, no markdown, lists or URLs. Numbers, dates and codes must be normalized before TTS (see text-to-speech).
  • Confirm consequential actions (payments, cancellations, bookings) by reading back the key values and getting an explicit “yes”.

Choose the architecture

OptionPick whenCosts
Cascaded: streaming STT -> LLM -> streaming TTSYou need a specific LLM, full text transcripts for compliance, per-stage vendor choice, or strict tool-calling controlMore moving parts; latency is the sum of streamed stages
Speech-to-speech realtime API (OpenAI Realtime, Gemini Live)Lowest latency and natural prosody matter most, and the provider’s model and tools fitLess control over the reasoning model and voice; transcripts are secondary outputs; audio tokens priced separately
Hybrid: realtime model for conversation, text LLM behind a tool for hard reasoningMost turns are simple, a few need a stronger modelTwo models to evaluate

Default: build on an orchestration framework (Pipecat or LiveKit Agents) rather than raw sockets. Both handle transport, VAD, interruption plumbing, and support cascaded and speech-to-speech modes, so switching later is a config change rather than a rewrite. Check the framework’s current docs for parameter names; they change between releases.

Workflow

- [ ] 1. Write the call flows and the latency budget
- [ ] 2. Pick architecture, transport and region
- [ ] 3. Configure turn detection
- [ ] 4. Implement barge-in and history truncation
- [ ] 5. Add tools with spoken filler and timeouts
- [ ] 6. Build the recorded-audio test suite
- [ ] 7. Measure per stage, optimize the largest stage first
- [ ] 8. Production: call recording, latency dashboards, failure transfer

1. Latency budget

Human turn gaps are short, so treat these as planning targets and tune with real calls:

StageTarget (p50)Notes
Endpointing (speech end -> turn committed)200-500 msLargest lever; fixed silence timers tend to sit at the top of this range
LLM time to first token200-500 msSmall or fast model, short prompt, cached prefix, no up-front reasoning
TTS time to first audio100-250 msStreaming TTS fed sentence by sentence
Transport and buffering50-150 msJitter buffers, telephony legs, region distance
Voice-to-voiceunder ~800 ms p50, under ~1.5 s p95Above about 1.5 s callers start talking again or assume the line dropped

Every tool call adds its full duration on top unless it is covered with speech (step 5).

2. Transport and region

ClientTransportNotes
Browser or mobile appWebRTCOpus codec, built-in echo cancellation, jitter buffer, NAT traversal
Phone (PSTN)SIP trunk from a carrier or CPaaS, or the CPaaS media-stream WebSocketNarrowband G.711 (mu-law/A-law, 8 kHz); resample to what STT expects and back for TTS
Server to serverWebSocketFine inside a data center; avoid over lossy networks
  • OpenAI’s Realtime API accepts WebRTC, WebSocket and SIP; inbound SIP calls are pointed at sip.api.openai.com and accepted from a realtime.call.incoming webhook. Confirm the current model ID on the provider’s models page before shipping.
  • Run the agent in the same region as the STT, LLM and TTS endpoints and the telephony media edge. Cross-ocean hops add latency to every turn.
  • Support DTMF for digits and menus, and implement warm transfer to a human (SIP REFER or the CPaaS transfer API) as the failure path.

3. Turn detection

MethodBehaviorUse
VAD + silence timer (e.g. Silero VAD, server VAD)Ends the turn after N ms of silenceBaseline; simple and predictable
Semantic turn detection (OpenAI semantic_vad, LiveKit turn detector model, STT endpointing such as Deepgram Flux or AssemblyAI)Uses words and prosody to decide the caller is doneDefault for open conversation; fewer cut-offs at the same speed
Push-to-talk / manual commitClient commits the turnNoisy environments, kiosks, walkie-talkie UX

OpenAI Realtime session config (GA shape; turn detection lives under audio.input):

{
  "type": "session.update",
  "session": {
    "type": "realtime",
    "audio": {
      "input": {
        "turn_detection": { "type": "server_vad", "threshold": 0.5,
                            "prefix_padding_ms": 300, "silence_duration_ms": 500 }
      }
    }
  }
}

Swap to {"type": "semantic_vad", "eagerness": "auto"} for semantic detection; set turn_detection to null to commit turns manually.

Tuning rules:

  • Callers cut off mid-sentence -> raise silence duration or lower eagerness; switch to semantic detection.
  • Dead air after the caller finishes -> lower silence duration; check that STT finals are not the bottleneck.
  • Collecting digits, addresses or card numbers -> lengthen the silence window for that step only (people pause between groups), or use DTMF.
  • Background noise triggers turns -> raise the VAD threshold, enable noise suppression on the client, never on already-compressed telephony audio twice.

4. Barge-in

When the caller starts speaking while the agent is talking:

  1. Stop playback immediately: flush the client or telephony output buffer, not just the generator. Audio already queued keeps playing otherwise.
  2. Cancel in-flight LLM generation and TTS synthesis.
  3. Truncate the assistant turn in history to what was played. Over WebSocket with OpenAI Realtime, send conversation.item.truncate with item_id, content_index and audio_end_ms; over WebRTC and SIP the server truncates unplayed audio itself. In a cascaded pipeline, track played duration and cut the text at the matching word.
  4. Ignore false interruptions: backchannels (“mm-hmm”, “right”), coughs and echo. Require a minimum speech duration or word count before cancelling, and resume if the caller goes silent.

Echo cancellation is mandatory on speakerphone and in-browser use. Without it the agent hears itself and interrupts its own sentences. WebRTC provides it; raw WebSocket audio from a browser does not unless you enable it in getUserMedia constraints.

5. Tools mid-conversation

  • Before any tool that may take longer than about 500 ms, speak a short filler (“Let me pull up your booking.”) in parallel with the call.
  • Give every tool a timeout well under the caller’s patience (a few seconds), then say what happened and offer a next step instead of going silent.
  • OpenAI Realtime flow: arguments stream in response.function_call_arguments.delta, the complete call is in response.done; reply with conversation.item.create of type function_call_output using the same call_id, then send response.create.
  • Keep the tool list small and descriptions short; every tool definition adds tokens to each turn’s prompt and therefore to TTFT.
  • Make write tools idempotent (a retried “book it” must not double-book) and gate them on a spoken confirmation.

6. Test with recorded audio

Build a corpus before tuning:

tests/audio/
  happy_path_booking.wav       # clean speech, expected outcome in a sidecar .json
  accent_variety_*.wav
  noisy_street.wav             # SNR well below office conditions
  long_pause_mid_sentence.wav  # "My account number is ... 4 4 7 ... 2 1"
  interrupt_during_answer.wav  # caller talks over the agent at a known offset
  backchannel_mmhmm.wav        # must NOT stop the agent
  silence_10s.wav              # agent should re-prompt, then end politely
  • Stream files into the pipeline at real-time pace in 20 ms frames. Feeding faster than real time breaks VAD timing and hides latency.
  • Include 8 kHz mu-law versions of each file if the agent answers phones.
  • For multi-turn coverage, drive the agent with a simulated caller (an LLM persona plus TTS) and grade the transcript for task completion and policy compliance with llm-evals.
  • Assert on outcomes (booking created, transfer triggered) and on latency per turn.

7. Measure and optimize

Log one record per agent turn with timestamps on one clock, then run:

python3 scripts/latency_report.py turns.jsonl --v2v-p50 800 --v2v-p95 1500 --barge-in-p95 300

It prints p50/p90/p95 for endpointing, LLM TTFT, TTS first audio, transport and voice-to-voice, plus tool and barge-in times, names the largest median stage, lists the slowest turns, and exits 1 if a budget is breached. Server timestamps miss the caller’s network leg: once per release, verify against stereo call recordings (caller and agent on separate channels) by measuring the gap in an audio editor or with a VAD script.

Optimization order: endpointing, then LLM TTFT (smaller model, shorter system prompt, prompt caching, no up-front reasoning on simple turns), then TTS (streaming, sentence-level chunking, warm connections), then region and transport.

Gotchas

  • Sending the LLM’s first sentence to TTS only after the full response finishes; stream sentence by sentence.
  • Opening a new TTS or STT connection per turn; keep connections warm for the call.
  • Text formatting reaching TTS: asterisks, bullet points and URLs are read aloud.
  • Clipping the first syllable because VAD prefix padding is too short.
  • Telephony audio resampled with naive decimation, producing artifacts that hurt STT.
  • History that keeps the full generated text after an interruption, so the agent later refers to things the caller never heard.
  • Recording calls without the consent notice your jurisdiction requires; announce it.
  • Testing only in quiet rooms with fast networks; production callers use speakerphones in cars.

Output contract

## Architecture: cascaded | speech-to-speech | hybrid (why), framework, transport, region
## Latency budget vs measured (scripts/latency_report.py: p50/p95 per stage)
## Turn detection settings and rationale
## Barge-in: stop path, truncation method, false-interruption rule
## Tools: list, filler phrases, timeouts, confirmations
## Recorded-audio suite: cases, pass/fail, outcome assertions
## Escalation: transfer path and triggers

Bundled files

Loaded only when a step in the skill needs them. Review them here before installing.

scripts/latency_report.py 134 lines · 5.7 KB
#!/usr/bin/env python3
"""Per-stage latency report for voice agent turns (JSONL), with budget checks.

One JSON object per agent turn. Timestamps are milliseconds on ONE clock (epoch ms or
ms since call start); missing stages are allowed and skipped:

    {"call_id": "c1", "turn": 3,
     "user_speech_end": 10200,   # VAD: user stopped speaking (last voiced frame)
     "turn_committed": 10650,    # end-of-turn decision made, response requested
     "llm_first_token": 11020,   # first text token (pipeline) or first response event (s2s)
     "tts_first_audio": 11210,   # first synthesized audio chunk produced
     "audio_out_start": 11290,   # first agent audio sample played to the caller
     "tool_ms": 0,               # optional: time spent in tool calls before speaking
     "barge_in_stop_ms": 180}    # optional: user started talking -> agent audio stopped

Stages reported: endpointing (speech_end->committed), llm_ttft (committed->first token),
tts_ttfb (first token->first audio), transport (first audio->played), and voice_to_voice
(speech_end->played), the number callers feel.

Usage:
    python3 latency_report.py turns.jsonl
    python3 latency_report.py turns.jsonl --v2v-p50 800 --v2v-p95 1500 --barge-in-p95 300

Exit 0 when within budgets, 1 when a budget is breached, 2 on bad input.
"""
import argparse
import json
import sys

STAGES = [
    ("endpointing", "user_speech_end", "turn_committed"),
    ("llm_ttft", "turn_committed", "llm_first_token"),
    ("tts_ttfb", "llm_first_token", "tts_first_audio"),
    ("transport", "tts_first_audio", "audio_out_start"),
    ("voice_to_voice", "user_speech_end", "audio_out_start"),
]


def pct(values, p):
    if not values:
        return None
    s = sorted(values)
    k = max(0, min(len(s) - 1, int(round(p / 100 * (len(s) - 1)))))
    return s[k]


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("log", help="JSONL, one object per agent turn")
    ap.add_argument("--v2v-p50", type=float, default=800, help="voice-to-voice p50 budget in ms (default 800)")
    ap.add_argument("--v2v-p95", type=float, default=1500, help="voice-to-voice p95 budget in ms (default 1500)")
    ap.add_argument("--barge-in-p95", type=float, default=300, help="barge-in stop p95 budget in ms (default 300)")
    args = ap.parse_args()

    data = {name: [] for name, _, _ in STAGES}
    tools, barge, negative, turns = [], [], 0, 0
    worst = []
    try:
        with open(args.log) as f:
            for n, line in enumerate(f, 1):
                line = line.strip()
                if not line:
                    continue
                try:
                    t = json.loads(line)
                except json.JSONDecodeError as e:
                    print(f"error: line {n}: {e}", file=sys.stderr)
                    return 2
                turns += 1
                for name, a, b in STAGES:
                    if a in t and b in t:
                        d = float(t[b]) - float(t[a])
                        if d < 0:
                            negative += 1
                            continue
                        data[name].append(d)
                        if name == "voice_to_voice":
                            worst.append((d, t.get("call_id", "?"), t.get("turn", n)))
                if "tool_ms" in t and float(t["tool_ms"]) > 0:
                    tools.append(float(t["tool_ms"]))
                if "barge_in_stop_ms" in t:
                    barge.append(float(t["barge_in_stop_ms"]))
    except OSError as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    if not turns:
        print("error: no turns found", file=sys.stderr)
        return 2

    print(f"turns: {turns}")
    print(f"{'stage':<16} {'n':>5} {'p50':>7} {'p90':>7} {'p95':>7} {'max':>7}")
    for name, _, _ in STAGES:
        v = data[name]
        if not v:
            print(f"{name:<16} {0:>5} {'-':>7} {'-':>7} {'-':>7} {'-':>7}")
            continue
        print(f"{name:<16} {len(v):>5} {pct(v,50):>7.0f} {pct(v,90):>7.0f} {pct(v,95):>7.0f} {max(v):>7.0f}")
    if tools:
        print(f"{'tool_calls':<16} {len(tools):>5} {pct(tools,50):>7.0f} {pct(tools,90):>7.0f} {pct(tools,95):>7.0f} {max(tools):>7.0f}")
    if barge:
        print(f"{'barge_in_stop':<16} {len(barge):>5} {pct(barge,50):>7.0f} {pct(barge,90):>7.0f} {pct(barge,95):>7.0f} {max(barge):>7.0f}")
    if negative:
        print(f"warning: {negative} stage durations were negative (clocks not aligned?) and were skipped")

    # Largest share of the median budget
    parts = {k: pct(data[k], 50) for k in ("endpointing", "llm_ttft", "tts_ttfb", "transport") if data[k]}
    if parts:
        top = max(parts, key=parts.get)
        print(f"largest median stage: {top} ({parts[top]:.0f} ms) - optimize this first")
    if worst:
        print("slowest turns (voice_to_voice ms, call, turn):")
        for d, c, tn in sorted(worst, reverse=True)[:5]:
            print(f"  {d:>7.0f}  {c}  {tn}")

    fails = []
    v2v = data["voice_to_voice"]
    if v2v:
        if pct(v2v, 50) > args.v2v_p50:
            fails.append(f"voice_to_voice p50 {pct(v2v,50):.0f} > {args.v2v_p50:.0f}")
        if pct(v2v, 95) > args.v2v_p95:
            fails.append(f"voice_to_voice p95 {pct(v2v,95):.0f} > {args.v2v_p95:.0f}")
    else:
        print("warning: no turn had both user_speech_end and audio_out_start; voice_to_voice not checked")
    if barge and pct(barge, 95) > args.barge_in_p95:
        fails.append(f"barge_in_stop p95 {pct(barge,95):.0f} > {args.barge_in_p95:.0f}")
    if fails:
        print("RESULT: over budget - " + "; ".join(fails))
        return 1
    print("RESULT: within budget")
    return 0


if __name__ == "__main__":
    sys.exit(main())

About this skill

Written and maintained by the Claude Skills editors to the same editorial standard as every skill in the directory: a selective trigger, a real workflow, known pitfalls and a verification step. First published , last updated . MIT licensed.

esc