OpenAI Agents SDK #16：Runner 跑完之后，你真的知道手里拿着什么吗？

print(result.final_output)

一、三层类继承结构

• input：本次运行的原始输入（str 或 list[TResponseInputItem]）
• new_items：运行过程中产生的所有事件列表（list[RunItem]）
• raw_responses：模型返回的原始响应列表（list[ModelResponse]）
• final_output：最终输出，类型为 Any
• input_guardrail_results、output_guardrail_results：输入/输出级 Guardrail 结果
• tool_input_guardrail_results、tool_output_guardrail_results：工具调用的 Guardrail 结果
• context_wrapper：运行上下文封装（RunContextWrapper[Any]）

二、final_output 的三种状态

# raise_if_incorrect_type=True 时非目标类型会抛 TypeError
output = result.final_output_as(MyOutputModel, raise_if_incorrect_type=True)

1. 运行被中断（HITL）：Agent 执行中触发了需要人工审批的操作，运行暂停，此时还没产生最终输出
2. 流式运行未消费完：RunResultStreaming 的 final_output 在流结束前始终为 None
3. 运行被强制取消：调用了 cancel(mode='immediate') 后 Agent 没能完成最后一轮

result = await runner.run(agent, input="请帮我分析这份数据")

if result.final_output is None:
    # 检查是否有中断项待审批
    if result.interruptions:
        print(f"运行暂停，等待审批: {result.interruptions}")
    else:
        print("运行未完成")
else:
    print(result.final_output)

三、new_items：13 种 RunItem 子类型

from agents.items import (
    MessageOutputItem,
    ToolCallItem,
    ToolCallOutputItem,
    HandoffCallItem,
)

# 提取 Agent 所有文本输出
messages = [
    item for item in result.new_items
    if isinstance(item, MessageOutputItem)
]

# 提取所有工具调用（便于日志审计）
tool_calls = [
    (item.tool_name, item.tool_input)
    for item in result.new_items
    if isinstance(item, ToolCallItem)
]

# 检查是否发生了 Handoff
handoffs = [
    item for item in result.new_items
    if isinstance(item, HandoffCallItem)
]
if handoffs:
    print(f"发生了 {len(handoffs)} 次 Agent 切换")

四、to_input_list() 的正确用法

• preserve_all（默认）：将 new_items 转换为完整的 input-item 历史，原样保留所有事件
• normalized：在 handoff 过滤或重写模型历史时使用规范化接续输入；对于没有发生 handoff 的普通运行，行为与 preserve_all 一致

# ✅ 正确：用 to_input_list() 构建下一轮的 input
input_messages = result.to_input_list()
input_messages.append({"role": "user", "content": "继续分析第二个维度"})
result2 = await runner.run(agent, input=input_messages)

# ❌ 错误：直接把 result 传给下一轮
result2 = await runner.run(agent, input=result)  # 这会报类型错误

# ❌ 错误：只传 final_output，丢失了工具调用历史
result2 = await runner.run(agent, input=result.final_output)

五、last_agent 与多 Agent 场景

result = await runner.run(triage_agent, input=user_message)

# 根据 last_agent 决定下一步
if result.last_agent.name == "billing_agent":
    # 账单问题，继续追问细节
    followup = "你的账单问题已初步处理，需要开具发票吗？"
elif result.last_agent.name == "tech_support_agent":
    followup = "技术问题处理中，请描述报错信息"
else:
    followup = "您的问题已收到，稍后跟进"

next_input = result.to_input_list() + [{"role": "user", "content": followup}]

六、Guardrail 结果读取

# 检查是否有 Guardrail 被触发
for gr in result.input_guardrail_results:
    if gr.output.tripwire_triggered:
        print(f"输入 Guardrail 触发: {gr.guardrail.__class__.__name__}")
        print(f"  原因: {gr.output.output_info}")

# 工具级别 Guardrail 审计
for gr in result.tool_output_guardrail_results:
    print(f"工具: {gr.guardrail.__class__.__name__}, "
          f"触发: {gr.output.tripwire_triggered}")

七、RunResultStreaming 专项

• current_agent：当前正在运行的 Agent（动态更新）
• current_turn：当前轮次编号
• is_complete：流是否已终止（bool）
• run_loop_exception：后台异常容器，用于消费流后检查静默失败
• stream_events()：异步迭代器，消费语义化流事件
• cancel(mode)：取消流运行

final_output、interruptions、raw_responses 的赋值，以及 Session 持久化等副作用，可能在最后一个可见 token 到达之后仍在后台处理中。

result = runner.run_streamed(agent, input="请生成一份分析报告")

# ✅ 正确：消费完整个迭代器
async with result.stream_events() as events:
    async for event in events:
        if event.type == "raw_response_event":
            # 处理流式 token
            pass

# 到这里 is_complete 才可靠为 True
print(result.final_output)  # 此时才有值

# ❌ 错误：提前 break 跳出，final_output 可能仍为 None
async for event in result.stream_events():
    if some_condition:
        break  # 后台任务未完成
print(result.final_output)  # 不可靠

# immediate：立即停止，取消所有后台任务，清空队列
# 代价：对话历史可能不完整
result.cancel(mode="immediate")

# after_turn：完成当前轮次后停止
# 好处：LLM 响应完成、工具调用结束、Session 状态正确保存
result.cancel(mode="after_turn")

# 注意：调用 cancel() 后仍需继续消费 stream_events()
# 才能让取消和清理操作正常完成
async for event in result.stream_events():
    pass  # 继续消费，直到迭代器自然终止

八、三条可落地的实践建议

from agents import RunResult

def extract_output(result: RunResult, expected_type=None):
    if result.final_output is None:
        if hasattr(result, "interruptions") and result.interruptions:
            raise RuntimeError("运行被 HITL 中断，需要人工审批后恢复")
        raise RuntimeError("运行未产生输出（可能被取消或中途出错）")

if expected_type is not None:
        return result.final_output_as(expected_type, raise_if_incorrect_type=True)

return result.final_output

• 非流式（run()）：任务型、批处理、代码驱动的确定性流水线——你只关心最终结果，不需要实时反馈
• 流式（run_streamed()）：面向用户的交互界面——需要逐 token 显示让用户感受响应速度；或者运行时间长、需要展示进度
• 别在服务器端批处理里用流式：流式的持续消费约定会增加代码复杂度，非流式更简洁可靠

小结