openai-agents-python Realtime 传输层选型指南服务端 WebSocket、SIP 电话接入与自定义端点【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python本篇技术指南围绕 openai-agents-pythonOpenAI Agents SDK for Python中 Realtime 语音 Agent 的**传输层Transport**展开帮助你在搭建实时语音应用前做出正确的技术选型默认的服务端 WebSocket 路径适用于服务器托管音频管线、工具执行与审批流的场景SIP Attach 路径用于将 Agent 接入电话/语音呼叫浏览器 WebRTC 则明确不在本 SDK 范围内。读完本文你将掌握RealtimeRunner、OpenAIRealtimeWebSocketModel、OpenAIRealtimeSIPModel与RealtimeModelConfig的组合方式并能通过transport_config精确调优底层连接参数。快速选型指南在深入代码之前先用一张决策表确定起点。这份表格来自官方文档 docs/realtime/transport.md韩文版见 docs/ko/realtime/transport.md目标起点原因构建服务器托管的实时应用快速入门默认 Python 路径是由RealtimeRunner管理的服务端 WebSocket 会话确定该选哪种传输方式与部署形态本文在敲定传输方式或部署形态之前先参考本文将 Agent 接入电话或 SIP 呼叫Realtime 指南 与 examples/realtime/twilio_sip仓库内置了以call_id驱动的 SIP Attach 流程核心结论先行传输方式决定连接建立方式但会话生命周期、工具执行、Guardrails 与 Handoff 等能力在所有传输方式下保持一致。Python SDK 的传输边界在开始选型前必须明确 Python SDK 的能力边界本 SDK 不包含浏览器 WebRTC 传输。本文只讨论 Python SDK 的两种传输选择服务端 WebSocket 与 SIP Attach 流程。浏览器 WebRTC 属于独立的平台话题其客户端流程与事件模型由官方的 Realtime API with WebRTC 指南维护不在本仓库范围内详见 Browser WebRTC is outside this SDK 一节。也就是说如果你的主客户端是浏览器请将浏览器侧 WebRTC 视为本 SDK 文档之外的内容若在浏览器 WebRTC 客户端之外还需要一条带外sideband服务器连接需参考官方的 Realtime 服务端控制指南。本仓库目前也不提供浏览器 WebRTC Python sideband的组合示例。服务端 WebSocket默认的 Python 路径拓扑结构在RealtimeRunner的构造中只要你不传入自定义的RealtimeModel默认就会使用OpenAIRealtimeWebSocketModel。这在 runner.py 中有直接体现self._model model if model is not None else OpenAIRealtimeWebSocketModel()因此标准的 Python 拓扑如下在 Python 服务中创建RealtimeRunner调用await runner.run()获得一个RealtimeSession将RealtimeSession作为异步上下文管理器进入然后发送文本、结构化消息或音频消费RealtimeSessionEvent事件将音频或转写文本转发给你的应用。这一拓扑被仓库中的核心演示应用、CLI 示例和 Twilio Media Streams 示例共同采用examples/realtime/appexamples/realtime/cliexamples/realtime/twilio当你的服务器负责音频管线、工具执行、审批流程与历史记录处理时就走这条路径。一次完整的最小会话结合 session.py 的接口一个最小会话长这样from agents.realtime import RealtimeAgent, RealtimeRunner agent RealtimeAgent(nameAssistant) runner RealtimeRunner(starting_agentagent) async with await runner.run() as session: await session.send_message(Hello) async for event in session: print(event)关键接口对应如下session.send_message()/session.send_audio()发送用户输入session.pyasync for event in session流式消费RealtimeSessionEvent包括RealtimeAudio、RealtimeAudioEnd、RealtimeAgentStartEvent、RealtimeAgentEndEvent、RealtimeToolStart/RealtimeToolEnd、RealtimeGuardrailTripped等事件类型清单见 events.pysession.interrupt()主动中断模型输出。进入async with时RealtimeSession.__aenter__会先把自身注册为模型监听器再调用await self._model.connect(model_config)建立连接任何连接失败都会自动移除监听器并抛出异常session.py。与纯文本Runner不同runner.run()不会立刻返回最终结果而是返回一个持续同步本地历史、后台工具执行、Guardrail 状态与当前 Agent 配置的活跃会话对象详见 Realtime agents guide。底层 WebSocket 调优transport_config当需要调整底层服务端 WebSocket 连接时把transport_config传给OpenAIRealtimeWebSocketModelfrom agents.realtime import ( OpenAIRealtimeWebSocketModel, RealtimeAgent, RealtimeRunner, ) agent RealtimeAgent(nameAssistant) model OpenAIRealtimeWebSocketModel( transport_config{ ping_interval: 20.0, ping_timeout: 60.0, handshake_timeout: 30.0, max_size: 8 * 1024 * 1024, } ) runner RealtimeRunner(starting_agentagent, modelmodel)支持的选项TransportConfig的完整字段定义位于 openai_realtime.pyping_interval客户端保活 ping 的间隔秒。默认通常为20.0设为None可禁用 ping。ping_timeout断开连接前等待 pong 的时间秒。设为None可容忍延迟的 pong不设置心跳超时。handshake_timeout等待初始连接握手完成的时间秒。max_size接收的 WebSocket 消息最大字节数。SDK 默认值为None不限制接收消息大小当需要约束每条消息的内存占用时例如长连接位于代理之后或处于内存受限的容器中设置一个显式上限。底层实现映射这些配置最终被映射到websockets库的connect()参数见 openai_realtime.py 的_create_websocket_connectionif ping_interval in transport_config: connect_kwargs[ping_interval] transport_config[ping_interval] if ping_timeout in transport_config: connect_kwargs[ping_timeout] transport_config[ping_timeout] if handshake_timeout in transport_config: connect_kwargs[open_timeout] transport_config[handshake_timeout] if max_size in transport_config: connect_kwargs[max_size] transport_config[max_size]需要注意两点实现细节handshake_timeout对应的是websockets.connect的open_timeout参数在websockets新版本中由open_timeout替代旧的timeout连接建立时 SDK 默认设置max_sizeNone允许任意大小的消息同时通过user_agent_header携带Agents/Python version的 UA 标识openai_realtime.py。重要边界这些参数配置的是客户端连接本身而不是 Realtime API 会话。端点endpoint、认证、呼叫接入call attachment与播放playback设置依然通过RealtimeModelConfig完成。SIP Attach电话路径对于仓库中记录的电话流程Python SDK 通过call_id接入一个已存在的实时呼叫。拓扑结构OpenAI 向你的服务发送诸如realtime.call.incoming的 Webhook你的服务通过 Realtime Calls API 接受该呼叫Python 服务启动RealtimeRunner(..., modelOpenAIRealtimeSIPModel())会话使用model_config{call_id: ...}连接之后与其他实时会话一样处理事件。该拓扑的完整示例见 examples/realtime/twilio_sipFastAPI 服务器实现位于 examples/realtime/twilio_sip/server.py。在示例中服务端收到 Webhook 后通过client.post(f/realtime/calls/{call_id}/accept, ...)调用 Realtime Calls API 接受呼叫并携带typerealtime、modelgpt-realtime-2.1与 instructions随后启动携带OpenAIRealtimeSIPModel的 Runner用call_id建立会话。OpenAIRealtimeSIPModel 的实现OpenAIRealtimeSIPModel继承自OpenAIRealtimeWebSocketModel其connect()强制要求配置call_id否则抛出UserErroropenai_realtime.pyclass OpenAIRealtimeSIPModel(OpenAIRealtimeWebSocketModel): async def connect(self, options: RealtimeModelConfig) - None: call_id options.get(call_id) if not call_id: raise UserError(OpenAIRealtimeSIPModel requires call_id in the model configuration.) sip_options options.copy() await super().connect(sip_options)在基类connect()中当提供call_id时URL 会以call_id查询参数拼接而非使用模型名openai_realtime.pyif call_id and model_name: raise UserError( Cannot specify both call_id and model_name when attaching to an existing realtime call. ) ... if call_id: url options.get(url, fwss://api.openai.com/v1/realtime?call_id{call_id}) else: url options.get(url, fwss://api.openai.com/v1/realtime?model{self.model})也就是说call_id与model_name互斥二选一指定否则直接报错。此外OpenAIRealtimeSIPModel还提供了静态方法build_initial_session_payload()用于在接入 SIP 呼叫时通过 Realtime Calls API 转发会话负载避免重复实现会话设置逻辑openai_realtime.py。更广泛的 Realtime API 也会在一些服务端控制模式中使用call_id但本仓库提供的接入示例是 SIP。浏览器 WebRTC在本 SDK 范围之外如果应用的主客户端是使用 Realtime WebRTC 的浏览器将其视为本仓库 Python SDK 文档范围之外的内容客户端流程与事件模型参考官方的 Realtime API with WebRTC 与 Realtime conversations 文档若在浏览器 WebRTC 客户端之外还需要一条带外服务器连接参考官方的 Realtime server-side controls 指南不要期待本仓库提供浏览器侧的RTCPeerConnection抽象或开箱即用的浏览器 WebRTC 示例。当前仓库也不提供浏览器 WebRTC Python sideband的组合示例。自定义端点与接入点RealtimeModelConfigRealtimeModelConfig是连接实时模型的统一配置接口TypedDict可以自定义默认传输行为。文档明确支持的字段如下url覆盖 WebSocket 端点。未设置时使用合理默认值例如 OpenAI Realtime 模型默认使用wss://api.openai.com/v1/realtime并按model或call_id拼接查询参数。headers提供显式请求头例如 Azure 认证头{api-key: your api key here}。注意一旦设置headersSDK 将不再自动设置 Authorization 头认证完全由你提供的头部承担。api_key直接传入 API 密钥或传入一个返回密钥的回调函数支持同步与异步。未设置时模型会尝试读取OPENAI_API_KEY环境变量openai_realtime.py 的get_api_key实现了字符串、可调用对象与环境变量的三级回退。默认路径下若无 API key连接会抛出UserError(API key is required but was not provided.)。call_id接入已存在的实时呼叫使用call_id查询参数而非模型名本仓库的接入示例为 SIP。playback_tracker报告真实的播放进度用于打断interruption处理。播放跟踪器的工作原理RealtimePlaybackTrackermodel.py用于在自定义播放逻辑、或音频以延迟/变速播放的场景下跟踪实际播放进度。你负责跟踪播放进度并在用户播放了部分音频时调用on_play_bytes或on_play_ms模型侧通过get_state()获取当前播放状态当前 item ID、内容索引与已播放毫秒数。默认实现假设音频以实时速度立即播放。在低延迟场景下这种假设完全够用但在电话呼叫或其他远程交互场景下模型生成音频的速度远快于实际播放速度发生打断时模型需要知道用户到底听到了多少内容。此时传入自定义playback_tracker可以让模型在conversation.item.truncate时精确到已播放的毫秒位置进行截断从而获得更自然的打断体验。SDK 还内置了ModelAudioTrackersrc/agents/realtime/_default_tracker.py作为内部默认实现。完整组合示例以下示例同时演示自定义端点、显式认证头与播放跟踪器from agents.realtime import ( OpenAIRealtimeWebSocketModel, RealtimeAgent, RealtimePlaybackTracker, RealtimeRunner, ) agent RealtimeAgent(nameAssistant) tracker RealtimePlaybackTracker() model OpenAIRealtimeWebSocketModel( transport_config{ ping_interval: 20.0, ping_timeout: 60.0, handshake_timeout: 30.0, max_size: 8 * 1024 * 1024, } ) runner RealtimeRunner( starting_agentagent, modelmodel, ) async with await runner.run( model_config{ url: wss://your-endpoint.example.com/v1/realtime?modelgpt-realtime-2.1, headers: {api-key: your-api-key}, playback_tracker: tracker, } ) as session: # 将播放进度回传给 tracker # tracker.on_play_ms(item_id, content_index, ms) async for event in session: ...注意url与headers的搭配一旦显式提供headersSDK 不再自动注入Authorization认证信息必须完整包含在头部中如 Azure OpenAI Realtime WebSocket 的api-key头。传输层之上的能力面选定拓扑之后传输层之上的会话生命周期与能力面是一致的均由RealtimeSession承载本地历史维护会话维护RealtimeItem历史列表随item_updated、item_deleted、input_audio_transcription_completed等模型事件增量更新session.py工具执行与审批工具调用触发RealtimeToolStart/RealtimeToolEnd需要审批时发出RealtimeToolApprovalRequired可通过approve_tool_call()恢复执行输出 Guardrails按响应批次对输出音频转写/文本增量进行 Guardrail 检查Handoff支持realtime_handoff在多个实时 Agent 之间切换连接状态服务端正常关闭 WebSocket 时模型传输层依次发出RealtimeModelConnectionStatusEvent(statusdisconnected)与RealtimeModelEndOfStreamEvent会话将其转发为raw_model_event后排空已排队事件并正常结束迭代由调用方主动session.close()则不会合成这两类事件意外的 WebSocket 失败走会话异常路径详见 docs/realtime/guide.md 的 Session lifecycle 一节。更完整的生命周期与功能面说明请阅读 Realtime agents guide韩文版 docs/ko/realtime/guide.md以及本仓库的 Realtime 参考文档。相关的单元测试覆盖可参考 tests/realtime/test_runner.py 与 tests/realtime/test_session.py它们验证了会话事件流转、工具输出与连接生命周期等行为。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考