用 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-python（OpenAI 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 安装 SDK：

pip install openai-agents

安装完成后，实时相关组件从agents.realtime包导入，该包对外公开了RealtimeAgent、RealtimeRunner、RealtimeSession、RealtimeModelConfig、RealtimePlaybackTracker以及全部会话事件类型，具体导出清单见 src/agents/realtime/init.py。

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

1. 导入实时组件

import asyncio from agents.realtime import RealtimeAgent, RealtimeRunner

RealtimeAgent是会话中使用的专用智能体类型，RealtimeRunner则是实时场景下的运行器。从源码看，RealtimeRunner是普通Runner在实时场景下的等价物：它通过维持与底层模型层的持久连接自动处理多轮对话，会话内部负责维护本地历史副本、执行工具、运行护栏（guardrails）并在智能体之间完成交接（handoffs），详见 src/agents/realtime/runner.py。

2. 定义开始智能体

agent = RealtimeAgent( name="Assistant", instructions="You 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_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", }, }, } }, )

这段配置的每个字段都有明确的类型约束（对应 src/agents/realtime/config.py 中RealtimeSessionModelSettings的 TypedDict 定义）：

RealtimeRunner的构造函数签名是RealtimeRunner(starting_agent, *, model=None, config=None)：不传model时默认使用OpenAIRealtimeWebSocketModel（即默认走服务端 WebSocket），config作为整次运行的覆盖参数，见 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(f"Error: {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, *, commit=False)，见 src/agents/realtime/session.py）；
• 会话以异步迭代器形式产出事件，退出async with时自动关闭连接（__aexit__调用close()，见 src/agents/realtime/session.py）；
• 与纯文本运行不同，runner.run()不会立刻返回最终结果，而是返回一个保持本地历史、后台工具执行、护栏状态与活动智能体配置实时同步的活跃会话对象。

理解会话事件流：RealtimeSession 与事件类型

RealtimeSession是到实时模型的双向连接：它把模型事件流式转发给你，同时允许你向模型发送消息与音频。在 src/agents/realtime/events.py 中，RealtimeSessionEvent定义了会话对外发出的全部事件类型：

事件循环中每个事件都带有info.context（RunContextWrapper），可读取上下文；当模型返回包含 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_name
• audio.input.format、audio.output.format
• audio.input.transcription
• audio.input.noise_reduction
• audio.input.turn_detection（自动话轮检测）
• audio.output.voice
• output_modalities（["text", "audio"]）
• tool_choice、prompt、tracing
• max_output_tokens（1 到 4096 的整数，或"inf"，服务端默认"inf"）

运行级（config 顶层）常用项（见 src/agents/realtime/config.py 的RealtimeRunConfig）：

• async_tool_calls：函数工具是否异步执行，默认True
• output_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_KEY="your-api-key-here"

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

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

model_config对应RealtimeModelConfig（见 src/agents/realtime/model.py），完整支持以下字段：

连接 Azure OpenAI

当连接 Azure OpenAI 时：将model_config["url"]设置为GA 版 Realtime 端点 URL（形如wss://<your-resource>.openai.azure.com/openai/v1/realtime?model=<deployment-name>），并显式传入头（{"api-key": "<your-azure-api-key>"}或 bearer token 形式{"authorization": f"Bearer {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/）、CLI（cli/）、Twilio Media Streams（twilio/）与 Twilio SIP（twilio_sip/）四套可直接运行的参考实现。

附：核心源码索引

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