OpenAI Agents SDK #19：拆开 RealtimeSession，那 20 种事件你只处理了 3 种

一、RealtimeAgent 和普通 Agent 的 4 个关键区别

• instructions：支持函数形式（动态指令），签名是 Callable[[RunContextWrapper, RealtimeAgent], MaybeAwaitable[str]]，在 Session 运行时被调用并返回字符串
• handoffs：只能转接给其他 RealtimeAgent，不能转接给普通 Agent
• voice：同一 Session 里第一个 Agent 发言后就锁定了，之后再改无效（这是坑，后面详说）

二、Runner 是工厂，Session 是连接

runner = RealtimeRunner(
    starting_agent=agent,
    config={...}     # 模型配置在这里
)

# runner.run() 返回 Session，但此时还没连
session = await runner.run()

# 进入上下文时才建立连接
async with session:
    await session.send_message("Say hello.")
    async for event in session:
        ...

三、Session 的 7 个核心 API

四、事件系统全景：近 20 种 event.type 分类解读

音频类（5 种）

async for event in session:
    if event.type == "audio":
        audio_player.write(event.audio.data)   # 直接转发给播放器
    elif event.type == "audio_interrupted":
        audio_player.flush()                   # 清空播放缓冲区
    elif event.type == "audio_done":
        pass                                   # 本轮完成，可以更新 UI 状态

对话历史类（5 种）

Agent 生命周期类（2 种）

工具类（3 种）

elif event.type == "tool_approval_required":
    # 在这里把 call_id 送去你的审批 UI / 队列
    pending_approvals[event.call_id] = event
    # 用户点「批准」后再调用：
    # await session.approve_tool_call(event.call_id)

Handoff / Guardrail / 底层类（5 种）

五、RealtimeRunConfig 的 6 个字段

class RealtimeRunConfig(TypedDict):
    model_settings:       RealtimeSessionModelSettings   # 模型配置（VAD、音频格式等）
    output_guardrails:    list[OutputGuardrail[Any]]     # 输出守护栏列表
    guardrails_settings:  RealtimeGuardrailsSettings     # 守护栏运行配置
    tracing_disabled:     bool                           # 禁用追踪，默认 False
    async_tool_calls:     bool                           # 工具调用是否异步，默认 True
    tool_error_formatter: ToolErrorFormatter             # 工具错误消息格式化回调

六、VAD 两种模式：语义 vs 能量

turn_detection: {
    "type": "semantic_vad",   # 或 "server_vad"
    "create_response": True,
    "eagerness": "auto",      # "low" / "medium" / "high" / "auto"
    "interrupt_response": True,
    "prefix_padding_ms": 300,
    "silence_duration_ms": 500,
    "threshold": 0.5,
}

• high：发言一停就快速响应，适合短问答场景
• low：允许更长的停顿，适合思考型对话
• auto：SDK 自动调整

七、Transport 层：SDK 只有两条路

# 默认就是 WebSocket，无需额外配置
runner = RealtimeRunner(starting_agent=agent)

session = await runner.run(
    model_config={"call_id": incoming_call_id}
)

八、6 个生产坑

try:
    async for event in session:
        handle_event(event)
except Exception as e:
    # Session 内部异常在这里捕获
    logger.error(f"Realtime session error: {e}")

session = await runner.run()
try:
    await session.enter()
    async for event in session:
        ...
finally:
    await session.close()   # 无论是否异常都要关

九、完整最小可用示例

import asyncio
from agents.realtime import RealtimeAgent, RealtimeRunner

agent = RealtimeAgent(
    name="VoiceBot",
    instructions="Keep responses concise and conversational.",
)

runner = RealtimeRunner(
    starting_agent=agent,
    config={
        "model_settings": {
            "model_name": "gpt-realtime-2",
            "audio": {
                "input": {
                    "format": "pcm16",
                    "transcription": {"model": "gpt-4o-mini-transcribe"},
                    "turn_detection": {
                        "type": "semantic_vad",
                        "interrupt_response": True,
                        "eagerness": "auto",
                    },
                },
                "output": {
                    "format": "pcm16",
                    "voice": "ash",   # 发言后锁定，无法更改
                },
            },
        },
        "async_tool_calls": False,   # 工具有共享状态时用串行
    },
)

async def main() -> None:
    session = await runner.run()
    async with session:
        await session.send_message("Say hello.")
        try:
            async for event in session:
                if event.type == "audio":
                    write_to_speaker(event.audio.data)
                elif event.type == "audio_interrupted":
                    flush_speaker_buffer()
                elif event.type == "transcript_delta":
                    print(event.delta, end="", flush=True)
                elif event.type == "tool_approval_required":
                    # HITL 审批逻辑
                    await session.approve_tool_call(event.call_id)
                elif event.type == "error":
                    print(f"错误: {event.error}")
                elif event.type == "agent_end":
                    break
        except Exception as e:
            print(f"Session 异常: {e}")

asyncio.run(main())

三条实践建议

下期预告 #20：进入 Tracing 模块——Span、Trace、TraceProvider 完整结构，以及如何把 Realtime 和 Voice Pipeline 的追踪数据打通送进自定义监控系统。