OpenRouter 教程:如何用 Python 构建语音 AI 智能体
How to Build a Voice AI Agent
OpenRouter 发布教程,用 Python 串起语音转写、LLM 工具调用和 TTS 三段流水线,构建一个餐厅订位语音智能体,并给出可运行代码与 LiveKit 实时版本。
完整代码与实测延迟、成本拆解,可据此判断语音智能体流水线中 TTS 与工具调用各自的代价。
语音智能体不能只生成一个好的回答。它必须理解对方说了什么,决定该做什么,调用任何需要的工具,并在停顿变得尴尬之前重新开始说话。
在本教程中,我们将用 Python 构建这个完整的循环。我们的示例是一个餐厅预订智能体,它转录来电者的语音,通过工具检查餐桌可用性,并将响应转换回语音。语音转文本、LLM 和文本转语音都通过一个 OpenRouter API 密钥运行,并且我们将每个完成的句子流式传输到 TTS,而不是等待整个回复。
在三段测试对话的全部 12 轮中,服务器端整体首次音频时间中位数为 1.9 秒,而在我们 10 月 8 日测试期间可用的促销价格下,四轮对话每处理一分钟语音的成本约为 0.013 美元。
TL;DR
-
文本转语音约占所测模型成本的 90%,这使得语音回复长度成为该智能体的重要成本因素。
-
在我们的测试中,服务器端首次音频时间估计为普通轮次 1.4 秒、工具轮次 2.4 秒,每轮计一次 LLM 请求。示例代码会发起额外的 LLM 请求,网络和播放还会增加时间。
-
姓名比电话号码更难转录。在采取行动之前,让来电者确认重要细节。
-
将完成的句子发送到文本转语音(TTS)可以减少对语音输出的等待,但实时播放和打断需要额外的编排。
语音 AI 智能体的一轮如何工作
语音智能体接收来电者的语音,将其转换为文本,传递给 LLM 以决定该做什么,并将响应转换回语音。以下是该过程的工作方式:

