
OpenAI Agents Python SDK Realtime 快速上手用 RealtimeAgent 构建服务端语音会话【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python本文基于 OpenAI Agents Python SDK 的 Realtime 快速入门讲解如何用RealtimeAgent、RealtimeRunner与RealtimeSession搭建服务端低延迟实时语音 Agent从安装依赖、创建起始 Agent、配置音频会话参数到启动会话、收发消息与处理事件流再到 API Key、自定义端点等连接选项。读完本文你将掌握一套可直接运行的服务端 WebSocket Realtime 会话骨架并理解其底层运行机制。Realtime Agent 是什么Realtime agents 是 Python SDK 中服务端、低延迟的 Agent构建在 OpenAI Realtime API 之上通过 WebSocket 传输。与普通文本 Agent 每次请求重建连接不同Realtime Agent 与 Realtime API 保持一条长连接模型可以增量处理文本与音频、流式输出音频、调用工具、处理打断而无需每一轮都重新发起一次请求。Python SDK 边界说明Python SDK不提供浏览器端 WebRTC 传输。本文只覆盖由 Python 管理的、基于服务端 WebSocket 的 realtime 会话。SDK 适用于服务端编排、工具调用、审批流程与电话SIP集成。传输选型详见 Realtime transport。从源码结构看realtime 层由四个核心组件构成见 src/agents/realtime 目录RealtimeAgentagent.py一个 realtime 专用 Agent携带指令、工具、输出护栏与交接handoff配置RealtimeRunnerrunner.py会话工厂把起始 Agent 与 realtime 传输层连接起来RealtimeSessionsession.py实时会话对象负责发送输入、接收事件、维护本地历史、执行工具RealtimeModelmodel.py传输抽象接口默认实现是 OpenAI 服务端 WebSocketOpenAIRealtimeWebSocketModel。前置条件Python 3.10 或更高版本OpenAI API Key对 OpenAI Agents SDK 有基本了解安装如果你还没有安装 OpenAI Agents SDK先执行pip install openai-agents创建服务端 Realtime 会话下面按快速入门给出的四步流程搭建一个最小可运行的实时语音会话。1. 导入 realtime 组件import asyncio from agents.realtime import RealtimeAgent, RealtimeRunner2. 定义起始 Agentagent RealtimeAgent( nameAssistant, instructionsYou are a helpful voice assistant. Keep responses short and conversational., )RealtimeAgent的instructions可以是静态字符串也可以是接收RunContextWrapper与 Agent 实例、返回字符串的动态函数见 agent.py 中get_system_prompt的实现。需要说明的是RealtimeAgent是刻意收窄过的 Agent 类型模型选择在会话级配置而不是 Agent 级不支持结构化输出voice可以在 Agent 级配置但一旦会话已产生过语音输出就不能再更改。3. 配置 Runner新代码优先使用嵌套的audio.input/audio.output会话设置结构新的 realtime Agent 建议从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, }, }, } }, )这段config就是源码中的RealtimeRunConfig见 config.py其中的model_settings对应RealtimeSessionModelSettings。从 runner.py 的实现看RealtimeRunner.__init__只做三件事保存起始 Agent、保存运行配置、在没有显式传入model时默认创建OpenAIRealtimeWebSocketModel()。4. 启动会话并发送输入runner.run()返回一个RealtimeSession。连接会在进入会话上下文async with时建立async def main() - None: 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: # Forward or play event.audio.data. pass elif event.type history_added: print(event.item) elif event.type agent_end: # One assistant turn finished. break elif event.type error: print(fError: {event.error}) if __name__ __main__: asyncio.run(main())运行机制拆解runner.run()runner.py并不会像普通Runner那样立即返回最终结果而是构造并返回一个RealtimeSession——它内部持有_history本地历史、后台工具执行、护栏状态与当前 Agent 配置并与传输层保持同步async with session会触发RealtimeSession.__aenter__session.py先从 Agent 解析初始模型设置把自己注册为模型监听器然后调用model.connect()建立 WebSocket 连接并发出第一个history_updated事件session.send_message()发送用户输入session.py可以接受纯字符串也可以接受结构化 realtime 消息如RealtimeUserInputMessage支持input_text与input_image内容项是向会话传入图片的主要途径见 config.py 中RealtimeUserInput类型定义async for event in session迭代事件队列。RealtimeSessionEvent的完整类型见 events.py常用事件包括audio模型生成的新音频通过event.audio.data获取字节流、audio_end、audio_interrupted被打断可据此停止本地播放或给出视觉提示、agent_start/agent_end一个助手回合开始/结束、tool_start/tool_end/tool_approval_required、handoff、history_added/history_updated暴露会话本地历史是最常用于驱动 UI 状态的事件、guardrail_tripped、input_audio_timeout_triggered、error、raw_model_event透传原始模型事件需要发送原始音频块时使用session.send_audio()session.py传入原始音频字节在关闭服务端自动轮转检测的场景下可用send_audio(audio_bytes, commitTrue)标记轮转边界。快速入门未涵盖的内容麦克风采集与扬声器播放代码本文示例只演示了会话与事件流采集/播放由你的应用层负责。可参考仓库示例 examples/realtime/app、examples/realtime/cli 与 examples/realtime/twilioSIP / 电话接入流程通过call_id附加到已存在的实时通话本仓库给出的文档化示例是 SIP详见 Realtime transport 与 Realtime agents guide 的 SIP 章节。关键设置基础会话跑通后最常调用的配置一旦基础会话工作正常下一步通常就是调整以下设置完整类型定义见 config.py 中的RealtimeSessionModelSettings与RealtimeRunConfigmodel_namerealtime 模型名。源码中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系列等可选值也允许任意字符串新项目建议从gpt-realtime-2.1开始audio.input.format、audio.output.format音频格式。RealtimeAudioFormat支持pcm16、g711_ulaw、g711_alaw后者常用于电话场景audio.input.transcription输入音频转写配置RealtimeInputAudioTranscriptionConfig可指定model如gpt-4o-mini-transcribe、gpt-live-transcribe、gpt-transcribe、gpt-realtime-whisper等、language/languages、prompt、keywords、delay等。例如低延迟增量转写用gpt-live-transcribe并在prompt中提供录音上下文、用keywords标注可能出现的字面术语、用languages声明预期输入语言注意gpt-live-transcribe使用复数languages不要与单数language同时发送delayminimal/low/medium/high/xhigh目前只支持gpt-realtime-whisper值越低越早产出部分文本值越高给转写模型更多音频上下文、准确率可能更好建议用真实音频基准测试audio.input.noise_reduction降噪配置type可为near_field或far_fieldaudio.input.turn_detection自动轮转检测RealtimeTurnDetectionConfig。type可为semantic_vad或server_vad可配interrupt_response是否允许打断助手回复、eagernessauto/low/medium/high、silence_duration_ms、prefix_padding_ms、threshold、idle_timeout_ms等设置为None则关闭自动检测此时需要应用层自己提交音频轮转并控制响应创建见下文手动响应控制audio.output.voice输出语音ash等会话产生过语音后不可再改tool_choice、prompt、tracing工具选择策略、Prompt 对象仅 OpenAI 模型可用、请求追踪配置RealtimeModelTracingConfig可设workflow_name、group_id、metadataasync_tool_calls默认True是否异步执行函数工具调用、tool_execution.pre_approval_tool_input_guardrails是否在发出待审批事件前先运行函数工具输入护栏、guardrails_settings.debounce_text_length护栏防抖阈值默认 100 字符累计文本每达到该值的 1 倍、2 倍、3 倍……运行一次护栏、tool_error_formatter格式化返回给模型的工具错误消息的回调。关于新旧配置形态旧的扁平别名input_audio_format、output_audio_format、input_audio_transcription、turn_detection仍然可用但新代码推荐使用嵌套的audio结构。从 session.py 中self._debounce_text_length self._run_config.get(guardrails_settings, {}).get(debounce_text_length, 100)可以看到运行配置在会话内部的实际消费方式。手动响应控制需要手动控制轮转时使用低层session.update/input_audio_buffer.commit/response.create流程在 Realtime API 层面即发送一个把turn_detection置为null的session.update然后自己发送input_audio_buffer.commit与response.create。在 SDK 中可通过session.model.send_event(...)发送原始客户端事件如RealtimeModelSendRawMessage详见 Realtime agents guide 的手动响应控制章节。完整配置 Schema 参见 RealtimeRunConfig 与 RealtimeSessionModelSettings。连接选项通过环境变量设置 API Keyexport OPENAI_API_KEYyour-api-key-here启动会话时直接传入session await runner.run(model_config{api_key: your-api-key})从源码看openai_realtime.py 的connect实现约 L588-L638get_api_key会优先使用显式传入的api_key也支持传回调函数未设置时回退到OPENAI_API_KEY环境变量并把Authorization: Bearer key头注入 WebSocket 握手请求。model_config还支持类型为RealtimeModelConfig见 model.pyurl自定义 WebSocket 端点headers自定义请求头。注意如果你显式传了headersSDK 将不再替你注入Authorization头如 Azure OpenAI 场景需要自己传{api-key: ...}或{authorization: Bearer token}call_id附加到已存在的 realtime 通话通过call_id查询参数连接而非模型名。在本仓库中文档化的附加流程是 SIP通过 Realtime Calls API 接入完整示例见 examples/realtime/twilio_sipplayback_trackerRealtimePlaybackTracker实例用于上报用户实际听到了多少音频。默认实现假设音频立即以实时速度播放在电话等远端/延迟播放场景下用它让打断后的响应在真实播放位置截断见 model.py 中RealtimePlaybackTracker的on_play_bytes/on_play_ms/on_interrupted方法。连接 Azure OpenAI连接 Azure OpenAI 时把model_config[url]设置为 GA 版 Realtime 端点 URL并显式传入请求头。避免在 realtime Agent 上使用遗留的 beta 路径/openai/realtime?api-version...。示例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}, } )更多细节见 Realtime agents guide 的低层访问与自定义端点章节。底层 WebSocket 调优进阶如果需要调节底层 WebSocket 连接参数可以不使用默认模型而是显式构造OpenAIRealtimeWebSocketModel并传入transport_config详见 transport.mdfrom agents.realtime import OpenAIRealtimeWebSocketModel 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 的间隔秒数设为None可禁用 pingping_timeout等待 pong 的秒数超时断开设为None可容忍延迟 pong 而不触发心跳超时handshake_timeout等待初始连接握手的秒数max_size最大入站 WebSocket 消息字节数。SDK 默认是None不限大小需要限制单条消息内存占用时再显式设置。这些设置配置的是客户端连接本身而不是 Realtime API 会话端点、认证、呼叫附加与播放追踪仍然走RealtimeModelConfig。下一步阅读 Realtime transport在服务端 WebSocket 与 SIP 之间做出传输选型阅读 Realtime agents guide深入了解生命周期、结构化输入、审批、交接handoff、护栏与低层控制浏览仓库示例 examples/realtime包含 app 演示、CLI 与 Twilio / Twilio SIP 电话接入。附一段可直接运行的完整示例把前文四步整合为完整脚本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())运行前请确保已设置OPENAI_API_KEY环境变量并已执行pip install openai-agents。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考