跳到正文
原文
OpenRouter:Blog(Announcements)·· 19 小时前精选AI 评分64

OpenRouter 教程:如何用 Python 搭建语音 AI 智能体

How to Build a Voice AI Agent

文章要点
  • 1OpenRouter 发布 Python 语音智能体教程,覆盖语音识别、工具调用与语音合成。
  • 2作者实测中,12 轮对话服务端首音中位时延 1.9 秒,无工具轮 1.4 秒、有工具轮 2.4 秒。
  • 3作者实测在促销价下四轮对话约 $0.013 每分钟处理语音,语音合成占成本约 90%。
  • 410 款语音转文字模型对比中,Scribe v2 与准确率速度兼备,受测 LLM 均能调用订座工具。
决策要点

语音智能体的搭建门槛已降到用公开 API 拼装的程度,真正的约束转向时延编排与以语音合成为主的成本结构。

  • 影响:预订、客服等电话场景可按每分钟处理语音约 $0.013 至 $0.025 的量级先做验证;成本大头在文本转语音而非 LLM;每次工具调用约多一秒时延,真实后端须确认操作成功再告知客户。
  • 参考:可以先用真实业务录音测试入围的语音转文字与语音合成模型,并在自身应用里测含网络与播放的端到端响应时间;试点时控制口语回复长度,比省 LLM token 对成本影响更大;接真实预订后端时确认操作成功再回复客户。
  • 依据:时延与成本数据均为 OpenRouter 自测(作者实测,暂无第三方验证),基于合成英语录音与其测试环境;ElevenLabs 促销价含 50% 上线折扣,2026 年 10 月 19 日结束。
正文 · AI 翻译

语音智能体需要做的远不止生成一个好的回答。它必须理解对方说了什么,决定要做什么,调用所需的工具,并在停顿变得尴尬之前重新开口说话。

在本教程中,我们将用 Python 构建完整的这一循环。我们的示例是一个餐厅订位智能体,它能转录来电者的语音,用工具检查餐桌空位情况,并将响应转换回语音。语音转文字、LLM 和文字转语音都通过同一个 OpenRouter API 密钥运行,而且我们将把每个完成的句子流式传输给 TTS,而不是等待整个回复。

在三次测试对话的全部 12 个回合中,服务端首段音频时间的总体中位数为 1.9 秒,而按照我们 10 月 8 日测试时可用的促销价格计算,一个四回合对话每分钟处理语音的成本约为 $0.013。

太长不看版

  • 文字转语音约占实测模型成本的 90%,这使得口语响应的长度成为该智能体的一项重要成本因素。

  • 在我们的测试中,按每个回合一次 LLM 请求计算,普通回合的服务端首段音频时间估计为 1.4 秒,工具回合为 2.4 秒。示例代码会产生额外的 LLM 请求,而网络和播放也会增加时间。

  • 人名比电话号码更难转录。在采取行动之前,让来电者确认重要细节。

  • 将完整的句子发送给文字转语音(TTS)可以减少口语输出的等待时间,但实时播放和打断需要额外的编排。

语音 AI 智能体回合的工作原理

语音智能体会接收来电者的语音,将其转换为文字,传递给 LLM 以决定要做什么,然后将响应转换回语音。该流程的工作方式如下:

Diagram of one voice AI agent turn: caller audio goes to speech-to-text, then the LLM, a sentence splitter, text-to-speech and agent audio. The LLM can call the booking tools check_availability and book_table and continue with the tool result.

例如,当有人致电我们的餐厅预订餐桌时,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.

这个示例展示了智能体先查询空位情况,然后提供另一个时间,并在来电者接受后调用 book_table。它还拒绝承诺发送文字确认,因为我们没有提供发送文字确认的工具。

这种设计为何有效,以及什么会破坏它

一条回复在聊天窗口中可能看起来完全没问题,但朗读出来却会显得奇怪。时间、电话号码和 markdown 都是典型的例子。这就是为什么我们的系统提示要求 LLM 保持回复简短、避免格式化,并将时间写成口语形式,比如用“eight fifteen”而不是 8:15。

同样的问题也可能反向发生。在我们的测试中,语音转文字将 “eight fifteen” 转成了 815,但 LLM 仍正确理解了来电者的回答,因为它拥有之前的对话作为上下文。我们在 messages 中维护这段历史并随每次请求一起发送,因为 Chat Completions 不会自动保留之前的轮次。因此,提示长度从第一轮的约 290 个 token 增长到了第四轮的 490 个。

另一个重要的考虑因素是确保智能体只确认实际发生的操作。在没有 book_table 的早期版本中,即使应用尚未创建预订,智能体也可能告诉来电者其预订已确认。

这就是为什么预订工具需要做的不仅仅是返回一条响应。当连接到真实的预订系统时,它应该验证空位情况、保存预订,并且只在操作成功时才返回确认。

语音 AI 智能体到底有多快?

在全部 12 轮对话中,服务端到首段音频的中位时间为 1.9 秒。按轮次类型划分,不调用工具时中位时间为 1.4 秒,智能体需要调用工具时为 2.4 秒。

我们测量了涵盖 12 轮的三段完整对话,使用我们的 生成元数据 来估算各阶段的服务端延迟。这让我们能把模型处理时间与测试环境中额外的网络开销区分开来。

以下是结果对比:

阶段有工具调用无工具
语音转文字(Scribe v2)549 ms391 ms
LLM 选择工具736 ms—
LLM 回答,首个 token685 ms706 ms
文字转语音,首字节262 ms208 ms
预计到首段音频的时间2.4 s1.4 s