例如,当有人致电我们的餐厅预订餐桌时,LLM 可能需要在响应之前调用 check_availability。你的应用程序执行该函数,将可用时间发送回模型,并等待它生成响应。这个额外的往返增加了来电者的等待时间。
为了减少这种等待,我们可以在 LLM 完成整个响应之前就开始生成语音。我们不再等待完整文本,而是流式传输响应,将其拆分为句子,并在模型继续生成其余内容的同时,将每个完成的句子发送到 TTS。
对于这个构建,我们将使用我们的 语音转文本 API 来转录来电者,使用 带工具调用的 Chat Completions 来处理对话和预订,并使用我们的 文本转语音 API 来生成语音响应。
转录 API 以 JSON 形式返回文本,而语音 API 返回音频字节。我们的 Python 示例将把生成的音频保存为 WAV 文件,稍后我们将使用 LiveKit 来处理实时音频和打断。
用 Python 构建语音智能体
让我们使用三个模型构建餐厅预订智能体:elevenlabs/scribe-v2 用于转录,openai/gpt-6-luna 用于对话和工具调用,elevenlabs/eleven-flash-v2.5 用于语音生成。
你需要 Python 3.10 或更高版本以及一个 OpenRouter API 密钥。安装 requests 库并将你的 API 密钥设置为环境变量:
python -m pip install requests
export OPENROUTER_API_KEY="your-api-key"为了使示例易于复现,我们将使用第二个 TTS 语音来模拟四个来电者轮次,而不是连接麦克风。我们还将使用模拟预订数据,以便智能体可以在没有真实预订系统的情况下检查可用性并返回预订详情。
将以下代码保存为 voice_agent.py:
import base64
import io
import json
import os
import re
import wave
from concurrent.futures import ThreadPoolExecutor
from datetime import date
import requests
API = "https://openrouter.ai/api/v1"
AUTH = {"Authorization": f"Bearer {os.environ['OPENROUTER_API_KEY']}"}
STT = "elevenlabs/scribe-v2"
LLM = "openai/gpt-6-luna"
TTS = "elevenlabs/eleven-flash-v2.5"
PCM_RATE = 24000
def system_prompt():
today = date.today() # read per call, so a long-running agent stays on the right day
return (
"You are the phone receptionist for Luigi's restaurant. "
f"Today is {today:%A, %B} {today.day}, {today.year}. Keep answers short and spoken: "
"no markdown, and say times in words. Check availability before "
"offering a table. Only book after the caller agrees and provides a name. "
"Do not claim to send confirmation texts."
)
def tool(name, description, properties):
return {
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": {
"type": "object",
"properties": properties,
"required": list(properties),
},
},
}
DATE = {"type": "string", "description": "YYYY-MM-DD"}
TIME = {"type": "string", "description": "HH:MM, 24-hour"}
PARTY = {"type": "integer", "minimum": 1}
TOOLS = [
tool(
"check_availability",
"Find open tables near the requested time",
{"date": DATE, "time": TIME, "party_size": PARTY},
),
tool(
"book_table",
"Book only after the caller accepts a time and gives a name",
{
"date": DATE,
"time": TIME,
"party_size": PARTY,
"name": {"type": "string"},
},
),
]
def run_tool(name, args):
# Mock booking service; replace with a real reservation backend.
if name == "check_availability":
return {
"open_times": (
["17:45", "20:15"]
if args["party_size"] >= 5
else [args["time"]]
)
}
if name == "book_table":
allowed = (
["17:45", "20:15"]
if args["party_size"] >= 5
else [args["time"]]
)
return (
{"booked": True, "reference": "XJ47Q"}
if args["time"] in allowed
else {"booked": False, "error": "Time unavailable"}
)
return {"error": f"Unknown tool: {name}"}
def to_wav(pcm, rate):
if len(pcm) % 2:
raise ValueError("PCM audio must contain whole 16-bit samples")
output = io.BytesIO()
with wave.open(output, "wb") as wav:
wav.setnchannels(1)
wav.setsampwidth(2)
wav.setframerate(rate)
wav.writeframes(pcm)
return output.getvalue()
def speak(text, voice="sarah"):
if not text.strip():
return b"", PCM_RATE
with requests.post(
f"{API}/audio/speech",
headers=AUTH,
stream=True,
timeout=60,
json={
"model": TTS,
"input": text,
"voice": voice,
"response_format": "pcm",
"provider": {
"options": {
"elevenlabs": {"output_format": "pcm_24000"}
}
},
},
) as response:
response.raise_for_status()
if not response.headers.get("Content-Type", "").startswith("audio/pcm"):
raise ValueError("Expected raw PCM audio from TTS")
pcm = b"".join(response.iter_content(4096))
if not pcm:
raise ValueError("TTS returned no audio")
return pcm, PCM_RATE
def transcribe(wav):
response = requests.post(
f"{API}/audio/transcriptions",
headers=AUTH,
timeout=60,
json={
"model": STT,
"language": "en",
"input_audio": {
"data": base64.b64encode(wav).decode("ascii"),
"format": "wav",
},
},
)
response.raise_for_status()
return response.json()["text"].strip()
def choose_tool_calls(messages):
# Resolve possible tools before starting any paid TTS requests.
response = requests.post(
f"{API}/chat/completions",
headers=AUTH,
timeout=60,
json={
"model": LLM,
"messages": messages,
"tools": TOOLS,
"tool_choice": "auto",
"reasoning": {"effort": "none"},
"max_tokens": 300,
},
)
response.raise_for_status()
message = response.json()["choices"][0]["message"]
return message.get("tool_calls") or []
def stream_reply(messages):
# This request has no tools, so sentences can safely go to TTS.
pending, finish = "", None
with requests.post(
f"{API}/chat/completions",
headers=AUTH,
stream=True,
timeout=60,
json={
"model": LLM,
"messages": messages,
"reasoning": {"effort": "none"},
"max_tokens": 300,
"stream": True,
},
) as response:
response.raise_for_status()
response.encoding = "utf-8"
for line in response.iter_lines(decode_unicode=True):
if not line or not line.startswith("data: {"):
continue
chunk = json.loads(line[6:])
if chunk.get("error"): # a provider error mid-stream, after some text may have arrived
raise RuntimeError(f"stream error: {chunk['error']}")
choices = chunk.get("choices")
if not choices:
continue
finish = choices[0].get("finish_reason") or finish
pending += (choices[0].get("delta") or {}).get("content") or ""
# Wait for whitespace; a buffer ending in "3." may still become "3.14".
while match := re.search(r"[.!?]\s", pending):
sentence = pending[:match.end()].strip()
if sentence:
yield sentence
pending = pending[match.end():]
if finish == "length": # hit max_tokens, so don't speak the clipped tail or save the reply as a full answer
raise RuntimeError("reply cut off at max_tokens")
if pending.strip():
yield pending.strip()
def run_turn(caller_wav, messages):
heard = transcribe(caller_wav)
messages.append({"role": "user", "content": heard})
# Tool selection can repeat when the model needs several tool results.
for _ in range(4):
calls = choose_tool_calls(messages)
if not calls:
break
messages.append({"role": "assistant", "content": None, "tool_calls": calls})
for call in calls:
try:
args = json.loads(call["function"]["arguments"])
result = run_tool(call["function"]["name"], args)
except (ValueError, TypeError, KeyError) as exc:
result = {"error": str(exc)}
messages.append({
"role": "tool",
"tool_call_id": call["id"],
"content": json.dumps(result),
})
else:
raise RuntimeError("Too many tool calls")
sentences, jobs = [], []
with ThreadPoolExecutor(max_workers=3) as pool:
for sentence in stream_reply(messages):
sentences.append(sentence)
jobs.append(pool.submit(speak, sentence))
audio = [job.result() for job in jobs]
if not sentences:
raise RuntimeError("LLM returned no spoken response")
rates = {rate for _, rate in audio}
if rates != {PCM_RATE}:
raise ValueError("Inconsistent TTS sample rates")
spoken = " ".join(sentences)
messages.append({"role": "assistant", "content": spoken})
pcm = b"".join(chunk for chunk, _ in audio)
return heard, spoken, to_wav(pcm, PCM_RATE)
if __name__ == "__main__":
caller_lines = [
"Hi, do you have a table for six this Saturday at seven thirty?",
"Eight fifteen works. The name is Siobhan Okafor.",
"Can you text the confirmation to four one five, "
"five five five, zero one eight two?",
"No, that's everything. Thanks, bye.",
]
messages = [{"role": "system", "content": system_prompt()}]
for turn, line in enumerate(caller_lines, 1):
pcm, rate = speak(line, voice="george")
heard, said, reply = run_turn(to_wav(pcm, rate), messages)
with open(f"agent_{turn}.wav", "wb") as file:
file.write(reply)
print(f"caller: {heard}\nagent: {said}\n")该脚本通过 run_turn() 连接三个 API,该函数会转写每个来电者的消息,将转写文本传给 LLM,并处理响应。如果 LLM 请求使用工具,该函数会执行该工具,将结果添加到对话历史中,并再次发起请求,以便模型继续。
看看 stream_reply() 如何处理工具调用。模型会以片段形式流式输出函数名和参数,因此我们在执行工具之前先将它们组装起来。对于语音回复,我们通过线程池将完整的句子发送给 TTS,使语音生成与剩余的 LLM 输出重叠进行。
然后我们按句子顺序合并音频,并将其保存为 WAV 文件。我们将在引入 LiveKit 时处理实际播放和打断。
你可以使用以下命令运行该脚本:
python voice_agent.py它应该会创建四个音频文件,agent_1.wav 到 agent_4.wav,并打印对话内容。以下是我们某次测试运行中的一段摘录。来电者行显示的是转写后的音频,因此数字可能与实际说出的方式不同。
caller: Hi, do you have a table for six this Saturday at 7:30?
agent: We don't have seven-thirty available for six,
but I can offer five forty-five or eight fifteen.
Would either work for you?
caller: 815 works. The name is Siobhan Okafor
agent: You're booked for six at eight fifteen this
Saturday under Siobhan Okafor.
Your reference is XJ47Q.
caller: Can you text the confirmation to 415-555-0182?
agent: I can't send texts from here, but your
reservation is confirmed under Siobhan Okafor
for six at eight fifteen this Saturday,
reference XJ47Q.该示例展示了 agent 在提供另一个时间之前先检查可用性,然后在来电者接受后调用 book_table。它同样拒绝承诺发送短信确认,因为我们没有提供用于发送短信的工具。
为什么这种设计有效,以及什么会破坏它
一条回复在聊天窗口中看起来可能完全正常,但朗读出来时却会显得奇怪。时间和电话号码就是很好的例子。这就是为什么我们的系统提示要求 LLM 保持回复简短、避免使用格式,并将时间写成口语化的词语,例如用“eight fifteen”而不是 8:15。
同样的问题也可能反向发生。在我们的测试中,语音转文本将“eight fifteen”转换成了 815,但 LLM 正确理解了来电者的回复,因为它拥有之前的对话作为上下文。我们在 messages 中维护该历史记录,并在每次请求时发送它,因为 Chat Completions 不会自动保留之前的轮次。结果,提示从第一轮的大约 290 个 token 增长到第四轮的 490 个。
另一个重要的考虑因素是确保 agent 只确认实际已发生的操作。在早期没有 book_table 的版本中,agent 可能会告诉来电者他们的预订已确认,即使应用程序并未创建预订。
这就是为什么预订工具需要做的不仅仅是返回响应。当连接到真实的预订系统时,它应该验证可用性、保存预订,并且仅在操作成功时返回确认。
语音 AI agent 有多快?
在所有 12 轮中,服务端到首段音频的中位时间为 1.9 秒。按轮次类型划分,不使用工具时的中位时间为 1.4 秒,而 agent 需要调用工具时为 2.4 秒。
我们测量了三段完整的对话,涵盖 12 轮,使用我们的 生成元数据 来估算每个阶段的服务端延迟。这使我们能够将模型处理时间与测试环境中的额外网络开销区分开来。
以下是结果的对比:
| 阶段 | 有工具调用 | 无工具 |
|---|---|---|
| 语音转文本(Scribe v2) | 549 毫秒 | 391 毫秒 |
| LLM 选择工具 | 736 毫秒 | — |
| LLM 回答,首个 token | 685 毫秒 | 706 毫秒 |
| 文本转语音,首字节 | 262 毫秒 | 208 毫秒 |
| 估算的首段音频时间 | 2.4 秒 | 1.4 秒 |
额外的 LLM 请求是造成差异的主要原因。当智能体需要检查可用性时,它必须等待工具返回结果后才能生成回答。
这些数字反映的是服务器端处理时间,而非呼叫方实际体验到的完整时长。网络传输、检测呼叫方何时说完以及音频播放都会带来额外延迟。各阶段的中位数也不一定加总等于总体中位数。这些估算还按每轮一次 LLM 请求计算。下面的示例代码每轮会发起两次请求:一次非流式请求用于检查工具调用,然后一次流式请求用于生成回复;工具轮在工具运行后还会再发起一次请求,因此预计耗时会比这些数字更长。
下面的模型对比使用了单独的测试运行和输入,因此其报告的延迟与这里展示的完整对话测量结果不同。
语音 AI 智能体的成本是多少?
在五次运行中,我们四轮的餐厅对话成本在 $0.0069 到 $0.0087 之间。每次通话包含约 38 秒语音,按 2026 年 10 月 8 日可用的价格计算,我们估算每处理一分钟语音的成本为 $0.013。
大部分成本来自生成智能体的语音。
| 阶段 | 占总成本比例 | 测试期间价格 |
|---|---|---|
| 文本转语音(Eleven Flash v2.5) | 90% | $20 / 百万字符 |
| 语音转文本(Scribe v2) | 6% | $0.11 / 音频小时 |
| LLM(GPT-6 Luna) | 4% | $0.10 / $0.50 每百万输入/输出 token |
上述 ElevenLabs 价格包含 50% 的发布折扣,该折扣已于 2026 年 10 月 19 日结束。按常规标价,Eleven Flash v2.5 为每百万字符 $40,而 Scribe v2 约为每音频小时 $0.22。假设使用量相同,我们的对话每处理一分钟语音的成本约为 $0.025。这是基于我们 10 月 8 日测量结果的计算,并非单独的计费测试。
成本明细仍然有用。在这次对话中,TTS 约占我们支出的 90%,因此保持智能体的语音回复简短,对成本的影响要大于小幅减少 LLM token 用量。
你还可以追踪单个请求的实际成本。转录响应包含 usage.cost,而语音响应提供 X-Generation-Id 响应头,你可以通过我们的 生成元数据 API 用它来获取成本。
你应该使用哪个语音转文本模型?
对于餐厅预订智能体而言,当呼叫方提到智能体需要据此采取行动的细节时,例如姓名、日期、电话号码和预订时间,转录准确性最为重要。
为了比较模型,我们使用三种合成语音生成了九条呼叫方语句,然后为每条录音创建了清晰版和电话音质版。这样我们得到了 54 个音频片段,测试中每个模型评分 90 个事实。
我们评估了十个 STT 模型。下表展示了五个模型,以说明我们观察到的准确性和延迟范围。
| 模型 | 保留事实数(共 90) | 服务器延迟中位数 | 测试日成本/小时 |
|---|---|---|---|
| microsoft/mai-transcribe-2 | 89 | 1,740 ms | $0.11 |
| elevenlabs/scribe-v2 | 88 | 379 ms | $0.11 |
| openai/whisper-large-v3 | 88 | 517 ms | $0.17 |
| x-ai/grok-stt-1.0 | 86 | 264 ms | $0.10 |
| deepgram/nova-3 | 83 | 83 ms | $0.26 |
MAI-Transcribe 2 保留的事实最多,但 Scribe v2 速度快得多,仅少保留一个事实。Whisper Large V3 的得分与 Scribe 相同,但响应时间更长。对于我们的预订代理,Scribe v2 在准确性和速度之间取得了良好平衡,因此我们选择它用于 Python 实现。
每个识别器都在不常见的名字上出错,即使它正确保留了电话号码。例如,“Siobhan Okafor”在转录文本中显示为“Shioban Okafor”、“Siobhan O'Keefe”和“Siobhan Akhefer”。
与其假设转录的名字是正确的,不如让代理与来电者确认,或在完成预订前请他们拼写出来。
请记住,这些结果来自合成的英语录音,而非不同口音和环境下的真实客户来电。你可以探索可用的转录模型,但值得用反映你自己用例的音频测试入围的模型。
你应该使用哪个 LLM 和文本转语音模型?
对于 LLM,我们考察了工具调用准确性、响应时间,以及生成的文本朗读起来有多自然。
我们比较中的每个 LLM 在全部五次运行中都请求了预期的预订工具。差异归结为响应速度和答案格式。
| 模型 | 以数字形式书写时间 | 首个答案 token 的时间(中位数) | 每轮成本 |
|---|---|---|---|
| DeepSeek V4.1 Flash | 0/5 | 1.9 秒(无推理时 1.4 秒) | $0.00017 |
| GPT-6 Luna | 0/5 | 2.3 秒(无推理时 1.6 秒) | $0.00008 |
| Claude Haiku 5.5 | 5/5 | 2.6 秒 | $0.00026 |
| Gemini 3.8 Flash | 0/5 | 4.0 秒 | $0.00129 |
| MiMo V2.6 Flash | 2/5 | 6.4 秒 | $0.00006 |
一个差异是模型如何处理口语时间。Claude Haiku 5.5 在每次运行中都把 7:30 写成数字,而非要求的口语形式,并且在一种配置中,它还将日期计算包含在可见响应中。当响应直接发送给 TTS 时,这两种行为都很重要。
这些结果来自单独的五次运行模型比较,而非上述完整的对话测量。它们测量的是在测试的模型设置下首个答案 token 的时间。
推理也影响了延迟。在我们的 GPT-6 Luna 比较中,禁用推理将测试工具轮次的首个答案 token 时间从 2.3 秒减少到 1.6 秒。对于相对受限的预订任务来说,这很有用,前提是模型继续做出正确的工具调用。
选择文本转语音模型
对于 TTS,我们关注每个模型开始返回音频的速度。我们向十二个模型发送了相同的 101 字符响应,每个模型重复请求五次。
下面展示了六个代表性结果:
| 模型 | 首字节服务器时间中位数 | 测试日期成本/百万字符 |
|---|---|---|
| Grok Voice TTS 1.0 | 80 毫秒 | $15 |
| Kokoro 82M | 98 毫秒 | $0.62 |
| Deepgram Aura-2 | 236 毫秒 | $30 |
| Eleven Flash v2.5 | 275 毫秒 | $20 促销价 |
| Eleven v4 Turbo | 1,245 毫秒 | $20 促销价 |
| Gemini 3.8 Flash TTS | 2,561 毫秒 | 无法直接比较(按 token 计价) |
Gemini 的定价使用不同的计费单位,因此我们将其排除在每字符成本比较之外,而不是将估算的换算视为报价费率。
在服务器端测量中,Grok Voice 和 Kokoro 返回首批音频字节最快,而 Eleven Flash v2.5 也表现良好。
然而,服务器报告的时间并不总是与我们的应用程序实际体验相符。Kokoro 记录的服务器延迟中位数低于 100 毫秒,但其音频到达我们测试环境需要三到八秒。
这就是为什么你应该在自己的应用中测量延迟,尤其是在构建对话式功能时。在选择模型时,你还需要试听生成的语音并检查发音,而不是仅凭速度做决定。
在这个实现中,我们使用了 Eleven Flash v2.5。你可以浏览可用的语音模型,并使用模型 slug 将其替换为其他选项。
使用 LiveKit Agents 运行同一个智能体
我们的 Python 脚本处理转录、工具调用和语音生成,但它不支持与智能体进行实时对话。要做到这一点,我们还需要处理麦克风输入、检测通话者何时说完,以及管理打断。
我们可以使用 LiveKit Agents 来实现这一点。它负责实时音频连接和对话流程,同时让我们继续通过 OpenRouter 使用同样的三个模型。
首先,安装带有 OpenAI 插件的 LiveKit Agents:
python -m pip install "livekit-agents[openai]~=1.8"你还需要从你的 LiveKit 项目中获取 LIVEKIT_URL、LIVEKIT_API_KEY 和 LIVEKIT_API_SECRET,以及我们之前配置的 OPENROUTER_API_KEY。如果你还没有设置项目,请按照 LiveKit 快速入门操作。
将以下内容保存为 livekit_agent.py,与 voice_agent.py 放在一起。我们将复用上一个示例中的餐厅指令和预订函数。
import os
from livekit import agents
from livekit.agents import (
Agent,
AgentServer,
AgentSession,
function_tool,
inference,
)
from livekit.plugins import openai
from voice_agent import run_tool, system_prompt
API = "https://openrouter.ai/api/v1"
KEY = os.environ["OPENROUTER_API_KEY"]
class Luigis(Agent):
def __init__(self):
super().__init__(instructions=system_prompt())
@function_tool
async def check_availability(
self, date: str, time: str, party_size: int
) -> dict:
"""Check available tables for a date, time, and party size."""
return run_tool(
"check_availability",
{"date": date, "time": time, "party_size": party_size},
)
@function_tool
async def book_table(
self, date: str, time: str, party_size: int, name: str
) -> dict:
"""Book a table after the caller agrees and gives a name."""
return run_tool(
"book_table",
{
"date": date,
"time": time,
"party_size": party_size,
"name": name,
},
)
server = AgentServer()
@server.rtc_session(agent_name="luigis")
async def luigis_session(ctx: agents.JobContext):
session = AgentSession(
vad=inference.VAD(model="silero"),
stt=openai.STT(
model="elevenlabs/scribe-v2",
base_url=API,
api_key=KEY,
),
llm=openai.LLM.with_openrouter(
model="openai/gpt-6-luna",
api_key=KEY,
reasoning_effort="none",
),
tts=openai.TTS(
model="elevenlabs/eleven-flash-v2.5",
voice="sarah",
response_format="pcm",
base_url=API,
api_key=KEY,
),
)
await session.start(room=ctx.room, agent=Luigis())
if __name__ == "__main__":
agents.cli.run_app(server)AgentSession 连接三个模型,而 Silero 语音活动检测(VAD)帮助识别通话者何时停止说话,以便 LiveKit 可以发送音频进行转录。LiveKit 还管理语音播放和打断。
注意,预订函数没有变化。LiveKit 的 @function_tool 装饰器使它们可供 LLM 使用,每个函数都会将通话者的详细信息传递给同一个模拟预订服务。
你可以使用麦克风在本地测试该智能体:
python livekit_agent.py console要将其连接到你的 LiveKit 项目以通过 Agent Console 进行测试,请运行:
python livekit_agent.py dev试着要求预订七点半的六人桌,然后接受八点十五分。在我们之前的文本驱动 LiveKit 测试中,这产生了以下工具调用:
check_availability(
date="2026-10-10",
time="19:30",
party_size=6
)
book_table(
date="2026-10-10",
time="20:15",
party_size=6,
name="Siobhan Okafor"
)此设置将预订工具连接到 LiveKit 音频会话,因此你可以使用麦克风测试同样的对话。你也可以使用我们的 LiveKit 集成指南来探索模型回退和提供商路由。
你可能会遇到的错误
在测试过程中,我们遇到了一些由缺少参数或不支持的模型设置引起的错误。这三个错误都返回了 HTTP 400 响应:
| 问题 | API 响应 | 如何修复 |
|---|---|---|
| 为 ElevenLabs TTS 模型省略了语音 | 此 TTS 提供商需要显式指定语音。 | 指定一个受支持的语音,例如 sarah。 |
| 向 STT 发送了 data:audio/wav;base64,… URI | 提供商无法处理音频输入(不支持的或格式错误的音频)。 | 发送带有正确音频格式的原始 base64 数据。 |
| 在强制要求推理的地方禁用了推理 | 此端点强制要求推理,无法禁用。 | 移除不支持的推理设置。 |
我们的语音转文本 API 期望音频字节以 base64 编码在 input_audio.data 中,不带 data:audio/…;base64, 前缀。你还必须指定音频格式,例如 wav 或 mp3。
语音名称也因提供商而异。例如,OpenAI 的 alloy 等语音不会自动适用于 ElevenLabs。你可以在我们的文本转语音文档中或通过模型 API 查看每个模型支持的语音。
我们遇到的另一个问题是,空的 TTS 输入会返回一段很短的音频片段。在发起语音请求前,最好先过滤掉空字符串。
你需要语音到语音的模型吗?
这个示例不需要。我们构建的语音转文本、LLM 和文本转语音的独立流水线足以处理我们的餐厅预订工作流。
我们还测试了将呼叫者的音频直接发送给四个支持音频的 LLM:Gemini 3.8 Flash、Qwen3.8 Omni Flash、MiMo V2.6 Flash 和 GPT Audio Mini。它们在全部 12 次测试运行中都选择了预期的预订工具。
这表明这些模型无需单独的转录请求就能处理我们示例中的工具调用部分。不过,我们并未测量完整的语音到语音对话,因此这些结果无法告诉我们该方法是否会更快或产生更好的语音回复。
采用三阶段方法,我们可以检查转录文本、找出错误发生的位置,并独立地更改转录、语言或语音模型。直接处理音频输入和输出的模型可能会简化流水线,但在选择之前,值得比较完整的响应时间、音频质量和中断处理能力。
结论
我们的 Python 示例只是一个起点。在将智能体连接到真实呼叫者之前,请用你自己的后端替换模拟预订函数,并用真实语音进行测试。只有系统提示会让智能体在调用 book_table 之前等待呼叫者同意,因此请让你的后端在写入预订之前先确认预订。要测量完整的响应时间,包括网络传输和播放,而不是仅依赖服务器端延迟。
如果你需要实时音频和中断处理,请使用 LiveKit,同时通过 OpenRouter 保持相同的模型。
常见问题
如何将语音 AI 智能体连接到真实电话号码?
你可以使用 LiveKit 的电话集成将语音 AI 智能体连接到电话号码。LiveKit 支持通过其电话号码服务或 SIP 提供商接听来电。配置所需的中继和调度规则,将来电路由到 LiveKit 房间,你的智能体可以在那里加入对话。有关设置步骤,请参阅 LiveKit 电话文档。
语音 AI 智能体能处理中断吗?
可以。像 LiveKit 这样的框架可以检测到有人在智能体回复时开始说话,暂停智能体的语音,并用新的输入继续对话。
请记住,中断语音并不会自动取消已经在运行的工具。对于预订或支付等操作,你的应用应谨慎处理取消和确认。
我可以为 AI 语音智能体使用自己的声音吗?
可以,只要你选择支持声音克隆的文本转语音模型。OpenRouter 上的一些模型通过 input_references 参数接受参考音频,并使用该声音生成语音,而无需单独的声音训练过程。
声音克隆支持因模型和提供商端点而异,因此在选择模型之前请查看 TTS 文档。仅使用你有权复制的录音。
我可以将语音智能体连接到真实的预订系统或其他 API 吗?
可以。我们示例中的 check_availability 和 book_table 函数使用模拟数据,但你可以将其实现替换为对预订 API 或数据库的请求。
LLM 仍然决定何时调用每个函数,而你的应用负责执行实际操作并返回结果。对于预订餐桌或发送确认等操作,请确保操作成功后再告知调用方已完成。
来源:OpenRouter Announcements · openrouter.ai