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 在语音到语音指数上得分 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。这次切换既是模型变更，也是价格变更：每分钟音频 $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 端点，整个对话都在这一个 socket 上以 JSON 事件流的形式进行。

事件生命周期

连接遵循固定流程：一连上服务器就会发送 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 开销并不是支持座席构建中的瓶颈。只有在有量化理由时再切换到 binary。

与 OpenAI Realtime API 的兼容性

如果您从未用过 OpenAI 的 Realtime API，可以跳过本节。其他读者请注意，该语音到语音 API 与 OpenAI Realtime API 的对齐程度足以让大多数客户端代码只需改 base URL 和密钥即可移植，但并非完美替换。

在这里，转写以 conversation.item.input_audio_transcription.updated 到达，而不是 OpenAI 的 delta；OpenAI 的部分事件未被支持；SpaceXAI 也加入了自家扩展：用于话术披露的 force_message、用于重连的 resumption，以及在 TTS 前修正品牌名读音的 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 语音模型写的提示，而非照搬。我给座席的提示是保持简短回复、一次只问一个问题、在执行写操作前先复述确认。口头确认是用户体验上的体贴，不是安全控制。您应用中的写操作仍需由自身授权校验。

有个让我意外的点：未识别的模型字符串在连接时不会报错，而是静默回退到 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 后，插话在服务器侧是自动处理的：一旦检测到来电者再次说话，它就会发出 input_audio_buffer.speech_started 并停止生成旧响应。您的工作是握手的客户端一半：清空已排队的音频，让座席安静下来，而不是把没人想听的句子说完。

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

对于手动、非 VAD 会话，response.cancel 可以按需完成同样的事。另有 conversation.item.truncate 可将助理项裁剪到实际被听到的部分。文档只确认它存在，但没有说明在实时插话中何时触发，所以请自行测试时机。

我用中途修改配送说明来测试：发起请求，在座席确认时插话修改地址的一部分。关键不是音频是否停下，而是座席有没有应用更正后的说明，而不是默默完成旧的那条。请对订单记录做断言，而不是对静音。文末的浏览器演示可以亲耳听到。

恢复已断开的会话

会话续接需要显式开启，而且它不是记忆。请在 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 分钟无活动后失效。我通过先询问订单、断开连接、再重连追问而不重复自己来测试；座席能正确接上 ETA。

有个未文档化的点：重放并非瞬时完成，所以在 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 阈值被接受而非拒绝——这类问题如果只测“快乐路径”就会悄悄上线。再加一个多语言测试，关于语言命名的一个小坑见文末 FAQs。

其中两项无法通过打字来测。 app_streamlit.py 是一个 Streamlit 页面，把实时通话放进浏览器：麦克风通过 WebRTC 流进同一个 WebSocket，座席的声音流回，全程保持连接。

streamlit run app_streamlit.py