额外的 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 价格包含了已于 2026 年 10 月 19 日结束的 50% 上线折扣。按常规标价计算,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-2891,740 ms$0.11
elevenlabs/scribe-v288379 ms$0.11
openai/whisper-large-v388517 ms$0.17
x-ai/grok-stt-1.086264 ms$0.10
deepgram/nova-38383 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 Flash0/51.9 s(不启用推理时 1.4 s)$0.00017
GPT-6 Luna0/52.3 s(不启用推理时 1.6 s)$0.00008
Claude Haiku 5.55/52.6 s$0.00026
Gemini 3.8 Flash0/54.0 s$0.00129
MiMo V2.6 Flash2/56.4 s$0.00006

一个差异在于模型如何处理口语化的时间表达。Claude Haiku 5.5 在每次运行中都写成了 7:30,而非所要求的口语形式;在一种配置下,它还将日期计算过程包含在了可见回复中。当回复直接送入 TTS 时,这两种行为都很重要。

这些结果来自单独进行的五次运行模型比较,而非上文完整的对话测量。它们衡量的是在所测模型设置下首个回答 token 的生成时间。

推理也会影响延迟。在我们的 GPT-6 Luna 比较中,在所测的工具调用轮次上,禁用推理将首个回答 token 的生成时间从 2.3 秒降低到了 1.6 秒。对于相对受限的预订任务而言这很有用,前提是模型仍能继续做出正确的工具调用。

选择文本转语音模型

对于 TTS,我们关注的是每个模型开始返回音频的速度。我们向十二个模型发送了同一段 101 字符的回复,每个模型重复请求五次。

以下展示了六个有代表性的结果:

模型服务器首字节时间中位数测试当日每百万字符成本
Grok Voice TTS 1.080 ms$15
Kokoro 82M98 ms$0.62
Deepgram Aura-2236 ms$30
Eleven Flash v2.5275 ms$20 促销价
Eleven v4 Turbo1,245 ms$20 促销价
Gemini 3.8 Flash TTS2,561 ms不直接可比(按 token 计价)

Gemini 的定价采用不同的计费单位,因此我们将其排除在每字符成本比较之外,而没有把估算换算当作报价费率。

在服务器端测量中,Grok Voice 和 Kokoro 返回首个音频字节的速度最快,Eleven Flash v2.5 的表现也不错。

然而,服务器报告的时间并不总是与我们的应用实际体验到的一致。Kokoro 记录到的服务器中位延迟低于 100 毫秒,但其音频到达我们的测试环境花了三到八秒。

这就是为什么你应该在自己的应用中测量延迟,尤其是在构建对话类应用时。在仅凭速度选择模型之前,你还应该试听生成的语音并检查发音。

在本实现中,我们使用了 Eleven Flash v2.5。你可以浏览可用的语音模型,并使用模型 slug 将其替换为其他选项。

使用 LiveKit Agents 运行同一个智能体

我们的 Python 脚本可以处理转录、工具调用和语音生成,但它不支持与智能体进行实时对话。要做到这一点,我们还需要处理麦克风输入、检测通话者何时说完话,并管理打断。

我们可以使用 LiveKit Agents 来实现这一点。它处理实时音频连接和对话流程,同时让我们继续通过 OpenRouter 使用同样的三个模型。

首先,安装 LiveKit Agents 和 OpenAI 插件:

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 模型缺少 voice 参数此 TTS 提供商需要明确指定 voice。指定一个受支持的语音,例如 sarah。
向 STT 发送了 data:audio/wav;base64,… URI提供商无法处理该音频输入(音频不受支持或格式错误)。发送原始 base64 数据并附带正确的音频格式。
在强制要求推理的情况下禁用了推理此端点强制要求推理,无法禁用。移除不受支持的 reasoning 设置。

我们的语音转文本 API 要求音频字节以 base64 编码放在 input_audio.data 中,且不带 data:audio/…;base64, 前缀。你还必须指定音频格式,例如 wav 或 mp3。

语音名称也因提供商而异。例如,OpenAI 的语音(如 alloy)无法自动用于 ElevenLabs。你可以在我们的文本转语音文档中或通过 models API 查看每个模型支持的语音。

我们遇到的另一个问题是,空的 TTS 输入会返回一段很短的音频。值得在发起语音请求之前先过滤掉空字符串。

你需要语音到语音(speech-to-speech)模型吗?

在这个示例中不需要。我们构建的由语音转文字、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 提供商支持呼入电话。配置所需的 trunk 和分发规则,将呼入电话路由到 LiveKit 房间,您的智能体即可加入对话。设置步骤请参见 LiveKit 电话文档。

语音 AI 智能体能处理打断吗?

可以。像 LiveKit 这样的框架能够检测到智能体回应时有人开始说话,暂停智能体的语音,并根据新输入继续对话。

请记住,打断语音并不会自动取消已经在运行的工具。对于预订或支付等操作,您的应用应谨慎处理取消和确认。

我可以为自己的 AI 语音智能体使用自己的声音吗?

可以,只要您选择支持声音克隆的文字转语音模型。OpenRouter 上的一些模型通过 input_references 参数接受参考音频,并使用该声音生成语音,无需单独的声音训练过程。

声音克隆的支持因模型和提供商端点而异,因此在选择模型之前请查看 TTS 文档。只能使用您有权复制的录音。

我可以将语音智能体连接到真实的预订系统或其他 API 吗?

可以。示例中的 check_availability 和 book_table 函数使用的是模拟数据,但您可以将它们的实现替换为对您的预订 API 或数据库的请求。

LLM 仍然决定何时调用每个函数,而你的应用程序负责执行实际操作并返回结果。对于预订餐桌或发送确认信息之类的操作,请确保操作成功后再告知调用方已完成。

来源:OpenRouter:Blog(Announcements) · openrouter.ai