跳转至

openai/openai-agents-python:OpenAI 官方的 Python 多智能体框架

先说人话:这是个什么东西

openai-agents-python 是 OpenAI 官方的 Python 智能体开发框架,帮你搭建"一个或多个 AI 角色互相协作完成任务"的系统。说白了:如果一个任务用一个 AI 角色(一份 System Prompt + 一堆工具)就能搞定,你只需要一个 Agent;如果任务复杂到需要按角色分工(比如客服场景里"先分诊、退款走退款流程、售前走售前话术"),这个框架负责把请求从一个 Agent 交接给另一个 Agent,并把整个对话跑到终止条件为止。

在看多智能体协作之前,先看一个最简单的场景:只有一个 Agent,挂一个工具,没有任何交接逻辑。

import asyncio
from agents import Agent, Runner, function_tool

# 用 @function_tool 装饰一个普通 Python 函数,SDK 会自动读取函数签名和
# docstring,生成模型能理解的工具 JSON Schema
@function_tool
def get_weather(city: str) -> str:
    """Get the current weather for a given city."""
    return f"The weather in {city} is sunny, 25°C."

agent = Agent(
    name="WeatherAgent",
    instructions="You help users check the weather. Use the get_weather tool when asked.",
    tools=[get_weather],
)

async def main():
    result = await Runner.run(agent, "北京今天天气怎么样?")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())

这里发生的事情很直白:Runner.run() 把用户输入和 Agent 的 instructions 一起发给模型;模型判断需要调用 get_weather,SDK 在本地执行这个 Python 函数,把返回值重新塞回历史消息,再让模型基于工具结果生成最终回复。整个过程只有一个 Agent,没有交接、没有路由——这是这套框架里最基础的用法,后面所有内容都是在这个基础上叠加"多个角色怎么分工"。

核心设计思路:任务需要分工时,怎么把控制权从一个 Agent 转交给另一个

但只用一个 Agent 很快会遇到瓶颈:在大模型复杂应用场景中,让单一智能体持有过多的系统指令与工具集,会导致提示词稀释(Prompt Dilution)、注意力失焦以及模型推理出错概率明显上升。还是拿客服举例,如果把"查账""退款""转人工"全塞进同一个 Agent 的 Instructions 和工具列表里,模型经常在意图判断上"选错科室",体验很差。

于是核心问题变成:当一个任务需要不同角色的 AI 分工处理时,怎么把任务从一个 Agent 转交给另一个 Agent,而不是让一个 Agent 硬扛所有职责?OpenAI 官方推出的正式生产级 SDK openai-agents-python 给出的答案是 Handoff(控制权交接)机制。需要说明的是,它的设计思想脱胎于 2024 年 OpenAI 发布的实验性项目 Swarm(该项目已停止维护,仅作教学参考),但两者的 API 完全不同:openai-agents-python 是有官方支持、持续发版的正式 SDK,核心类是 AgentRunner,并以显式的 handoff() 原语替代了 Swarm 里"工具函数返回 Agent 对象"的隐式约定。本文后续内容均以 openai-agents-python 当前正式 API 为准。

下面分两节展开:Handoff 具体是怎么把控制权从一个 Agent 移交给另一个的,以及驱动整个对话往前推进的 Run 循环长什么样。


1. 分工是怎么落地的:Handoff 控制权交接机制

Agents SDK 将智能体协作模型简化为:智能体(Agents)交接(Handoffs)。Handoff 在 SDK 内部被建模为一种特殊的工具(tool),会被追加进当前 Agent 暴露给模型的工具列表中;模型只要"调用"这个工具,Runner 就会把控制权切换到目标 Agent。

1.1 什么是 Handoff 控制权交接?

