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

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

结语

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

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

加入讨论

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

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