Grok Voice Think Fast 2.0 API 教程：用 Python 构建实时语音代理

SpaceXAI 的 Grok Voice Think Fast 2.0 是一款语音到语音模型。您通过 WebSocket 发送音频，它返回音频；在此期间，它可以一边推理一边继续说话，同时已决定要调用的函数已在运行。无需单独的语音转文本步骤，也无需单独的文本转语音步骤。

SpaceXAI 于 2026 年 7 月 29 日发布 Think Fast 2.0：更快的首包音频、更稳的全双工行为（边说边听，不再严格轮流），以及在轮次早期即触发的工具调用。基准测试我只会一笔带过，教程真正重要的是代码需要改什么。

我们将构建一个网店客服语音代理。来电者可以查询订单、在没有订单号的情况下用邮箱找订单、修改派送指示、取消订单、在坐席说话时插话，以及断线后继续对话。这是 API 路径，不是我们在 Grok Voice Agent Builder 教程中演示的零代码构建器。若走控制台优先的版本，请从那里开始。

Grok Voice Think Fast 2.0 是什么？

Grok Voice Think Fast 2.0 是 SpaceXAI 面向 语音到语音 API的最新模型，这是大家常说的 Grok Voice 背后的产品名。如果您还把这家公司记作 xAI，是同一拨人：2026 年 7 月 6 日并入 SpaceX 并更名为 SpaceXAI。API 没跟着改名，所以下面的各类标识依然是 xai，从 XAI_API_KEY 变量到 api.x.ai 主机名。

传统语音栈要串三项服务：语音转文本、语言模型、再文本转语音，每一跳都会增加时延、也容易丢上下文。Think Fast 2.0 把它们收敛为一个模型，通过同一连接接收音频或文本，并输出音频或文本。

语音到语音 WebSocket 与三服务流水线架构对比。作者供图。

对于会“行动”的代理而非只会“说话”的代理而言，关键在于推理与语音并行。SpaceXAI 表示，工具调用“通常”会在坐席说完第一句话之前就开始执行，这个“通常”一词并非虚言。

按 SpaceXAI 引用 Artificial Analysis 的基准，Think Fast 2.0 在 Speech to Speech Index 上得分 82.9%，而 1.0 为 75.7%，首包音频时间从 1.25 秒降至 0.70 秒。厂商在通用基准上的数字只是对您来电流程的一个假设，而非测试计划。

您会看到三个模型字符串：grok-voice-latest、grok-voice-think-fast-2.0 和 grok-voice-think-fast-1.0。别名在原型期很方便，但不足以支撑其他用途。

我在 2026 年 8 月 4 日测试时，grok-voice-latest 仍解析为 grok-voice-think-fast-1.0，而 SpaceXAI 的 版本说明 计划第二天切换到 Think Fast 2.0。此切换既是模型变化，也涉及价格变化：2.0 每分钟音频 $0.08，而 1.0 为 $0.05，因此未固定版本的别名会在您代码未改一行的情况下变贵。上线部署务必固定版本字符串。

我们要构建什么

该代理覆盖客服热线常见诉求：查询订单、在来电者没有订单号时用邮箱查找、修改派送指示、取消订单、创建或查询工单，并可转接人工。过程中会出现插话与断线。

我们用少量小文件而非一个脚本，因为每个部分职能不同，且您需要分别测试。结构如下：

• config.py 加载 API 密钥，保存模型字符串、采样率与端点 URL

• voice_client.py 封装 WebSocket，跟踪计费，并提供发送/接收辅助方法

• tools.py 定义订单函数，以及一个内存级订单存储来代替真实数据库

• assistant.py 保存系统提示、会话配置，以及将一切串联起来的事件循环

• token_server.py 是一个用于签发临时令牌的小型 FastAPI 端点

• app_streamlit.py 将同一客户端置于实时浏览器通话之后，测试章节后我会回到它

教学路径从终端开始；演示版再加上麦克风。

先决条件

您需要一个带 API 密钥的 SpaceXAI 账户、已充值的计费安排（没有永久免费层，且新账号的促销额度不够支撑）、以及对 asyncio 和 WebSocket 的足够熟悉，能在没有逐行讲解 await 的情况下跟上。

SpaceXAI 的快速上手示例使用原生 websockets 包而非专用 SDK，我们也如此。文档未说明所需 Python 版本。我在 3.11 上测试通过。

将 API 密钥保存在服务器端。如果浏览器或移动端直接调用 Voice API，应使用临时令牌，详见下方安全章节。

项目搭建

下列每个文件都在项目仓库中，您可以直接克隆而非手动拷贝片段：

git clone https://github.com/KhalidAbdelaty/grok-voice-think-fast-2.0.git
cd grok-voice-think-fast-2.0
pip install -r requirements.txt