在传统的 Agent 编排中,如果用户要从"查账"转向"退款",通常需要一个中央路由节点(Router)去判断分配。而在 Agents SDK 中,交接逻辑被抽象为:Agenthandoffs 参数里声明可以移交给哪些下游 Agent,模型通过调用对应的 handoff 工具来触发切换

 ┌────────────────────────────────────────────────────────┐
 │ 1. 处于 triage_agent (分流智能体) 控制周期内            │
 └────────────────────────┬───────────────────────────────┘
                          │ 用户输入: "我想申请退款"
                          v
 ┌────────────────────────────────────────────────────────┐
 │ 2. 模型调用 handoff 工具,选中 refund_agent             │
 └────────────────────────┬───────────────────────────────┘
                          │ Runner 内部切换 current_agent
                          v
 ┌────────────────────────────────────────────────────────┐
 │ 3. Handoff 发生!控制权完全移交                         │
 │    - 卸载旧 Agent 的 Instructions / Tools               │
 │    - 装载 refund_agent 专属系统 Prompt 与工具集         │
 └────────────────────────────────────────────────────────┘

这张图和前面单 Agent 例子的区别只有一步:多了一个"模型调用 handoff 工具、Runner 切换 current_agent"的动作,其他采样、执行、写回历史的流程是一样的。这一步切换具体带来什么好处,落到工程上是这两点:

1.2 Handoff 的工程价值:

  • 注意力域物理裁剪:一旦 Handoff 触发,运行上下文将彻底换载,模型每次前向传导(Forward Pass)只看当前活跃 Agent 的 Instructions 和 Tools,消除了无关 Prompt 的噪声污染,大幅降低了运行成本与幻觉率
  • 无状态会话解耦:Agent 对象之间高度自治,可以平滑替换而不依赖沉重的中央图管理器。

2. 一次对话是怎么跑起来的:Run 循环生命周期

前面提到 Handoff 只是"多了一次切换",那么切换前后,驱动整个对话往前走的循环本身是什么样的?答案落在 Runneragents 包中的核心类,参见仓库 src/agents/run.py)身上。它以类方法 Runner.run() / Runner.run_sync() / Runner.run_streamed() 的形式对外暴露,无需实例化即可直接调用。其单轮会话的执行生命周期如下:

  1. 首轮采样:根据当前活跃 Agent 的系统 Prompt 与消息历史,调用大模型 API。
  2. 意图拦截:如果大模型返回的是普通文本,则直接输出并终止本轮 Loop;如果返回的是 tool_calls(工具调用意图),则进入本地执行管道。
  3. 结果求值:逐一调用本地 Tools。如果调用的是某个 handoff 工具,则立即触发 Handoff,Runner 内部将活跃 Agent 切换为目标 Agent 实例。
  4. 循环迭代:将工具执行结果与 Handoff 变化更新入历史消息,立即自动发起新一轮模型采样,直到没有工具请求或达到 max_turns(防死循环硬限制)为止。

也就是说,无论有没有发生 Handoff,这个循环本身不变——Handoff 只是第 3 步里"恰好调用的工具是一个特殊工具"而已,这正是它被称为"轻量级"的原因:没有额外引入一套单独的路由框架。


3. 完整例子:把分工逻辑接到最开头的单 Agent 例子上

有了前两节的概念,再回头看最开头那个只有一个 Agent 的例子就可以直接扩展:把 WeatherAgent 换成多个各司其职的 Agent,再加一个负责分流的 triage_agent,通过 handoffs 参数声明"谁能交接给谁"。下面是一个更完整的例子,展示了路由分流、handoffs 声明式配置与 @function_tool 工具注册怎么组合成一个闭环:

import asyncio
from agents import Agent, Runner, function_tool, handoff, RunContextWrapper
from pydantic import BaseModel

# 1. 定义本地工具,使用 @function_tool 装饰器自动生成 JSON Schema
@function_tool
def execute_refund(item_id: str) -> str:
    """Execute refund logic for a given item identifier."""
    print(f"[SYSTEM TOOL] Refund executed for item: {item_id}")
    return f"Success: Refund processed for item '{item_id}'."

# 2. 声明 handoff 交接时的结构化输入与回调(可选)
class EscalationData(BaseModel):
    reason: str

async def on_escalation(ctx: RunContextWrapper[None], input_data: EscalationData):
    print(f"[SYSTEM HANDOFF] Escalation triggered: {input_data.reason}")

# 3. 预声明各专属 Agent,绑定各自的指令与工具集
sales_agent = Agent(name="SalesAgent", instructions="Provide product features and pricing info. Be persuasive.")

