Voice & Realtime Agents

# script_id: day_081_voice_realtime/latency_budget
# Toy latency budget for one voice turn. Stdlib only; the numbers are illustrative, not measured.
STAGES_MS = {"stt": 700, "llm_first_token": 900, "llm_full": 2500,
             "tts_full": 1200, "tts_first_audio": 300}
FIRST_SENTENCE_MS = 400   # time from first token until the first sentence is complete


def wait_for_everything(s: dict[str, int]) -> int:
    # Each stage waits for the previous one to finish completely.
    return s["stt"] + s["llm_first_token"] + s["llm_full"] + s["tts_full"]


def stream_each_stage(s: dict[str, int]) -> int:
    # Start speaking as soon as the first sentence of text exists.
    return s["stt"] + s["llm_first_token"] + FIRST_SENTENCE_MS + s["tts_first_audio"]


for name, fn in [("wait for everything", wait_for_everything), ("stream each stage", stream_each_stage)]:
    ms = fn(STAGES_MS)
    print(f"{name:20} first audio after {ms:5,} ms  {'OK' if ms <= 2000 else 'over a 2 s budget'}")

wait for everything  first audio after 5,300 ms  over a 2 s budget
stream each stage    first audio after 2,300 ms  over a 2 s budget

# script_id: day_081_voice_realtime/make_audio_fixtures
# Turn 3 qa_gold questions into WAV fixtures (PCM16, 24 kHz, mono): no microphone needed.
# Install: nothing (stdlib). Needs an OS speech engine: macOS `say` or Linux `espeak-ng`.
# No engine? Use speak() from atlas_voice.py ("Speaking the Answer"), which uses the Realtime API.
import array, json, os, shutil, subprocess, wave
from pathlib import Path

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
OUT, RATE = Path("audio"), 24_000            # the Realtime API's audio/pcm rate
QUESTIONS = {"q1": "qa-001", "q2": "qa-004", "q3": "qa-024"}


