Grok Voice Agent Builder：Python 实战指南

xAI 发布了Voice Agent Builder，一款用于创建语音代理的控制台。您可以描述呼叫流程、附加文档和工具，并选择语音。

测试语音代理控制台时，我不太在意发布说明，更关心我需要接入代码的部分：如何配置 WebSocket 会话、音频如何传输、工具调用发生在何处、通话成本是多少、以及其他应用如何调用该工作流。

下面的代码直接基于 Voice Agent API 重建了该流程。具体来说，我们将使用一个门诊预约助手：它会检查可用性、用语音回复、跟踪成本、处理工具故障，并通过 FastAPI 暴露一个端点。

什么是 Grok Voice Agent Builder？

Voice Agent Builder 是 xAI 在Grok Voice 上用于创建和部署语音代理的控制台。它于 2026 年 7 月 1 日以测试版发布。相比分别使用语音转文本、语言模型和文本转语音服务，它采用一条语音模型路径。

该控制台包含电话功能、文档检索、工具与连接器、护栏、远程 MCP 服务器，以及带录音、转录与追踪的通话日志。

音频按分钟计费。控制台仍处于测试阶段，因此我们直接使用 API。

Builder 之下的 Grok Voice Agent API 如何工作

控制台之下是Voice Agent API，这是一个实时 WebSocket API，暴露了与 Builder 相同的运行时。

Builder 构建于 Voice API 之上。作者制图。

此处使用的模型为 grok-voice-think-fast-1.0。别名 grok-voice-latest 指向最新模型。本文使用该别名，但在部署应用中我会锁定带版本号的名称。xAI 报告该模型在 τ-voice Bench 榜单上得分 67.3%；我将其视为一个数据点，而非保证。

兼容性说明：该 API 与OpenAI Realtime API 兼容。如果您已有与 OpenAI 实时端点通信的代码，基本只需更换基 URL 和密钥即可。

项目概览：我们将构建什么

该门诊助手接收语音输入、以生成语音回复、提出追问、在提供时间段前检查可用性，并在需要时转人工。核心示例使用一个工具；Streamlit 演示增加了预订、转接与结束通话操作。

核心教程拆分为四个文件，每个文件各司其职：

• voice_client.py 存放 WebSocket 客户端、音频辅助方法与成本跟踪

• tools.py 存放 check_availability，以及 Streamlit 使用的其他演示工具

• assistant.py 存放系统提示、会话配置与工作流

• app.py 通过 FastAPI 提供服务

这四个文件是本文的主线。代码库还包含用于可视化演示的 app_streamlit.py 和作为 Windows 启动器的 run.py，但我们会在核心流程跑通后再介绍它们。

前置条件

在运行代码之前，您需要 Python 3.10 或更高版本、xAI 账户、来自console.x.ai 的 API 密钥、预付额度，以及对环境变量、JSON 与 WebSocket 的基本了解。

项目初始化

创建文件夹与虚拟环境，然后安装依赖包：

mkdir appointment-agent
cd appointment-agent
python -m venv .venv
.venv\Scripts\activate       # macOS/Linux: source .venv/bin/activate
pip install websockets python-dotenv fastapi uvicorn pydantic httpx numpy streamlit

将这些包固定在 requirements.txt 中，以便全新检出时使用相同的环境。

在 Python 文件旁创建一个 .env 文件：

XAI_API_KEY=xai-your-key-here

将 .env 加入 .gitignore。API 密钥应保存在服务器上。

构建语音代理

开始动手构建。

通过 WebSocket 连接 Grok Voice Agent API

第一步是打开连接。将模型作为查询参数传递，并在握手时将您的密钥作为 Bearer 令牌发送：

import asyncio
import json
import os
import websockets

async def voice_agent():
    url = "wss://api.x.ai/v1/realtime?model=grok-voice-latest"
    async with websockets.connect(
        url,
        additional_headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
    ) as ws:
        async for message in ws:
            print(json.loads(message)["type"])

asyncio.run(voice_agent())

使用有效密钥时，您首先会看到 session.created 事件，表示套接字已打开并可进行配置。

Session created 事件确认了连接。作者制图。

配置语音会话

一个已连接的套接字并非已配置的代理。您需要发送 session.update 事件并附带 session 对象来进行设置。

语音、音频格式与指令

最常调整的三个设置是语音、音频格式和系统提示。实时 API 提供五种命名语音： eve、 ara、rex、sal 和 leo，以及任何自定义克隆。音频默认为 audio/pcm 、采样率 24000 Hz，输入与输出可分别配置。

以下是助手使用的会话配置，定义在 assistant.py：

