A voice agent with tool calls

"""A phone-style voice agent: speech in, a tool call, speech out, with every stage timed.

pip install openai
RUNCRATE_API_KEY=... python voice_agent.py caller.wav
The reply is spoken sentence by sentence: the first sentence goes to TTS while the model is still writing the rest.
"""
import json, os, re, subprocess, sys, time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

from openai import OpenAI

client = OpenAI(base_url="https://api.runcrate.ai/v1", api_key=os.environ["RUNCRATE_API_KEY"], max_retries=2)
ASR = "openai/whisper-large-v3-turbo"
LLM = os.environ.get("VOICE_LLM", "anthropic/claude-haiku-5.5")
TTS, VOICE = "hexgrad/Kokoro-82M", "af_heart"

# Your system of record. Here, a dict; in production, your orders API.
ORDERS = {"A-1042": {"status": "out for delivery", "eta": "today before 6 pm", "carrier": "the local courier"}}
TOOLS = [{
    "type": "function",
    "function": {
        "name": "lookup_order",
        "description": "Look up an order's delivery status by its id, such as A-1042.",
        "parameters": {"type": "object", "properties": {"order_id": {"type": "string"}}, "required": ["order_id"]},
    },
}]
SYSTEM = ("You are a phone support agent. Replies are spoken aloud: short sentences, no lists, no markdown. "
          "Always look an order up before saying anything about it. Order ids look like A-1042; callers may say them digit by digit.")

t_start = time.time()
marks: dict[str, float] = {}
costs: dict[str, float] = {}

def mark(name: str) -> None:
    marks[name] = round(time.time() - t_start, 3)

def speak(i: int, sentence: str) -> Path:
    raw = client.audio.speech.with_raw_response.create(model=TTS, voice=VOICE, input=sentence, response_format="mp3")
    costs[f"tts_{i}"] = float(raw.headers.get("x-llm-cost", 0))
    path = Path("out") / f"reply-{i:02d}.mp3"
    path.write_bytes(raw.content)
    if i == 0:
        mark("first_audio_ready")
    return path

def stream_turn(messages: list, label: str, on_text) -> list[dict]:
    """Stream one model turn, passing text to on_text as it arrives. Returns the tool calls it asked for."""
    calls: dict[int, dict] = {}
    stream = client.chat.completions.create(model=LLM, messages=messages, tools=TOOLS, stream=True,
                                            stream_options={"include_usage": True}, max_tokens=300)
    for chunk in stream:
        if chunk.usage:
            costs[label] = float(getattr(chunk.usage, "cost", 0) or 0)
        for choice in chunk.choices:
            d = choice.delta
            if f"{label}_first_token" not in marks and (d.content or d.tool_calls):
                mark(f"{label}_first_token")
            for tc in d.tool_calls or []:
                slot = calls.setdefault(tc.index, {"id": "", "name": "", "arguments": ""})
                slot["id"] = tc.id or slot["id"]
                if tc.function:
                    slot["name"] = tc.function.name or slot["name"]
                    slot["arguments"] += tc.function.arguments or ""
            if d.content:
                on_text(d.content)
    return [calls[k] for k in sorted(calls)]

