首页/资讯中心/详情

Voice Agent 实战：用 OpenAI Realtime API + Twilio 复刻一个“全双工”的 AI 电话客服

Voice Agent 实战：用 OpenAI Realtime API + Twilio 复刻一个“全双工”的 AI 电话客服
📅 发布时间：2026/6/20 21:56:28

🚀 前言：为什么现在的电话机器人这么“傻”？

如果你接过营销电话，那种机械感是掩盖不住的。传统的架构是ASR (识别) -> NLP (处理) -> TTS (合成)的“回合制”游戏。

而OpenAI Realtime API (GPT-4o Audio)将这三步合二为一。它直接处理音频输入，直接输出音频，中间不再有文本转换的损耗。这不仅让延迟降低到了毫秒级，更重要的是：它可以听出你的语气（愤怒、犹豫），也能用带有情感的语调回复你。

今天，我们将构建这样一个系统：

  1. 用户：拨打一个真实的电话号码。
  2. Twilio：接听电话，并将音频流（Media Stream）通过 WebSocket 推送给我们的服务器。
  3. Server：作为中继，将 Twilio 的音频转发给 OpenAI，并将 OpenAI 的回复音频转发回 Twilio。
  4. OpenAI：实时思考并说话。

🏗️ 一、 架构设计：WebSocket 是核心

这是一个典型的**双向流式（Bi-directional Streaming）**架构。

数据流向图 (Mermaid):

中继服务器 (Python/Node)

WebSocket (G.711音频)

WebSocket (PCM音频)

用户 (PSTN电话)

Twilio 电话网关

中间件逻辑

OpenAI Realtime API

关键难点：

  • 协议转换：Twilio 输出的是mulaw格式音频（Base64 编码），OpenAI 需要的是pcm16。
  • 打断机制 (Interruption)：当用户说话时，服务器必须通过 OpenAI 的input_audio_buffer.speech_started事件，立刻发送指令告诉 Twilio“清空播放缓存”，实现打断效果。

🛠️ 二、 准备工作

  1. OpenAI API Key：需要有访问gpt-4o-realtime-preview的权限。
  2. Twilio 账号：注册并购买一个电话号码（试用号也可以，大概 $1）。
  3. 公网服务器：或者使用ngrok将本地端口暴露到公网（Twilio 需要回调）。

📞 三、 第一步：配置 Twilio TwiML

当电话打进来时，我们需要告诉 Twilio：“别自己处理，把音频流通过 WebSocket 扔给我的服务器。”

在 Twilio 后台创建一个TwiML Bin，或者直接在代码中返回以下 XML：

<Response><Connect><Streamurl="wss://your-domain.com/media-stream"><Parametername="customerId"value="12345"/></Stream></Connect></Response>

🐍 四、 第二步：编写中继服务器 (Python FastAPI)

我们需要一个能够同时处理Twilio WebSocket和OpenAI WebSocket的服务。

核心代码 (server.py)：

