OpenAI Agents SDK #2：Runner 到底在跑什么？Agent Loop 全解析

Agent Loop：一个你绕不开的 while True

1. 调用当前 Agent 的 LLM
2. 检查返回结果——
   • 有 final_output（纯文本输出且无工具调用）→ 退出循环，返回结果
   • 有 handoff（要移交给另一个 Agent）→ 切换当前 Agent，重新跑循环
   • 有工具调用 → 执行工具，把结果拼回输入，继续循环
3. 超过 max_turns 上限 → 抛出 MaxTurnsExceeded

三种运行方式，怎么选

from agents import Agent, Runner

agent = Agent(name="Assistant", instructions="You are a helpful assistant")
result = await Runner.run(agent, "Write a haiku about recursion in programming.")
print(result.final_output)

RunConfig：一个 run 的全局控制台

from agents import Agent, Runner, RunConfig
from agents.run import CallModelData, ModelInputData

def drop_old_messages(data: CallModelData[None]) -> ModelInputData:
    # 只保留最近 5 条消息
    trimmed = data.model_data.input[-5:]
    return ModelInputData(input=trimmed, instructions=data.model_data.instructions)

result = Runner.run_sync(
    agent,
    "Explain quines",
    run_config=RunConfig(call_model_input_filter=drop_old_messages),
)

对话管理：4 种策略，一次说清

策略 1：result.to_input_list() — 手动拼接

agent = Agent(name="Assistant", instructions="Reply very concisely.")

result = await Runner.run(agent, "What city is the Golden Gate Bridge in?")
print(result.final_output)  # San Francisco

# 手动把上轮输出 + 新问题拼在一起
new_input = result.to_input_list() + [{"role": "user", "content": "What state is it in?"}]
result = await Runner.run(agent, new_input)
print(result.final_output)  # California

策略 2：session — 自动历史管理

from agents import Agent, Runner, SQLiteSession

agent = Agent(name="Assistant", instructions="Reply very concisely.")
session = SQLiteSession("conversation_123")

# 第一轮
result = await Runner.run(agent, "What city is the Golden Gate Bridge in?", session=session)
# 第二轮 — Agent 自动记住上下文
result = await Runner.run(agent, "What state is it in?", session=session)

策略 3：conversation_id — 服务端命名对话

from openai import AsyncOpenAI

client = AsyncOpenAI()
conversation = await client.conversations.create()
conv_id = conversation.id

while True:
    user_input = input("You: ")
    result = await Runner.run(agent, user_input, conversation_id=conv_id)
    print(f"Assistant: {result.final_output}")

策略 4：previous_response_id — 轻量链式对话

agent = Agent(name="Assistant", instructions="Reply very concisely.")
previous_response_id = None

while True:
    user_input = input("You: ")
    result = await Runner.run(
        agent,
        user_input,
        previous_response_id=previous_response_id,
        auto_previous_response_id=True,
    )
    previous_response_id = result.last_response_id
    print(f"Assistant: {result.final_output}")

流式输出：run_streamed 的双层事件

from openai.types.responses import ResponseTextDeltaEvent

result = Runner.run_streamed(agent, input="Tell me 5 jokes.")
async for event in result.stream_events():
    if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
        print(event.data.delta, end="", flush=True)

from agents import Agent, ItemHelpers, Runner, function_tool

async for event in result.stream_events():
    if event.type == "raw_response_event":
        continue  # 忽略低层 token 事件
    elif event.type == "agent_updated_stream_event":
        print(f"切换到 Agent: {event.new_agent.name}")
    elif event.type == "run_item_stream_event":
        if event.item.type == "tool_call_item":
            print("-- 工具被调用")
        elif event.item.type == "tool_call_output_item":
            print(f"-- 工具输出: {event.item.output}")
        elif event.item.type == "message_output_item":
            print(f"-- 消息: {ItemHelpers.text_message_output(event.item)}")

6 种异常，分清楚才能优雅处理

from agents import Agent, RunErrorHandlerInput, RunErrorHandlerResult, Runner

def on_max_turns(_data: RunErrorHandlerInput[None]) -> RunErrorHandlerResult:
    return RunErrorHandlerResult(
        final_output="任务太复杂了，请把请求拆得更细一些。",
        include_in_history=False,  # 不把这条 fallback 加入对话历史
    )

result = Runner.run_sync(
    agent,
    "分析这份超长报告",
    max_turns=3,
    error_handlers={"max_turns": on_max_turns},
)
print(result.final_output)

三条落地建议