speech-assistant-openai-realtime-api-python
用电话直接对话 AI:通过 Twilio 电话网络 + OpenAI Realtime API 实现打电话与 GPT-4o 实时语音对话
加载项目详情…
本应用为开源项目,仅供学习研究,请遵守其开源协议。
用电话直接对话 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 连接,它们像三条管道一样同时运作:
用户电话 → Twilio 媒体流:用户拨打 Twilio 电话号码后,Twilio 将通话音频以 WebSocket 媒体流(Media Streams)的形式实时推送给你的服务器。这条管道传输的是原始音频数据,格式为 Opus 或 PCMU 编码。
服务器 → OpenAI Realtime API:服务器将 Twilio 收到的音频 base64 编码后,通过 WebSocket 发送给 OpenAI Realtime API(模型为 gpt-realtime)。OpenAI 会实时转录、理解,并生成语音回复。
OpenAI 回复 → Twilio → 用户:OpenAI 返回的音频数据通过 WebSocket 回传给你的服务器,服务器再将其转发给 Twilio,Twilio 播放给用户听。
用户手机 ──(PSTN电话线)──► Twilio ──(WebSocket/MediaStream)──► 你的服务器
│
┌──────────────────────────────┘
▼
OpenAI Realtime API (gpt-realtime)
│
(WebSocket/音频双向流)
│
▼
AI 语音回复
代码使用 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 事件),代码会执行以下操作:
conversation.item.truncate 事件,告知 OpenAI 截断 AI 正在播放的回复(精确到毫秒级:记录 speech_started 时的 latest_media_timestamp,计算与 AI 回复起始时间的差值)clear 事件,清除 Twilio 端的音频缓冲这段打断逻辑体现了实时语音系统的核心挑战:双向音频流的时序同步。时间戳(stream_sid、latest_media_timestamp)是协调两条音频管道的关键。
| 组件 | 技术选型 | 作用 |
|---|---|---|
| Web 框架 | FastAPI + Starlette | HTTP/WebSocket 端点,异步处理 |
| 电话网络 | Twilio Voice + Media Streams | 电话呼入、WebSocket 媒体流 |
| AI 接口 | OpenAI Realtime API | 实时语音对话(gpt-realtime 模型) |
| 异步运行时 | Python asyncio + aiohttp | 高并发 WebSocket 管理 |
| 协议库 | websockets 15.0 | WebSocket 客户端/服务端 |
| 数据验证 | Pydantic 2 | API 请求/响应模型校验 |
项目没有提供 Dockerfile,只能手动部署。完整流程分五步:
ngrok http 5050,生成公网 HTTPS URL(因为 Twilio 要求 Webhook 必须是 HTTPS)pip install -r requirements.txt{ngrok_url}/incoming-call.env 中填入 OPENAI_API_KEYpython main.py,服务器监听 5050 端口项目使用 Google Chirp3-HD-Aoede 语音(TTS)作为 Twilio 侧的提示语(TwiML <Say>),用 WebSocket Media Stream 进行双向实时音频。
这个项目代表了一个重要趋势:将 AI 能力通过电信网络无门槛地触达用户。不需要 App、不需要互联网连接——任何一部电话都可以成为 AI 交互终端。对于老年人群体、残障人士,或网络条件差的地区,这可能是最自然的 AI 交互入口。Twilio 和 OpenAI 的这次握手,正在重新定义「打电话给 AI」这个场景的可能性。