openai-agents-python 实时智能体快速入门：基于 WebSocket 的服务端低延迟语音会话实战指南

openai-agents-python 实时智能体快速入门：基于 WebSocket 的服务端低延迟语音会话实战指南

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python

导读

本指南面向 openai-agents-python（OpenAI Agents SDK）开发者，讲解如何在 Python 服务端构建基于 OpenAI Realtime API 的低延迟实时语音智能体。你将掌握RealtimeAgent、RealtimeRunner、RealtimeSession三个核心组件的完整用法，学会配置嵌套式audio.input/audio.output会话参数、驱动事件循环消费音频与历史记录，并了解 API 密钥、自定义 WebSocket 端点、SIP 电话接入等连接选项。读完本文，你可以独立跑通一个由 Python 管理的服务端实时语音会话，并在此基础上扩展工具调用、审批与守卫（guardrails）等能力。

适用范围：Python SDK 的服务端边界

在开始之前，需要明确 Python SDK 的能力边界：它不提供浏览器 WebRTC 传输。本文（以及 实时传输指南）只覆盖两类由 Python 管理的实时路径：

• 服务端 WebSocket：RealtimeRunner默认使用的标准 Python 路径，适合服务端编排、工具执行、审批流程与电话集成；
• SIP / 电话接入：通过call_id挂接到已有实时通话的电信路径。

浏览器端 WebRTC 属于另一平台话题，不在本 SDK 范围内。若你的前端是 WebRTC 客户端，需要自行处理客户端侧流程与事件模型。

前提条件

• Python 3.10 或更高版本；
• OpenAI API 密钥；
• 基本熟悉 OpenAI Agents SDK（了解Agent、Runner等基础概念即可）。

安装 SDK

如果尚未安装，使用 pip 安装：

pip install openai-agents

安装后，实时相关模块位于agents.realtime包内，源码对应仓库目录 src/agents/realtime，其公开导出包含RealtimeAgent、RealtimeRunner、RealtimeSession以及底层模型与事件类型。

创建服务端实时会话：四步走

实时会话的搭建分为四步：导入组件、定义起始智能体、配置运行器、启动会话并发送输入。下面逐一展开。

1. 导入实时组件

import asyncio from agents.realtime import RealtimeAgent, RealtimeRunner

2. 定义起始智能体

RealtimeAgent是面向实时语音场景的专用智能体类型，比普通Agent更收敛：

agent = RealtimeAgent( name="Assistant", instructions="You are a helpful voice assistant. Keep responses short and conversational.", )

从 agent.py 的源码可以看出它的约束：model、modelSettings、outputType、toolUseBehavior均不受支持（所有RealtimeAgent在同一个RealtimeSession内由同一个模型处理，且实时智能体不支持结构化输出）；voice可以在智能体级配置，但一旦会话中已有智能体发声，就不能再更改。instructions可以是字符串，也可以是返回字符串的可调用函数（支持异步），用于动态生成系统提示词。

3. 配置运行器

RealtimeRunner是实时场景下的Runner等价物。从 runner.py 看，它会为整个运行维护与底层模型的持久连接，并自动处理多轮对话：会话负责本地历史副本、工具执行、守卫运行以及智能体间交接（handoff）。若未显式传入model，默认使用OpenAIRealtimeWebSocketModel，即服务端 WebSocket 实现。

对新代码，建议采用嵌套的audio.input/audio.output会话设置结构；对新实时智能体，模型从gpt-realtime-2.1开始：

