OpenAI Agents SDK #9：让 Agent「边跑边说」——Streaming 流式输出全解析

一句话说清楚 Streaming 是什么

run_streamed() 接口

Runner.run_streamed(
    starting_agent: Agent[TContext],
    input: str | list[TResponseInputItem] | RunState[TContext],
    context: TContext | None = None,
    max_turns: int = DEFAULT_MAX_TURNS,
    hooks: RunHooks[TContext] | None = None,
    run_config: RunConfig | None = None,
    previous_response_id: str | None = None,
    auto_previous_response_id: bool = False,
    conversation_id: str | None = None,
    session: Session | None = None,
    *,
    error_handlers: RunErrorHandlers[TContext] | None = None,
) -> RunResultStreaming

• stream_events() —— 异步迭代器，消费事件流的核心入口
• interruptions —— 用于 Human-in-the-Loop 工具审批的中断列表
• cancel() / cancel(mode="after_turn") —— 中止任务（立即 vs 当前轮结束后）
• to_state() —— 序列化当前状态，用于持久化或恢复
• is_complete —— 布尔属性，表示流是否已结束

StreamEvent 三种类型全解析

完整代码示例：从 token 级到语义级

示例一：实现打字机效果（token 粒度）5

import asyncio
from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner

agent = Agent(
    name="chat_agent",
    instructions="你是一个友好的助手，用简洁的中文回答问题。",
)

async def main():
    # run_streamed() 立即返回，不等 Agent Loop 完成
    result = await Runner.run_streamed(agent, input="用三句话介绍一下量子计算。")

print("=== 开始流式输出 ===")
    async for event in result.stream_events():
        # 只关心原始 token 事件
        if event.type == "raw_response_event" and isinstance(
            event.data, ResponseTextDeltaEvent
        ):
            # 逐 token 打印，end="" 不换行，flush=True 立即刷新缓冲区
            print(event.data.delta, end="", flush=True)

print("\n=== 流式完成 ===")

asyncio.run(main())

示例二：追踪工具调用与 Agent 切换（语义粒度）6

import asyncio
from agents import Agent, Runner
from agents.items import ToolCallItem, ToolCallOutputItem, MessageOutputItem

# 假设已定义一个带工具调用的 agent
agent = Agent(
    name="research_agent",
    instructions="你是一个研究助手，使用工具查找信息后综合回答。",
    tools=[search_web],  # 自定义工具
)

async def main():
    result = await Runner.run_streamed(agent, input="搜索 OpenAI 最新动态")

current_agent_name = agent.name
    print(f"▶ 开始（Agent: {current_agent_name}）")

async for event in result.stream_events():
        # 跳过 token 级噪音，只看语义事件
        if event.type == "raw_response_event":
            continue

# Agent 切换事件 —— 更新当前 Agent 名称
        if event.type == "agent_updated_stream_event":
            current_agent_name = event.new_agent.name
            print(f"\n🔀 Agent 切换 → {current_agent_name}")
            continue

# 语义事件处理
        if event.type == "run_item_stream_event":
            item = event.item

if isinstance(item, ToolCallItem):
                # 工具被触发：打印工具名和参数
                print(f"\n🔧 工具调用: {item.raw_item.name}")
                print(f"   参数: {item.raw_item.arguments}")

elif isinstance(item, ToolCallOutputItem):
                # 工具执行完毕：打印返回结果（截断避免刷屏）
                output_preview = str(item.output)[:100]
                print(f"   ✓ 结果: {output_preview}...")

elif isinstance(item, MessageOutputItem):
                # 最终消息输出完成
                print(f"\n💬 消息输出完成")

print("\n◼ 完成")

asyncio.run(main())

处理工具调用事件的细节

流式 Guardrail 的特殊行为

两种实际应用场景

• 收到 tool_called → 在 UI 显示「正在搜索...」
• 收到 tool_output → 更新为「搜索完成，正在分析...」
• 收到 agent_updated_stream_event → 显示「当前 Agent：Summarizer」
• 收到 message_output_created → 显示「任务完成」

3 条落地建议

try:
    async for event in result.stream_events():
        # 处理事件
        pass
    # 流正常结束后，访问最终结果
    final_output = result.final_output
except OutputGuardrailTripwireTriggered as e:
    # 输出检查失败
    handle_guardrail_failure(e)
except MaxTurnsExceeded:
    # 超过最大轮次
    handle_timeout()

下一篇预告