OpenAI Agents Python SDK 源码拆解:800 行代码搞定 Agent 循环、工具调用与 Handoff

我没直接回答，反手把 OpenAI 官方那个 openai-agents-python 仓库丢给他。这是 OpenAI 在 2025 年初开源的"轻量级 Agent SDK"，截至现在 GitHub 星数已经过 1.5 万。它的代码量出奇地小——核心 Runner 加上 Agent 主流程不到 800 行，比 LangChain 一个小模块还少。

"轻"是它的卖点，但更值得读的是它的"取舍"——OpenAI 官方下场写 Agent SDK，他们怎么定义"Agent 的最小闭环"？把哪些抽象砍了，把哪些坚守住了？

这一篇就把 Runner.run() 的核心循环、工具调用机制、Handoff 多 Agent 协作这三个最关键的部分，一行行扒一遍。代码全部来自 openai/openai-agents-python 主分支，做了少量精简但语义不变。

一、整体架构：四个核心抽象

先看一眼骨架。SDK 的核心目录结构很扁平：

src/agents/
├── agent.py           # Agent 配置类（dataclass）
├── run.py             # Runner，全局执行器（核心 ~600 行）
├── _run_impl.py       # 单轮执行细节，~700 行
├── tool.py            # @function_tool 装饰器
├── handoffs.py        # 多 Agent 切换
├── guardrail.py       # 输入/输出守卫
├── result.py          # RunResult 数据结构
└── models/            # 模型适配层
    ├── openai_responses.py
    ├── openai_chatcompletions.py
    └── interface.py

扒下来就四个核心抽象，每个都对应一个清晰的工程意图：

最关键的设计取舍是：Agent 是只读配置，Runner 是无状态执行器。这一点跟 LangChain 那种"Agent 同时持有运行状态"的设计完全不同。

这种分离带来一个直接好处——同一个 Agent 实例可以在多个并发请求里安全复用，不用担心状态污染。代码里 Agent 类直接用 @dataclass 修饰，连方法都几乎没有，纯数据载体。

二、Agent 循环：一个 while True 拆解

所有 Agent 框架里，最值钱的其实就是一段循环。这段循环回答两个问题：什么时候继续调模型？什么时候停下来返回结果？

OpenAI Agents SDK 把这段循环放在 Runner._run_impl() 里，去掉异常处理、追踪、超时控制后，骨架是这样：

async def _run_impl(
    starting_agent: Agent,
    input: str | list[TResponseInputItem],
    max_turns: int = 10,
    context: TContext | None = None,
) -> RunResult:

    current_agent = starting_agent
    generated_items: list[RunItem] = []
    input_items = ItemHelpers.input_to_items(input)
    current_turn = 0

    while True:
        current_turn += 1
        if current_turn > max_turns:
            raise MaxTurnsExceeded(...)

        # 1. 跑一轮：包括"调模型 + 解析输出"
        turn_result = await _run_single_turn(
            agent=current_agent,
            all_input=input_items + generated_items,
            context=context,
        )
        generated_items.extend(turn_result.new_items)

        # 2. 决策：根据本轮结果决定下一步
        next_step = turn_result.next_step

        if isinstance(next_step, NextStepFinalOutput):
            return RunResult(
                final_output=next_step.output,
                new_items=generated_items,
                last_agent=current_agent,
            )

        elif isinstance(next_step, NextStepHandoff):
            current_agent = next_step.new_agent
            # 不 break，下一轮 while 继续

        elif isinstance(next_step, NextStepRunAgain):
            # 工具已经调用完，结果已经入 generated_items
            # 让模型再看一眼
            continue

这段循环的精妙之处全在 next_step 的三态设计上：

• NextStepFinalOutput——模型给出了纯文本回答（没调工具、没切 Agent），结束。
• NextStepHandoff——模型调了一个 handoff 工具，把控制权移交给另一个 Agent，current_agent 换人，循环继续。
• NextStepRunAgain——模型调了普通工具，工具结果已经被附加到 input 里，模型需要再看一眼这些结果。

为什么这个三态设计很关键？

我自己写过几版手撸的 Agent，最早就是一个布尔标志 should_continue，后来发现处理多 Agent 切换的时候很别扭——你得在循环外面塞额外的"current agent"变量，还要保证它不被新一轮覆盖。

SDK 这种"把控制流编码成数据"的方式，让循环本身保持极简：循环只看 next_step 的类型来分发，状态变更放在 step 对象里。这是经典的"State as Data"设计。

另一个细节：max_turns 默认值是 10。这个数字看起来朴素，但反映了一个朴素的工程原则——必须给死循环设兜底。LLM 是有几率自己绕进去的，没有 max_turns 兜底，线上随时可能因为模型一次抽风而把 token 烧穿。

三、单轮执行：模型调用 + 工具分发

展开 _run_single_turn()，里面做的事情其实可以再拆成两步：把上下文打包发给模型，把模型返回的每一项分类处理。

