引言:从 Swarm 到 Agents SDK 的演进
2024 年底,OpenAI 发布了 Swarm——一个轻量级的 Agent 编排实验框架。Swarm 以其极简的设计理念和优雅的 handoff 机制迅速获得社区关注,但它始终定位为"实验性"项目,不适合生产使用。2025 年,OpenAI 在 Swarm 的设计理念基础上正式推出了 Agents SDK——一个生产就绪的 Agent 开发框架,继承了 Swarm 的简洁性,同时增加了企业级所需的关键特性。
本文将从实际开发角度出发,深入解析 Agents SDK 的核心概念、架构设计、常见模式和实战经验,帮助你快速构建生产级的多 Agent 应用。
一、核心概念解析
1.1 Agent
Agent 是 Agents SDK 的基本构建单元。一个 Agent 封装了:
from agents import Agent
customer_service_agent = Agent(
name="客服助手",
model="gpt-4o",
instructions="""你是一个专业的客服助手。
职责:回答用户问题、处理投诉、引导使用产品。
原则:耐心、准确、礼貌。不确定时坦诚告知并转交人工。""",
tools=[search_knowledge_base, create_ticket, check_order_status],
)
与 Swarm 相比,Agents SDK 的 Agent 新增了以下能力:
- 模型路由:可以为不同 Agent 指定不同模型(gpt-4o、gpt-4o-mini 等),按需平衡性能和成本
- 结构化输出:支持通过 Pydantic 模型定义 Agent 的输出格式
- Guardrails:为 Agent 添加输入/输出安全检查
- 生命周期钩子:在 Agent 执行的不同阶段注入自定义逻辑
1.2 Handoff(交接)
Handoff 是 Agents SDK 最核心的设计模式,它定义了 Agent 之间如何转移控制权:
from agents import Agent, handoff
triage_agent = Agent(
name="分诊助手",
instructions="根据用户问题类型决定由哪个 specialist 处理。",
handoffs=[
handoff(billing_agent, description="处理账单和支付相关问题"),
handoff(technical_agent, description="处理技术支持和故障排查"),
handoff(general_agent, description="处理一般性咨询"),
],
)
Handoff 的关键设计:
- 语义化触发:Agent 通过 LLM 理解 handoff 描述,自主决定何时交接
- 上下文传递:交接时自动携带完整对话历史
- 工具化实现:handoff 本质上是一个特殊工具,LLM “调用"它来触发交接
- 可定制过滤:可以自定义交接时传递的上下文内容
1.3 Tool(工具)
Agents SDK 提供了灵活的工具定义方式:
函数工具:最常用的方式,将普通 Python 函数转为 Agent 工具。
from agents import function_tool
@function_tool
def search_orders(user_id: str, status: str = "all") -> list[dict]:
"""搜索用户订单。
Args:
user_id: 用户 ID
status: 订单状态筛选 (all/active/completed/cancelled)
Returns:
匹配的订单列表
"""
# 实际查询逻辑
return orders
关键细节:
- 函数的 docstring 和类型注解会自动转化为工具的 JSON Schema
- 参数类型支持 str、int、float、bool、list、dict 以及 Pydantic 模型
- 返回值会被序列化为字符串传递给 LLM
- 支持异步函数(
async def)
计算机使用工具:Agents SDK 集成了 OpenAI 的 Computer Use 能力,可以让 Agent 操控计算机界面。
内置工具:提供文件搜索、代码解释器等 OpenAI 原生工具的封装。
1.4 Runner(运行器)
Runner 是 Agent 执行的引擎,管理整个 Agent 生命周期:
from agents import Runner
result = await Runner.run(
triage_agent,
input="我的订单 #12345 还没到,已经超时 3 天了",
context=user_context,
max_turns=20,
)
Runner 的工作流程:
- 将用户输入和 Agent 指令传递给 LLM
- LLM 决定调用哪些工具或执行哪些 handoff
- 执行工具调用,将结果返回 LLM
- 重复 2-3 直到 LLM 不再请求工具调用,或者达到 max_turns
- 返回最终结果
同步执行也支持,但建议生产环境使用异步执行以获得更好的并发性能。
二、架构设计模式
2.1 分诊模式(Triage Pattern)
最基础的多 Agent 模式。一个中心 Agent 负责理解用户意图,然后将控制权交给专门的 Agent 处理。
用户 → 分诊 Agent →
├→ 账单 Agent (处理支付问题)
├→ 技术 Agent (处理技术问题)
├→ 退换 Agent (处理退换货)
└→ 通用 Agent (处理其他问题)
适用场景:客服系统、多领域问答、复杂工作流入口。
优势:用户只需与一个入口交互,内部分发透明。
注意点:分诊 Agent 的指令需要精心设计,确保它能准确识别用户意图。可以提供 handoff 的详细描述来辅助决策。
2.2 管线模式(Pipeline Pattern)
Agent 按顺序处理任务,每个 Agent 负责一个阶段:
输入 → 研究 Agent → 写作 Agent → 审校 Agent → 输出
实现方式是利用 handoff 在 Agent 间传递:
researcher = Agent(
name="研究员",
instructions="收集和分析相关信息,整理成结构化笔记。",
tools=[web_search, read_url],
handoffs=[handoff(writer, description="将研究结果写成文章")],
)
writer = Agent(
name="撰稿人",
instructions="基于研究结果撰写高质量文章。",
handoffs=[handoff(editor, description="审校文章")],
)
editor = Agent(
name="审校编辑",
instructions="审校文章,修正错误,优化表达。",
)
适用场景:内容生产、代码开发、数据处理管线。
优势:每个 Agent 职责单一,指令清晰,输出质量高。
注意点:需要合理设计 Agent 间的信息传递格式,避免信息损失。
2.3 协作模式(Collaboration Pattern)
多个 Agent 并行工作,最后汇总结果:
from agents import Agent, Runner
import asyncio
# 并行执行多个专家 Agent
results = await asyncio.gather(
Runner.run(security_reviewer, input=code_diff),
Runner.run(performance_reviewer, input=code_diff),
Runner.run(style_reviewer, input=code_diff),
)
# 汇总 Agent 整合所有结果
summary = await Runner.run(
summarizer,
input=f"安全审查: {results[0].final_output}\n"
f"性能审查: {results[1].final_output}\n"
f"风格审查: {results[2].final_output}",
)
适用场景:代码审查、多角度分析、多语言翻译。
优势:并行执行效率高,多角度视角提升结果质量。
注意点:汇总 Agent 需要处理可能冲突的建议,指令中应包含冲突解决策略。
2.4 循环改进模式(Refinement Loop)
Agent 的输出经过评审后返回改进,形成循环:
@function_tool
def review_code(code: str) -> str:
"""评审代码质量,返回改进建议。如果有严重问题,返回 'NEEDS_REVISION'。"""
issues = analyze_code(code)
if has_critical_issues(issues):
return f"NEEDS_REVISION: {issues}"
return f"APPROVED: {issues}"
developer = Agent(
name="开发者",
instructions="""编写代码,使用 review_code 工具评审。
如果返回 NEEDS_REVISION,根据反馈修改后再次评审。
如果返回 APPROVED,结束任务。""",
tools=[review_code, write_file],
)
适用场景:代码生成、文档撰写、方案设计。
优势:自我纠错能力强,输出质量稳定。
注意点:必须设置 max_turns 防止无限循环。
三、实战案例:智能客服系统
下面通过一个完整的客服系统案例展示 Agents SDK 的实际应用:
3.1 系统架构
用户 → 分诊 Agent →
├→ 订单查询 Agent (工具: 查询订单系统)
├→ 退换货 Agent (工具: 创建退换货单)
├→ 技术支持 Agent (工具: 搜索知识库)
└→ 投诉处理 Agent (工具: 创建工单, 升级处理)
3.2 核心实现
from agents import Agent, Runner, function_tool, handoff
from pydantic import BaseModel
# 上下文模型
class UserContext(BaseModel):
user_id: str
user_name: str
vip_level: int
recent_orders: list[dict] = []
# 工具定义
@function_tool
def query_order(order_id: str, ctx: UserContext) -> str:
"""查询订单详情。"""
order = order_service.get(ctx.user_id, order_id)
return f"订单 {order_id}: 状态={order.status}, 预计送达={order.eta}"
@function_tool
def create_return_request(order_id: str, reason: str, ctx: UserContext) -> str:
"""创建退货申请。"""
request = return_service.create(
user_id=ctx.user_id,
order_id=order_id,
reason=reason,
)
return f"退货申请已创建,编号 {request.id},预计 1-3 工作日审核。"
@function_tool
def escalate_to_human(priority: str, summary: str, ctx: UserContext) -> str:
"""升级至人工客服。"""
ticket = ticket_service.create(
user_id=ctx.user_id,
priority=priority,
summary=summary,
vip=ctx.vip_level >= 3,
)
return f"已创建工单 {ticket.id},VIP 通道预计 5 分钟内响应。"
# Agent 定义
order_agent = Agent(
name="订单助手",
model="gpt-4o-mini", # 简单查询用小模型,控制成本
instructions="""你是订单查询助手。帮助用户查询订单状态和物流信息。
如果涉及退换货,转交给退换货助手。""",
tools=[query_order],
handoffs=[handoff(return_agent, description="处理退换货请求")],
)
return_agent = Agent(
name="退换货助手",
model="gpt-4o",
instructions="""你是退换货处理助手。帮助用户提交退换货申请。
需要收集:订单号、退换原因、期望处理方式。
VIP 用户提醒可享受极速退款服务。""",
tools=[create_return_request],
)
complaint_agent = Agent(
name="投诉处理助手",
model="gpt-4o",
instructions="""你是投诉处理助手。认真倾听用户诉求,表达同理心。
能解决的问题直接处理,不能解决的升级至人工。""",
tools=[escalate_to_human, create_return_request],
)
triage_agent = Agent(
name="客服分诊",
model="gpt-4o-mini",
instructions="""你是客服分诊助手。根据用户问题类型分发给对应的专门助手。
- 订单查询/物流追踪 → 订单助手
- 退货/换货/退款 → 退换货助手
- 投诉/纠纷/不满 → 投诉处理助手
不确定时选择最相关的助手。""",
handoffs=[
handoff(order_agent, description="订单查询和物流追踪"),
handoff(return_agent, description="退换货和退款处理"),
handoff(complaint_agent, description="投诉和纠纷处理"),
],
)
3.3 运行与部署
# 使用 FastAPI 部署
from fastapi import FastAPI
from agents import Runner
app = FastAPI()
@app.post("/chat")
async def chat_endpoint(message: str, user_id: str):
ctx = UserContext(
user_id=user_id,
user_name=user_service.get_name(user_id),
vip_level=user_service.get_vip_level(user_id),
)
result = await Runner.run(
triage_agent,
input=message,
context=ctx,
max_turns=15,
)
return {"response": result.final_output}
四、高级特性
4.1 Guardrails(安全护栏)
Guardrails 在 Agent 执行前后进行安全检查:
from agents import Agent, GuardrailFunctionOutput, input_guardrail
@input_guardrail
async def check_toxic_content(ctx, agent, input):
"""检测输入中的有害内容。"""
toxic_score = await moderation_service.check(input)
return GuardrailFunctionOutput(
output_info={"toxic_score": toxic_score},
tripwire_triggered=toxic_score > 0.8,
)
agent = Agent(
name="安全助手",
instructions="...",
input_guardrails=[check_toxic_content],
)
当 guardrail 触发 tripwire 时,Agent 执行会被立即中断。
4.2 生命周期钩子
from agents import AgentHooks
class LoggingHooks(AgentHooks):
async def on_start(self, ctx, agent, input):
logger.info(f"Agent {agent.name} started with input: {input[:100]}")
async def on_tool_start(self, ctx, agent, tool):
logger.info(f"Tool {tool.name} called by {agent.name}")
async def on_tool_end(self, ctx, agent, tool, result):
logger.info(f"Tool {tool.name} returned: {result[:100]}")
async def on_end(self, ctx, agent, output):
logger.info(f"Agent {agent.name} finished: {output[:100]}")
agent = Agent(name="带日志的Agent", instructions="...", hooks=LoggingHooks())
4.3 流式输出
result = Runner.run_streamed(agent, input="用户问题")
async for event in result.stream_events():
if event.type == "tool_output":
print(f"工具结果: {event.output}")
elif event.type == "message_output":
print(f"Agent: {event.delta}", end="")
流式输出对于用户体验至关重要,特别是需要实时展示 Agent 思考过程的交互场景。
4.4 Tracing(追踪)
Agents SDK 内置了 OpenAI Tracing 集成,可以可视化 Agent 执行的全过程:
from agents import enable_tracing
enable_tracing(
project_name="客服系统",
endpoint="https://api.openai.com/v1/traces",
)
Tracing 会记录每个 Agent 调用、工具执行、handoff 事件,便于调试和性能分析。
五、生产部署最佳实践
5.1 错误处理
from agents import RunError
try:
result = await Runner.run(agent, input=user_msg, context=ctx)
except RunError as e:
# Agent 执行错误(如达到 max_turns)
logger.error(f"Agent error: {e}")
return "抱歉,处理过程中遇到问题,已为您转接人工客服。"
except Exception as e:
# 系统错误
logger.error(f"System error: {e}")
return "系统暂时不可用,请稍后重试。"
5.2 成本控制
- 模型分级:分诊等简单任务用 gpt-4o-mini,专业处理用 gpt-4o
- Token 预算:设置 max_turns 避免无限循环消耗
- 缓存:对重复查询实现语义缓存,避免重复调用 LLM
- 监控:追踪每次请求的 token 消耗,设置异常告警
5.3 测试策略
# 单元测试工具函数
def test_query_order():
result = query_order("12345", ctx=test_ctx)
assert "订单状态" in result
# 集成测试 Agent 行为
async def test_triage_routes_to_return():
result = await Runner.run(
triage_agent,
input="我想退货,订单 12345",
context=test_ctx,
)
# 验证是否正确 handoff 到退换货 Agent
assert "退货申请" in result.final_output or "退换" in result.last_agent.name
5.4 可观测性
生产环境必须关注以下指标:
- 响应延迟分布:P50、P95、P99 响应时间
- Agent 路由准确率:分诊 Agent 是否正确分发
- 工具调用成功率:各工具的成功/失败率
- Token 消耗:每次请求的输入/输出 token 数
- Handoff 链长:平均 handoff 次数,过长可能说明分诊不准
六、与 LangGraph 的对比
| 维度 | Agents SDK | LangGraph |
|---|---|---|
| 设计理念 | 简洁优先,LLM 驱动 | 显式图结构,开发者控制 |
| Handoff | 语义化,LLM 自主决定 | 显式定义边,开发者指定 |
| 状态管理 | 自动(对话历史) | 手动(State 对象) |
| 流程控制 | LLM 自主 | 开发者预定义 |
| 适合场景 | 开放对话、灵活路由 | 固定流程、严格控制 |
| 学习曲线 | 低 | 中高 |
| 生产成熟度 | 较新,快速迭代中 | 成熟,社区活跃 |
选择建议:如果你的 Agent 流程高度灵活,需要 LLM 自主决策路由,Agents SDK 是更好的选择。如果你的流程是固定的、需要严格控制每一步,LangGraph 更合适。
结语
OpenAI Agents SDK 继承了 Swarm 的简洁设计理念,同时补齐了生产所需的关键特性。它不是最强大的 Agent 框架,但可能是最容易上手的。对于大多数 Agent 应用场景,“简单"本身就是一种强大——更少的代码意味着更少的 bug、更快的迭代、更低的维护成本。
Agent 技术仍在快速演进,框架也会持续迭代。最重要的不是选择哪个框架,而是深入理解 Agent 的核心设计模式:分诊、管线、协作、循环。这些模式不依赖于特定框架,是构建任何 Agent 系统的基础。
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞,认知差异创造无限可能。