websockets 负责实时连接， python-dotenv 读取您的密钥。其余依赖用于令牌端点与浏览器演示。将密钥写入 .env：

XAI_API_KEY=xai-your-key-here

这基本完成了设置。连接部分才是重点。

理解 Grok Voice 实时 API

Grok Voice 是产品名。您真正要对接的是 wss://api.x.ai/v1/realtime 的 WebSocket 端点，整段对话都以 JSON 事件流的形式在这一个 socket 上完成。

事件生命周期

连接遵循固定流程：一连上，服务器会发送 session.created 与 conversation.created，您发送 session.update 配置语音与工具，服务器用 session.updated 确认，之后您创建对话项并请求回复。我用真密钥验证，顺序与文档完全一致。

• session.update （客户端）配置语音、指令、工具与音频格式

• conversation.item.create （客户端）添加用户消息、助手消息或工具结果

• response.create （客户端）请求模型说话；服务器的 VAD 会为您自动发送该事件

• response.output_audio.delta 与 response.output_audio_transcript.delta （服务器）在生成时流式传输回复

• response.done （服务器）结束当前轮次

有两点容易踩坑。我前面链接的语音到语音文档页在会话续接时提到了 conversation.item.created 事件，但权威的 事件参考 只列了 conversation.item.added，而我所有测试中到达的也是它，所以应基于它编写代码。您还会在多数连接的几秒后看到一个未文档化的 ping 事件，仅提一嘴，以免将其误读为错误。

音频格式与传输

编解码与传输是两件事。编解码通过 audio.input.format 与 audio.output.format 设置，可选 audio/pcm（Linear16，默认 24000 Hz）、audio/pcmu 或 audio/pcma （G.711 8 kHz，面向电话）、或 audio/opus（24 kHz）。传输决定这些字节如何“上线”：

• json（默认）在 input_audio_buffer.append 与 response.output_audio.delta 中将音频作为 base64 文本发送，便于日志与调试

• binary 将原始编解码字节作为 WebSocket 二进制帧发送，省掉 base64 开销，但接收循环需按消息类型分支

先用 JSON。文档中的示例全用它，检查起来很直观，且 base64 开销并非客服代理构建的瓶颈。若测到确有必要，再迁移到二进制。

与 OpenAI Realtime API 的兼容性

如果您没用过 OpenAI 的 Realtime API，可跳过本节。对其他人而言，语音到语音 API 与 OpenAI Realtime API 的演进基本同步，大多数客户端代码改改基址 URL 与密钥就能迁移，但并非完全可替换。

此处转写到达为 conversation.item.input_audio_transcription.updated，而非 OpenAI 的 delta；有些 OpenAI 事件不受支持，SpaceXAI 也加入了自家扩展：用于脚本化披露语的 force_message、用于重连的 resumption、以及在文本转语音前修正品牌名发音的 replace。

构建实时语音代理

协议聊够了。下面是与之对话的客户端。

连接并配置会话

连接使用 Bearer 令牌与模型查询参数打开，您发出的第一条消息配置代理的全部行为：

import asyncio
import json
import os
import websockets

MODEL = "grok-voice-think-fast-2.0"  # pin the version, not grok-voice-latest

async def connect():
    url = f"wss://api.x.ai/v1/realtime?model={MODEL}"
    ws = await websockets.connect(
        url, additional_headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"}
    )
    await ws.send(json.dumps({
        "type": "session.update",
        "session": {
            "voice": "eve",
            "instructions": SYSTEM_PROMPT,
            "turn_detection": {"type": "server_vad"},
            "tools": ORDER_TOOLS,
            "resumption": {"enabled": True},
        }
    }))

return ws

instructions 是系统提示，该模型偏好简短的提示。SpaceXAI 的迁移说明建议精简为 GPT 时代语音模型编写的提示，而非照搬。我给代理的指令是保持简短回答、一次只问一个问题、执行写操作前复述确认。口头确认是 UX 讲究，不是安全控制；您的应用仍需在实际写入时强制鉴权。

有一点让我意外：无法识别的模型字符串在连接时不会报错，而是默默回退到 grok-voice-think-fast-1.0。仅因拼写错误就降级付费请求且不提示，默认行为颇为奇怪。请在启动时记录一次 session.created 的 session.model 字段，核对是否拿到所求模型。

终端输出显示连接后收到 session.created。作者供图。

流式发送用户音频

当 turn_detection.type 设为 server_vad 时，您只需持续追加音频。服务器会判断来电者何时停止说话，并为您触发响应。若将其设为 null，则该决定由您掌控，在您认为说完时显式提交缓冲区。