runner = RealtimeRunner( starting_agent=agent, 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.py）

• model_name：实时模型名。源码中RealtimeModelName列出的可选值包括gpt-realtime、gpt-realtime-1.5、gpt-realtime-2、gpt-realtime-2.1、gpt-realtime-2.1-mini、gpt-realtime-mini系列以及gpt-4o-realtime-preview系列等，同时允许传入任意字符串以适配新模型。
• audio.input.format/audio.output.format：音频格式，RealtimeAudioFormat支持pcm16、g711_ulaw、g711_alaw等。PCM16 是通用的原始音频格式，适合自建播放管线；G.711 常用于电话场景。
• audio.input.transcription：输入音频转录配置。model可选gpt-transcribe、gpt-live-transcribe、gpt-4o-transcribe、gpt-4o-mini-transcribe、gpt-realtime-whisper、whisper-1等；还支持language、prompt、keywords、languages、delay等可选字段。其中delay仅在gpt-realtime-whisper下受当前 SDK 固定的 OpenAI 客户端版本支持，取值为minimal/low/medium/high/xhigh。
• audio.input.noise_reduction：输入降噪，type可选near_field（近场）或far_field（远场）。
• audio.input.turn_detection：自动轮次检测，type可选semantic_vad（语义 VAD）或server_vad（服务器 VAD）；interrupt_response控制是否允许打断助手的当前回复；此外还支持create_response、eagerness（auto/low/medium/high）、prefix_padding_ms、silence_duration_ms、threshold、idle_timeout_ms、model_version等。置为None可完全禁用自动轮次检测。
• audio.output.voice：输出音色，RealtimeVoice可以是字符串、自定义音色对象（{"id": ...}）或映射；示例中使用的ash为内置音色之一。
• audio.output.speed：回复语速（浮点数）。

旧的扁平别名仍然可用

input_audio_format、output_audio_format、input_audio_transcription、input_audio_noise_reduction、turn_detection等扁平字段在RealtimeSessionModelSettings中依旧存在并继续生效，但新代码推荐使用嵌套的audio结构，二者不可混用时以嵌套结构为准。

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(f"Error: {event.error}") if __name__ == "__main__": asyncio.run(main())

会话的输入方式

• session.send_message()：接受纯字符串或结构化实时消息（RealtimeUserInputMessage，可携带input_text与input_image内容，是实时对话中传入图片的主要途径）；
• session.send_audio(audio_bytes, commit=False)：发送原始音频块；当服务端轮次检测被禁用时，可用commit=True标记音频轮次的结束边界；
• 低层控制：通过session.model.send_event(...)可直接发送input_audio_buffer.commit、response.create、session.update等 Realtime API 客户端事件（详见 实时智能体指南 的手动轮次控制小节）。

事件循环如何工作

RealtimeSession实现异步迭代器（见 session.py 的__aiter__），从内部事件队列持续产出RealtimeSessionEvent。常用事件类型（定义在 events.py）：

• audio、audio_end、audio_interrupted：音频流输出、结束与被中断（被打断时你的播放器应立即停止本地播放）；
• agent_start、agent_end：智能体回合开始与结束（示例中以agent_end作为"一个助手回合完成"的退出条件）；
• tool_start、tool_end、tool_approval_required：工具调用与人工审批事件；
• handoff：智能体交接；
• history_added、history_updated：本地历史变更，通常是对 UI 状态最有用的两类事件，其item/history字段为RealtimeItem对象（用户消息、助手消息、工具调用等）；
• guardrail_tripped：输出守卫触发；
• error：错误事件，event.error携带错误详情；
• raw_model_event：底层模型层原始事件透传，可用于用量统计等高级需求。

本快速入门未包含的内容

• 麦克风采集与扬声器播放代码：本指南只覆盖会话侧逻辑，采集/播放需自行实现。可参考 examples/realtime 中的实时示例：
  • examples/realtime/app：核心演示应用（含浏览器端 audio worklet 采集与播放）；
  • examples/realtime/cli：CLI 演示；
  • examples/realtime/twilio：Twilio Media Streams 电话流示例。
• SIP / 电话接入流程：见 实时传输 与 实时智能体指南的 SIP 小节。

关键设置：基本会话跑通后优先调优的参数

基础会话工作后，大多数人接下来会用到的设置包括：

层级	设置项	作用
模型	model_name	选择实时模型（新项目从gpt-realtime-2.1起步）
音频	audio.input.format/audio.output.format	输入/输出音频编码格式
音频	audio.input.transcription	输入音频转写模型与参数
音频	audio.input.noise_reduction	输入降噪模式（near_field/far_field）
音频	audio.input.turn_detection	自动轮次检测（semantic_vad/server_vad）
音频	audio.output.voice	输出音色
会话	tool_choice、prompt、tracing	工具选择策略、提示词对象、请求追踪
运行	async_tool_calls	函数工具是否异步执行（默认True）
运行	tool_execution.pre_approval_tool_input_guardrails	是否在发出审批事件前先运行工具输入守卫（默认False）
运行	guardrails_settings.debounce_text_length	输出守卫防抖的字符阈值（默认100，累积到 1x/2x/3x 倍数时才触发检查）
运行	tool_error_formatter	返回给模型的工具错误信息格式化回调

完整类型化配置面见 config.py 中的RealtimeRunConfig与RealtimeSessionModelSettings，其中RealtimeRunConfig还包含output_guardrails（输出守卫列表）与tracing_disabled（是否禁用本次运行的追踪）。

手动轮次控制：当你需要完全掌控"何时让模型响应"时，可关闭自动轮次检测，改用底层session.update（将turn_detection置为null）→input_audio_buffer.commit→response.create的低层流程，详见 实时智能体指南。该模式适用于：检测到用户输入后才决定响应、需要在触发响应前对输入进行门控、或需要为带外响应定制提示词等场景。

连接选项

配置 API 密钥

方式一：环境变量

export OPENAI_API_KEY="your-api-key-here"

方式二：启动会话时直接传入

session = await runner.run(model_config={"api_key": "your-api-key"})

api_key也支持传入可调用对象（返回密钥的函数），便于从密钥管理服务动态获取（见 model.py 中RealtimeModelConfig的api_key字段定义）。未设置时，OpenAI 实时模型会回退到OPENAI_API_KEY环境变量。

model_config支持的全部选项

RealtimeModelConfig（定义于 model.py）还支持：

• url：自定义 WebSocket 端点（如 Azure OpenAI 的 GA Realtime 端点）；
• headers：自定义请求头。注意：一旦显式传入headers，SDK 将不再自动注入Authorization头，认证完全由你负责；
• initial_model_settings：连接时使用的初始模型设置；
• call_id：挂接到已有实时通话（而不是新建会话）。传入后，传输层使用call_id查询参数连接而非模型名；本仓库内置的挂接示例是 SIP（通过 Realtime Calls API）；
• playback_tracker：RealtimePlaybackTracker实例，用于向模型报告用户实际听到的音频量。默认实现假设音频即时、按实时速度播放；在电话或远端播放等延迟场景下，应在播放时调用on_play_bytes/on_play_ms上报进度，使中断处理能按真实播放位置截断回复（见 model.py 中RealtimePlaybackTracker的源码）。

连接 Azure OpenAI

连接 Azure OpenAI 时，将model_config["url"]设置为 GA Realtime 端点 URL，并显式传入请求头：

session = await runner.run( model_config={ "url": "wss://<your-resource>.openai.azure.com/openai/v1/realtime?model=<deployment-name>", "headers": {"api-key": "<your-azure-api-key>"}, } )

使用令牌认证时：

session = await runner.run( model_config={ "url": "wss://<your-resource>.openai.azure.com/openai/v1/realtime?model=<deployment-name>", "headers": {"authorization": f"Bearer {token}"}, } )

务必避免与实时智能体一起使用旧版 beta 路径（/openai/realtime?api-version=...），详见 实时智能体指南的低层访问与自定义端点小节。

底层 WebSocket 调优

当需要调优底层服务端 WebSocket 连接时，可显式构造OpenAIRealtimeWebSocketModel并传入transport_config：

from agents.realtime import ( OpenAIRealtimeWebSocketModel, RealtimeAgent, RealtimeRunner, ) agent = RealtimeAgent(name="Assistant") model = OpenAIRealtimeWebSocketModel( transport_config={ "ping_interval": 20.0, "ping_timeout": 60.0, "handshake_timeout": 30.0, "max_size": 8 * 1024 * 1024, } ) runner = RealtimeRunner(starting_agent=agent, model=model)

支持的选项（详见 实时传输指南）：

• ping_interval：客户端保活 ping 间隔（秒），设为None可禁用；
• ping_timeout：断开前等待 pong 的时间（秒），设为None可容忍延迟 pong；
• handshake_timeout：等待初始握手完成的时间（秒）；
• max_size：接收 WebSocket 消息的最大字节数，SDK 默认None（不限制），需要限制单条消息内存占用时显式设置。

注意：这些选项配置的是客户端连接本身，端点、认证、通话挂接与播放设置仍通过RealtimeModelConfig完成。

从快速入门走向生产：下一步

• 选择传输方式：在服务端 WebSocket 与 SIP 之间做选择，阅读 实时传输指南。默认 Python 路径下，RealtimeRunner会使用OpenAIRealtimeWebSocketModel；电信场景下，通过RealtimeRunner(..., model=OpenAIRealtimeSIPModel())并以model_config={"call_id": ...}挂接通话，完整流程见 examples/realtime/twilio_sip/server.py。
• 深入生命周期与高级能力：生命周期、结构化输入、审批、交接（handoff）、守卫与低层控制，阅读 实时智能体指南。函数工具、tool_approval_required+session.approve_tool_call()审批循环、realtime_handoff交接、输出守卫的防抖检查（触发时发guardrail_tripped并中断当前响应）都可以在实时会话中直接使用。
• 参考完整示例：浏览 examples/realtime 下的 app、cli、twilio、twilio_sip 四个示例；其中 examples/realtime/app/server.py 展示了服务端审批循环与结构化input_image消息转发，是理解"Python 服务端编排 + 实时语音"生产形态的最佳起点。

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python