def resample(pcm: bytes, src: int) -> bytes:
    # Nearest-sample resampling: fine for speech fixtures, not for music.
    a = array.array("h", pcm)
    return array.array("h", (a[min(i * src // RATE, len(a) - 1)] for i in range(len(a) * RATE // src))).tobytes()


def synthesize(text: str, path: Path) -> None:
    if shutil.which("say"):                  # macOS can write 24 kHz PCM16 directly
        subprocess.run(["say", "-o", str(path), "--file-format=WAVE", "--data-format=LEI16@24000", text], check=True)
    elif shutil.which("espeak-ng"):          # writes 22,050 Hz, so resample afterwards
        subprocess.run(["espeak-ng", "-w", str(path), text], check=True)
    else:
        raise SystemExit("No speech engine found: install espeak-ng or use speak() from atlas_voice.py")
    with wave.open(str(path), "rb") as w:
        rate, pcm = w.getframerate(), w.readframes(w.getnframes())
    with wave.open(str(path), "wb") as w:    # rewrite as mono PCM16 at 24 kHz
        w.setnchannels(1); w.setsampwidth(2); w.setframerate(RATE)
        w.writeframes(resample(pcm, rate) if rate != RATE else pcm)


rows = {}
for line in (DATA_DIR / "eval" / "qa_gold.jsonl").read_text(encoding="utf-8").splitlines():
    if line.strip():
        row = json.loads(line)
        rows[row["id"]] = row["question"]
OUT.mkdir(exist_ok=True)
for name, qid in QUESTIONS.items():
    path = OUT / f"{name}.wav"
    synthesize(rows[qid].replace("_", " "), path)   # say "lease seconds", as a person would
    with wave.open(str(path), "rb") as w:
        print(f"wrote {path} ({w.getnframes() / w.getframerate():.1f} s) <- {rows[qid]}")

wrote audio/q1.wav (5.3 s) <- What are the default values of lease_seconds and heartbeat_seconds in orbit.toml?
wrote audio/q2.wav (3.1 s) <- What does ORB-1202 mean and how should I fix it?
wrote audio/q3.wav (4.6 s) <- How should I set terminationGracePeriodSeconds for Orbit workers on Kubernetes?

# script_id: day_081_voice_realtime/first_voice_turn
# Minimal chained turn: WAV -> transcript -> grounded text answer, with stage timings.
# Install: uv add "openai>=3.46,<4"   Env: OPENAI_API_KEY, ATLAS_MODEL, ORBIT_DATA. Input: audio/q1.wav
import os, time
from pathlib import Path
from openai import OpenAI

client = OpenAI()
MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
DOC = (DATA_DIR / "docs" / "configuration-reference.md").read_text(encoding="utf-8")

t0 = time.perf_counter()
with open("audio/q1.wav", "rb") as f:
    heard = client.audio.transcriptions.create(model="gpt-transcribe", file=f, languages=["en"],
                                               keywords=["Orbit", "orbit.toml", "lease_seconds", "heartbeat_seconds"]).text
t1 = time.perf_counter()
resp = client.responses.create(model=MODEL, reasoning={"effort": "none"},
                               instructions="Answer in at most 2 short spoken sentences, using only the doc.",
                               input=f"Doc:\n{DOC}\n\nQuestion: {heard}")
t2 = time.perf_counter()
print(f"heard : {heard}\nanswer: {resp.output_text}\nstt {t1 - t0:.1f} s | llm {t2 - t1:.1f} s")

heard : What are the default values of lease_seconds and heartbeat_seconds in orbit.toml?
answer: In the worker section, lease_seconds defaults to 300 and heartbeat_seconds to 30.
stt 0.8 s | llm 1.6 s

# script_id: day_081_voice_realtime/break_it
# Same audio, with and without transcription hints; plus a size guard. Stage timings printed.
# Install: uv add "openai>=3.46,<4"   Env: OPENAI_API_KEY. Input: audio/q2.wav
import time
from pathlib import Path
from openai import OpenAI

client = OpenAI()
MAX_UPLOAD_BYTES = 25 * 1024 * 1024          # the transcription endpoint's file limit


def transcribe(path: Path, keywords: list[str]) -> str:
    if path.stat().st_size > MAX_UPLOAD_BYTES:
        raise ValueError(f"{path} is {path.stat().st_size:,} bytes; split it below 25 MB first")
    with path.open("rb") as f:
        kwargs = {"keywords": keywords} if keywords else {}
        return client.audio.transcriptions.create(model="gpt-transcribe", file=f, languages=["en"], **kwargs).text


for hints in ([], ["Orbit", "ORB-1202", "PayloadTooLarge", "max_payload_kb"]):
    t0 = time.perf_counter()
    text = transcribe(Path("audio/q2.wav"), hints)
    print(f"{len(hints)} hints, {time.perf_counter() - t0:.1f} s: {text}")

0 hints, 0.7 s: What does orb twelve oh two mean and how should I fix it?
4 hints, 0.7 s: What does ORB-1202 mean and how should I fix it?

# script_id: day_081_voice_realtime/retiring_speech_api
# fragment
# RETIRING PATH: tts-1, tts-1-hd and gpt-4o-mini-tts shut down 2027-01-06 (as of 2026-11, verify at
# https://developers.openai.com/api/docs/deprecations). Shown so you recognise it in older code.
with client.audio.speech.with_streaming_response.create(
        model="gpt-4o-mini-tts", voice="marin", input="Lease seconds defaults to 300.") as r:
    r.stream_to_file("answer.mp3")

# script_id: day_081_voice_realtime/atlas_voice
# atlas/voice.py: Atlas's voice channel, chained and speech-to-speech (Day 81).
# Install: uv add "openai[realtime]>=3.46,<4"   (the extra adds websockets)
# Env: OPENAI_API_KEY, ATLAS_MODEL (gpt-6-luna), ATLAS_REALTIME_MODEL, ORBIT_DATA
# Inputs: audio/q1-q3.wav from make_audio_fixtures.py. Writes audio/*_chained.wav and *_s2s.wav.
from __future__ import annotations

import asyncio, base64, json, os, re, time, wave
from collections.abc import Iterator
from dataclasses import dataclass, field
from pathlib import Path

from openai import AsyncOpenAI, OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
REALTIME_MODEL = os.environ.get("ATLAS_REALTIME_MODEL", "gpt-realtime-2.1-mini")
STT_MODEL = "gpt-transcribe"                 # files; live sessions use gpt-live-transcribe
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
RATE = 24_000                                # audio/pcm: 24 kHz, 16-bit, mono
BYTES_PER_MS = RATE * 2 // 1000              # 48 bytes of PCM16 per millisecond
KEYWORDS = ["Orbit", "orbit.toml", "lease_seconds", "heartbeat_seconds", "ORB-1101", "ORB-1202"]


def read_pcm(wav: Path) -> bytes:
    with wave.open(str(wav), "rb") as w:
        got = (w.getframerate(), w.getsampwidth(), w.getnchannels())
        if got != (RATE, 2, 1):              # wrong format = chipmunk or garbled audio
            raise ValueError(f"{wav}: need (24000 Hz, 2 bytes, 1 channel), got {got}")
        return w.readframes(w.getnframes())


def pcm_chunks(wav: Path, ms: int = 100) -> Iterator[str]:
    """Yield base64 chunks of `ms` milliseconds, the way a live mic would deliver them."""
    pcm, step = read_pcm(wav), ms * BYTES_PER_MS
    for i in range(0, len(pcm), step):
        yield base64.b64encode(pcm[i:i + step]).decode()


def write_wav(pcm: bytes, path: Path, rate: int = RATE) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    with wave.open(str(path), "wb") as w:
        w.setnchannels(1); w.setsampwidth(2); w.setframerate(rate); w.writeframes(pcm)
    return path

# script_id: day_081_voice_realtime/atlas_voice
# USD per 1M tokens, as of 2026-11; verify at https://developers.openai.com/api/docs/pricing
PRICES = {"rt_audio_in": 10.0, "rt_audio_out": 20.0, "rt_text_in": 0.60, "rt_text_out": 2.40,
          "llm_in": 0.10, "llm_out": 0.50}
STT_PER_MIN = 0.0045                         # gpt-transcribe, per audio minute (same caveat)


@dataclass
class VoiceTurn:
    transcript: str
    answer: str
    audio_out: Path | None
    timings_ms: dict[str, float] = field(default_factory=dict)
    cost_usd: float = 0.0


@dataclass(frozen=True)
class LatencyBudget:
    stt_ms: float = 1000
    llm_first_token_ms: float = 1000
    first_audio_ms: float = 2000             # from "user stopped talking" to "Atlas starts talking"


def check_budget(t: VoiceTurn, b: LatencyBudget) -> list[str]:
    limits = {"stt": b.stt_ms, "llm_first_token": b.llm_first_token_ms, "first_audio": b.first_audio_ms}
    return [f"{k} {t.timings_ms[k]:,.0f} ms > {v:,.0f} ms"
            for k, v in limits.items() if t.timings_ms.get(k, 0) > v]


def realtime_cost(usage) -> float:
    """Price one Realtime response from response.done usage (cached-token discounts ignored)."""
    if usage is None:
        return 0.0
    i, o = usage.input_token_details, usage.output_token_details
    return ((i.audio_tokens or 0) * PRICES["rt_audio_in"] + (i.text_tokens or 0) * PRICES["rt_text_in"]
            + (o.audio_tokens or 0) * PRICES["rt_audio_out"] + (o.text_tokens or 0) * PRICES["rt_text_out"]) / 1e6

# script_id: day_081_voice_realtime/atlas_voice
client, aclient = OpenAI(), AsyncOpenAI()
AUDIO_FORMAT = {"type": "audio/pcm", "rate": RATE}
READ_ALOUD = "Read the user's text aloud exactly as written. Do not add, drop or change words."


async def speak_async(text: str, out: Path, *, voice: str = "marin") -> tuple[Path, float, float]:
    """Return (wav path, ms until the first audio chunk, cost in USD)."""
    audio, said, t0, first, cost = bytearray(), [], time.perf_counter(), 0.0, 0.0
    async with aclient.realtime.connect(model=REALTIME_MODEL) as conn:
        await conn.session.update(session={"type": "realtime", "output_modalities": ["audio"],
                                           "audio": {"output": {"format": AUDIO_FORMAT, "voice": voice}}})
        await conn.response.create(response={"conversation": "none", "instructions": READ_ALOUD, "input": [
            {"type": "message", "role": "user", "content": [{"type": "input_text", "text": text}]}]})
        async for ev in conn:
            if ev.type == "response.output_audio.delta":
                first = first or (time.perf_counter() - t0) * 1000
                audio += base64.b64decode(ev.delta)
            elif ev.type == "response.output_audio_transcript.delta":
                said.append(ev.delta)        # the model's own text of what it spoke
            elif ev.type == "error":         # errors arrive as events, not exceptions
                raise RuntimeError(f"realtime error: {ev.error.message}")
            elif ev.type == "response.done":
                cost = realtime_cost(ev.response.usage)
                break
    words = lambda s: re.findall(r"[a-z0-9_]+", s.lower())
    if words("".join(said)) != words(text):  # read-aloud is an instruction, not a guarantee
        print(f"warning: spoke {''.join(said)!r}, expected {text!r}")
    return write_wav(bytes(audio), out), first, cost


def speak(text: str, out: Path, *, voice: str = "marin") -> Path:
    """Sync wrapper for scripts. Inside a running event loop, await speak_async() instead."""
    return asyncio.run(speak_async(text, out, voice=voice))[0]

# script_id: day_081_voice_realtime/atlas_voice
import math
from collections import Counter

DOCS = {p.stem: p.read_text(encoding="utf-8") for p in sorted((DATA_DIR / "docs").glob("*.md"))}
TOKEN = re.compile(r"[a-z0-9_]+(?:[.-][a-z0-9_]+)*")       # keeps orbit.toml, ORB-1202, lease_seconds
STOP = set("what which when where does how should could would mean the and are for with "
           "this that from your can fix a an of in on to is it i my".split())


def _terms(doc: str) -> Counter:                            # words in headings count 5x
    heads = " ".join(line for line in doc.splitlines() if line.startswith("#"))
    boost = Counter({w: 4 * c for w, c in Counter(TOKEN.findall(heads.lower())).items()})
    return Counter(TOKEN.findall(doc.lower())) + boost


TF = {n: _terms(d) for n, d in DOCS.items()}
DF = Counter(w for tf in TF.values() for w in tf)
AVG_LEN = sum(sum(tf.values()) for tf in TF.values()) / len(TF)


def retrieve(question: str, k: int = 2) -> list[str]:
    def score(name: str) -> float:
        tf, norm = TF[name], 1.2 * (0.25 + 0.75 * sum(TF[name].values()) / AVG_LEN)
        return sum(math.log(1 + (len(TF) - DF[w] + 0.5) / (DF[w] + 0.5)) * tf[w] * 2.2 / (tf[w] + norm)
                   for w in set(TOKEN.findall(question.lower())) - STOP if w in tf)
    return sorted(DOCS, key=score, reverse=True)[:k]

# script_id: day_081_voice_realtime/atlas_voice
VOICE_RULES = ("You are Atlas, a voice assistant for Orbit maintainers. Answer in at most 3 short spoken "
               "sentences: no markdown, lists or code. Name the doc page you used, "
               "for example 'see the leases and heartbeats page'.")


def transcribe(audio: Path) -> str:          # compact copy of Day 14's helper
    with audio.open("rb") as f:
        return client.audio.transcriptions.create(model=STT_MODEL, file=f, keywords=KEYWORDS,
                                                  languages=["en"]).text


def ask(question: str) -> tuple[str, float, float]:     # stand-in for Day 28's ask()
    """Return (answer, ms until the first text token, cost in USD)."""
    context = "\n\n".join(f"[page: {n}]\n{DOCS[n][:6000]}" for n in retrieve(question))
    t0, first, parts = time.perf_counter(), 0.0, []
    with client.responses.stream(model=MODEL, instructions=VOICE_RULES, reasoning={"effort": "none"},
                                 input=f"{context}\n\nQuestion: {question}") as stream:
        for ev in stream:
            if ev.type == "response.output_text.delta":
                first = first or (time.perf_counter() - t0) * 1000
                parts.append(ev.delta)
        u = stream.get_final_response().usage
    return "".join(parts), first, (u.input_tokens * PRICES["llm_in"] + u.output_tokens * PRICES["llm_out"]) / 1e6

# script_id: day_081_voice_realtime/atlas_voice
def voice_turn(audio_in: Path) -> VoiceTurn:
    """Chained turn: transcribe -> ask -> speak. Waits for the full answer before speaking."""
    t0 = time.perf_counter()
    heard = transcribe(audio_in)
    stt_ms = (time.perf_counter() - t0) * 1000
    answer, ttft_ms, llm_cost = ask(heard)
    llm_ms = (time.perf_counter() - t0) * 1000 - stt_ms
    out, speak_first_ms, tts_cost = asyncio.run(
        speak_async(answer, audio_in.with_name(f"{audio_in.stem}_chained.wav")))
    minutes = len(read_pcm(audio_in)) / BYTES_PER_MS / 60_000
    timings = {"stt": stt_ms, "llm_first_token": ttft_ms, "llm_total": llm_ms,
               "first_audio": stt_ms + llm_ms + speak_first_ms}
    return VoiceTurn(heard, answer, out, timings, minutes * STT_PER_MIN + llm_cost + tts_cost)

# script_id: day_081_voice_realtime/atlas_voice
def s2s_session(tools: list[dict]) -> dict:
    return {"type": "realtime", "instructions": VOICE_RULES, "output_modalities": ["audio"], "tools": tools,
            "audio": {"input": {"format": AUDIO_FORMAT, "turn_detection": None,      # manual commit
                                "transcription": {"model": "gpt-live-transcribe"}},  # a text log of the user
                      "output": {"format": AUDIO_FORMAT, "voice": "marin"}}}


async def realtime_turn(audio_in: Path, *, tools: list[dict] = ()) -> VoiceTurn:
    audio, said, heard, cost, first = bytearray(), [], None, 0.0, 0.0
    tool_pending = done = False
    async with asyncio.timeout(90), aclient.realtime.connect(model=REALTIME_MODEL) as conn:
        await conn.session.update(session=s2s_session(list(tools)))
        for chunk in pcm_chunks(audio_in):
            await conn.input_audio_buffer.append(audio=chunk)
        await conn.input_audio_buffer.commit()   # "the user has finished speaking"
        t_end = time.perf_counter()
        await conn.response.create()
        async for ev in conn:
            if ev.type == "response.output_audio.delta":
                first = first or (time.perf_counter() - t_end) * 1000
                audio += base64.b64decode(ev.delta)
            elif ev.type == "response.output_audio_transcript.delta":
                said.append(ev.delta)
            elif ev.type == "conversation.item.input_audio_transcription.completed":
                heard = ev.transcript
            elif ev.type == "conversation.item.input_audio_transcription.failed":
                heard = ""
            elif ev.type == "response.function_call_arguments.done":
                await run_tool(conn, ev)             # defined in "Tools in a Voice Agent"
                tool_pending = True
            elif ev.type == "error":
                raise RuntimeError(f"realtime error: {ev.error.message}")
            elif ev.type == "response.done":
                cost += realtime_cost(ev.response.usage)
                if tool_pending:                     # tool output is in; ask the model to continue
                    tool_pending = False
                    await conn.response.create()
                else:
                    done = True
            if done and heard is not None:           # the user transcript can arrive after response.done
                break
    out = write_wav(bytes(audio), audio_in.with_name(f"{audio_in.stem}_s2s.wav"))
    return VoiceTurn(heard or "", "".join(said), out, {"first_audio": first}, cost)

# script_id: day_081_voice_realtime/replay_events
# Replay a recorded speech-to-speech turn offline: first-audio latency, transcripts, cost.
# Stdlib only, no key. Event shapes match the Realtime API; timings and token counts are toy values.
import base64

AUDIO_200MS = base64.b64encode(bytes(48 * 200)).decode()   # 200 ms of silent PCM16 at 24 kHz
LOG = [  # (ms since input_audio_buffer.commit, event)
    (0, {"type": "input_audio_buffer.committed"}),
    (610, {"type": "response.output_audio.delta", "delta": AUDIO_200MS}),
    (612, {"type": "response.output_audio_transcript.delta", "delta": "Lease seconds defaults to 300 "}),
    (690, {"type": "conversation.item.input_audio_transcription.completed",
           "transcript": "What are the default values of lease seconds and heartbeat seconds in orbit.toml?"}),
    (700, {"type": "response.output_audio.delta", "delta": AUDIO_200MS}),
    (705, {"type": "response.output_audio_transcript.delta", "delta": "and heartbeat seconds to 30."}),
    (880, {"type": "response.done", "response": {"status": "completed", "usage": {
        "input_token_details": {"audio_tokens": 120, "text_tokens": 300},
        "output_token_details": {"audio_tokens": 240, "text_tokens": 30}}}}),
]
PRICE = {"audio_in": 10.0, "text_in": 0.60, "audio_out": 20.0, "text_out": 2.40}  # $/1M, as of 2026-11, verify

first = next(t for t, e in LOG if e["type"] == "response.output_audio.delta")
heard = next(e["transcript"] for _, e in LOG if e["type"].endswith("transcription.completed"))
said = "".join(e["delta"] for _, e in LOG if e["type"] == "response.output_audio_transcript.delta")
audio_ms = sum(len(base64.b64decode(e["delta"])) for _, e in LOG if e["type"] == "response.output_audio.delta") // 48
usage = next(e["response"]["usage"] for _, e in LOG if e["type"] == "response.done")
i, o = usage["input_token_details"], usage["output_token_details"]
cost = (i["audio_tokens"] * PRICE["audio_in"] + i["text_tokens"] * PRICE["text_in"]
        + o["audio_tokens"] * PRICE["audio_out"] + o["text_tokens"] * PRICE["text_out"]) / 1e6
print(f"heard: {heard}\nsaid : {said}\nfirst audio {first} ms (budget 2,000: {'OK' if first <= 2000 else 'OVER'}) "
      f"| {audio_ms} ms of audio | ${cost:.5f}")

heard: What are the default values of lease seconds and heartbeat seconds in orbit.toml?
said : Lease seconds defaults to 300 and heartbeat seconds to 30.
first audio 610 ms (budget 2,000: OK) | 400 ms of audio | $0.00625

# script_id: day_081_voice_realtime/barge_in_demo
# Barge-in with files: ask q1, "hear" 1.5 s of the answer, interrupt with q3.
# Install: uv add "openai[realtime]>=3.46,<4"   Env: OPENAI_API_KEY. Inputs: audio/q1.wav, audio/q3.wav
import asyncio, base64, os, time, wave
from pathlib import Path
from openai import AsyncOpenAI

REALTIME_MODEL = os.environ.get("ATLAS_REALTIME_MODEL", "gpt-realtime-2.1-mini")
FMT, BYTES_PER_MS, HEARD_MS = {"type": "audio/pcm", "rate": 24_000}, 48, 1500
SESSION = {"type": "realtime", "output_modalities": ["audio"],
           "audio": {"input": {"format": FMT, "turn_detection": None}, "output": {"format": FMT, "voice": "marin"}}}


async def ask_turn(conn, wav: Path) -> None:
    with wave.open(str(wav), "rb") as w:
        pcm = w.readframes(w.getnframes())
    for i in range(0, len(pcm), 100 * BYTES_PER_MS):
        await conn.input_audio_buffer.append(audio=base64.b64encode(pcm[i:i + 100 * BYTES_PER_MS]).decode())
    await conn.input_audio_buffer.commit()
    await conn.response.create()


async def main() -> None:
    t0, received, item_id, interrupted, asked_q3 = time.perf_counter(), 0, "", False, False
    log = lambda msg: print(f"{(time.perf_counter() - t0) * 1000:6.0f} ms  {msg}")
    async with asyncio.timeout(90), AsyncOpenAI().realtime.connect(model=REALTIME_MODEL) as conn:
        await conn.session.update(session=SESSION)
        await ask_turn(conn, Path("audio/q1.wav")); log("asked q1")
        async for ev in conn:
            if ev.type == "response.output_audio.delta" and not asked_q3:
                item_id, received = ev.item_id, received + len(base64.b64decode(ev.delta))
                if not interrupted and received >= HEARD_MS * BYTES_PER_MS:
                    interrupted = True; log(f"user talks over Atlas at {HEARD_MS} ms -> response.cancel")
                    await conn.response.cancel()
            elif ev.type == "response.done":
                log(f"response.done status={ev.response.status}")
                if not interrupted or asked_q3:
                    break
                await conn.conversation.item.truncate(item_id=item_id, content_index=0, audio_end_ms=HEARD_MS)
                await ask_turn(conn, Path("audio/q3.wav")); asked_q3 = True; log("asked q3")
            elif ev.type in ("conversation.item.truncated", "error"):
                log(ev.type if ev.type != "error" else f"error: {ev.error.message}")


asyncio.run(main())

     0 ms  asked q1
   840 ms  user talks over Atlas at 1500 ms -> response.cancel
   910 ms  response.done status=cancelled
   960 ms  conversation.item.truncated
  1180 ms  asked q3
  2950 ms  response.done status=completed

# script_id: day_081_voice_realtime/atlas_voice
DOCS_SEARCH_TOOL = {"type": "function", "name": "docs_search",
                    "description": "Search the Orbit docs (read-only). Returns page names and excerpts.",
                    "parameters": {"type": "object", "properties": {"query": {"type": "string"}},
                                   "required": ["query"]}}


def docs_search(query: str) -> str:
    return json.dumps([{"page": n, "excerpt": DOCS[n][:1500]} for n in retrieve(query)])


READ_ONLY_TOOLS = {"docs_search": docs_search}   # side-effecting tools never go in this dict


async def run_tool(conn, ev) -> None:
    fn = READ_ONLY_TOOLS.get(ev.name)
    try:
        output = fn(**json.loads(ev.arguments)) if fn else json.dumps({"error": f"unknown tool {ev.name}"})
    except Exception as exc:                     # the model hears the error and can recover
        output = json.dumps({"error": str(exc)})
    await conn.conversation.item.create(item={"type": "function_call_output", "call_id": ev.call_id,
                                              "output": output})

# script_id: day_081_voice_realtime/atlas_voice
if __name__ == "__main__":
    wav, budget = Path("audio/q1.wav"), LatencyBudget()
    chained = voice_turn(wav)
    s2s = asyncio.run(realtime_turn(wav, tools=[DOCS_SEARCH_TOOL]))
    print(f"heard  : {chained.transcript}\nchained: {chained.answer}\ns2s    : {s2s.answer}\n")
    for name, t in (("chained", chained), ("speech-to-speech", s2s)):
        stages = " | ".join(f"{k} {v:,.0f}" for k, v in t.timings_ms.items())
        verdict = check_budget(t, budget) or "within budget"
        print(f"{name:16} {stages} ms | ${t.cost_usd:.4f} | {verdict} | {t.audio_out}")

heard  : What are the default values of lease_seconds and heartbeat_seconds in orbit.toml?
chained: In the worker section, lease_seconds defaults to 300 and heartbeat_seconds to 30. See the configuration reference page.
s2s    : Lease seconds defaults to 300 and heartbeat seconds to 30, per the configuration reference page.

chained          stt 780 | llm_first_token 640 | llm_total 1,420 | first_audio 3,050 ms | $0.0021 | ['first_audio 3,050 ms > 2,000 ms'] | audio/q1_chained.wav
speech-to-speech first_audio 720 ms | $0.0094 | within budget | audio/q1_s2s.wav