async def send_audio_chunk(ws, pcm_bytes: bytes):
    await ws.send(json.dumps({
        "type": "input_audio_buffer.append",
        "audio": base64.b64encode(pcm_bytes).decode(),
    }))

服务器端 VAD 有三个可调参数，设置不当是我见过最常让语音代理“感觉坏了”的原因——而日志并不会报错。默认情况下，这些参数都不会回显在 session.updated 中，所以请以文档为准，不要想当然。

• threshold（0.1–0.9，默认 0.85）：音量达到多大才算语音；环境嘈杂就调高，讲话很轻被漏检就调低

• silence_duration_ms：来电者安静多久后服务器结束其轮次；过短会打断思考中的人，过长显得迟钝

• prefix_padding_ms（默认 333）：在检测到语音之前保留一小段音频，避免首个音节被截断

若来电者在思考停顿时总被打断，优先调 silence_duration_ms。这通常是我先下手的参数。

接收与播放回复

音频会以小片段的 response.output_audio.delta 到达；流式的意义就在于每片一到便立刻播放，而不是等到 response.done。

async def play_response(ws):
    async for message in ws:
        event = json.loads(message)
        if event["type"] == "response.output_audio.delta":
            chunk = base64.b64decode(event["delta"])
            speaker.write(chunk)  # your playback call goes here
        elif event["type"] == "response.output_audio_transcript.delta":
            print(event["delta"], end="", flush=True)

即便在生产环境也要保留转写。这是当来电者说“坐席刚才讲得怪怪的”时最廉价的调试工具。

为语音代理添加工具

只会说话的语音代理不过是带麦克风的聊天机器人。

创建订单工具

每个工具由一个 JSON 架构加我们这边的一个普通 Python 函数组成。模型从不直接接触数据库，它只会看到我们函数返回的内容。

ORDER_TOOLS = [
    {
        "type": "function",
        "name": "check_order_status",
        "description": "Look up the status, ETA, and delivery instructions for an order.",
        "parameters": {
            "type": "object",
            "properties": {
                "order_number": {"type": "string", "description": "e.g. ORD-1042"},
            },
            "required": ["order_number"],
        },
    },
    # find_orders, update_delivery_instructions, cancel_order,
    # create_support_ticket, check_ticket_status and transfer_to_human
    # all follow the same shape
]

读取类操作如 check_order_status，若超时可安全重试。写入类则不行：在不明确的超时后重试 update_delivery_instructions 可能导致同一变更被应用两次。提示里的确认语并不能阻止此事，因此应给写入操作加幂等键或重复检查。

把拒绝逻辑也写进函数里。cancel_order 在订单已发货时返回原因与备选方案，而不是直接取消，因为“永不取消已发货订单”的提示只是建议，函数中的拒绝才是硬约束。

处理工具调用循环

四个步骤，顺序比看起来更重要：模型发送 response.function_call_arguments.done，您的代码执行函数，您将结果作为 function_call_output 项发回，之后才请求模型继续。

async def handle_tool_call(ws, event):
    args = json.loads(event["arguments"])
    result = execute(event["name"], args)  # never raises; errors come back as {"error": ...}
    await ws.send(json.dumps({
        "type": "conversation.item.create",
        "item": {
            "type": "function_call_output",
            "call_id": event["call_id"],
            "output": json.dumps(result),
        },
    }))

若该请求需要多个工具，模型会在任何音频播放前连发多个 function_call_arguments.done 事件。您需全部处理完毕并发送每个结果，之后再发一个 response.create。若发得太早，模型会在缺少仍在路上的调用上下文时就作答。

这里有个 SpaceXAI 文档提过、我首轮仍中招的点：在工具结果刚发出的一瞬间就发送 response.create，可能会与代理仍在播放的开场句重叠。有次它一边说“我马上帮您查询 ORD-1042 的状态”，一边在半句时就调用了工具；若立即请求继续，就会与自己的开场话重叠。

等当前轮次音频播放完，再展示一个简短的“思考中”状态。

在继续回复前的工具调用流程。作者供图。

处理打断与会话状态

这里是两个独立问题：来电者在坐席回复中插话；以及 WebSocket 断开后需要续接。

支持自然插话

开启 server_vad 后，服务器端会自动处理 barge-in：一旦检测到来电者再次说话，它会发出 input_audio_buffer.speech_started 并停止生成旧回复。您的职责是完成握手的客户端一侧，清空已排队的待播音频，让代理立刻安静下来，而不是把没人想听的句子说完。

if event["type"] == "input_audio_buffer.speech_started":
    playback_queue.clear()