def main(audio: str) -> None:
    Path("out").mkdir(exist_ok=True)
    with open(audio, "rb") as f:
        raw = client.audio.transcriptions.with_raw_response.create(model=ASR, file=f, language="en")
    costs["asr"] = float(raw.headers.get("x-llm-cost", 0))
    heard = raw.parse().text.strip()
    mark("asr_done")
    print(f"caller: {heard}")

    messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": heard}]
    pool = ThreadPoolExecutor(max_workers=4)
    futures, said = [], []
    buf = ""

    def on_text(piece: str) -> None:
        nonlocal buf
        buf += piece
        # Hand each finished sentence to TTS at once, while the model keeps writing.
        while m := re.search(r"[.!?](\s+)", buf):
            sentence, buf = buf[: m.start() + 1].strip(), buf[m.end():]
            if sentence:
                futures.append(pool.submit(speak, len(futures), sentence)); said.append(sentence)

    for turn in range(3):  # at most two tool rounds, then the answer
        calls = stream_turn(messages, f"llm_{turn}", on_text)
        if not calls:
            break
        messages.append({"role": "assistant", "content": None, "tool_calls": [
            {"id": c["id"], "type": "function", "function": {"name": c["name"], "arguments": c["arguments"]}} for c in calls]})
        for c in calls:
            oid = json.loads(c["arguments"] or "{}").get("order_id", "").upper().replace(" ", "")
            result = ORDERS.get(oid, {"error": f"no order {oid}"})
            messages.append({"role": "tool", "tool_call_id": c["id"], "content": json.dumps(result)})
        mark(f"tool_{turn}_done")
    if buf.strip():
        futures.append(pool.submit(speak, len(futures), buf.strip())); said.append(buf.strip())
    mark("llm_done")
    parts = [f.result() for f in futures]
    mark("all_audio_ready")
    print(f"agent: {' '.join(said)}")

    listing = Path("out/parts.txt"); listing.write_text("".join(f"file '{p.resolve()}'\n" for p in parts))
    subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-f", "concat", "-safe", "0", "-i", str(listing), "out/reply.mp3"], check=True)
    total = sum(costs.values())
    print(json.dumps({"model": LLM, "seconds_from_start": marks, "cost_usd": costs, "total_usd": total}, indent=2))
    Path("run.json").write_text(json.dumps({"model": LLM, "heard": heard, "said": " ".join(said), "marks": marks, "costs": costs, "total_usd": total}, indent=2))

if __name__ == "__main__":
    main(sys.argv[1])

// A phone-style voice agent: speech in, a tool call, speech out, with every stage timed.
//
// npm install openai
// RUNCRATE_API_KEY=... npx tsx voice-agent.ts caller.wav
// The reply is spoken sentence by sentence: the first sentence goes to TTS while the model is still writing the rest.
import { execFileSync } from "node:child_process";
import { createReadStream, mkdirSync, writeFileSync } from "node:fs";
import { resolve } from "node:path";
import OpenAI from "openai";
import type { ChatCompletionMessageParam, ChatCompletionTool } from "openai/resources/chat/completions";

const client = new OpenAI({ baseURL: "https://api.runcrate.ai/v1", apiKey: process.env.RUNCRATE_API_KEY, maxRetries: 2 });
const ASR = "openai/whisper-large-v3-turbo";
const LLM = process.env.VOICE_LLM ?? "anthropic/claude-haiku-5.5";
const TTS = "hexgrad/Kokoro-82M";
const VOICE = "af_heart";

// Your system of record. Here, a map; in production, your orders API.
const ORDERS: Record<string, object> = { "A-1042": { status: "out for delivery", eta: "today before 6 pm", carrier: "the local courier" } };
const TOOLS: ChatCompletionTool[] = [
  {
    type: "function",
    function: {
      name: "lookup_order",
      description: "Look up an order's delivery status by its id, such as A-1042.",
      parameters: { type: "object", properties: { order_id: { type: "string" } }, required: ["order_id"] },
    },
  },
];
const SYSTEM =
  "You are a phone support agent. Replies are spoken aloud: short sentences, no lists, no markdown. " +
  "Always look an order up before saying anything about it. Order ids look like A-1042; callers may say them digit by digit.";

const tStart = Date.now();
const marks: Record<string, number> = {};
const costs: Record<string, number> = {};
const mark = (name: string) => (marks[name] = (Date.now() - tStart) / 1000);

async function speak(i: number, sentence: string): Promise<string> {
  const { data, response } = await client.audio.speech
    .create({ model: TTS, voice: VOICE, input: sentence, response_format: "mp3" })
    .withResponse();
  costs[`tts_${i}`] = Number(response.headers.get("x-llm-cost") ?? 0);
  const path = `out/reply-${String(i).padStart(2, "0")}.mp3`;
  writeFileSync(path, Buffer.from(await data.arrayBuffer()));
  if (i === 0) mark("first_audio_ready");
  return path;
}