refund_agent = Agent(
    name="RefundAgent",
    instructions="Handle user refund queries. If item_id is provided, execute the refund.",
    tools=[execute_refund],
)

escalation_agent = Agent(name="EscalationAgent", instructions="Handle escalations that need a human.")

# 4. 分流 Agent 通过 handoffs 参数声明可交接的下游 Agent
#    普通 Agent 可直接放入列表;需要自定义回调/输入契约时用 handoff() 包裹
triage_agent = Agent(
    name="TriageAgent",
    instructions="Determine the user's intent and transfer them to the correct agent.",
    handoffs=[
        sales_agent,
        refund_agent,
        handoff(agent=escalation_agent, on_handoff=on_escalation, input_type=EscalationData),
    ],
)

# 5. 生产级驱动流
async def main():
    print("--- Agents SDK Run Loop Started ---")
    # 6. 通过 Runner.run() 类方法启动 Run Loop,初始入口为 triage_agent
    result = await Runner.run(
        triage_agent,
        "我想申请退款,商品单号是 item_998",
        max_turns=5,  # 设置超级步最大深度防御
    )

    # 7. 查看最终会话状态
    print("\n--- Execution Finished ---")
    print(f"Final response: {result.final_output}")

if __name__ == "__main__":
    asyncio.run(main())

4. 上线之后会遇到的坑:常见故障模式与排查

上面的例子逻辑上很干净,但 Handoff 机制自己也会在生产环境里引入几种新的故障模式,排查起来跟单 Agent 场景不太一样:

故障模式 底层诱因 系统级表现 预防与排查手段
乒乓穿梭死循环 (Ping-pong Routing Loop) 两个 Agent 的 Handoff 条件冲突,导致请求在 A 与 B 之间无限循环投递。 耗尽 Token 额度,API 请求卡死,最终触发 max_turns 限制被迫截断。 1. 严格设置 max_turns 硬限制(生产建议 \(\le 5\))。
2. 优化 Instructions,确保交接边界单向清晰,避免互斥模糊。
上下文断裂丢失 (Context Truncation) 多次 Handoff 后,对话消息历史累积过长,超出当前活跃小模型的 Context Window。 模型开始遗忘最初的 User Question,胡乱生成错误工具参数。 1. 采用滑窗剪枝(Sliding Window)清理无害历史。
2. 在 Handoff 时使用 Meta-summary 精炼前置会话的核心结论,附带传给新 Agent。
上下文对象误用 (Context Misuse) Handoff 发生后,业务代码把不该跨 Agent 共享的状态塞进了通用的 RunContextWrapper 新 Agent 拿到了非预期的历史数据,导致决策执行失常。 1. 明确区分"模型可见的对话历史"与"代码可见的 context 依赖对象"。
2. 对 context 中的字段做最小化设计,只放真正需要跨 Agent 传递的数据。

5. 资深系统架构师面试表达方案

面试提问:在设计多智能体系统时,你是如何解决单模型上下文太重、容易遗忘或者幻觉的问题的?OpenAI Agents SDK 的 Handoff 机制解决了什么工程痛点?

回答模版: 在复杂的多步骤任务里,让一个 Agent 同时背负所有 Prompt 和工具集并不划算——工具越多,模型在每一步要筛选的候选项越多,出错概率也会上升。我们早期一个客服机器人把退款、售前咨询、转人工全塞进一个 Agent 的工具列表里,模型经常在意图判断上"选错科室",体验很差。

后来改用 openai-agents-python 的 Handoff 机制把职责拆开:每个 Agent 只声明自己能交接给哪些下游 Agent,触发交接后 Runner 会切换到新 Agent 的 Instructions 和工具集,旧 Agent 的工具就不再出现在下一轮请求里,模型选错的概率明显降了下来,顺带也省了一部分不必要的 Token。Runner.run() 本身没有额外的持久化开销,每次都是从传入的消息历史出发执行到终止条件,协作靠的是标准化的 handoff 调用,而不是某种隐藏的共享状态,配合 max_turns 兜底和必要时对历史做摘要压缩,长任务里的上下文膨胀基本能控制在可预期范围内。