引言:从 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 的工作流程:

  1. 将用户输入和 Agent 指令传递给 LLM
  2. LLM 决定调用哪些工具或执行哪些 handoff
  3. 执行工具调用,将结果返回 LLM
  4. 重复 2-3 直到 LLM 不再请求工具调用,或者达到 max_turns
  5. 返回最终结果

同步执行也支持,但建议生产环境使用异步执行以获得更好的并发性能。

二、架构设计模式

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 SDKLangGraph
设计理念简洁优先,LLM 驱动显式图结构,开发者控制
Handoff语义化,LLM 自主决定显式定义边,开发者指定
状态管理自动(对话历史)手动(State 对象)
流程控制LLM 自主开发者预定义
适合场景开放对话、灵活路由固定流程、严格控制
学习曲线中高
生产成熟度较新,快速迭代中成熟,社区活跃

选择建议:如果你的 Agent 流程高度灵活,需要 LLM 自主决策路由,Agents SDK 是更好的选择。如果你的流程是固定的、需要严格控制每一步,LangGraph 更合适。

结语

OpenAI Agents SDK 继承了 Swarm 的简洁设计理念,同时补齐了生产所需的关键特性。它不是最强大的 Agent 框架,但可能是最容易上手的。对于大多数 Agent 应用场景,“简单"本身就是一种强大——更少的代码意味着更少的 bug、更快的迭代、更低的维护成本。

Agent 技术仍在快速演进,框架也会持续迭代。最重要的不是选择哪个框架,而是深入理解 Agent 的核心设计模式:分诊、管线、协作、循环。这些模式不依赖于特定框架,是构建任何 Agent 系统的基础。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞,认知差异创造无限可能。