type Call = { id: string; name: string; arguments: string };

/** Stream one model turn, passing text to onText as it arrives. Returns the tool calls it asked for. */
async function streamTurn(messages: ChatCompletionMessageParam[], label: string, onText: (s: string) => void): Promise<Call[]> {
  const calls = new Map<number, Call>();
  const stream = await client.chat.completions.create({
    model: LLM,
    messages,
    tools: TOOLS,
    stream: true,
    stream_options: { include_usage: true },
    max_tokens: 300,
  });
  for await (const chunk of stream) {
    if (chunk.usage) costs[label] = Number((chunk.usage as { cost?: number }).cost ?? 0);
    for (const choice of chunk.choices) {
      const d = choice.delta;
      if (!(`${label}_first_token` in marks) && (d.content || d.tool_calls)) mark(`${label}_first_token`);
      for (const tc of d.tool_calls ?? []) {
        const slot = calls.get(tc.index) ?? { id: "", name: "", arguments: "" };
        slot.id = tc.id ?? slot.id;
        slot.name = tc.function?.name ?? slot.name;
        slot.arguments += tc.function?.arguments ?? "";
        calls.set(tc.index, slot);
      }
      if (d.content) onText(d.content);
    }
  }
  return [...calls.entries()].sort(([a], [b]) => a - b).map(([, c]) => c);
}

mkdirSync("out", { recursive: true });
const tr = await client.audio.transcriptions.create({ model: ASR, file: createReadStream(process.argv[2]), language: "en" }).withResponse();
costs.asr = Number(tr.response.headers.get("x-llm-cost") ?? 0);
const heard = tr.data.text.trim();
mark("asr_done");
console.log(`caller: ${heard}`);

const messages: ChatCompletionMessageParam[] = [
  { role: "system", content: SYSTEM },
  { role: "user", content: heard },
];
const parts: Promise<string>[] = [];
const said: string[] = [];
let buf = "";
const onText = (piece: string) => {
  buf += piece;
  // Hand each finished sentence to TTS at once, while the model keeps writing.
  for (let m = buf.match(/[.!?](\s+)/); m && m.index !== undefined; m = buf.match(/[.!?](\s+)/)) {
    const sentence = buf.slice(0, m.index + 1).trim();
    buf = buf.slice(m.index + m[0].length);
    if (sentence) {
      parts.push(speak(parts.length, sentence));
      said.push(sentence);
    }
  }
};

for (let turn = 0; turn < 3; turn++) {
  // at most two tool rounds, then the answer
  const calls = await streamTurn(messages, `llm_${turn}`, onText);
  if (calls.length === 0) break;
  messages.push({
    role: "assistant",
    content: null,
    tool_calls: calls.map((c) => ({ id: c.id, type: "function", function: { name: c.name, arguments: c.arguments } })),
  });
  for (const c of calls) {
    const oid = String(JSON.parse(c.arguments || "{}").order_id ?? "").toUpperCase().replace(/\s/g, "");
    messages.push({ role: "tool", tool_call_id: c.id, content: JSON.stringify(ORDERS[oid] ?? { error: `no order ${oid}` }) });
  }
  mark(`tool_${turn}_done`);
}
if (buf.trim()) {
  parts.push(speak(parts.length, buf.trim()));
  said.push(buf.trim());
}
mark("llm_done");
const files = await Promise.all(parts);
mark("all_audio_ready");
console.log(`agent: ${said.join(" ")}`);

writeFileSync("out/parts.txt", files.map((f) => `file '${resolve(f)}'\n`).join(""));
execFileSync("ffmpeg", ["-y", "-loglevel", "error", "-f", "concat", "-safe", "0", "-i", "out/parts.txt", "out/reply.mp3"]);
const total = Object.values(costs).reduce((a, b) => a + b, 0);
console.log(JSON.stringify({ model: LLM, seconds_from_start: marks, cost_usd: costs, total_usd: total }, null, 2));
writeFileSync("run.json", JSON.stringify({ model: LLM, heard, said: said.join(" "), marks, costs, total_usd: total }, null, 2));