async def _run_single_turn(agent, all_input, context):
    # 把 Agent 的工具/handoff 都序列化成 OpenAI 格式
    tools_schema = [t.to_openai_tool() for t in agent.tools]
    handoff_tools = [_handoff_to_tool(h) for h in agent.handoffs]

    # 调模型
    model_response = await agent.model.get_response(
        system_instructions=agent.instructions,
        input=all_input,
        tools=tools_schema + handoff_tools,
        output_schema=agent.output_type,
    )

    new_items: list[RunItem] = []
    for output_item in model_response.output:

        if isinstance(output_item, ResponseFunctionToolCall):
            # 工具调用，可能是普通工具或 handoff
            tool_name = output_item.name
            if tool_name in {h.tool_name for h in agent.handoffs}:
                return SingleTurnResult(
                    new_items=new_items,
                    next_step=NextStepHandoff(_resolve_handoff(...)),
                )

            tool = _find_tool(agent, tool_name)
            tool_result = await tool.invoke(output_item.arguments, context)
            new_items.append(ToolCallItem(...))
            new_items.append(ToolCallOutputItem(tool_result))

        elif isinstance(output_item, ResponseOutputMessage):
            # 纯文本输出
            new_items.append(MessageOutputItem(output_item))

    # 都处理完没遇到 handoff，看模型是否还想继续
    has_tool_calls = any(isinstance(i, ToolCallItem) for i in new_items)

    if has_tool_calls:
        return SingleTurnResult(
            new_items=new_items,
            next_step=NextStepRunAgain(),
        )
    else:
        final_output = ItemHelpers.text_message_outputs(new_items)
        return SingleTurnResult(
            new_items=new_items,
            next_step=NextStepFinalOutput(final_output),
        )

注意几个关键设计：

① Handoff 复用工具调用通道

Handoff 在序列化时被转换成一个名字像 transfer_to_billing_agent 的工具，跟普通工具混在同一份 tools 列表里发给模型。

这个设计的好处是——不用改模型的接口协议，直接复用 function calling。模型不需要知道"什么是 Agent 切换"，它只看到一个特殊命名的工具，调用它即可。SDK 在收到这个调用后，做了一次"截胡"：在执行工具之前判断这是个 handoff，把 next_step 设成 NextStepHandoff，控制权立刻切到下一个 Agent。

代价是：handoff 的描述需要写在 tool description 里，让模型理解"什么时候该转交"。我看了一下 SDK 的默认实现，handoff 的 description 长这样："Handoff to the {agent_name} agent to handle the request. The reason for handing off is..."

② 工具结果立刻入上下文，下轮可见

工具调用完，结果通过 ToolCallOutputItem 立刻 append 到 new_items，回到 _run_impl 的循环顶部时再拼到 input_items 上发给模型。

这里有个坑要注意——SDK 是把所有 ToolCallItem + ToolCallOutputItem 完整堆进 input 的。如果工具返回大段文本（比如爬一个网页 50KB），多调几次 context 就爆了。生产环境用这个 SDK 一定要给工具结果加截断。

四、@function_tool 装饰器：怎么把 Python 函数变工具

这是 SDK 里我最喜欢的部分。一个装饰器，把任意 Python 函数变成 OpenAI 兼容的工具，连参数 schema 都自动生成。看下精简后的实现：

def function_tool(
    func: Callable | None = None,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
):
    def _decorate(f: Callable):
        # 1. 用 inspect 拿到签名
        sig = inspect.signature(f)
        type_hints = typing.get_type_hints(f)

        # 2. 用 pydantic 动态构造一个参数模型
        fields = {}
        for param_name, param in sig.parameters.items():
            if param_name == "context":
                continue  # context 由 SDK 注入，不算参数
            type_ = type_hints.get(param_name, str)
            default = param.default if param.default is not inspect.Parameter.empty else ...
            fields[param_name] = (type_, default)

        ParamModel = pydantic.create_model(
            f"{f.__name__}_Args",
            **fields,
        )

        # 3. JSON schema 直接从 pydantic 拿
        schema = ParamModel.model_json_schema()

        # 4. 包一层 invoke：解析参数 → 调原函数
        async def invoke(json_args: str, context: TContext) -> str:
            args_dict = json.loads(json_args)
            validated = ParamModel(**args_dict)
            kwargs = validated.model_dump()
            if "context" in sig.parameters:
                kwargs["context"] = context
            result = f(**kwargs)
            if inspect.iscoroutine(result):
                result = await result
            return json.dumps(result) if not isinstance(result, str) else result

        return FunctionTool(
            name=name_override or f.__name__,
            description=description_override or f.__doc__ or "",
            params_json_schema=schema,
            on_invoke=invoke,
        )

    return _decorate(func) if func is not None else _decorate

真正在做活的就三步：

第一步是用 inspect.signature 反射拿到原函数的签名和类型注解。第二步是用 pydantic.create_model 动态创建一个参数 schema。第三步是包一层 invoke，做 JSON 解析、参数校验、可选 context 注入。

