openai-agents-python 实战用 Twilio Media Streams 打通电话与 OpenAI Realtime API 的实时语音 Agent【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python本指南以仓库examples/realtime/twilio示例为蓝本讲解如何用 openai-agents-python 的 realtime 语音栈把 Twilio 电话来电接入 OpenAI Realtime API实现电话端的实时语音对话。读完你可以掌握如何启动 FastAPI 媒体流服务器、如何用 ngrok 暴露公网地址并配置 Twilio 电话号码、TwilioHandler如何完成 Twilio WebSocket 与 OpenAI Realtime 会话之间的音频转发、中断处理与工具调用天气、当前时间以及RealtimeAgent、RealtimeRunner、RealtimeSession、RealtimePlaybackTracker在其中的底层分工。示例概览一条从电话到 AI 的实时音频链路examples/realtime/twilio/README.md描述了示例的核心目标把 OpenAI Realtime API 接到一通真实的电话呼叫上通过 Twilio Media Streams 在 Twilio 与 OpenAI Realtime API 之间流式传输音频从而让 AI Agent 能在电话里与人进行实时语音对话。仓库中的两个核心文件实现了这条链路examples/realtime/twilio/server.py基于 FastAPI 的 Web 服务器暴露/incoming-call返回 TwiML与/media-streamWebSocket 端点两个入口examples/realtime/twilio/twilio_handler.py核心的TwilioHandler负责在 Twilio WebSocket 与 OpenAI Realtime 会话之间搬运音频并处理协议差异。整体架构可以用 README 中的链路图概括Phone Call → Twilio → WebSocket → TwilioRealtimeTransportLayer → OpenAI Realtime API ↓ RealtimeAgent with Tools ↓ Audio Response → Twilio → Phone Call需要说明的是README 中把这一层抽象地称为TwilioRealtimeTransportLayer而从当前仓库源码看这一职责实际由twilio_handler.py中的TwilioHandler类承担仓库中TwilioRealtimeTransportLayer符号仅出现在 README 文档里。它正如 README 所描述的那样接管握手后的 Twilio WebSocket、运行自己的消息循环处理所有 Twilio 消息、处理 Twilio 与 OpenAI 之间的协议差异、自动为 Twilio 兼容性设置 G.711 μ-law 音频格式、维护音频块追踪以支持打断并且以包装而非继承的方式使用 OpenAI realtime 模型。环境准备与依赖README 列出的前置条件如下Python 3.10具备 Realtime API 访问权限的 OpenAI API key拥有一个电话号码的 Twilio 账户用于把本地服务暴露到公网的隧道工具如 ngrok示例的依赖清单见 examples/realtime/twilio/requirements.txtopenai-agents fastapi uvicorn[standard] websockets python-dotenv其中openai-agents提供 realtime 语音栈RealtimeAgent、RealtimeRunner、RealtimeSession等fastapi与uvicorn[standard]支撑 HTTP 与 WebSocket 服务python-dotenv便于从.env加载OPENAI_API_KEY等环境变量。安装依赖后在示例目录下启动即可。三步启动服务器、公网隧道与 Twilio 配置1. 启动本地服务器uv run server.py服务器默认监听8000端口见 server.py 的uvicorn.run(app, host0.0.0.0, portport)端口由PORT环境变量控制。启动后访问http://localhost:8000/会返回{message: Twilio Media Stream Server is running!}用于健康检查。2. 用 ngrok 暴露公网地址ngrok http 8000记下 ngrok 分配的公网 URL形如https://abc123.ngrok.io。Twilio 需要通过该公网地址回调你的本地服务。3. 配置 Twilio 电话号码登录 Twilio Console选择你要用的电话号码将呼入来电的 Webhook URL 设置为https://your-ngrok-url.ngrok.io/incoming-call将 HTTP 方法设置为 POST。完成以上三步后拨打你的 Twilio 号码会听到语音提示 Hello! Youre now connected to an AI assistant. You can start talking!随后即可开始实时对话。核心调用链来电如何变成一次 Realtime 会话/incoming-call返回 TwiML把呼叫导向媒体流server.py 中的incoming_call同时注册了 POST 与 GET 两个方法它从请求头中读取Host即你的 ngrok 公网域名动态拼出 WebSocket 地址并返回 TwiML?xml version1.0 encodingUTF-8? Response SayHello! Youre now connected to an AI assistant. You can start talking!/Say Connect Stream urlwss://{host}/media-stream / /Connect /Response这段 TwiML 做了两件事Say播放开场白ConnectStream让 Twilio 建立到wss://{host}/media-stream的 WebSocket 连接用于双向音频流。/media-stream接管 WebSocket 并启动会话server.py 中的 WebSocket 端点流程非常清晰通过TwilioWebSocketManager.new_session创建TwilioHandler然后依次handler.start()建立 OpenAI Realtime 会话、接受 Twilio 握手与handler.wait_until_done()阻塞等待消息循环结束并分别处理WebSocketDisconnect与一般异常。README 提到在真实应用中你还需要在通话结束时清理/关闭 handler——TwilioWebSocketManager保留了active_handlers字典正是为这类会话管理预留的扩展点。TwilioHandler.start()组装实时语音会话twilio_handler.py 中的start()是整条链路的组装点创建RealtimeRunner(agent)——这是 realtime 版Runner自动通过底层模型的持久连接处理多轮对话见 src/agents/realtime/runner.py 的类说明从OPENAI_API_KEY环境变量读取 API key缺失则抛ValueError调用runner.run(model_config...)得到RealtimeSession并传入关键的模型配置见下文await session.enter()进入会话等价于async with上下文管理器见 src/agents/realtime/session.py接受 Twilio WebSocket 握手启动三个并发任务realtime 会话事件循环、Twilio 消息循环、音频缓冲刷新循环。Realtime 模型配置G.711 μ-law 与语义 VADrunner.run的model_config是本例与普通语音示例最大的不同点self.session await runner.run( model_config{ api_key: api_key, initial_model_settings: { model_name: gpt-realtime-2.1, input_audio_format: g711_ulaw, output_audio_format: g711_ulaw, turn_detection: { type: semantic_vad, interrupt_response: True, create_response: True, }, }, playback_tracker: self.playback_tracker, } )各字段的含义可以结合 src/agents/realtime/config.py 中的类型定义来理解model_name指定 realtime 模型类型RealtimeModelName支持的取值包括gpt-realtime、gpt-realtime-1.5、gpt-realtime-2、gpt-realtime-2.1、gpt-realtime-2.1-mini、gpt-4o-realtime-preview系列、gpt-realtime-mini系列等见 config.pyinput_audio_format/output_audio_format音频格式RealtimeAudioFormat限定为pcm16、g711_ulaw、g711_alaw见 config.py。Twilio 的 Media Streams 原生使用 G.711 μ-lawmulaw因此这里必须设置为g711_ulaw避免格式转换带来的额外开销turn_detection端点检测配置type可取semantic_vad或server_vad见 config.py。本例使用语义 VAD并开启interrupt_response允许用户打断助手应答与create_response检测到端点即生成响应playback_tracker传入RealtimePlaybackTracker实例用于回放进度追踪详见下文打断与回放追踪一节。双向音频搬运Twilio ↔ OpenAI 的协议桥Twilio → OpenAI缓冲、分块与启动预热Twilio 会通过 WebSocket 持续推送 JSON 消息_twilio_message_looptwilio_handler.py解析出事件类型connected、start从中记录streamSid、media携带音频负载、mark回放确认与stop。音频上行路径在_handle_media_event与_flush_audio_buffertwilio_handler.py从media.payload取出 base64 编码的 μ-law 音频并解码追加到_audio_buffer当缓冲达到BUFFER_SIZE_BYTES时触发一次发送。块大小由常量决定twilio_handler.pyCHUNK_LENGTH_S 0.0550ms 一块、SAMPLE_RATE 8000Twilio g711_ulaw 为 8kHz于是BUFFER_SIZE_BYTES 8000 * 0.05 400字节/块。_flush_audio_buffer还实现了确定性启动预热机制在会话建立初期先攒够TWILIO_STARTUP_BUFFER_CHUNKS默认 3个块再一次性发给 OpenAI以替代简单sleep的粗暴做法让模型在开始处理前已有足够的音频上下文预热完成后_startup_warmed True则立即逐块发送。两个相关环境变量均可在twilio_handler.py中配置TWILIO_STARTUP_BUFFER_CHUNKS默认3设为 0 视为立即预热与TWILIO_STARTUP_DELAY_S默认0.0因为缓冲方案已优先通常无需额外延迟。此外_buffer_flush_looptwilio_handler.py每 50ms 检查一次缓冲若缓冲非空且距上次发送超过CHUNK_LENGTH_S * 2100ms则强制冲刷避免零星音频滞留造成延迟。发送时调用的是session.send_audio(audio)——即 src/agents/realtime/session.py 中向模型层发送RealtimeModelSendAudio事件的入口。OpenAI → Twilio音频、mark 与 clear 事件下行路径在_handle_realtime_eventtwilio_handler.py它遍历RealtimeSession产生的RealtimeSessionEventaudio事件将模型输出的音频 base64 编码后以{event: media, streamSid: ..., media: {payload: ...}}的 Twilio 协议格式发回随后发送一个mark事件携带自增的mark_id并记录该 mark 对应的(item_id, content_index, byte_count)到_mark_data字典为回放追踪做准备audio_interrupted事件当用户打断导致输出中断时向 Twilio 发送{event: clear, streamSid: ...}让 Twilio 清空播放缓冲立即停止当前播报audio_end与其余事件记录日志或忽略。事件类型RealtimeAudio、RealtimeAudioInterrupted、RealtimeAudioEnd分别对应 src/agents/realtime/events.py、events.py、events.py 中的定义RealtimeAudio携带模型层的RealtimeModelAudioEvent及item_id、content_indexRealtimeAudioInterrupted专供上层停止播放或给出视觉提示。打断与回放追踪让语音助手可被插话成为现实电话场景最反直觉的一点是当用户开口打断时模型已经生成了部分音频且这些音频可能已部分播出。RealtimePlaybackTrackersrc/agents/realtime/model.py正是为此设计当你有自定义播放逻辑、或音频以延迟/不同速度播放时需要创建它并传入会话由你负责在播放进度发生时调用on_play_bytes或on_play_ms报告进度。twilio_handler.py的实现思路是Twilio 播放完一段音频后会回发mark事件_handle_mark_eventtwilio_handler.pyhandler 用与下行发送时记录一致的mark_id反查_mark_data得到(item_id, item_content_index, byte_count)再用占位字节b\x00 * byte_count调用playback_tracker.on_play_bytes(...)。这样模型层就能精确获知哪一条响应音频已经实际播出、播出了多少从而在打断发生时给出正确的播放状态RealtimePlaybackStatecurrent_item_id、current_item_content_index、elapsed_ms见 model.py。定义带工具的语音 Agenttwilio_handler.py顶部用tool装饰器定义了两个示例工具并用RealtimeAgent组装twilio_handler.pytool def get_weather(city: str) - str: Get the weather in a city. return fThe weather in {city} is sunny. tool def get_current_time() - str: Get the current time. return fThe current time is {datetime.now().strftime(%H:%M:%S)} agent RealtimeAgent( nameTwilio Assistant, instructions( You are a helpful assistant that starts every conversation with a creative greeting. Keep responses concise and friendly since this is a phone conversation. ), tools[get_weather, get_current_time], )RealtimeAgent是专用于RealtimeSession的语音 Agent 子类src/agents/realtime/agent.py。它的类文档明确标注了与普通Agent的差异不支持model选择、modelSettings、outputType结构化输出与toolUseBehavior配置因为这些都由同一个 realtime 模型统一处理voice可以在 Agent 级配置但一旦会话中第一个 Agent 开口后便不可再更改。instructions既可以是字符串也可以是返回字符串的同步/异步函数见 agent.py 的get_system_prompt实现。由于真实电话对话要求应答简短README 与示例的 instructions 都刻意强调保持简洁友好README 中助手拥有天气与当前时间等工具的能力即来自上面的tools[get_weather, get_current_time]。配置项速查配置项环境变量默认值说明服务端口PORT8000uvicorn 监听端口见 server.pyOpenAI API KeyOPENAI_API_KEY无缺失即报错Realtime 会话所需见 twilio_handler.py启动预热块数TWILIO_STARTUP_BUFFER_CHUNKS3预热阶段攒够 N 个 50ms 块再发送0表示立即预热启动延迟TWILIO_STARTUP_DELAY_S0.0可选额外启动延迟秒缓冲方案已优先通常保持 0Agent 指令代码内instructions见示例修改 twilio_handler.py 中的RealtimeAgent配置工具集代码内toolsget_weather、get_current_time在 twilio_handler.py 增删tool函数常见问题排查README 的 Troubleshooting 一节给出了四类高频问题的定位方向结合源码可进一步细化WebSocket 连接问题确认 ngrok URL 正确且公网可达且 Twilio Console 中 Webhook 地址指向https://your-ngrok-url.ngrok.io/incoming-call、方法为 POST。另外注意 server.py 对TwilioHandler做了相对导入与直接导入的双重兼容脚本方式运行uv run server.py与包方式导入均可正常工作。音频质量Twilio 以 mulaw 格式、8kHz 采样率传输音频这是电话网络的固有约束本示例通过g711_ulaw输入/输出格式直接对接避免了转码损失但 8kHz 采样率本身仍会限制音质上限。延迟端到端延迟由 Twilio ↔ 你的服务器 ↔ OpenAI 三段的网络往返共同决定。可以关注_buffer_flush_loop的 50ms 检查周期与启动预热机制——缓冲既是平滑音频的必需品也是延迟的主要来源可通过TWILIO_STARTUP_BUFFER_CHUNKS调节。日志控制台会打印关键状态如 Twilio WebSocket connection accepted、Media stream started with SID: ...、Playback tracker updated: ... 以及各类异常信息是定位问题如OPENAI_API_KEY缺失、JSON 解析失败的第一现场。小结与扩展方向这个示例完整演示了 openai-agents-python realtime 语音栈在真实电话场景下的落地方式RealtimeRunner负责维持与模型的持久连接与多轮对话RealtimeSession提供send_audio/interrupt/enter等会话原语RealtimeAgent承载指令与工具RealtimePlaybackTracker解决打断场景下的播放状态一致性而TwilioHandler则是把这些能力与 Twilio Media Streams 协议对接起来的运输层。在此基础上你可以继续探索仓库中更丰富的 realtime 能力会话级配置项如RealtimeSessionModelSettings中的input_audio_transcription、input_audio_noise_reduction、modalities、voice、speed等见 src/agents/realtime/config.py、输出护栏output_guardrails与guardrails_settingsconfig.py、Agent 之间的realtime_handoffsrc/agents/realtime/handoffs.py以及 realtime 相关文档 docs/realtime/guide.md 与 docs/realtime/quickstart.md。若你不需要电话能力只想在浏览器或 CLI 里体验实时语音仓库还提供了 examples/realtime/cli 与 examples/realtime/app 两个参考实现。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考