openai-agents-python 实时智能体快速入门:用 RealtimeRunner 构建服务端 WebSocket 语音会话

openai-agents-python 实时智能体快速入门:用 RealtimeRunner 构建服务端 WebSocket 语音会话 openai-agents-python 实时智能体快速入门用 RealtimeRunner 构建服务端 WebSocket 语音会话【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python导读本文基于 OpenAI Agents SDKPython的实时Realtime模块讲解如何在服务端创建低延迟实时语音智能体会话。实时智能体基于通过 WebSocket 传输的 OpenAI Realtime API 构建由 Python 服务端统一编排音频管线、工具调用、审批流程与历史记录适用于服务端编排、工具、审批和电话SIP集成等场景。读完本文你将掌握RealtimeAgent、RealtimeRunner、RealtimeSession三个核心组件的完整用法包括会话创建、音频/文本输入、事件流处理、关键模型设置与连接选项并了解仓库内对应的源码与可运行示例。适用边界Python SDK不提供浏览器 WebRTC 传输。本文仅介绍通过服务端 WebSocket、由 Python 管理的实时会话若需在服务端 WebSocket 与 SIP 电话接入之间做选择请阅读实时传输。前提条件Python 3.10 或更高版本OpenAI API 密钥基本熟悉 OpenAI Agents SDK安装如果尚未安装请安装 OpenAI Agents SDKpip install openai-agents安装完成后可确认实时模块可被正常导入from agents.realtime import RealtimeAgent, RealtimeRunner核心组件概览在动手编码前先理解实时层的四个核心组件对应 src/agents/realtime 目录RealtimeAgent一个实时专职智能体承载 instructions系统提示词、函数工具、输出安全护栏output guardrails与任务转移handoffs。源码位于 src/agents/realtime/agent.py。RealtimeRunner会话工厂负责把起始智能体与实时传输层transport绑定。源码位于 src/agents/realtime/runner.py。RealtimeSession一个活动会话负责发送输入、接收事件、维护本地历史、执行工具与护栏。源码位于 src/agents/realtime/session.py。RealtimeModel传输层抽象。默认实现是 OpenAI 服务端 WebSocketOpenAIRealtimeWebSocketModel见 src/agents/realtime/openai_realtime.py。与纯文本运行不同runner.run()不会立刻返回最终结果而是返回一个与传输层保持同步的实时会话对象。默认情况下RealtimeRunner使用OpenAIRealtimeWebSocketModel即标准 Python 路径是一条指向 Realtime API 的服务端 WebSocket 连接。服务端实时会话的创建下面按快速入门中的四步走逐步搭建一个可运行的实时会话。1. 实时组件的导入import asyncio from agents.realtime import RealtimeAgent, RealtimeRunner2. 起始智能体的定义RealtimeAgent相比普通Agent有意做了收窄见 src/agents/realtime/agent.py 的类注释不支持模型选择同一会话内的所有RealtimeAgent共享同一个底层模型模型在会话级配置不支持结构化输出outputTypevoice 可在智能体级配置但一旦会话内已有智能体说过话就不能再更改instructions、函数工具、handoffs、hooks、输出护栏等仍然全部可用。agent RealtimeAgent( nameAssistant, instructionsYou are a helpful voice assistant. Keep responses short and conversational., )instructions既可以是字符串也可以是接收(RunContextWrapper, RealtimeAgent)并返回字符串的同步/异步函数用于动态生成系统提示词。3. 运行器的配置对于新代码建议采用嵌套的audio.input/audio.output会话设置结构。对于新的实时智能体请从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定义见 src/agents/realtime/config.pymodel_settings对应RealtimeSessionModelSettings。下面结合源码逐一说明关键字段的取值与含义音频格式audio.input.format/audio.output.formatRealtimeAudioFormat支持pcm16、g711_ulaw、g711_alaw三种字面量也接受 OpenAI 客户端中的RealtimeAudioFormats对象。pcm16是默认的 16 位 PCM 编码适合服务端直接处理g711_ulaw/g711_alaw常用于电话PSTN场景。输入转录audio.input.transcriptionRealtimeInputAudioTranscriptionConfig支持以下字段源码 src/agents/realtime/config.pymodel转录模型可取值包括gpt-transcribe、gpt-live-transcribe、gpt-4o-transcribe、gpt-4o-mini-transcribe、gpt-realtime-whisper、whisper-1等prompt引导转录的提示词keywords音频中可能出现的字面术语列表languages预期的输入语言列表注意gpt-live-transcribe使用复数languages不要与单数language同时发送delay流式转录的延迟/精度取舍取值minimal、low、medium、high、xhigh仅gpt-realtime-whisper支持。轮次检测audio.input.turn_detectionRealtimeTurnDetectionConfig的type支持semantic_vad语义级人声活动检测与server_vad服务端 VAD。常用字段create_response检测到轮次后是否自动创建响应eagerness检测轮次边界的积极程度auto、low、medium、highinterrupt_response是否允许打断智能体的回复快速入门示例中设为True支持边说边打断prefix_padding_ms/silence_duration_msVAD 的前置填充与静音时长毫秒threshold人声活动检测阈值idle_timeout_ms用户静默多久后触发响应。将turn_detection设为None可关闭自动轮次检测此时需要应用自行提交音频轮次并控制响应创建见后文手动响应控制。输出语音audio.output.voiceRealtimeVoice接受字符串或自定义语音对象RealtimeCustomVoice含id字段。快速入门使用ash输出还支持speed语速浮点数等字段。4. 会话的启动与输入的发送runner.run()返回一个RealtimeSession。进入会话上下文时连接将建立__aenter__内部会调用self._model.connect(model_config)建立 WebSocket 连接见 src/agents/realtime/session.py。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())session.send_message()接受纯字符串或结构化实时消息。结构化消息是向实时会话传递图片输入的主要方式——RealtimeUserInputMessage的content是input_text与input_image条目的列表input_image支持image_url与detail字段detail可取auto/low/highfrom agents.realtime import RealtimeUserInputMessage 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)对于原始音频块请使用session.send_audio()await session.send_audio(audio_bytes)如果服务端轮次检测被禁用你需要自行标记轮次边界高层便捷方式是await session.send_audio(audio_bytes, commitTrue)send_audio内部会构造RealtimeModelSendAudio事件发送给底层模型传输层commitTrue时会在发送音频后提交输入音频缓冲。会话事件流详解RealtimeSession作为异步可迭代对象向你暴露高层 SDK 事件完整定义见 src/agents/realtime/events.py。高频事件包括audio、audio_end、audio_interrupted音频输出、输出结束、输出被打断agent_start、agent_end智能体回合开始/结束tool_start、tool_end、tool_approval_required工具调用生命周期与审批请求handoff智能体间任务转移history_added、history_updated本地历史更新对 UI 状态最有价值事件携带RealtimeItem列表/条目guardrail_tripped输出护栏被触发input_audio_timeout_triggered检测到用户静默超时error错误raw_model_event透传的底层模型事件。会话关闭语义源码 src/agents/realtime/session.py当 Realtime API 服务端正常关闭默认 WebSocket 连接时模型传输层会依次发出disconnected连接状态事件与end_of_stream事件RealtimeSession会把两者透传进raw_model_event排空已排队的事件后正常结束异步迭代不抛异常由调用方发起的session.close()不会合成这类服务端断开事件意外的 WebSocket 故障则走会话的异常路径抛出。本快速入门未包含的内容麦克风采集和扬声器播放代码。请参阅 examples/realtime 中的实时功能代码示例——其中 examples/realtime/cli/demo.py 是一个完整的命令行语音对话实现40ms 分块采集、24kHz 采样率、播放回调、打断淡出与回声门控examples/realtime/app/server.py 是一个 FastAPI 服务端 WebSocket 转发示例。SIP / 电话接入流程。请参阅实时传输和实时智能体指南中的 SIP 与电话章节docs/realtime/guide.md仓库在 examples/realtime/twilio_sip 中提供了完整示例。关键设置基本会话正常运行后大多数人接下来会用到以下设置model_name实时模型名称。源码中RealtimeModelName类型src/agents/realtime/config.py列举了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输入/输出音频格式pcm16、g711_ulaw、g711_alawaudio.input.transcription输入音频转录配置audio.input.noise_reduction输入音频降噪near_field/far_field两种模式near_field适合贴近麦克风的场景far_field适合远场拾音用于自动轮次检测的audio.input.turn_detectionsemantic_vad/server_vad及其参数audio.output.voice输出语音tool_choice、prompt、tracing工具选择策略如auto、提示词对象Prompt仅 OpenAI 模型可用、请求追踪配置RealtimeModelTracingConfig支持workflow_name、group_id、metadataasync_tool_calls函数工具调用是否异步执行默认True见 src/agents/realtime/config.py 中RealtimeRunConfigtool_execution.pre_approval_tool_input_guardrails是否在发出审批事件前先运行工具输入护栏审批后执行前仍会再检查一次guardrails_settings.debounce_text_length输出文本/转录增量达到多少字符才运行输出护栏默认 100每次累积达到该阈值的 1x、2x、3x… 倍数时运行tool_error_formatter把工具错误信息格式化后返回给模型的可选回调。较旧的扁平别名例如input_audio_format、output_audio_format、input_audio_transcription和turn_detection仍然可用但对于新代码建议使用嵌套的audio设置。对于手动轮次控制请使用实时智能体指南中手动响应控制章节#manual-response-control介绍的底层session.update/input_audio_buffer.commit/response.create流程。例如通过session.model.send_event()发送原始客户端事件from agents.realtime.model_inputs import RealtimeModelSendRawMessage await session.model.send_event( RealtimeModelSendRawMessage( message{ type: response.create, } ) )这种模式适用于关闭了turn_detection后自行决定模型何时响应、在触发响应前检查/门控用户输入、或为带外响应提供自定义提示词。仓库中 Twilio SIP 示例examples/realtime/twilio_sip/server.py就用原始response.create强制开场问候。有关完整 schema请参阅 RealtimeRunConfig 与 RealtimeSessionModelSettings 的源码定义。连接选项在环境中设置 API 密钥export OPENAI_API_KEYyour-api-key-here或者在启动会话时直接传入session await runner.run(model_config{api_key: your-api-key})model_config对应RealtimeModelConfig定义见 src/agents/realtime/model.py还支持url自定义 WebSocket 端点默认使用 OpenAI 默认 WebSocket URLheaders自定义请求标头例如 Azure 的{api-key: ...}认证头如果显式传入headersSDK 将不会自动注入Authorization标头api_key直接传入 API 密钥或传入一个返回密钥的同步/异步回调函数未设置时默认读取OPENAI_API_KEY环境变量call_id接入现有的实时通话在此代码仓库中文档介绍的接入流程为 SIP通过 Realtime Calls API 将智能体会话挂接到call_idplayback_tracker报告用户实际听到的音频量RealtimePlaybackTracker见 src/agents/realtime/model.py。模型生成音频的速度远快于实时播放速度因此在打断场景中需要知道用户实际听到的位置低延迟本地播放默认假设立即以实时速度播放通常足够但在电话等远程/延迟播放场景应传入自定义 playback tracker通过on_play_bytes/on_play_ms上报播放进度使被打断的响应在实际播放位置截断。底层 WebSocket 调优如需调节连接本身而非会话参数可给OpenAIRealtimeWebSocketModel传transport_configfrom 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禁用、ping_timeout等待 pong 超时秒数None容忍延迟 pong、handshake_timeout握手超时秒数、max_size最大入站消息字节数SDK 默认None表示不限大小。连接 Azure OpenAI 时请将model_config[url]设置为正式发布版 Realtime 端点 URL并显式传入标头使用实时智能体时请避免使用旧版 beta 路径/openai/realtime?api-version...。有关详细信息请参阅实时智能体指南中的底层访问与自定义端点章节。例如基于 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}, } )基于令牌的认证则把 bearer token 放进headerssession await runner.run( model_config{ url: wss://your-resource.openai.azure.com/openai/v1/realtime?modeldeployment-name, headers: {authorization: fBearer {token}}, } )进阶方向从快速入门到生产快速入门之外的常用进阶能力均已在仓库源码与示例中落地函数工具RealtimeAgent支持在实时对话中执行函数工具from agents.decorators import tool装饰的函数可直接挂到tools[...]。工具审批工具可要求人工审批。会话会发出tool_approval_required并暂停工具执行直到你调用session.approve_tool_call(call_id)或session.reject_tool_call(call_id)相关实现见 src/agents/realtime/session.py审批循环示例见 examples/realtime/app/server.py。任务转移handoffsrealtime_handoff(...)可把一个实时会话转交给另一个专职智能体RealtimeAgent直接作为 handoff 使用时会自动包装且不支持普通 handoff 的input_filter。输出护栏RealtimeAgent支持输出护栏检查按去抖后的输出文本/音频转录增量运行触发时发出guardrail_tripped事件并中断当前响应底层逻辑见 src/agents/realtime/session.py 的_record_output_guardrail_delta等实现。用量统计模型返回的每次响应用量会以RealtimeModelUsageEvent形式出现在raw_model_event中同时累加到共享的RunContextWrapper.usage可在agent_end等事件中通过event.info.context.usage读取累计用量。打断处理用户打断时会话发出audio_interrupted并更新历史若使用RealtimePlaybackTracker服务端会话会与用户实际听到的内容对齐。后续步骤阅读实时传输以便在服务端 WebSocket 和 SIP 之间进行选择。阅读实时智能体指南了解生命周期、结构化输入、审批、任务转移、安全防护措施和底层控制。浏览 examples/realtime 中的代码示例核心演示应用examples/realtime/app、命令行示例examples/realtime/cli、Twilio 媒体流示例examples/realtime/twilio与 SIP 接入示例examples/realtime/twilio_sip。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考