用起来是这样：

@function_tool
def get_weather(city: str, unit: str = "celsius") -> str:
    """查询某城市当前天气。"""
    return f"{city} 当前 23°{unit[0].upper()}"

# 直接传给 Agent
agent = Agent(
    name="weather_bot",
    tools=[get_weather],
)

和 LangChain 那种要手写 args_schema 加 BaseModel 比，写法清爽不少。但代价是——对类型注解的依赖很硬。如果你有个参数没写类型，schema 直接退化成 str，模型就只能瞎猜。我建议团队里用这个 SDK 的，强制配上 mypy 检查。

五、Handoff 实战：客服分流的完整代码

前面拆完原理，看一段真实跑得起来的代码。这是我自己写的一个客服分流 demo，模拟"前台 → 计费/技术支持"的两级分流：

from agents import Agent, Runner, function_tool

@function_tool
def lookup_invoice(invoice_id: str) -> str:
    """根据发票号查询发票状态。"""
    return f"发票 {invoice_id} 状态：已开具，金额 ¥1280。"

@function_tool
def restart_service(service_name: str) -> str:
    """重启一个服务。"""
    return f"{service_name} 已重启完毕。"

billing_agent = Agent(
    name="billing",
    instructions="你是计费专员，处理发票/退款/充值类问题。",
    tools=[lookup_invoice],
)

tech_agent = Agent(
    name="tech_support",
    instructions="你是技术支持，处理服务故障类问题。",
    tools=[restart_service],
)

triage_agent = Agent(
    name="triage",
    instructions=(
        "你是前台客服，判断用户问题类型。"
        "账单/发票相关 → 转交 billing；"
        "服务故障/异常 → 转交 tech_support。"
    ),
    handoffs=[billing_agent, tech_agent],
)

# 跑起来
async def main():
    result = await Runner.run(
        triage_agent,
        input="我的服务昨晚开始 502，能帮我重启一下吗？服务名 order-svc。",
    )
    print(result.final_output)
    print("最后一个 Agent：", result.last_agent.name)

输出会是这样：

order-svc 已重启完毕。如果还有问题，请告诉我具体表现。
最后一个 Agent： tech_support

流程上发生的事情：triage 收到问题 → 调用 transfer_to_tech_support（handoff 工具）→ Runner 截胡，把 current_agent 切到 tech_support → 进入下一轮 → tech_support 调用 restart_service → 拿到结果 → 输出 final_output。

整个过程对外看就是一个 await Runner.run() 调用，背后是两次 LLM 请求 + 一次工具调用。

六、源码读完，几个我自己的看法

① "Agent 框架的本质"被它说清了

读完一遍 OpenAI Agents SDK，再回头看那些动辄上万行代码的 Agent 框架，会觉得它们大部分代码都不在"Agent 这件事"上——而是在做工具集成、记忆管理、检索、追踪面板这些外围工程。

"Agent 的本质"其实就两个东西：一个会用工具的循环 + 一种把工具描述给模型看的方式。这两个东西加起来不到 800 行 Python。剩下的都是封装好不好看、生态全不全的问题。

② "无状态 Runner"是个被低估的设计

把执行器和配置分开，Runner.run(agent, input) 而不是 agent.run(input)，看起来只是一字之差，工程含义完全不同。

线上 web 服务里，每个 Agent 配置加载一次就行，每次请求只 new 一个 RunResult，配置可以放进进程级缓存，甚至跨请求复用。LangChain 那种"agent 同时持运行状态"的写法在并发场景需要小心避免共享。

③ Handoff 借道 function calling 是巧思也是限制

把"切 Agent"建模成"调一个特殊工具"，复用 function calling 协议——这招很聪明，没改模型接口就实现了多 Agent 协作。

但限制是：handoff 的"转交时机"完全交给模型自己判断，写在 tool description 里。如果你想做更精细的控制（比如"必须先收集到 X 信息再转交"），你得自己加 guardrail——SDK 提供了 @input_guardrail，但用起来有点啰嗦。这地方比起 LangGraph 那种"显式状态机"的设计要弱一点。

④ 我会在生产里用它吗？

看场景。简单的"单 Agent + 工具调用"或者"两三层 handoff 的客服分流"，我会直接上这个 SDK，省事。但如果是复杂的状态机（比如订单审批流，固定步骤、固定分支），我宁可用 LangGraph 显式建图，或者干脆自己写——因为复杂业务最怕的就是"模型自己决定下一步"。

读源码最大的收获从来不是学 API，而是看到设计者在哪些地方做了取舍。OpenAI Agents SDK 把"轻"做到了极致，代价是"灵活性下沉到模型"。这是一个非常清晰的工程判断——它假设你信任模型，所以才敢把这么多决策权交给它。

如果你也信任模型，这个 SDK 是当前 Python 生态里最优雅的选择之一。如果你不信任模型，那就得加一堆 guardrail，到那时候，"轻"就变成"裸奔"了。

—— 工程师写给工程师的源码笔记 ——