OpenAI Agents SDK Realtime 智能体完全指南会话生命周期、工具审批、Handoff 与低层控制【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python本篇指南深入剖析 OpenAI Agents SDKopenai-agents-python中 Realtime 智能体Realtime Agents的完整工作方式SDK 如何把 Realtime 层映射到 OpenAI Realtime API以及在原生 Realtime API 之上额外提供了哪些行为。读完本文你将掌握 RealtimeAgent / RealtimeRunner / RealtimeSession 的核心组件关系、会话生命周期、输入转写与手动响应控制、事件与历史同步、中断处理、工具审批、Handoff、输出护栏、SIP 电话接入以及自定义端点等底层能力并能在自己的服务端代码中直接落地。Realtime 层概述长连接、增量处理与四大核心组件与文本对话中每一轮都发起一次新请求不同Realtime 智能体在客户端与服务端之间维持一条长连接模型可以增量地处理文本与音频、流式输出音频、调用工具、处理打断全程无需为每一轮重新开启请求。SDK 的 Realtime 层由四个核心组件构成src/agents/realtime/组件职责RealtimeAgent描述一个 Realtime 专项智能体指令instructions、工具、输出护栏与 handoffRealtimeRunner会话工厂把起始智能体与一个 Realtime 传输层绑定RealtimeSession活跃会话发送输入、接收事件、维护本地历史、执行工具RealtimeModel传输抽象默认实现是 OpenAI 服务端 WebSocket 模型从源码结构看RealtimeRunner是普通Runner的 Realtime 等价物src/agents/realtime/runner.py中的 docstring 明确说明它通过维持与模型层的持久连接自动处理多轮对话会话负责本地历史副本管理、工具执行、护栏与智能体间 handoff。由于这段代码运行在你的服务器上因此默认走 WebSocket你也可以实现RealtimeModel接口来自定义模型层。会话生命周期从创建智能体到流式消费事件一个典型的 Realtime 会话流程如下创建一个或多个RealtimeAgent用起始智能体创建RealtimeRunnerawait runner.run()获得一个RealtimeSession通过async with session:或await session.enter()进入会话连接在此刻建立用send_message()或send_audio()发送用户输入迭代消费会话事件直到对话结束。与纯文本运行不同runner.run()不会立刻产生最终结果。它返回一个实时会话对象该对象在传输层之上同步维护本地历史、后台工具执行、护栏状态与当前活跃智能体配置。RealtimeSession的构造逻辑可以在src/agents/realtime/session.py中看到它持有_history列表、事件队列asyncio.Queue、待处理工具调用表与护栏状态追踪等内部结构。默认情况下RealtimeRunner使用OpenAIRealtimeWebSocketModel见src/agents/realtime/runner.py中self._model model if model is not None else OpenAIRealtimeWebSocketModel()所以默认的 Python 路径是到 Realtime API 的服务端 WebSocket 连接。如果你传入不同的RealtimeModel会话生命周期与智能体功能保持不变只有连接机制会改变。关于正常断开的行为当 Realtime API 服务端正常关闭默认 WebSocket 连接时模型传输层会先发出disconnected状态的RealtimeModelConnectionStatusEvent随后发出RealtimeModelEndOfStreamEvent二者均定义于src/agents/realtime/model_events.py。RealtimeSession会把这两个事件都转发到raw_model_event排空已排队的事件然后正常结束异步迭代——不会抛异常。而由调用方发起的session.close()不会合成这些服务端断开事件意外的 WebSocket 失败则会走会话的异常路径而不是像正常服务端关闭那样结束迭代。智能体与会话配置更窄的 RealtimeAgent 与模型设置RealtimeAgent相比普通Agent是有意做得更窄的src/agents/realtime/agent.py的类 docstring 列出了限制模型选择在会话级别配置而不是每个智能体一个模型——同一个RealtimeSession中的所有 RealtimeAgent 由同一个模型处理不支持结构化输出outputType语音voice可以配置但一旦会话已经产生过语音音频就不能再更改指令instructions、函数工具、handoff、hooks 与输出护栏仍然全部可用。RealtimeSessionModelSettings同时支持较新的嵌套式audio配置和旧的扁平别名。新代码请优先使用嵌套结构新 Realtime 智能体建议从gpt-realtime-2.1模型开始。这一点与源码一致在src/agents/realtime/openai_realtime.py中DEFAULT_REALTIME_MODEL gpt-realtime-2.1且DEFAULT_MODEL_SETTINGS默认使用voice: ash、modalities: [audio]、pcm16音频格式、gpt-4o-mini-transcribe转写与semantic_vad自动断句。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, } }, )会话级常用设置model_settings内audio.input.format、audio.output.format输入/输出音频格式源码类型别名支持pcm16、g711_ulaw、g711_alaw等src/agents/realtime/config.py中的RealtimeAudioFormataudio.input.transcription输入音频转写配置audio.input.noise_reduction降噪模式支持near_field近场与far_field远场audio.input.turn_detection自动断句检测设为None可禁用audio.output.voice、audio.output.speed输出语音与语速output_modalities输出模态text/audiotool_choice模型如何选择要调用的工具prompt会话提示词tracing请求追踪配置。运行级常用设置RealtimeRunner(config...)内async_tool_calls函数工具调用是否异步执行源码默认值为True见src/agents/realtime/config.py中RealtimeRunConfig的说明output_guardrails作用于智能体响应的输出护栏列表guardrails_settings.debounce_text_length输出护栏的防抖阈值默认100即累积文本每达到该阈值的 1 倍、2 倍、3 倍……就运行一次护栏tool_error_formatter把工具错误消息格式化后返回给模型的可选回调tracing_disabled是否关闭本次运行的追踪。完整的类型化配置面可查阅RealtimeRunConfig与RealtimeSessionModelSettings对应agents.realtime.config模块。旧版扁平别名如input_audio_format、output_audio_format、input_audio_transcription、turn_detection仍然可用但新代码应优先使用嵌套的audio结构。输入转写设置低延迟增量转写 vs 提交后转写输入转写配置在audio.input.transcription下完成。根据你的延迟与准确性需求选择转写模型gpt-live-transcribe用于低延迟的增量转写边说话边出文本gpt-transcribeWebSocket 场景下转写从一次音频轮次提交committed之后才开始或者当应用需要检测语言输出时使用。SDK 会把模型特定的 GA 转写设置以嵌套会话配置的形式转发。以下示例同时给出了prompt、keywords与languages的用法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-live-transcribe而言prompt提供自由格式的录制上下文keywords列出音频中可能出现的字面术语languages列出预期的输入语言。注意该模型使用复数languages字段而不是单数language不要同时发送两个字段。本 SDK 锁定的 OpenAI 客户端版本只在gpt-realtime-whisper上支持delay参数用于配置该模型的延迟/准确性权衡runner RealtimeRunner( starting_agentagent, config{ model_settings: { audio: { input: { transcription: { model: gpt-realtime-whisper, delay: low, }, turn_detection: None, } } } }, )delay的取值包括minimal、low、medium、high、xhigh这也是RealtimeInputAudioTranscriptionConfig中delay字段的字面量。取值越低越可能更早产出部分文本取值越高转写模型获得越多的音频上下文识别准确率可能越好。由于不同音频的实际时序不同请以代表性音频做基准测试而不要假设某个档位有固定延迟。gpt-transcribe仅在转写应在一次已提交的音频轮次之后开始或应用需要检测语言输出时才应在 WebSocket 的 Realtime 会话中使用。该模型会自动把此前已转写的轮次作为上下文其 completion 事件会在languages输出字段中报告检测到的语言——注意这个输出字段与gpt-live-transcribe上面展示的预期语言输入字段是不同的东西。把audio.input.turn_detection设为None会禁用自动断句此时应用必须自行提交音频轮次并控制响应创建详见下文手动响应控制一节。输入与输出文本、结构化消息、音频与手动响应控制文本与结构化用户消息用session.send_message()发送纯文本或结构化 Realtime 消息对应agents.realtime.session.RealtimeSession.send_message源码实现为向模型传输层发送RealtimeModelSendUserInput事件见src/agents/realtime/session.pyfrom 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)结构化消息是在 Realtime 对话中携带图片输入的主要方式。RealtimeUserInputMessage的结构定义于src/agents/realtime/config.pycontent是RealtimeUserInputTextinput_text与RealtimeUserInputImageinput_image支持detail取auto/low/high的列表。仓库中的 Web 演示 examples/realtime/app/server.py 正是以这种方式转发input_image消息的。音频输入用session.send_audio()流式发送原始音频字节await session.send_audio(audio_bytes)如果服务端断句检测被禁用你需要自己标记轮次边界。高层便捷用法是await session.send_audio(audio_bytes, commitTrue)对应源码中RealtimeModelSendAudio(audioaudio, commitcommit)见src/agents/realtime/session.py的send_audio方法。如果需要更低层的控制你也可以直接通过底层模型传输层发送 Realtime API 客户端事件例如input_audio_buffer.commit。手动响应控制session.send_message()走高层路径发送用户输入并自动开始一次响应但在某些配置下原始音频缓冲不会自动做同样的事。在 Realtime API 层面手动轮次控制意味着发送一个把turn_detection置为null的session.update事件然后自行发送input_audio_buffer.commit与response.create。如果正在手动管理轮次可以通过模型传输层发送原始客户端事件from agents.realtime.model_inputs import RealtimeModelSendRawMessage await session.model.send_event( RealtimeModelSendRawMessage( message{ type: response.create, } ) )这个模式在以下场景很有用turn_detection被禁用你想自己决定模型何时响应你想在触发响应之前检查或门控用户输入你需要为一次带外out-of-band响应提供自定义提示词。examples/realtime/twilio_sip/server.py 中的 SIP 示例就用一条原始的response.create来强制播放开场问候语。事件、历史与中断RealtimeSession会发出高层 SDK 事件同时在需要时转发原始模型事件两类事件的类型定义分别见src/agents/realtime/events.py与src/agents/realtime/model_events.py。高价值的会话事件包括audio、audio_end、audio_interrupted音频流、音频结束、音频被打断agent_start、agent_end一个智能体回合开始/结束分别对应模型层的turn_started/turn_endedtool_start、tool_end、tool_approval_required工具调用开始/结束、需要人工审批handoff智能体切换history_added、history_updated会话历史新增/整体更新guardrail_tripped输出护栏被触发input_audio_timeout_triggered检测到用户静默超时error错误raw_model_event原始模型事件转发。对 UI 状态来说最有用的通常是history_added与history_updated它们把会话的本地历史以RealtimeItem对象暴露出来包括用户消息、助手消息与工具调用。从src/agents/realtime/session.py的on_event处理逻辑可以看到会话在收到item_updated、input_audio_transcription_completed、item_deleted等模型事件时都会同步_history并据此发出history_added或history_updated。用量统计Usage当一条已完成的模型响应包含用量信息时SDK 的 OpenAIRealtimeModel传输层会在raw_model_event中发出RealtimeModelUsageEvent定义于src/agents/realtime/model_events.py。其usage字段包含该响应的 token 计数input_tokens_details与output_tokens_details提供可选的模态细分文本/音频/图片/cached token 等。会话还会把每次响应的用量累加到共享的RunContextWrapper.usage上源码中on_event收到usage事件时执行self._context_wrapper.usage.add(event.usage)。可以在后续高层事件如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并更新历史使服务端对话与用户实际听到的内容保持一致。在低延迟的本地播放场景中默认的播放追踪器通常就够用了。但在远程或延迟播放场景尤其是电话中应使用RealtimePlaybackTrackersrc/agents/realtime/model.py让被打断的响应在实际播放位置处截断而不是假设所有已生成的音频都已被听到。RealtimePlaybackTracker由你负责在播放进度发生时调用on_play_bytes或on_play_ms并在中断时调用on_interrupted模型生成音频的速度远快于实时播放因此知道用户实际听到了多少音频对正确处理打断至关重要。examples/realtime/twilio/twilio_handler.py 中的 Twilio 示例展示了这一模式。工具、审批、Handoff 与护栏函数工具Realtime 智能体在实时对话中支持函数工具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_approval_required并暂停工具运行直到你调用approve_tool_call()或reject_tool_call()源码中审批事件会携带call_id会话据此在_pending_tool_calls中查找待处理的调用见src/agents/realtime/session.py。如果该工具还带有输入护栏这些护栏默认在审批之后、执行之前立即运行。若希望在发出审批事件之前就运行它们请用如下方式创建 runnerrunner RealtimeRunner( starting_agentagent, config{tool_execution: {pre_approval_tool_input_guardrails: True}}, )通过这一预审批检查的调用在审批后、执行前仍会再次被检查源码中_pre_approval_tool_input_guardrails_enabled()读取该配置项并在_maybe_request_tool_approval中实现这一流程。async for event in session: if event.type tool_approval_required: await session.approve_tool_call(event.call_id)一个完整的服务端审批循环可参考 examples/realtime/app/server.py人机协同Human in the loop 文档也会引用这一流程。HandoffRealtime handoff 允许一个智能体把实时对话转移给另一个专项智能体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, ) ], )直接作为 handoff 使用的RealtimeAgent对象会被自动包装realtime_handoff(...)允许你自定义名称、描述、校验、回调与可用性。注意Realtime handoff不支持普通 handoff 的input_filter从src/agents/realtime/handoffs.py的实现可以看到其专门的处理逻辑。护栏GuardrailsRealtime 智能体支持对智能体响应做输出护栏、对函数工具调用做输入护栏。输出护栏检查是防抖的每次检查运行在累积的输出文本与音频转写增量上而不是每个局部增量都跑一次触发时发出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语义同时必须覆写RealtimeModel.send_event_if()以支持纯文本输出路径的恢复消息——实现必须在传输层真正的事件提交边界重新检查所给条件或将条件检查与事件提交串行化。默认实现会安全地跳过恢复消息因为如果先检查一次条件再单独发送事件检查与提交之间可能有另一个响应开始不过响应取消与guardrail_tripped事件仍会发生send_event_if的默认行为见src/agents/realtime/model.py。SIP 与电话接入Python SDK 通过OpenAIRealtimeSIPModelsrc/agents/realtime/openai_realtime.py提供一等公民的 SIP 挂接attach流程。当一通电话通过 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: ...如果需要先接听accept来电并希望 accept 的 payload 与由智能体派生的会话配置一致可以使用OpenAIRealtimeSIPModel.build_initial_session_payload(...)。完整流程见 examples/realtime/twilio_sip/server.py。底层访问与自定义端点可以通过session.model访问底层传输对象。这在以下场景中需要通过session.model.add_listener(...)添加自定义监听器发送response.create或session.update等原始客户端事件通过model_config自定义url、headers或api_key处理通过call_id挂接到已有的 Realtime 通话。RealtimeModelConfigsrc/agents/realtime/model.py支持api_keyAPI 密钥或返回密钥的函数未设置时模型会使用合理默认值例如 OpenAI Realtime 模型读取OPENAI_API_KEY环境变量url自定义 WebSocket 端点未设置时使用默认 OpenAI WebSocket URLheaders自定义请求头。注意一旦你显式传入headersSDK 就不会再自动注入Authorization头例如 Azure OpenAI Realtime WebSocket 连接使用{api-key: ...}initial_model_settings连接时使用的初始模型设置playback_tracker用于报告用户实际听到了多少音频远程/电话场景call_id挂接到已有 Realtime 通话本仓库打包的示例是 SIP。本仓库内置的call_id示例是 SIP。更广义的 Realtime API 也会在某些服务端控制流程中使用call_id但这里没有打包成 Python 示例。连接 Azure OpenAI连接 Azure OpenAI 时传入 GA Realtime 端点 URL 与显式请求头例如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}, } )基于令牌的认证则使用 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...。快速上手的完整最小示例结合 docs/realtime/quickstart.md 与上文所有配置一个完整的服务端 Realtime 会话如下import asyncio from agents.realtime import RealtimeAgent, RealtimeRunner async def main() - None: agent RealtimeAgent( nameAssistant, instructionsYou are a helpful voice assistant. Keep responses short and conversational., ) 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}, }, } }, ) session await runner.run() async with session: await session.send_message(Say hello in one short sentence.) async for event in session: if event.type audio: # 转发或播放 event.audio.data pass elif event.type history_added: print(event.item) elif event.type agent_end: # 一个助手回合结束 break elif event.type error: print(fError: {event.error}) if __name__ __main__: asyncio.run(main())API 密钥既可以通过环境变量OPENAI_API_KEY提供也可以在启动会话时直接传入session await runner.run(model_config{api_key: your-api-key})进一步阅读Realtime transport 指南在服务端 WebSocket 与 SIP 之间做选择Realtime 快速入门最简可用路径人机协同Human in the loop审批流程的完整说明examples/realtime仓库内置的 Web 演示、CLI 演示、Twilio 与 Twilio SIP 电话接入示例。需要注意的是Python SDK不提供浏览器 WebRTC 传输本仓库的 Realtime 能力全部面向服务端 WebSocket含 SIP 电话接入适合服务端编排、工具调用、审批与电话集成的场景。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考