openai-agents-python 实战：用 Twilio Media Streams 打通电话与 OpenAI Realtime API 的实时语音 Agent

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-stream（WebSocket 端点）两个入口；
• 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.txt：

openai-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, host="0.0.0.0", port=port)，端口由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! You're 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 version="1.0" encoding="UTF-8"?> <Response> <Say>Hello! You're now connected to an AI assistant. You can start talking!</Say> <Connect> <Stream url="wss://{host}/media-stream" /> </Connect> </Response>

这段 TwiML 做了两件事：<Say>播放开场白，<Connect><Stream>让 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()是整条链路的组装点：

1. 创建RealtimeRunner(agent)——这是 realtime 版Runner，自动通过底层模型的持久连接处理多轮对话（见 src/agents/realtime/runner.py 的类说明）；
2. 从OPENAI_API_KEY环境变量读取 API key，缺失则抛ValueError；
3. 调用runner.run(model_config=...)得到RealtimeSession，并传入关键的模型配置（见下文）；
4. await session.enter()进入会话（等价于async with上下文管理器，见 src/agents/realtime/session.py）；
5. 接受 Twilio WebSocket 握手；
6. 启动三个并发任务：realtime 会话事件循环、Twilio 消息循环、音频缓冲刷新循环。

Realtime 模型配置：G.711 μ-law 与语义 VAD

runner.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.py）；
• input_audio_format/output_audio_format：音频格式，RealtimeAudioFormat限定为pcm16、g711_ulaw、g711_alaw（见 config.py）。Twilio 的 Media Streams 原生使用 G.711 μ-law（mulaw），因此这里必须设置为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_loop（twilio_handler.py）解析出事件类型：connected、start（从中记录streamSid）、media（携带音频负载）、mark（回放确认）与stop。

音频上行路径在_handle_media_event与_flush_audio_buffer（twilio_handler.py）：

1. 从media.payload取出 base64 编码的 μ-law 音频并解码；
2. 追加到_audio_buffer；
3. 当缓冲达到BUFFER_SIZE_BYTES时触发一次发送。

块大小由常量决定（twilio_handler.py）：CHUNK_LENGTH_S = 0.05（50ms 一块）、SAMPLE_RATE = 8000（Twilio 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_loop（twilio_handler.py）每 50ms 检查一次缓冲，若缓冲非空且距上次发送超过CHUNK_LENGTH_S * 2（100ms），则强制冲刷，避免零星音频滞留造成延迟。

发送时调用的是session.send_audio(audio)——即 src/agents/realtime/session.py 中向模型层发送RealtimeModelSendAudio事件的入口。

OpenAI → Twilio：音频、mark 与 clear 事件

下行路径在_handle_realtime_event（twilio_handler.py），它遍历RealtimeSession产生的RealtimeSessionEvent：

• audio事件：将模型输出的音频 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_index；RealtimeAudioInterrupted专供上层停止播放或给出视觉提示。

打断与回放追踪：让"语音助手可被插话"成为现实

电话场景最反直觉的一点是：当用户开口打断时，模型已经生成了部分音频，且这些音频可能已部分播出。RealtimePlaybackTracker（src/agents/realtime/model.py）正是为此设计：当你有自定义播放逻辑、或音频以延迟/不同速度播放时，需要创建它并传入会话，由你负责在播放进度发生时调用on_play_bytes或on_play_ms报告进度。

twilio_handler.py的实现思路是：Twilio 播放完一段音频后会回发mark事件（_handle_mark_event，twilio_handler.py），handler 用与下行发送时记录一致的mark_id反查_mark_data，得到(item_id, item_content_index, byte_count)，再用占位字节（b"\x00" * byte_count）调用playback_tracker.on_play_bytes(...)。这样模型层就能精确获知"哪一条响应音频已经实际播出、播出了多少"，从而在打断发生时给出正确的播放状态（RealtimePlaybackState：current_item_id、current_item_content_index、elapsed_ms，见 model.py）。

定义带工具的语音 Agent

twilio_handler.py顶部用@tool装饰器定义了两个示例工具，并用RealtimeAgent组装（twilio_handler.py）：

@tool def get_weather(city: str) -> str: """Get the weather in a city.""" return f"The weather in {city} is sunny." @tool def get_current_time() -> str: """Get the current time.""" return f"The current time is {datetime.now().strftime('%H:%M:%S')}" agent = RealtimeAgent( name="Twilio 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]。

配置项速查

配置项	环境变量	默认值	说明
服务端口	PORT	8000	uvicorn 监听端口，见 server.py
OpenAI API Key	OPENAI_API_KEY	无（缺失即报错）	Realtime 会话所需，见 twilio_handler.py
启动预热块数	TWILIO_STARTUP_BUFFER_CHUNKS	3	预热阶段攒够 N 个 50ms 块再发送，0表示立即预热
启动延迟	TWILIO_STARTUP_DELAY_S	0.0	可选额外启动延迟（秒），缓冲方案已优先，通常保持 0
Agent 指令	代码内instructions	见示例	修改 twilio_handler.py 中的RealtimeAgent配置
工具集	代码内tools	get_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_settings（config.py）、Agent 之间的realtime_handoff（src/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