对于手动（非 VAD）会话，response.cancel 可按需完成同样的工作。同时还有 conversation.item.truncate 可将助手项裁剪为用户实际听到的长度。文档只确认它存在，但未说明在实时 barge-in 中何时触发，因此需要自行测试时机。

我用在回复中途修改派送指示来测试：启动请求，在坐席确认中途以不同地址插话。关键不在于音频是否停止，而在于代理是否应用了更正后的指示，而非悄悄完成旧的那条。请针对订单记录断言，而不是针对“是否安静”。文末的浏览器演示能让您听到这一点。

续接断开的会话

会话续接需主动开启，但它不是“记忆”。在 session.update 上设 resumption.enabled: true，从 conversation.created 事件中拿到 ID；若 socket 掉线，重连时在 URL 中带上 ?conversation_id=<id>，并在新连接上再次选择加入。

async def reconnect(conversation_id):
    url = f"wss://api.x.ai/v1/realtime?model={MODEL}&conversation_id={conversation_id}"
    ws = await websockets.connect(url, additional_headers=auth_header)
    await ws.send(json.dumps({"type": "session.update", "session": {"resumption": {"enabled": True}}}))
    return ws

缓存的轮次、转写、工具调用与结果会在您下一个问题前回放，且缓存会在 30 分钟无活动后清除。我用查询订单、断开连接、重连后不重复提问直接追问来测试；代理正确接上了先前的到达时间。

一个未文档化的注意点：回放不是瞬间到达，因此在 socket 打开瞬间发问，可能会“抢跑”并在缺少先前轮次记忆的情况下作答。先等一会儿再怀疑续接机制。

已续接会话的终端日志。作者供图。

不要用它来代替将订单状态保存到自家数据库。若缓存过期或来电者改日回拨，您将从零上下文开始——这正是设计使然。

保障与监控语音代理

切勿将永久 API 密钥放在浏览器或移动端代码中。若客户端直接连接而非经由您的服务器，请签发短期令牌：

from fastapi import FastAPI
import httpx, os

app = FastAPI()

@app.post("/session")
async def create_session():
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.x.ai/v1/realtime/client_secrets",
            headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
            json={"expires_after": {"seconds": 300}},
        )
    return response.json()  # {"value": "xai-realtime-client-secret-...", "expires_at": ...}

浏览器无法在 WebSocket 握手中自定义 Authorization 头，因此它会将令牌放在 sec-websocket-protocol 头中，并以 xai-client-secret. 前缀传递。

服务器签发令牌，浏览器加入通话。作者供图。

计费有两种计量。音频（发送或接收）按我之前提到的每分钟 $0.08 计费，即每小时 $4.80；每个非音频且非 function_call_output 的 conversation.item.create 固定 $0.004。response.create 不计费。每个 response.done 都带有一个 usage 对象，我的测试中报告了 output_audio_seconds 与单独的 billable_audio_seconds。请基于这些字段计费，而非预估数。

语音到语音 API 的已文档化限制为每个团队 10 个并发会话、会话上限 120 分钟，均在 us-east-1。不要用 Voice Agent API 的限额来规划产能，那一套不同。

关于隐私，请表述准确。SpaceXAI 的安全常见问答称 API 请求与响应会因滥用监控而加密保留 30 天，且未经许可不会用于训练；团队可开启零数据留存（ZDR），但 ZDR 会停用持久化的语音代理对话历史，因此与续接不兼容。

若您需要披露通话被录音或由 AI 处理，可用我前面提到的 force_message 扩展。该句会按原文播放，而不是任由模型转述。

测试语音代理

WebSocket 握手返回 200 并不能说明代理做对了事。测试结果，而非仅测试连接。

• 一次干净的订单查询，核对口播答案与记录一致，而非仅确认“有回复”
• 一次被打断的回复，确认播放停止且代理响应新请求
• 一次需要确认的派送更新，与订单记录核对
• 一次拒绝（如取消已发货订单），代理应解释规则而非道歉
• 一个未知订单号，确保代理明确说明未知，而非编造状态
• 一个返回错误的工具，检查代理会将错误说出而非卡住
• 重连与续接，包括我前面遇到的回放窗口
• 嘈杂音频、快速语速、以及拼读号码与地址的来电者

我在写作过程中用真密钥跑了大多数用例。有趣的失败在于行为而非错误：上述续接时机，以及一个越界的 VAD 阈值被接受而非拒绝——这类问题若只测“快乐路径”就会带着缺陷上线。也加一项多语言测试，FAQ 中还提到语言命名的一个小坑。

其中两项无法靠打字测试。 app_streamlit.py 是一个 Streamlit 页面，将实时通话放在浏览器中：麦克风通过 WebRTC 流入同一个 WebSocket，代理语音反向流回，socket 全程保持打开。

streamlit run app_streamlit.py