
用 OpenAI Agents SDK 搭建多智能体系统:agent-as-a-tool 与 handoff 两种委派模式
用 OpenAI Agents SDK 搭建多智能体系统:agent-as-a-tool 与 handoff 两种委派模式
当单个智能体难以同时处理调研、评估与决策时,可以把职责拆给多个专职智能体。本教程用 OpenAI Agents SDK 实现两种基础委派模式:把专家智能体包装成工具(agent-as-a-tool),以及把控制权移交给专家(handoff),并以电动车购车顾问与充电支持两个场景给出可运行的完整代码。
单个大模型智能体可以独立完成不少任务,但当应用逐渐变大,它往往需要同时应付调研、评估、决策、答疑等性质完全不同的工作。这时把职责拆开、交给多个专职智能体,再让它们以某种方式协作,通常比把所有指令塞进一个提示词更清晰、也更容易维护。本教程围绕 OpenAI Agents SDK 提供的两种基础委派机制展开:agent-as-a-tool(把专家智能体当作工具调用)与 handoff(把控制权移交给专家智能体)。掌握这两块积木之后,更复杂的多智能体拓扑都可以由它们组合出来。
准备工作
在动手写代码之前,先把运行环境与依赖确认清楚。下面的清单对应示例代码实际用到的部分。
- Python 运行环境:示例使用
async/await语法,需要在支持异步的 Python 版本下运行,并在异步函数中执行。 - OpenAI Agents SDK:提供
Agent、Runner、ModelSettings、OpenAIResponsesModel、WebSearchTool等对象。 - OpenAI Python 客户端:示例通过
AsyncAzureOpenAI构造客户端,说明它走的是 Azure 风格的端点配置;如果你使用其他接入方式,替换客户端构造部分即可,后续智能体代码不变。 - dotenv:用于从
.env文件读取环境变量。 - 三个环境变量:
OPENAI_API_KEY、OPENAI_API_BASE、OPENAI_API_VERSION。示例直接以os.environ[...]读取,缺任何一个都会在启动时报错。 - 联网搜索能力:三个研究型智能体都挂载了
WebSearchTool,需要所用模型与接入方式支持该工具。
把依赖装好、把 .env 填好之后,先写一段共享的客户端与模型配置,后面所有智能体都复用它。
import os
from agents import ModelSettings, OpenAIResponsesModel
from dotenv import load_dotenv
from openai import AsyncAzureOpenAI
load_dotenv()
client = AsyncAzureOpenAI(
api_key=os.environ["OPENAI_API_KEY"],
azure_endpoint=os.environ["OPENAI_API_BASE"],
api_version=os.environ["OPENAI_API_VERSION"],
)
model = OpenAIResponsesModel(
model="gpt-5.4",
openai_client=client,
)
model_settings = ModelSettings(
reasoning={"effort": "medium"},
)
这里把模型名与推理强度集中在一处,是为了让所有智能体行为一致;具体可用的模型名称、推理强度取值以及各接入方式的配额,请以所用平台官网的当前信息为准。
操作步骤
第一步:理解两种委派方式的差别
多智能体系统的第一步永远是拆分职责,第二步是让拆开的智能体重新协作,而协作的核心就是「委派」。实践中常见两种委派方式,它们的区别不在调用形式,而在谁对最终答案负责。
agent-as-a-tool:主智能体接到请求后,发现自己需要某位专家的帮助,于是像调用普通工具一样调用该专家智能体。专家只处理一个有明确边界的子任务,完成后把结果交回给调用方。调用方始终保持活跃,并负责把专家结果与其他信息合并成最终答复。从调用方视角看,这位专家就是一个工具。由于可以挂多个专家工具,主智能体能够根据任务自行决定咨询哪几位专家。
handoff:初始智能体接到请求后,判断该由哪位专家接手,然后把控制权整体转移过去。专家成为当前活跃智能体,最终答复由专家自己产出。初始智能体在这里更像一个路由器,只负责把请求送到正确的专家手上。
一句话概括:agent-as-a-tool 是「专家交回结果」,handoff 是「专家接管任务」。选哪种,取决于最终答案需要被汇总,还是本来就属于某一位专家的领域。
第二步:用 agent-as-a-tool 搭建购车顾问
第一个场景:一个三口之家住在苏黎世,每年行驶约 15,000 公里,预算 45,000 瑞士法郎,租住公寓、无法安装私人充电桩,需要为儿童座椅和行李留出空间。他们想知道电动车是否适合自己,如果适合,请推荐至多三款车型并说明取舍。
这个问题天然可以拆成几条相互独立的研究线:找合适车型、评估充电可行性、弄清购置与持有成本。三条线的结论只有合并成一份建议才有意义,因此 agent-as-a-tool 是自然的选择:保留一个购车顾问智能体掌握全局,让三位专家分别调研,最后由顾问汇总。
先定义三位专家,每位都挂载联网搜索工具:
from agents import Agent, WebSearchTool
vehicle_research_agent = Agent(
name="Vehicle researcher",
instructions=(
"Research current electric vehicles that fit the user's "
"needs and constraints."
),
model=model,
model_settings=model_settings,
tools=[WebSearchTool(search_context_size="medium")],
)
charging_research_agent = Agent(
name="Charging researcher",
instructions=(
"Assess electric-vehicle charging feasibility using current "
"infrastructure and pricing information."
),
model=model,
model_settings=model_settings,
tools=[WebSearchTool(search_context_size="medium")],
)
cost_research_agent = Agent(
name="Ownership-cost researcher",
instructions=(
"Research current purchase and ownership costs relevant to "
"an electric-vehicle decision."
),
model=model,
model_settings=model_settings,
tools=[WebSearchTool(search_context_size="medium")],
)
关键一步是把这三位专家暴露成工具。调用 as_tool() 时,tool_name 与 tool_description 是主智能体判断「什么时候该找谁」的唯一依据,所以要写得具体;max_turns 限制专家在单次调用内最多进行几轮推理与工具调用,防止某个子任务无限展开。
vehicle_research_tool = vehicle_research_agent.as_tool(
tool_name="research_ev_models",
tool_description="Research current EV models that fit the user's needs.",
max_turns=5,
)
charging_research_tool = charging_research_agent.as_tool(
tool_name="research_charging",
tool_description="Research charging access and current prices.",
max_turns=5,
)
cost_research_tool = cost_research_agent.as_tool(
tool_name="research_ownership_costs",
tool_description="Research purchase and ownership costs.",
max_turns=5,
)
然后定义负责协调的顾问智能体,把三个工具挂上去:
ev_purchase_adviser = Agent(
name="EV purchase adviser",
instructions=(
"Help the user decide whether an electric vehicle suits their "
"needs. Specialist tools are available for vehicle options, "
"charging, and ownership costs."
),
model=model,
model_settings=model_settings,
tools=[
vehicle_research_tool,
charging_research_tool,
cost_research_tool,
],
)
注意顾问的 instructions 里点明了有哪些专家工具可用,这能帮助模型在合适的时机发起调用,而不是自己硬答。
第三步:运行顾问并检查调用轨迹
用一段贴近真实用户口吻的请求来测试:
purchase_request = """
We are a family of three living in Zurich.
We drive about 15,000 km per year, mostly for commuting and weekend trips,
plus several long drives across Europe.
We rent an apartment and cannot install a private charger.
Our budget is CHF 45,000, and we need room for a child seat and luggage.
Would an electric vehicle make sense for us?
If so, recommend up to three current models and explain the trade-offs.
"""
通过 Runner.run 执行,max_turns 控制整个运行过程的最大轮数:
from agents import Runner
purchase_result = await Runner.run(
ev_purchase_adviser,
purchase_request,
max_turns=10,
)
print(purchase_result.final_output)
运行结果中,顾问会给出「电动车在行驶需求上可行,但缺少私人充电是主要制约」这类判断,并比较若干车型、说明取舍。真正值得关注的是它怎么得出这个答案。可以从返回结果里提取工具调用记录与最终负责的智能体:
agent_tool_calls = [
getattr(item.raw_item, "name", None)
for item in purchase_result.new_items
if item.type == "tool_call_item"
]
print("Agent tools called:", agent_tool_calls)
print("Final agent:", purchase_result.last_agent.name)
轨迹会显示顾问依次调用了车型、充电、持有成本三位研究员,而 last_agent 仍然是购车顾问本身。这正是 agent-as-a-tool 的特征:专家贡献了各自的调研结果,管理者保留了对最终答复的责任。
第四步:用 handoff 搭建充电支持智能体
第二个场景换了个方向:用户已经开上电动车,完全依赖苏黎世的公共充电网络,面对不同运营商、订阅方案和价格感到困惑,希望得到一个可落地的充电方案。这类请求主要落在单一专家领域内,不需要把多份报告拼在一起,因此 handoff 更合适。
先定义三位可能的接手专家。与前面不同,这里每位专家都设置了 handoff_description,它告诉路由智能体「什么问题该转给你」:
charging_support_agent = Agent(
name="Charging specialist",
handoff_description=(
"Handles questions about home and public EV charging."
),
instructions=(
"Help electric-vehicle owners solve charging questions using "
"current information."
),
model=model,
model_settings=model_settings,
tools=[WebSearchTool(search_context_size="medium")],
)
route_planning_agent = Agent(
name="Route planning specialist",
handoff_description=(
"Handles questions about long-distance EV route planning."
),
instructions=(
"Help electric-vehicle owners plan journeys using current "
"route and charging information."
),
model=model,
model_settings=model_settings,
tools=[WebSearchTool(search_context_size="medium")],
)
vehicle_care_agent = Agent(
name="Vehicle care specialist",
handoff_description=(
"Handles questions about EV maintenance and daily operation."
),
instructions=(
"Help electric-vehicle owners with maintenance and operating "
"questions using current information."
),
model=model,
model_settings=model_settings,
tools=[WebSearchTool(search_context_size="medium")],
)
再用 handoffs=[...] 显式列出支持智能体可以移交的对象:
ev_support_agent = Agent(
name="EV support agent",
instructions=(
"Help users with electric-vehicle ownership questions. "
"Transfer the conversation to the relevant specialist "
"when appropriate."
),
model=model,
model_settings=model_settings,
handoffs=[
charging_support_agent,
route_planning_agent,
vehicle_care_agent,
],
)
这个支持智能体不需要合并多份调研报告,它的职责只是识别出该由哪位专家对请求负责。因此它的 instructions 强调「在合适的时候转交」,而不是「自己回答」。
第五步:运行支持智能体并检查移交路径
用一段充电相关的请求测试:
support_request = """
I recently bought an electric car and rely entirely on public charging in Zurich.
The different networks, subscriptions, and prices are confusing.
Help me work out a practical charging setup.
"""
support_result = await Runner.run(
ev_support_agent,
support_request,
max_turns=10,
)
print(support_result.final_output)
运行后,请求会被移交给充电专家,由它检索当前网络与价格信息并产出最终答复。要确认移交确实发生,可以遍历结果中的移交记录:
handoff_path = [
f"{item.source_agent.name} -> {item.target_agent.name}"
for item in support_result.new_items
if item.type == "handoff_output_item"
]
print("Handoff:", handoff_path)
print("Final agent:", support_result.last_agent.name)
输出形如:
Handoff: EV support agent -> Charging specialist
Final agent: Charging specialist
与第一个场景对比:agent-as-a-tool 的 last_agent 是协调者,handoff 的 last_agent 是接手专家。这个字段是判断委派模式是否按预期生效的最直接信号。
一个完整示例
下面把两个场景串成一个可运行的最小脚本。它复用同一套客户端与模型配置,先跑购车顾问,再跑充电支持,并打印各自的轨迹,便于对照两种模式。
import os
import asyncio
from dotenv import load_dotenv
from openai import AsyncAzureOpenAI
from agents import (
Agent,
ModelSettings,
OpenAIResponsesModel,
Runner,
WebSearchTool,
)
load_dotenv()
client = AsyncAzureOpenAI(
api_key=os.environ["OPENAI_API_KEY"],
azure_endpoint=os.environ["OPENAI_API_BASE"],
api_version=os.environ["OPENAI_API_VERSION"],
)
model = OpenAIResponsesModel(model="gpt-5.4", openai_client=client)
model_settings = ModelSettings(reasoning={"effort": "medium"})
def make_researcher(name, instructions):
return Agent(
name=name,
instructions=instructions,
model=model,
model_settings=model_settings,
tools=[WebSearchTool(search_context_size="medium")],
)
async def main():
# ---- 场景一:agent-as-a-tool ----
vehicle = make_researcher(
"Vehicle researcher",
"Research current electric vehicles that fit the user's needs and constraints.",
)
charging = make_researcher(
"Charging researcher",
"Assess electric-vehicle charging feasibility using current infrastructure and pricing information.",
)
cost = make_researcher(
"Ownership-cost researcher",
"Research current purchase and ownership costs relevant to an electric-vehicle decision.",
)
adviser = Agent(
name="EV purchase adviser",
instructions=(
"Help the user decide whether an electric vehicle suits their needs. "
"Specialist tools are available for vehicle options, charging, and ownership costs."
),
model=model,
model_settings=model_settings,
tools=[
vehicle.as_tool(
tool_name="research_ev_models",
tool_description="Research current EV models that fit the user's needs.",
max_turns=5,
),
charging.as_tool(
tool_name="research_charging",
tool_description="Research charging access and current prices.",
max_turns=5,
),
cost.as_tool(
tool_name="research_ownership_costs",
tool_description="Research purchase and ownership costs.",
max_turns=5,
),
],
)
purchase_request = (
"We are a family of three living in Zurich. "
"We drive about 15,000 km per year, mostly for commuting and weekend trips, "
"plus several long drives across Europe. "
"We rent an apartment and cannot install a private charger. "
"Our budget is CHF 45,000, and we need room for a child seat and luggage. "
"Would an electric vehicle make sense for us? "
"If so, recommend up to three current models and explain the trade-offs."
)
purchase_result = await Runner.run(adviser, purchase_request, max_turns=10)
print(purchase_result.final_output)
called = [
getattr(item.raw_item, "name", None)
for item in purchase_result.new_items
if item.type == "tool_call_item"
]
print("Agent tools called:", called)
print("Final agent:", purchase_result.last_agent.name)
# ---- 场景二:handoff ----
charging_support = Agent(
name="Charging specialist",
handoff_description="Handles questions about home and public EV charging.",
instructions="Help electric-vehicle owners solve charging questions using current information.",
model=model,
model_settings=model_settings,
tools=[WebSearchTool(search_context_size="medium")],
)
route_planning = Agent(
name="Route planning specialist",
handoff_description="Handles questions about long-distance EV route planning.",
instructions="Help electric-vehicle owners plan journeys using current route and charging information.",
model=model,
model_settings=model_settings,
tools=[WebSearchTool(search_context_size="medium")],
)
vehicle_care = Agent(
name="Vehicle care specialist",
handoff_description="Handles questions about EV maintenance and daily operation.",
instructions="Help electric-vehicle owners with maintenance and operating questions using current information.",
model=model,
model_settings=model_settings,
tools=[WebSearchTool(search_context_size="medium")],
)
support = Agent(
name="EV support agent",
instructions=(
"Help users with electric-vehicle ownership questions. "
"Transfer the conversation to the relevant specialist when appropriate."
),
model=model,
model_settings=model_settings,
handoffs=[charging_support, route_planning, vehicle_care],
)
support_request = (
"I recently bought an electric car and rely entirely on public charging in Zurich. "
"The different networks, subscriptions, and prices are confusing. "
"Help me work out a practical charging setup."
)
support_result = await Runner.run(support, support_request, max_turns=10)
print(support_result.final_output)
path = [
f"{item.source_agent.name} -> {item.target_agent.name}"
for item in support_result.new_items
if item.type == "handoff_output_item"
]
print("Handoff:", path)
print("Final agent:", support_result.last_agent.name)
asyncio.run(main())
运行这个脚本时,重点看三行输出:Agent tools called 是否列出了三位研究员、Handoff 是否显示从支持智能体到充电专家、两次的 Final agent 是否分别是顾问与专家。这三项对上了,说明两种委派模式都按预期工作。
从两种模式组合出更大的系统
有了 agent-as-a-tool 与 handoff 这两块积木,就可以组合出更多形态的多智能体系统。为了不把组合方式搞混,可以从三个维度来思考:
- 委派方式:被委派的智能体是交回结果,还是接管任务?这就是本文覆盖的两种模式。
- 拓扑结构:智能体之间怎么连接?常见的有通过中心协调者连接、串成链、组成层级,或点对点互联。
- 执行方式:智能体是顺序执行、并行执行,还是反复迭代?
把这三个维度放在一起,很多常见的多智能体模式就变得容易理解了。例如 fan-out / fan-in:一个协调者面对若干相互独立的子任务,或者同一个子任务但希望由不同智能体分别处理,于是把工作分发出去,再收回结果汇总。这种结构在委派方式上对应 agent-as-a-tool,在执行方式上对应并行,在拓扑上对应中心协调者。
反过来,如果一条请求从头到尾都属于某个专家的领域,用 handoff 把控制权交出去,往往比让协调者反复调用工具再拼答案更直接。判断标准始终是同一个问题:最终答复需要被汇总,还是本来就该由某位专家负责?
注意事项
- 工具描述决定路由质量:
as_tool()的tool_name与tool_description、handoff 目标的handoff_description,是上层智能体做选择的唯一依据。写得含糊,就会出现该调用的没调用、该移交的没移交。 - 两层轮数限制要分别设置:
as_tool(max_turns=...)限制单个专家在一次调用内的轮数,Runner.run(..., max_turns=...)限制整个运行过程的轮数。只设一层容易导致子任务失控或整体提前截断。 - 用
last_agent验证模式:agent-as-a-tool 运行结束后last_agent应仍是协调者;handoff 运行结束后last_agent应是接手专家。这是最省事的自检手段。 - 结果对象的结构要按类型过滤:工具调用记录在
new_items中类型为tool_call_item的条目里,移交记录在类型为handoff_output_item的条目里,两者不要混在一起统计。 - 联网搜索带来时效性,也带来不确定性:示例中的价格、车型、充电网络信息都来自实时检索,同一问题在不同时间运行可能得到不同结论。涉及价格、配额、模型可用性等易变信息时,请以相关平台官网的当前信息为准。
- 环境变量缺失会直接失败:示例用
os.environ[...]读取三个变量,没有默认值,.env不完整时会在构造客户端阶段就报错。 - 异步上下文:
Runner.run是异步调用,必须在异步函数中await,或像完整示例那样用asyncio.run包一层。