speech-assistant-openai-realtime-api-python

用电话打给 AI 助手：Twilio + OpenAI Realtime API 语音通话实战

场景切入：当电话遇见 GPT-4o

想象这样一个场景：你走在路上，突然想起一个问题想问问 AI——打开手机 App 太麻烦，耳机又不在身边。这时候，你只需要拿起手机，拨一个电话号码，AI 助手就直接接起来，像真人一样和你对话。这不是科幻，这是 Twilio + OpenAI Realtime API 正在实现的事。

twilio-samples/speech-assistant-openai-realtime-api-python 是 Twilio 官方维护的示例项目，展示了如何用 Python 将传统电话网络与现代 AI 实时语音能力连接起来。项目获得了 347 个 GitHub Star、193 个 Fork，代码质量成熟，属于 Twilio 生态中兼具教育意义和实用价值的标杆项目。

技术架构：三条 WebSocket 管道的交响乐

整个系统的核心是三条并行的 WebSocket 连接，它们像三条管道一样同时运作：

1. 用户电话 → Twilio 媒体流：用户拨打 Twilio 电话号码后，Twilio 将通话音频以 WebSocket 媒体流（Media Streams）的形式实时推送给你的服务器。这条管道传输的是原始音频数据，格式为 Opus 或 PCMU 编码。

2. 服务器 → OpenAI Realtime API：服务器将 Twilio 收到的音频 base64 编码后，通过 WebSocket 发送给 OpenAI Realtime API（模型为 gpt-realtime）。OpenAI 会实时转录、理解，并生成语音回复。

3. OpenAI 回复 → Twilio → 用户：OpenAI 返回的音频数据通过 WebSocket 回传给你的服务器，服务器再将其转发给 Twilio，Twilio 播放给用户听。

用户手机 ──(PSTN电话线)──► Twilio ──(WebSocket/MediaStream)──► 你的服务器
                                                                       │
                                        ┌──────────────────────────────┘
                                        ▼
                              OpenAI Realtime API (gpt-realtime)
                                        │
                                    (WebSocket/音频双向流)
                                        │
                                        ▼
                                    AI 语音回复

核心实现：AsyncIO 驱动的并发音频管道

代码使用 Python asyncio + websockets 库实现全异步架构。主入口 /media-stream WebSocket 处理函数中，启动两个并行的异步任务：

• receive_from_twilio()：持续监听 Twilio 发来的文本消息（包括 media 事件携带的音频数据）。当收到音频时，将 base64 编码的 payload 提取出来，构造 input_audio_buffer.append 消息发送给 OpenAI。

• send_to_twilio()：持续监听 OpenAI WebSocket 的消息。当收到 response.output_audio.delta 事件（增量音频数据）时，提取 delta 字段，构造 Twilio 的 media 事件发回给 Twilio，让用户听到 AI 的回复。

async with websockets.connect(
    f"wss://api.openai.com/v1/realtime?model=gpt-realtime",
    additional_headers={"Authorization": f"Bearer {OPENAI_API_KEY}"}
) as openai_ws:
    await initialize_session(openai_ws)
    # 并行运行两个异步任务
    await asyncio.gather(
        receive_from_twilio(),
        send_to_twilio()
    )

这种 asyncio.gather() 并发模型确保了双向音频流的低延迟——Twilio 音频到达和 OpenAI 音频返回两条链路互不阻塞。

打断处理：让对话更自然的关键细节

真实通话中，用户往往会打断 AI。项目中实现了优雅的打断处理机制：当 OpenAI 检测到用户开始说话（input_audio_buffer.speech_started 事件），代码会执行以下操作：

1. 构造 conversation.item.truncate 事件，告知 OpenAI 截断 AI 正在播放的回复（精确到毫秒级：记录 speech_started 时的 latest_media_timestamp，计算与 AI 回复起始时间的差值）
2. 向 Twilio 发送 clear 事件，清除 Twilio 端的音频缓冲
3. 清空标记队列，防止陈旧的音频继续播放

这段打断逻辑体现了实时语音系统的核心挑战：双向音频流的时序同步。时间戳（stream_sid、latest_media_timestamp）是协调两条音频管道的关键。

技术栈解析

部署方式：ngrok 隧道 + Twilio Webhook

项目没有提供 Dockerfile，只能手动部署。完整流程分五步：

1. ngrok 隧道：本地运行 ngrok http 5050，生成公网 HTTPS URL（因为 Twilio 要求 Webhook 必须是 HTTPS）
2. 安装依赖：pip install -r requirements.txt
3. 配置 Twilio：在 Twilio Console 中将购买的电话号码的「A call comes in」Webhook 指向 {ngrok_url}/incoming-call
4. 配置环境变量：在 .env 中填入 OPENAI_API_KEY
5. 启动服务：python main.py，服务器监听 5050 端口

项目使用 Google Chirp3-HD-Aoede 语音（TTS）作为 Twilio 侧的提示语（TwiML <Say>），用 WebSocket Media Stream 进行双向实时音频。

局限与注意

• 仅支持呼入：项目只处理接听电话，不支持主动外呼。外呼功能需要额外实现。
• ngrok 每次重启 URL 变化：生产环境需要固定公网 IP 和域名。
• OpenAI Realtime API 需要单独申请：不是所有 OpenAI 账号默认开通，需要在 Platform 平台申请权限。
• 无 Web UI：所有交互通过电话完成，无法在浏览器中测试。

行业意义：电信级 AI 语音的民主化

这个项目代表了一个重要趋势：将 AI 能力通过电信网络无门槛地触达用户。不需要 App、不需要互联网连接——任何一部电话都可以成为 AI 交互终端。对于老年人群体、残障人士，或网络条件差的地区，这可能是最自然的 AI 交互入口。Twilio 和 OpenAI 的这次握手，正在重新定义「打电话给 AI」这个场景的可能性。