def build_session_config(voice="ara", instructions=SYSTEM_PROMPT, sample_rate=24000):
    # The model needs to know "today" or it guesses the year for a date like "July 6th".
    instructions = f"{instructions}\nToday's date is {date.today().isoformat()}."
    return {
        "voice": voice,
        "instructions": instructions,
        "turn_detection": None,  # manual turns for file-based input
        "audio": {
            "input": {"format": {"type": "audio/pcm", "rate": sample_rate}},
            "output": {"format": {"type": "audio/pcm", "rate": sample_rate}},
        },
        "tools": [CHECK_AVAILABILITY_TOOL],
    }

instructions 字段是系统提示。此门诊提示保持简短，因为冗长的语音回复难以跟上：

You are a voice appointment assistant for a small clinic. Help callers book,
reschedule, cancel, or ask questions about appointments, services, and hours.
Answer whatever the caller asks that relates to the clinic. Keep responses short
and natural for a phone conversation. Ask one question at a time. Confirm
important details before taking action. Use the availability tool before offering
a time slot. Escalate to a human for medical, urgent, sensitive, or unclear
requests. If a caller asks about something unrelated to the clinic, say briefly
that it is outside what you can help with, then steer back to booking. If you
cannot make out what the caller said, ask them to repeat it instead of repeating
your last message.

升级转人工的说明有助于让门诊代理避免提供医疗建议。最后两行有助于保持范围并在来电者表达不清时避免循环。配置中还追加了今天的日期，因为在我的实测中，模型可能会对类似 “July 6th” 这样的日期猜错年份。

调整回合检测

回合检测用于判断来电者是否已说完。将 turn_detection.type 设为 server_vad 时，服务器会在静音后结束回合。保持为 null 则由您通过提交音频缓冲来控制回合，这也是我在文件流程中使用的方法。

服务器端 VAD 有三个值得了解的设置：threshold 控制被视为语音的音量阈值，silence_duration_ms 控制多长的停顿会结束回合，prefix_padding_ms 保留语音开始前的一小段音频。如果您的代理打断来电者，请优先提高 silence_duration_ms。

向代理发送音频

现在发送来电者的语音。音频必须与会话格式匹配：单声道 16 位 PCM、24000 Hz，以 base64 编码并分块发送。

客户端以切片流式传输文件，然后提交缓冲以标记回合结束：

async def send_audio(self, pcm_bytes, chunk_ms=100, commit=True):
    bytes_per_chunk = int(self._sample_rate * 2 * chunk_ms / 1000)
    for start in range(0, len(pcm_bytes), bytes_per_chunk):
        chunk = pcm_bytes[start:start + bytes_per_chunk]
        await self._t.send({
            "type": "input_audio_buffer.append",
            "audio": base64.b64encode(chunk).decode(),
        })
    if commit:
        await self._t.send({"type": "input_audio_buffer.commit"})
    self.cost.audio_seconds += pcm_seconds(pcm_bytes, self._sample_rate)

如果您的采样率或编码与 session.update 不匹配，您可能会听到噪声或静音，而不是清晰的报错。音频通过 input_audio_buffer.append 发送，因此按时长计费，而非按消息。

接收语音回复

请求回复后，音频以 response.output_audio.delta 到达，转录以 response.output_audio_transcript.delta 到达，response.done 则关闭该回合。

客户端在一个 async 循环中收集这些内容：

async def _collect_response(self):
    audio = bytearray()
    transcript, calls = [], []
    while True:
        event = await self._recv()
        etype = event["type"]
        if etype == "response.output_audio.delta":
            audio += base64.b64decode(event["delta"])
        elif etype == "response.output_audio_transcript.delta":
            transcript.append(event.get("delta", ""))
        elif etype == "response.function_call_arguments.done":
            calls.append(event)
        elif etype == "response.done":
            break
    return bytes(audio), "".join(transcript), calls

解码音频增量，按顺序追加，并将结果写入 response.wav 文件。若要捕获来电者原话，请设置 audio.input.transcription 并读取 conversation.item.input_audio_transcription.completed。

构建预约助手工作流

现在将各部分串成一段对话：预约请求、澄清问题、可用性检查、提供时间段、确认。为在回合之间传递上下文，每个新回合都会携带会话 ID 重新连接，并选择恢复会话。

为语音代理添加工具调用

对门诊而言，代理必须在承诺时间前检查可用性。自定义工具是模型触达您代码的方式：它发出请求，您的应用运行函数，然后您将结果发回。

该工具是一个普通函数加一个进入会话配置的 JSON 架构。以下是来自 tools.py 的架构：

CHECK_AVAILABILITY_TOOL = {
    "type": "function",
    "name": "check_availability",
    "description": "Look up open appointment slots for a service on a given date. "
                   "Always call this before offering the caller a time.",
    "parameters": {
        "type": "object",
        "properties": {
            "service": {"type": "string", "description": "Service requested."},
            "date": {"type": "string", "description": "Requested date as YYYY-MM-DD."},
        },
        "required": ["service", "date"],
    },
}

