指南

本指南深入介绍了使用 OpenAI Agents SDK 的实时功能构建语音驱动的 AI Agent。

Beta 功能

实时 Agent 处于 Beta 阶段。随着我们改进实现，预计会出现一些破坏性更改。

概述

实时 Agent 允许进行对话流程，实时处理音频和文本输入，并以实时音频进行响应。它们与 OpenAI 的实时 API 保持持久连接，从而实现低延迟的自然语音对话，并能够优雅地处理中断。

架构

核心组件

实时系统由几个关键组件组成

RealtimeAgent：配置了指令、工具和交接的 Agent。
RealtimeRunner：管理配置。你可以调用 runner.run() 来获取会话。
RealtimeSession：单个交互会话。通常，每次用户开始对话时创建一个会话，并在对话结束之前保持其活跃状态。
RealtimeModel：底层的模型接口（通常是 OpenAI 的 WebSocket 实现）

会话流程

典型的实时会话遵循以下流程

创建你的 RealtimeAgent(s)，配置指令、工具和交接。
设置 RealtimeRunner，配置 Agent 和配置选项
启动会话，使用 await runner.run()，它会返回一个 RealtimeSession。
发送音频或文本消息到会话，使用 send_audio() 或 send_message()
监听事件，通过迭代会话 - 事件包括音频输出、转录、工具调用、交接和错误
处理中断，当用户打断 Agent 时，自动停止当前的音频生成

会话维护对话历史记录，并管理与实时模型的持久连接。

Agent 配置

RealtimeAgent 的工作方式与常规 Agent 类类似，但有一些关键区别。有关完整的 API 详细信息，请参阅 RealtimeAgent API 参考。

与常规 Agent 的主要区别

模型选择在会话级别配置，而不是 Agent 级别。
不支持结构化输出 (outputType 不受支持)。
语音可以为每个 Agent 配置，但 Agent 第一次说话后无法更改。
其他所有功能，如工具、交接和指令，都以相同的方式工作。

会话配置

模型设置

会话配置允许你控制底层的实时模型行为。你可以配置模型名称（例如 gpt-realtime）、语音选择（alloy、echo、fable、onyx、nova、shimmer）以及支持的模态（文本和/或音频）。可以为输入和输出设置音频格式，默认值为 PCM16。

音频配置

音频设置控制会话处理语音输入和输出的方式。你可以使用 Whisper 等模型配置输入音频转录，设置语言偏好，并提供转录提示以提高特定领域术语的准确性。语音检测设置控制 Agent 何时应该开始和停止响应，提供语音活动检测阈值、静音持续时间和检测到的语音周围的填充选项。

工具和函数

添加工具

与常规 Agent 一样，实时 Agent 支持在对话期间执行的函数工具

from agents import function_tool

@function_tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    # Your weather API logic here
    return f"The weather in {city} is sunny, 72°F"

@function_tool
def book_appointment(date: str, time: str, service: str) -> str:
    """Book an appointment."""
    # Your booking logic here
    return f"Appointment booked for {service} on {date} at {time}"

agent = RealtimeAgent(
    name="Assistant",
    instructions="You can help with weather and appointments.",
    tools=[get_weather, book_appointment],
)

交接

创建交接

交接允许在专门的 Agent 之间转移对话。

from agents.realtime import realtime_handoff

# Specialized agents
billing_agent = RealtimeAgent(
    name="Billing Support",
    instructions="You specialize in billing and payment issues.",
)

technical_agent = RealtimeAgent(
    name="Technical Support",
    instructions="You handle technical troubleshooting.",
)

# Main agent with handoffs
main_agent = RealtimeAgent(
    name="Customer Service",
    instructions="You are the main customer service agent. Hand off to specialists when needed.",
    handoffs=[
        realtime_handoff(billing_agent, tool_description="Transfer to billing support"),
        realtime_handoff(technical_agent, tool_description="Transfer to technical support"),
    ]
)

事件处理

会话会流式传输你可以通过迭代会话对象来监听的事件。事件包括音频输出块、转录结果、工具执行开始和结束、Agent 交接和错误。需要处理的关键事件包括

audio：来自 Agent 响应的原始音频数据
audio_end：Agent 停止说话
audio_interrupted：用户打断了 Agent
tool_start/tool_end：工具执行生命周期
handoff：Agent 交接发生
error：处理过程中发生错误

有关完整的事件详细信息，请参阅 RealtimeSessionEvent。

Guardrails

仅支持实时 Agent 的输出防护栏。这些防护栏会进行节流并定期运行（而不是每个单词运行一次），以避免实时生成期间的性能问题。默认的节流长度为 100 个字符，但可以进行配置。

防护栏可以直接附加到 RealtimeAgent，也可以通过会话的 run_config 提供。来自两个来源的防护栏将一起运行。

from agents.guardrail import GuardrailFunctionOutput, OutputGuardrail

def sensitive_data_check(context, agent, output):
    return GuardrailFunctionOutput(
        tripwire_triggered="password" in output,
        output_info=None,
    )

agent = RealtimeAgent(
    name="Assistant",
    instructions="...",
    output_guardrails=[OutputGuardrail(guardrail_function=sensitive_data_check)],
)

当防护栏被触发时，它会生成一个 guardrail_tripped 事件，并可以中断 Agent 当前的响应。节流行为有助于平衡安全性和实时性能要求。与文本 Agent 不同，实时 Agent 在防护栏被触发时**不会**引发异常。

音频处理

使用 session.send_audio(audio_bytes) 将音频发送到会话，或使用 session.send_message() 发送文本。

对于音频输出，监听 audio 事件，并通过你喜欢的音频库播放音频数据。请务必监听 audio_interrupted 事件，以便立即停止播放并在用户打断 Agent 时清除任何排队的音频。

SIP 集成

你可以将实时 Agent 附加到通过 Realtime Calls API 传入的电话呼叫。SDK 提供了 OpenAIRealtimeSIPModel，它重用了相同的 Agent 流程，同时通过 SIP 协商媒体。

要使用它，请将模型实例传递给 runner，并在启动会话时提供 SIP call_id。呼叫 ID 由指示来电的 webhook 传递。

from agents.realtime import RealtimeAgent, RealtimeRunner
from agents.realtime.openai_realtime import OpenAIRealtimeSIPModel

runner = RealtimeRunner(
    starting_agent=agent,
    model=OpenAIRealtimeSIPModel(),
)

async with await runner.run(
    model_config={
        "call_id": call_id_from_webhook,
        "initial_model_settings": {
            "turn_detection": {"type": "semantic_vad", "interrupt_response": True},
        },
    },
) as session:
    async for event in session:
        ...

当呼叫者挂断电话时，SIP 会话结束，实时连接自动关闭。有关完整的电话示例，请参阅 examples/realtime/twilio_sip。

直接模型访问

你可以访问底层的模型以添加自定义监听器或执行高级操作

# Add a custom listener to the model
session.model.add_listener(my_custom_listener)

这为你提供了对 RealtimeModel 接口的直接访问权限，用于需要对连接进行更低级别控制的高级用例。

示例

有关完整的示例，请查看 examples/realtime 目录，其中包含带有和不带有 UI 组件的演示。