importosimportjsonimportasyncioimportwebsocketsfromfastapiimportFastAPI,WebSocketfromstarlette.websocketsimportWebSocketDisconnect# 你的 OpenAI KeyOPENAI_API_KEY=os.getenv("OPENAI_API_KEY")VOICE="alloy"# AI 的声音SYSTEM_PROMPT="你是一个专业、幽默的电话客服。请用中文简短回答。"app=FastAPI()@app.websocket("/media-stream")asyncdefhandle_media_stream(websocket:WebSocket):awaitwebsocket.accept()print("Twilio 连接成功")# 1. 连接 OpenAI Realtime APIurl="wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"headers={"Authorization":f"Bearer{OPENAI_API_KEY}","OpenAI-Beta":"realtime=v1",}asyncwithwebsockets.connect(url,extra_headers=headers)asopenai_ws:# 2. 初始化 Sessionsession_update={"type":"session.update","session":{"voice":VOICE,"instructions":SYSTEM_PROMPT,"input_audio_format":"g711_ulaw",# Twilio 默认格式"output_audio_format":"g711_ulaw","turn_detection":{"type":"server_vad"}# 开启服务端语音活动检测}}awaitopenai_ws.send(json.dumps(session_update))# 3. 定义双向转发任务stream_sid=Noneasyncdefreceive_from_twilio():nonlocalstream_sidtry:whileTrue:message=awaitwebsocket.receive_text()data=json.loads(message)ifdata["event"]=="media":# 收到 Twilio 音频 -> 转发给 OpenAIaudio_append={"type":"input_audio_buffer.append","audio":data["media"]["payload"]}awaitopenai_ws.send(json.dumps(audio_append))elifdata["event"]=="start":stream_sid=data["start"]["streamSid"]print(f"Stream 开始:{stream_sid}")exceptWebSocketDisconnect:print("Twilio 断开连接")asyncdefreceive_from_openai():try:asyncformessageinopenai_ws:response=json.loads(message)# A. 收到 AI 音频 -> 转发给 Twilio 播放ifresponse["type"]=="response.audio.delta"andresponse.get("delta"):audio_payload={"event":"media","streamSid":stream_sid,"media":{"payload":response["delta"]}}awaitwebsocket.send_text(json.dumps(audio_payload))# B. 关键点：用户打断处理# 当 OpenAI 检测到用户开始说话时，我们需要让 Twilio 立刻闭嘴ifresponse["type"]=="input_audio_buffer.speech_started":print("检测到用户插话，清空播放缓存...")clear_msg={"event":"clear","streamSid":stream_sid,}awaitwebsocket.send_text(json.dumps(clear_msg))# 同时也告诉 OpenAI 别继续生成刚才没说完的话了awaitopenai_ws.send(json.dumps({"type":"response.cancel"}))exceptExceptionase:print(f"OpenAI 错误:{e}")# 并发运行两个任务awaitasyncio.gather(receive_from_twilio(),receive_from_openai())if__name__=="__main__":importuvicorn# 必须运行在 0.0.0.0 才能被 ngrok 访问uvicorn.run(app,host="0.0.0.0",port=5000)

⚡ 五、 关键技术点解析

1. 为什么不需要转码？

OpenAI Realtime API 最近更新支持了g711_ulaw格式。这正是传统电话网络（PSTN）使用的格式。

  • 以前：Twilio (ulaw) -> Server (转pcm) -> LLM (文本) -> TTS (pcm) -> Server (转ulaw) -> Twilio。
  • 现在：Twilio (ulaw) -> OpenAI (ulaw) -> Twilio。
    这省去了大量的编解码 CPU 开销和延迟。
2. VAD（语音活动检测）的妙用

在代码中，input_audio_buffer.speech_started是核心。
旧的电话机器人最傻的地方就是“抢话”。
有了这个事件，只要用户发出一声“哎等等”，OpenAI 毫秒级检测到，通过response.cancel和 Twilio 的clear指令，机器人会瞬间闭嘴，等待你的新指令。

3. 工具调用 (Function Calling)

Realtime API 同样支持 Function Calling。
你可以在session.update中定义一个check_order_status工具。当用户问“我的快递到哪了”，OpenAI 会暂停生成音频，向你的 Server 发送函数调用请求，你查库后返回结果，OpenAI 再把结果念给用户听。
这一切都在同一个 WebSocket 连接中完成。

🎯 总结

通过OpenAI Realtime API + Twilio，我们构建的不再是一个简单的“IVR 语音导航”，而是一个具有类人交互能力的数字员工。

它没有明显的延迟，它可以被打断，它甚至能听出你感冒了并表示关心。
语音交互（Voice UI）的 iPhone 时刻，可能真的已经到来了。

Next Step:

  1. 申请一个 Twilio 号码。
  2. 用ngrok http 5000把你的服务暴露出去。
  3. 把 Twilio 的 Voice Webhook 指向你的wss://xxxx.ngrok-free.app/media-stream。
  4. 打个电话过去，感受一下未来的样子。