OpenAI Agents SDK Realtime 智能体完全指南会话生命周期、音频配置与电话集成实战【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python本指南基于 OpenAI Agents Python SDK 的 Realtime 层系统讲解它如何映射到 OpenAI Realtime API以及 Python SDK 在此基础上增加的智能体级行为——包括会话生命周期、音频输入输出与转录配置、事件与中断处理、工具审批与任务转移、安全防护措施以及 SIP 电话接入和自定义端点连接。读完本文你将掌握用RealtimeAgent/RealtimeRunner/RealtimeSession构建低延迟语音对话应用含 WebSocket 与 SIP 两条路径的完整能力。概述Realtime 层与核心组件Realtime 智能体会与 Realtime API 保持长期连接使模型能够以增量方式处理文本和音频、以流式方式输出音频、调用工具并处理中断而无需在每个轮次都重新发起请求。这与纯文本运行每轮重新调用Runner.run()有本质区别Realtime 层把「一次请求-一次响应」的范式转变为「一条持久连接 持续双向事件流」。Python SDK 中 Realtime 层的四个主要组件源码分别位于 src/agents/realtime/ 目录组件职责源码位置RealtimeAgent一个 Realtime 专用智能体的指令、工具、输出安全防护措施和任务转移src/agents/realtime/agent.pyRealtimeRunner将起始智能体连接到 Realtime 传输层的会话工厂src/agents/realtime/runner.pyRealtimeSession发送输入、接收事件、追踪历史记录并执行工具的实时会话src/agents/realtime/session.pyRealtimeModel传输抽象默认实现是 OpenAI 的服务器端 WebSocketsrc/agents/realtime/model.py从源码结构看RealtimeRunner是「会话工厂」其run()方法并不立即返回最终结果而是构造并返回一个RealtimeSession见 runner.py由会话对象负责与模型层的双向通信。RealtimeModel则是一个抽象基类abc.ABC定义了connect()、add_listener()、remove_listener()、send_event()、close()等接口见 model.py因此可以替换传输实现而不影响上层智能体逻辑。入门指引如果要使用默认的 Python 路径请先阅读快速入门。如果正在决定应用应使用服务器端 WebSocket 还是 SIP请阅读 Realtime 传输方式。浏览器 WebRTC 传输不属于 Python SDK——Python SDK 适用于服务端编排、工具、审批和电话集成。会话生命周期典型的 Realtime 会话流程如下创建一个或多个RealtimeAgent。使用起始智能体创建RealtimeRunner。调用await runner.run()获取RealtimeSession。使用async with session:或await session.enter()进入会话。使用send_message()或send_audio()发送用户输入。迭代处理会话事件直到对话结束。与纯文本运行不同runner.run()不会立即生成最终结果。它会返回一个实时会话对象使本地历史记录、后台工具执行、安全防护措施状态和当前智能体配置与传输层保持同步。这一点在 runner.py 的 docstring 中也有明确说明会话负责维护本地历史副本、执行工具、运行安全防护措施并促成智能体之间的任务转移。默认情况下RealtimeRunner使用OpenAIRealtimeWebSocketModel见 runner.py因此默认 Python 路径是与 Realtime API 建立服务器端 WebSocket 连接。如果传入不同的RealtimeModel仍可使用相同的会话生命周期和智能体功能但连接机制可以改变。连接的正常关闭与异常退出当 Realtime API 服务器正常关闭默认 WebSocket 连接时模型传输层会依次发出两个底层模型事件类型定义见 model_events.pydisconnected状态的RealtimeModelConnectionStatusEventRealtimeModelEndOfStreamEvent表示事件流永久结束、不再产生新事件。RealtimeSession会在raw_model_event中转发这两个事件处理完已进入队列的事件然后结束异步迭代且不引发异常。需要注意由调用方发起的session.close()不会生成这些服务器断开连接事件意外的 WebSocket 故障仍会进入会话的异常处理路径而不会像服务器正常关闭一样结束迭代。智能体与会话配置RealtimeAgent的功能范围有意设计得比常规Agent类型更窄存在以下限制与能力边界模型选择在会话级别配置而不是按智能体配置不支持 structured outputs可以配置语音但会话生成语音音频后无法更改指令、函数工具、任务转移、钩子和输出安全防护措施仍然全部可用。RealtimeSessionModelSettings同时支持较新的嵌套audio配置和旧版扁平别名input_audio_format、output_audio_format、input_audio_transcription、turn_detection等见 config.py。新代码应优先使用嵌套结构并对新的 Realtime 智能体使用gpt-realtime-2.1作为起点runner RealtimeRunner( starting_agentagent, config{ model_settings: { model_name: gpt-realtime-2.1, audio: { input: { format: pcm16, transcription: {model: gpt-4o-mini-transcribe}, turn_detection: {type: semantic_vad, interrupt_response: True}, }, output: {format: pcm16, voice: ash}, }, tool_choice: auto, } }, )会话级设置一览从 config.py 的RealtimeSessionModelSettings与RealtimeAudioConfig类型定义可以整理出完整的会话级设置model_nameRealtime 模型名称。类型别名RealtimeModelName列出了 SDK 已知的模型名包括gpt-realtime、gpt-realtime-1.5、gpt-realtime-2、gpt-realtime-2.1、gpt-realtime-2.1-mini、gpt-realtime-2025-08-28、gpt-4o-realtime-preview系列以及gpt-realtime-mini系列见 config.pyaudio.input.format、audio.output.format音频格式支持pcm16、g711_ulaw、g711_alawaudio.input.transcription输入转录配置audio.input.noise_reduction降噪模式near_field或far_fieldaudio.input.turn_detection轮次检测配置支持semantic_vad或server_vad可设置create_response、eagernessauto/low/medium/high、interrupt_response、prefix_padding_ms、silence_duration_ms、threshold、idle_timeout_ms等设为None可禁用自动轮次检测audio.output.voice、audio.output.speed输出语音与语速output_modalities输出模态text/audiotool_choice模型如何选择要调用的工具prompt会话提示词tracing请求追踪配置workflow_name、group_id、metadata。运行级设置一览RealtimeRunner(config...)上常用的运行级设置RealtimeRunConfig见 config.py包括async_tool_calls函数工具调用是否异步执行默认Trueoutput_guardrails作用于智能体响应的输出安全防护措施列表guardrails_settings.debounce_text_length输出文本或转录增量累积到多少字符后才运行安全防护措施默认 100且每累积到该阈值的 1x、2x、3x… 倍数时都会运行tool_execution.pre_approval_tool_input_guardrails是否在发出待审批事件之前先运行函数工具的输入安全防护措施审批后执行前仍会再次检查tool_error_formatter格式化返回给模型的工具错误消息的回调tracing_disabled是否禁用本次运行的追踪。有关完整的类型化接口请参阅RealtimeRunConfig和RealtimeSessionModelSettingssrc/agents/realtime/config.py。输入转录设置在audio.input.transcription下配置输入转录类型见 RealtimeInputAudioTranscriptionConfig。SDK 会在嵌套会话配置中转发特定于模型的 GA 转录设置。低延迟增量转录gpt-live-transcribe使用gpt-live-transcribe可获得低延迟的增量转录。它支持三个附加字段prompt自由格式的录音上下文keywords列出音频中可能出现的确切词语languages列出预期的输入语言注意此模型使用复数languages而不是单数language请勿同时发送这两个字段。runner RealtimeRunner( starting_agentagent, config{ model_settings: { audio: { input: { transcription: { model: gpt-live-transcribe, prompt: A support call about the OpenAI Agents SDK., keywords: [RunState, MCPServerManager], languages: [en, ja], }, turn_detection: None, } } } }, )延迟与准确率权衡gpt-realtime-whisper 的 delay此 SDK 固定使用的 OpenAI 客户端版本仅支持将delay与gpt-realtime-whisper搭配使用。delay设置接受minimal、low、medium、high或xhigh较低的值可以更早生成部分文本较高的值会为转录模型提供更多音频上下文并可能提高识别准确率。runner RealtimeRunner( starting_agentagent, config{ model_settings: { audio: { input: { transcription: { model: gpt-realtime-whisper, delay: low, }, turn_detection: None, } } } }, )应使用具有代表性的音频进行基准测试而不要假定任何级别都有固定的处理时长。按轮次转录gpt-transcribe仅当转录应在提交音频轮次后开始或应用需要输出检测到的语言时才应在通过 WebSocket 建立的 Realtime 会话中使用gpt-transcribe。该模型会自动将之前已转录的轮次用作上下文。gpt-transcribe完成事件会在其languages输出字段中报告检测到的语言——这个输出字段不同于上文gpt-live-transcribe的预期语言输入字段。禁用自动轮次检测将audio.input.turn_detection设置为None会禁用自动轮次检测。之后应用必须按照下文「手动响应控制」中的说明提交音频轮次并控制响应创建。输入与输出文本与结构化用户消息使用session.send_message()src/agents/realtime/session.py发送纯文本或结构化 Realtime 消息。结构化消息是在 Realtime 对话中加入图像输入的主要方式——消息内容可以混合input_text与input_image类型类型定义见 config.py 的 RealtimeUserInputMessagefrom agents.realtime import RealtimeUserInputMessage await session.send_message(Summarize what we discussed so far.) message: RealtimeUserInputMessage { type: message, role: user, content: [ {type: input_text, text: Describe this image.}, {type: input_image, image_url: image_data_url, detail: high}, ], } await session.send_message(message)examples/realtime/app/server.py 中的 Web 演示代码示例会以这种方式转发input_image消息。input_image的detail支持auto、low、high。音频输入使用session.send_audio()以流式方式发送原始音频字节await session.send_audio(audio_bytes)如果禁用了服务器端轮次检测则需要自行标记轮次边界。高层便捷方式如下await session.send_audio(audio_bytes, commitTrue)如果需要更底层的控制也可以通过底层模型传输对象直接发送 Realtime API 客户端事件例如input_audio_buffer.commit。手动响应控制session.send_message()会通过高层路径发送用户输入并自动开始响应。但在某些配置中原始音频缓冲不会自动执行相同操作。在 Realtime API 层面手动轮次控制是指发送一个session.update事件将turn_detection设置为null然后自行发送input_audio_buffer.commit和response.create。在 Python SDK 中可以通过模型传输对象发送原始客户端事件RealtimeModelSendRawMessage定义见 model_inputs.pyfrom agents.realtime.model_inputs import RealtimeModelSendRawMessage await session.model.send_event( RealtimeModelSendRawMessage( message{ type: response.create, } ) )此模式适用于以下情况禁用了turn_detection并且希望自行决定模型何时响应希望在触发响应前检查或拦截用户输入需要为带外响应使用自定义提示词。examples/realtime/twilio_sip/server.py 中的 SIP 代码示例使用原始response.create强制生成开场问候语。事件、历史记录与中断RealtimeSession会发出更高层的 SDK 事件同时仍会在需要时转发原始模型事件完整事件联合类型RealtimeSessionEvent见 events.py。重要的会话事件包括audio、audio_end、audio_interruptedagent_start、agent_endtool_start、tool_end、tool_approval_requiredhandoffhistory_added、history_updatedguardrail_trippedinput_audio_timeout_triggerederrorraw_model_event对于 UI 状态而言通常最有用的事件是history_added和history_updated。它们将会话的本地历史记录公开为RealtimeItem对象类型定义见 src/agents/realtime/items.py包括用户消息、助手消息和工具调用——前端可以直接据此渲染对话界面。用量统计当已完成的模型响应包含用量信息时SDK 的 OpenAIRealtimeModel传输层会在raw_model_event中发出RealtimeModelUsageEvent。其usage字段包含该响应的 token 数量而input_tokens_details和output_tokens_details则提供可选的模态明细文本 / 音频 / 图像 token 分解含缓存 token 信息。会话还会将每个响应的用量添加到共享的RunContextWrapper.usagesrc/agents/run_context.py。可从后续高层事件例如agent_end的event.info.context.usage中读取以查看实时会话的累计用量from agents.realtime import RealtimeModelUsageEvent async for event in session: if event.type raw_model_event and isinstance( event.data, RealtimeModelUsageEvent ): response_usage event.data.usage print(Response tokens:, response_usage.total_tokens) print(Input modalities:, event.data.input_tokens_details) print(Output modalities:, event.data.output_tokens_details) elif event.type agent_end: session_usage event.info.context.usage print(Session tokens:, session_usage.total_tokens)注意两点仅当模型提供方在已完成的响应中包含用量信息时才会报告用量累计值涵盖该RealtimeSession收到的响应不是跨会话总计。中断与播放进度追踪当用户打断助手时会话会发出audio_interrupted并更新历史记录使服务器端对话与用户实际听到的内容保持一致。对于低延迟本地播放默认播放追踪器通常已足够其假设音频立即以实时速度播放。但在远程或延迟播放场景中尤其是电话场景请使用RealtimePlaybackTracker以便在实际播放位置截断被中断的响应而不是假定所有已生成的音频都已播放给用户。RealtimePlaybackTracker的用法要点见 model.py通过on_play_bytes(item_id, item_content_index, bytes)或on_play_ms(item_id, item_content_index, ms)报告实际播放进度通过session.model/model_config[playback_tracker]注入自定义追踪器实例模型传输层会调用get_state()查询当前播放位置当前 item、内容索引、已播放毫秒数用于精确截断被中断的响应。examples/realtime/twilio/twilio_handler.py 中的 Twilio 代码示例展示了此模式。工具、批准、任务转移与安全防护措施函数工具Realtime 智能体支持在实时对话期间使用函数工具与常规Agent一样通过tool装饰器定义from agents.decorators import tool tool def get_weather(city: str) - str: Get current weather for a city. return fThe weather in {city} is sunny, 72F. agent RealtimeAgent( nameAssistant, instructionsYou can answer weather questions., tools[get_weather], )工具调用期间会话会依次发出tool_start与tool_end事件见 events.py工具执行在后台进行不影响音频流。工具批准函数工具可以要求在执行前获得人工批准。发生这种情况时会话会发出tool_approval_required并暂停工具运行直到调用approve_tool_call()或reject_tool_call()async for event in session: if event.type tool_approval_required: await session.approve_tool_call(event.call_id)审批机制有两个细节需要注意如果该工具还有输入安全防护措施这些安全防护措施会在批准后的执行前立即运行若要在发出批准事件之前运行它们请使用RealtimeRunner(..., config{tool_execution: {pre_approval_tool_input_guardrails: True}})创建运行器。通过此批准前检查的调用在获批后、执行前仍会再次接受检查这正是 RealtimeToolExecutionConfig 所描述的行为。有关具体的服务器端批准循环请参阅 examples/realtime/app/server.py。人工参与流程文档也会指向此流程。任务转移Realtime 任务转移允许一个智能体将实时对话转交给另一个专用智能体from agents.realtime import RealtimeAgent, realtime_handoff billing_agent RealtimeAgent( nameBilling Support, instructionsYou specialize in billing issues., ) main_agent RealtimeAgent( nameCustomer Service, instructionsTriage the request and hand off when needed., handoffs[ realtime_handoff( billing_agent, tool_description_overrideTransfer to billing support, ) ], )直接用作任务转移的RealtimeAgent对象会被自动包装而realtime_handoff(...)可用于自定义名称、描述、验证、回调和可用性。任务转移发生时会话发出handoff事件含from_agent与to_agent并伴随agent_end/agent_start事件。注意Realtime 任务转移不支持常规任务转移的input_filter。安全防护措施Realtime 智能体支持针对智能体响应的输出安全防护措施以及针对函数工具调用的输入安全防护措施。输出安全防护措施检查会进行防抖处理每次检查都针对累积的输出文本和音频转录增量运行而不是针对每个部分增量运行默认累积 100 字符起检可由guardrails_settings.debounce_text_length调整并发出guardrail_tripped而不是引发异常from agents.guardrail import GuardrailFunctionOutput, OutputGuardrail def sensitive_data_check(context, agent, output): return GuardrailFunctionOutput( tripwire_triggeredpassword in output, output_infoNone, ) agent RealtimeAgent( nameAssistant, instructions..., output_guardrails[OutputGuardrail(guardrail_functionsensitive_data_check)], )安全防护措施触发时的行为细节当 Realtime 输出安全防护措施因音频转录而触发时会话会中断当前响应强制执行response.cancel发出guardrail_tripped发送一条指出已触发安全防护措施名称的后续用户消息以便模型生成替代响应。此时音频播放器仍应监听audio_interrupted并立即停止本地播放因为触发安全防护措施时部分音频可能已进入缓冲区。其余边界行为使用内置 OpenAI Realtime 传输方式时如果安全防护措施检查在被检查的响应结束后才完成会话只会中断该响应的缓冲播放而不会取消之后开始的任何响应对于纯文本输出会话会改为发送一个限定于该响应的response.cancel由于没有需要停止的音频播放因此不会发出audio_interrupted。使用内置 OpenAI Realtime 模型时纯文本路径也会发出相同的guardrail_tripped事件和后续用户消息。自定义传输层的实现要求自定义RealtimeModel传输方式必须遵循RealtimeModelSendInterrupt.response_id和playback_only定义见 model_inputs.py才能提供同样限定于源响应的音频中断行为。它们还必须覆盖RealtimeModel.send_event_if()见 model.py以支持纯文本输出路径的恢复消息。实现必须在传输层实际提交事件的边界重新检查所提供的条件或者将条件检查与事件提交串行化。默认实现会安全地跳过恢复消息因为如果只检查一次条件、然后单独发送事件在检查与事件提交之间可能会启动另一个响应不过响应取消和guardrail_tripped事件仍会发生。SIP 与电话Python SDK 通过OpenAIRealtimeSIPModel提供原生支持的SIP 挂接流程对应RealtimeModelConfig.call_id见 model.py。当通话通过 Realtime Calls API 到达并且希望将智能体会话挂接到生成的call_id时请使用此流程from agents.realtime import RealtimeRunner from agents.realtime.openai_realtime import OpenAIRealtimeSIPModel runner RealtimeRunner(starting_agentagent, modelOpenAIRealtimeSIPModel()) async with await runner.run( model_config{ call_id: call_id_from_webhook, } ) as session: async for event in session: ...如果需要先接听通话并希望接听请求体与从智能体生成的会话配置保持一致请使用OpenAIRealtimeSIPModel.build_initial_session_payload(...)。完整流程请参阅 examples/realtime/twilio_sip/server.py其中也演示了用原始response.create强制生成开场问候语的模式。底层访问与自定义端点可以通过session.model访问底层传输对象。在以下情况下可使用此对象通过session.model.add_listener(...)添加自定义监听器RealtimeModelListener接口见 model.py发送原始客户端事件例如response.create或session.update通过model_config自定义处理url、headers或api_key使用call_id挂接到现有 Realtime 通话。RealtimeModelConfigmodel.py支持以下字段字段说明api_keyAPI 密钥或返回密钥的函数未设置时使用默认值OpenAI 模型回退到OPENAI_API_KEY环境变量url连接 URL未设置时使用默认 WebSocket URLheaders连接请求头设置后 SDK 不会自动添加Authorizationinitial_model_settings连接时的初始模型设置playback_tracker播放进度追踪器默认实现假设音频立即以实时速度播放call_id挂接到现有 Realtime 通话本仓库打包的示例为 SIP此代码仓库随附的call_id代码示例使用 SIP。更广泛的 Realtime API 也会在某些服务器端控制流程中使用call_id但此处未将其打包为 Python 代码示例。连接 Azure OpenAI连接 Azure OpenAI 时请传入GA Realtime 端点 URL和显式请求头。例如使用 API 密钥认证session await runner.run( model_config{ url: wss://your-resource.openai.azure.com/openai/v1/realtime?modeldeployment-name, headers: {api-key: your-azure-api-key}, } )若使用基于 token 的身份验证请在headers中使用 bearer tokensession await runner.run( model_config{ url: wss://your-resource.openai.azure.com/openai/v1/realtime?modeldeployment-name, headers: {authorization: fBearer {token}}, } )两个关键提醒如果传入headersSDK不会自动添加Authorization请求头请勿对 Realtime 智能体使用旧版 beta 路径/openai/realtime?api-version...应使用正式发布版GA端点。延伸阅读Realtime 传输方式WebSocket 与 SIP 选型Realtime 快速入门Realtime 会话源码实现 与事件类型定义Realtime 运行配置与模型设置类型完整可运行示例examples/realtime/appWeb 演示、examples/realtime/twilioTwilio 电话、examples/realtime/twilio_sipSIP 接入、examples/realtime/cli命令行演示人工参与流程文档工具审批循环的端到端流程【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考