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实时智能体Realtime Agent与普通文本 Agent 最大的区别在于传输方式它需要在低延迟、双向、流式的前提下承载音频、文本与结构化消息。本指南基于 openai-agents-python 仓库的 docs/zh/realtime/transport.md 编写聚焦于 Python SDK 的两种官方传输路径——服务器端 WebSocket默认路径与基于call_id的 SIP 电话接入——并给出transport_config调优、RealtimeModelConfig自定义端点等完整实战参数。读完本文你将能够根据应用形态做出正确的传输选型并在仓库示例基础上搭建可运行的实时语音应用。传输决策指南先选形态再写代码在动手编码之前先明确你的应用由谁管理实时会话。仓库文档给出了三档决策入口目标入门资源原因构建由服务器管理的实时应用实时快速入门默认的 Python 路径是由RealtimeRunner管理的服务器端 WebSocket 会话了解应选择的传输方式和部署形态本页面docs/zh/realtime/transport.md在确定传输方式或部署形态之前请先阅读本页面将智能体接入电话或 SIP 通话实时指南 与examples/realtime/twilio_sip该仓库提供了由call_id驱动的 SIP 接入流程需要特别注意的是 Python SDK 的边界SDK 不包含浏览器 WebRTC 传输。本仓库的实时传输能力限定在服务端 WebSocket 与 SIP 接入两条路径上浏览器端的 WebRTC 属于独立的平台主题。默认路径服务器端 WebSocket除非传入自定义的RealtimeModel否则RealtimeRunner默认使用OpenAIRealtimeWebSocketModel。从源码看该默认模型由 src/agents/realtime/openai_realtime.py 定义其内部持有websockets库的ClientConnection并在connect()中建立到wss://api.openai.com/v1/realtime的连接。标准 Python 拓扑如下你的 Python 服务创建一个RealtimeRunner。await runner.run()返回一个RealtimeSession。将RealtimeSession作为异步上下文管理器进入然后发送文本、结构化消息或音频。消费RealtimeSessionEvent项并将音频或转录文本转发到你的应用程序。该拓扑由仓库中的三个示例共同验证分别覆盖演示应用、CLI 与电话媒体流examples/realtime/app核心演示应用examples/realtime/cli命令行交互示例examples/realtime/twilioTwilio Media Streams 示例当你的服务器负责音频管线、工具执行、审批流程和历史记录处理时请使用此路径。RealtimeRunner的run()方法签名见 src/agents/realtime/runner.py也印证了这一点它接受可选的context与model_config返回RealtimeSession使服务端可以在同一进程中托管完整的智能体逻辑。底层 WebSocket 调优transport_config当默认的连接参数无法满足你的网络环境如代理、弱网、内存受限容器时可将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)支持的选项及其语义如下参数含义默认行为ping_interval客户端保活 ping 之间的秒数通常为20.0设置为None可禁用 pingping_timeout断开连接前等待 pong 的秒数设置为None可容忍延迟的 pong不触发心跳超时handshake_timeout等待初始连接握手的秒数对应websockets库的open_timeoutmax_size传入 WebSocket 消息的最大字节数SDK 默认为None即不限制传入消息的大小如需限制每条消息的内存使用量请设置明确的上限从源码层面可以确认这些参数的真实去向TransportConfig是一个TypedDict在_create_websocket_connection()src/agents/realtime/openai_realtime.py中被逐项映射到websockets.connect()的关键字参数ping_interval、ping_timeout、max_size原样透传而handshake_timeout被映射为open_timeout。这意味着你可以直接用websockets库的语义来理解这些参数。一个关键的设计边界是这些设置配置的是客户端连接而不是 Realtime API 会话。端点、身份验证、通话接入和播放设置仍应使用RealtimeModelConfig详见下文。电话通信路径SIP 接入对于仓库中记录的电话通信流程Python SDK 通过call_id接入现有的实时通话而不是自行发起新会话。此拓扑如下OpenAI 向你的服务发送 Webhook例如realtime.call.incoming。你的服务通过 Realtime Calls API 接听通话。你的 Python 服务启动一个RealtimeRunner(..., modelOpenAIRealtimeSIPModel())。会话通过model_config{call_id: ...}建立连接然后像其他实时会话一样处理事件。examples/realtime/twilio_sip完整展示了此拓扑其核心实现在server.py中Webhook 校验与分发/openai/webhook端点使用client.webhooks.unwrap(body, request.headers)校验签名仅对realtime.call.incoming事件响应examples/realtime/twilio_sip/server.py。接听通话accept_call()通过client.post(/realtime/calls/{call_id}/accept, ...)调用 Realtime Calls API并将起始 Agent 的静态 instructions 直接透传进接听载荷对 404来电方已挂断做无害化处理examples/realtime/twilio_sip/server.py。附加会话observe_call()创建RealtimeRunner(assistant_agent, modelOpenAIRealtimeSIPModel())以model_config{call_id: call_id, initial_model_settings: {...}}进入会话随后消费history_added、error等事件并把来电者/助手文本记录到日志examples/realtime/twilio_sip/server.py。重复 Webhook 防护通过active_call_tasks字典跟踪后台任务避免 Twilio 重试投递导致重复的会话观察者examples/realtime/twilio_sip/server.py。在模型层OpenAIRealtimeSIPModel是OpenAIRealtimeWebSocketModel的子类其connect()强制要求options中必须携带call_id否则抛出UserError。此外它还提供build_initial_session_payload()静态方法用于在接听 SIP 来电时把会话配置载荷转发给 Realtime Calls API避免重复实现会话初始化逻辑。运行该示例的完整步骤见 examples/realtime/twilio_sip/README.md要点包括配置指向https://your-public-host/openai/webhook的 Webhook、在 Twilio Elastic SIP Trunking 中添加sip:proj_your_project_idsip.api.openai.com;transporttls的 Origination URI然后执行export OPENAI_API_KEYsk-... export OPENAI_WEBHOOK_SECRETwhsec_... uv run uvicorn examples.realtime.twilio_sip.server:app --host 0.0.0.0 --port 8000更广泛的 Realtime API 也会将call_id用于某些服务器端控制模式但本仓库提供的接入示例使用的是 SIP。SDK 范围之外浏览器 WebRTC如果你的应用主要使用 Realtime WebRTC 浏览器客户端请将其视为不在本仓库 Python SDK 文档的范围内客户端流程与事件模型请查阅 OpenAI 官方的 Realtime API 与 WebRTC、实时对话相关文档。如果除浏览器 WebRTC 客户端外还需要旁路服务器连接请参考官方实时服务器端控制指南。不要期望本仓库提供浏览器端RTCPeerConnection抽象或现成的浏览器 WebRTC 示例本仓库目前也未提供浏览器 WebRTC 与 Python 旁路连接结合使用的示例。换言之选择 WebRTC 意味着你的实时层由浏览器直接管理Python 服务无法复用本文介绍的服务端 WebSocket/SIP 会话管理能力。自定义端点与接入点RealtimeModelConfigRealtimeModelConfig是传输配置的核心接口它允许你在不更换模型类的前提下自定义默认传输行为字段作用源码说明url覆盖 WebSocket 端点未设置时使用默认的 OpenAI WebSocket URLsrc/agents/realtime/openai_realtime.pyheaders提供显式请求头例如 Azure 身份验证请求头设置后 SDK 不会在底层自动附加 Authorization 头api_key直接传入 API 密钥或通过回调传入未设置时回退到OPENAI_API_KEY环境变量get_api_key()见 src/agents/realtime/openai_realtime.pycall_id接入现有实时通话本仓库记录的示例使用 SIP设置后连接 URL 变为wss://api.openai.com/v1/realtime?call_id...playback_tracker报告实际播放进度以便处理中断未设置时使用音频即时按实时速度播放的默认假设自定义传输行为的实现细节从 OpenAIRealtimeWebSocketModel.connect() 的源码可以提炼出几个容易踩坑的细节call_id与model_name互斥若同时指定两者SDK 会抛出UserErrorCannot specify bothcall_idandmodel_name...因为附加现有通话不需要也不允许指定模型名。认证头策略当headers被显式提供时SDK 只合并你传入的请求头不再自动写入Authorization: Bearer api_key——这正是 Azure 等需要api-key头的场景所依赖的行为反之未提供headers且拿不到 API key 时connect()会直接抛错。playback_tracker的用途模型生成音频的速度远快于真实播放速度因此发生中断时模型需要知道用户实际听完了多少音频。RealtimePlaybackTrackersrc/agents/realtime/model.py允许你在自定义播放逻辑如电话或远程交互场景中通过on_play_bytes/on_play_ms回报真实进度并在on_interrupted()时重置状态低延迟场景下则可以放心使用默认实现。选择拓扑后请继续阅读实时智能体指南以了解详细的生命周期与功能接口并结合实时快速入门完成第一个可运行的实时会话。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考