循环有固定形态。当模型需要工具时，会发送 response.function_call_arguments.done 并附上参数。您运行函数，返回 function_call_output，然后发送 response.create 使代理继续。若遗漏最后的 response.create，代理会沉默。

工具调用往返流程示意。作者制图。

此类自定义函数在您的代码中运行。Streamlit 演示从同一文件注册了另外三个： book_appointment、transfer_to_human 和 end_call。内置工具（例如 网页搜索、X 搜索、集合 搜索以及远程 MCP 工具）在 xAI 服务器上执行。

处理工具失败

工具会失败，而一个假设成功的语音代理可能会承诺不存在的时间段。我的 ToolRegistry.execute 从不抛出异常：查找失败会以 {"error": ...} 字典返回。

def execute(self, name, arguments):
    handler = self._handlers.get(name)
    if handler is None:
        return {"error": f"unknown tool: {name}"}
    try:
        return handler(**arguments)
    except ToolError as exc:
        return {"error": str(exc)}

显式的错误状态可防止代理将失败的工具调用当作成功。

添加成本跟踪

在向任何人提供服务之前，先弄清一通电话的成本。音频按每分钟 $0.05 计费，包含您发送与接收的音频。文本输入事件按每次 $0.004 计费。function_call_output 结果与 response.create 事件不计费。

客户端在过程中进行跟踪，因此您可随时读取成本属性：

@property
def audio_usd(self):
    rate = 0.05 + (0.01 if self.telephony else 0.0)
    return self.audio_seconds / 60 * rate

@property
def total_usd(self):
    return self.audio_usd + self.text_usd + self.tool_usd

使用 xAI 提供的号码会增加每分钟 $0.01 的电话附加费，当您设置 telephony=True 时，辅助方法会考虑该项。由 xAI 托管的工具单独计费：网页搜索与 X 搜索约为每千次 $5，文件搜索约为 $2.50。

处理错误与边界情况

大多数失败都属于以下少数几类：

• 缺失或无效的 API 密钥会在握手时返回 401，因此先检查密钥

• 被封禁的团队返回 403，触发速率限制返回 429，应使用退避重试

• 格式错误的会话配置返回 400，通常是字段名拼写错误

• 不受支持的音频格式会产生噪声而非错误，因此请匹配会话采样率

• 工具返回结果后缺少 response.create 会导致代理挂起

• 重复的预订尝试可能造成严重问题，因此不要盲目重试

重试读取型操作如 check_availability 是安全的，但重试写入型操作如实际预订可能造成双重预订。任何改变数据的动作都需要先进行幂等性检查。

在客户端应用中使用短期令牌

以上均假设代码在您的服务器上运行，API 密钥也应放在这里。如果浏览器或移动端直接连接，请使用短期令牌。

您的服务器调用 POST https://api.x.ai/v1/realtime/client_secrets 并携带密钥，获得令牌响应后将令牌值传给客户端。在我的运行中，响应包含 value 与 expires_at：

@app.post("/session")
async def create_session():
    async with httpx.AsyncClient() as client:
        response = await client.post(
            CLIENT_SECRETS_URL,
            headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
            json={"expires_after": {"seconds": 300}},
        )
    return response.json()

浏览器无法设置自定义 WebSocket 头，因此令牌会附加在带有 xai-client-secret. 前缀的 sec-websocket-protocol 头中。

将工作流封装为 FastAPI 端点

通过端点让前端或其他服务调用工作流。该路由使用 Pydantic 模型校验请求体，接收文本消息或音频路径，并返回转录、回复音频、工具日志、时延与预估成本。

@app.post("/appointments/voice")
async def appointments_voice(body: VoiceRequest):
    fail = {"check_availability"} if body.simulate_tool_failure else None
    assistant = AppointmentAssistant(voice=body.voice, telephony=body.telephony, fail_tools=fail)
    if body.text:
        result = await assistant.run_live(text=body.text, conversation_id=body.conversation_id)
    else:
        pcm = load_wav_as_pcm(body.audio_path, 24000)
        result = await assistant.run_live(pcm, conversation_id=body.conversation_id)
    return {
        "transcript": result.transcript,
        "audio_wav_base64": base64.b64encode(encode_wav_bytes(result.audio, 24000)).decode(),
        "tool_calls": result.tool_calls,
        "latency_seconds": round(result.latency_s, 3),
        "estimated_cost_usd": round(result.cost.total_usd, 6),
        "audio_seconds": round(result.cost.audio_seconds, 2),
        "conversation_id": result.conversation_id,
    }

使用 uvicorn app:app --reload 运行，并打开 http://localhost:8000/docs。从服务器环境读取 XAI_API_KEY ，切勿从请求体接收。