ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

用 OpenAI Agents SDK 构建服务端实时语音智能体:RealtimeAgent 快速入门实战指南

用 OpenAI Agents SDK 构建服务端实时语音智能体:RealtimeAgent 快速入门实战指南 用 OpenAI Agents SDK 构建服务端实时语音智能体RealtimeAgent 快速入门实战指南【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python本文是基于开源项目openai-agents-pythonOpenAI Agents SDK for Python编写的实时Realtime智能体快速入门指南。SDK 中的实时智能体是一类运行在服务端、基于 WebSocket 传输的低延迟智能体它建立在 OpenAI Realtime API 之上能够一边接收文本与音频输入、一边增量生成语音输出并支持工具调用、人工审批与电话SIP集成。读完本文你将掌握从安装依赖、定义RealtimeAgent、配置RealtimeRunner到启动RealtimeSession并消费事件流的完整四步流程同时理解嵌套式audio会话配置、连接选项与源码级实现细节能够立刻在自有服务中搭建第一个可运行的实时语音会话。Python SDK 的实时能力边界先看清适用场景在开始之前需要明确 Python SDK 在实时领域的能力边界。该 SDK不提供面向浏览器的 WebRTC 传输——浏览器端 WebRTC 属于独立的平台主题。本文只覆盖通过服务端 WebSocket 由 Python 管理的实时会话SDK 的定位是承担服务端编排orchestration、工具调用、人工审批与电话telephony集成等职责。这意味着典型拓扑是Python 服务创建RealtimeRunner→await runner.run()返回RealtimeSession→ 以异步上下文管理器进入会话 → 发送文本/结构化消息或音频 → 消费RealtimeSessionEvent事件并转发音频或转录文本到你的应用。这一拓扑正是仓库中核心演示应用、CLI 示例与 Twilio Media Streams 示例所采用的方式见 examples/realtime。前提条件与安装开始前请确保满足以下条件Python 3.10 及以上版本OpenAI API 密钥对 OpenAI Agents SDK 有基本了解了解Agent、工具与运行器的基本概念即可若尚未安装使用 pip 安装 SDKpip install openai-agents安装完成后实时相关组件从agents.realtime包导入该包对外公开了RealtimeAgent、RealtimeRunner、RealtimeSession、RealtimeModelConfig、RealtimePlaybackTracker以及全部会话事件类型具体导出清单见 src/agents/realtime/init.py。创建服务端实时会话四步走1. 导入实时组件import asyncio from agents.realtime import RealtimeAgent, RealtimeRunnerRealtimeAgent是会话中使用的专用智能体类型RealtimeRunner则是实时场景下的运行器。从源码看RealtimeRunner是普通Runner在实时场景下的等价物它通过维持与底层模型层的持久连接自动处理多轮对话会话内部负责维护本地历史副本、执行工具、运行护栏guardrails并在智能体之间完成交接handoffs详见 src/agents/realtime/runner.py。2. 定义开始智能体agent RealtimeAgent( nameAssistant, instructionsYou are a helpful voice assistant. Keep responses short and conversational., )RealtimeAgent是一个专门用于RealtimeSession构建语音智能体的类。与普通Agent相比它刻意收窄了部分能力model选择在会话层面统一配置、不支持结构化输出outputType、voice可以在智能体级配置但在会话产生第一段语音后便无法更改而instructions、函数工具、交接、hooks 与输出护栏等均正常支持详见 src/agents/realtime/agent.py。其中instructions既可以是字符串也可以是一个接收(RunContextWrapper, RealtimeAgent)并返回字符串支持 async的动态生成函数——get_system_prompt()会先调用再按需await实现动态系统提示词见 src/agents/realtime/agent.py。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, }, }, } }, )这段配置的每个字段都有明确的类型约束对应 src/agents/realtime/config.py 中RealtimeSessionModelSettings的 TypedDict 定义配置项含义取值范围 / 说明model_name使用的实时模型源码中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系列等见 src/agents/realtime/config.py新代码建议从gpt-realtime-2.1开始audio.input.format输入音频编码pcm16、g711_ulaw、g711_alaw见 src/agents/realtime/config.pyaudio.input.transcription输入音频转录配置model可选gpt-transcribe、gpt-live-transcribe、gpt-4o-transcribe、gpt-4o-mini-transcribe、gpt-realtime-whisper、whisper-1等还可配language/languages、prompt、keywords、delay见 src/agents/realtime/config.pyaudio.input.turn_detection自动话轮检测type取semantic_vad语义 VAD或server_vad服务端 VAD支持create_response、eagernessauto/low/medium/high、interrupt_response是否允许打断助手回复、prefix_padding_ms、silence_duration_ms、threshold、idle_timeout_ms等见 src/agents/realtime/config.pyaudio.input.noise_reduction输入降噪type取near_field或far_field见 src/agents/realtime/config.pyaudio.output.format输出音频编码与输入相同pcm16、g711_ulaw、g711_alawaudio.output.voice输出音色字符串音色名如ash或带id的自定义音色RealtimeCustomVoice见 src/agents/realtime/config.pyaudio.output.speed输出语速浮点数RealtimeRunner的构造函数签名是RealtimeRunner(starting_agent, *, modelNone, configNone)不传model时默认使用OpenAIRealtimeWebSocketModel即默认走服务端 WebSocketconfig作为整次运行的覆盖参数见 src/agents/realtime/runner.py。4. 启动会话并发送输入runner.run()返回一个RealtimeSession。进入会话上下文时连接才会真正建立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{type: message, role: user, content: [{type: input_text, ...}, {type: input_image, ...}]}见 src/agents/realtime/config.py结构化消息是实时会话中携带图片输入的主要方式发送原始音频块请使用session.send_audio()send_audio(audio, *, commitFalse)见 src/agents/realtime/session.py会话以异步迭代器形式产出事件退出async with时自动关闭连接__aexit__调用close()见 src/agents/realtime/session.py与纯文本运行不同runner.run()不会立刻返回最终结果而是返回一个保持本地历史、后台工具执行、护栏状态与活动智能体配置实时同步的活跃会话对象。理解会话事件流RealtimeSession 与事件类型RealtimeSession是到实时模型的双向连接它把模型事件流式转发给你同时允许你向模型发送消息与音频。在 src/agents/realtime/events.py 中RealtimeSessionEvent定义了会话对外发出的全部事件类型事件类型触发时机agent_start/agent_end某个智能体开始 / 结束一轮handoff智能体交接给另一个智能体tool_start/tool_end工具调用开始 / 结束tool_approval_required工具调用需要人工审批audio/audio_end/audio_interrupted新音频生成 / 音频生成结束 / 音频被中断history_added/history_updated本地历史新增条目 / 历史整体更新对 UI 状态最有用的两类事件guardrail_tripped输出护栏被触发并中断了智能体input_audio_timeout_triggered模型检测到用户一段静默/无活动error发生错误raw_model_event转发底层模型的原始事件需要细粒度控制时使用事件循环中每个事件都带有info.contextRunContextWrapper可读取上下文当模型返回包含 usage 的完整响应时SDK 还会把 token 用量累加到共享的RunContextWrapper.usage上你可以在agent_end等后续事件中通过event.info.context.usage读取会话累计用量源码位于 src/agents/realtime/session.py 的 usage 分支。此外RealtimeSession还提供了若干主动控制方法interrupt()中断模型、update_agent()切换当前活动智能体并应用其设置、approve_tool_call(call_id)审批挂起的工具调用全部实现在 src/agents/realtime/session.py。关键设置一览嵌套结构 vs 传统别名基本会话跑通后下一步最常接触的设置在RealtimeRunner(config{model_settings: {...}})的model_settings与运行级config两个层面会话级model_settings常用项model_nameaudio.input.format、audio.output.formataudio.input.transcriptionaudio.input.noise_reductionaudio.input.turn_detection自动话轮检测audio.output.voiceoutput_modalities[text, audio]tool_choice、prompt、tracingmax_output_tokens1 到 4096 的整数或inf服务端默认inf运行级config 顶层常用项见 src/agents/realtime/config.py 的RealtimeRunConfigasync_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本次运行是否关闭追踪值得强调的是input_audio_format、output_audio_format、input_audio_transcription、turn_detection这类扁平的传统别名仍然可用兼容旧代码但新代码一律推荐使用嵌套的audio配置。完整类型定义可查阅RealtimeRunConfig与RealtimeSessionModelSettings均在 src/agents/realtime/config.py 中。如果你需要手动控制话轮例如关闭自动话轮检测后自行决定响应时机可参考 实时智能体指南 中描述的底层session.update/input_audio_buffer.commit/response.create流程——对应到 SDK 中是通过session.model.send_event(RealtimeModelSendRawMessage(message{...}))直接向模型传输发送原始客户端事件。连接选项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完整支持以下字段字段说明api_keyAPI 密钥或返回密钥的函数/回调未设置时模型会使用合理默认值例如 OpenAI Realtime 模型读取OPENAI_API_KEY环境变量url自定义 WebSocket 端点未设置时使用 OpenAI 默认 WebSocket URLheaders自定义请求头。注意一旦显式传入headersSDK 不会再自动注入Authorization头需要你自行提供鉴权信息initial_model_settings连接时使用的初始模型设置与会话级model_settings合并initial_model_settings优先级更高见 src/agents/realtime/session.pycall_id附加到已存在的实时通话而非新建会话。本仓库中记录的附加流程是 SIP传输层通过call_id查询字符串参数连接而非模型名见 src/agents/realtime/model.pyplayback_tracker报告用户实际听到的音频量RealtimePlaybackTracker实例。默认实现假定音频被立即以实时速度播放在电话等远端播放场景中传入 tracker 可让模型在中断时按真实播放位置截断响应见 src/agents/realtime/model.py连接 Azure OpenAI当连接 Azure OpenAI 时将model_config[url]设置为GA 版 Realtime 端点 URL形如wss://your-resource.openai.azure.com/openai/v1/realtime?modeldeployment-name并显式传入头{api-key: your-azure-api-key}或 bearer token 形式{authorization: fBearer {token}}。使用实时智能体时避免旧的 beta 路径/openai/realtime?api-version...。详细说明见 实时智能体指南。本快速入门未涵盖的内容麦克风采集与扬声器播放代码本文只演示了服务端会话逻辑。完整的客户端音频输入输出实现含 24kHz 采样、40ms 分块、能量阈值抢话、播放追踪器与淡出处理等细节可参考仓库 examples/realtime/cli/demo.py 以及 examples/realtime 下的其他示例SIP / 电话连接流程如何通过call_id附加到电话通话参见 实时传输说明 与 实时智能体指南的 SIP 章节仓库中的完整实现位于 examples/realtime/twilio_sip使用OpenAIRealtimeSIPModel。下一步阅读 实时传输说明在服务端 WebSocket与SIP两种传输方式之间做出选择阅读 实时智能体指南深入了解生命周期、结构化输入、审批、交接、护栏与低层控制浏览 examples/realtime 下的示例代码包括演示应用app/、CLIcli/、Twilio Media Streamstwilio/与 Twilio SIPtwilio_sip/四套可直接运行的参考实现。附核心源码索引关注点源码位置实时智能体类型RealtimeAgentsrc/agents/realtime/agent.py会话/模型/护栏全部配置 TypedDictsrc/agents/realtime/config.py运行器RealtimeRunner与会话工厂src/agents/realtime/runner.py会话生命周期与工具/审批/护栏处理src/agents/realtime/session.py会话事件类型定义src/agents/realtime/events.py传输抽象与连接配置RealtimeModelConfigsrc/agents/realtime/model.py公开导出清单src/agents/realtime/init.py【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表