手把手教你用 OpenAI Swarm 搭建高可用多智能体流水线：从故障重试到动态协同

摘要

在构建复杂数据处理系统时，开发者常面临两大痛点：如何让多个 AI 智能体高效协同工作，以及如何应对不可预见的运行时故障。OpenAI Swarm 框架提供了一种轻量级的多智能体编排方案，通过函数调用机制实现动态任务分配，并内置故障恢复能力。本文通过一个模拟数据处理流水线的 Demo，展示如何用不到 100 行代码搭建一个由 Analyst、Reporter、Notifier 三个智能体组成的协作系统，每个步骤以 30% 概率模拟故障，Swarm 自动重试最多 2 次。你将看到多智能体如何分工合作、动态传递上下文，以及如何优雅处理异常。读完本文，你就能在自己的项目中快速落地 Swarm 框架。

问题背景：当单智能体不够用，故障处理成噩梦

假设你正在开发一个企业级数据处理平台，需要完成“分析销售数据 → 生成报告 → 发送通知”这条流水线。传统做法是写一个串行脚本，但很快你会发现几个棘手问题：

职责耦合：所有逻辑堆在一个函数里，分析、报告、通知混在一起，难以维护和扩展。
故障处理粗暴：任何一个步骤失败，整个流程崩溃，没有重试机制。
智能体缺乏上下文：每个步骤需要知道前一步的结果，但手动传递参数容易出错。

更糟糕的是，当你尝试引入 AI 能力时，比如用 GPT 分析数据、生成报告，你会陷入“如何让多个 AI 模型协同工作”的泥潭。每个模型需要不同的 prompt 和上下文，调用链一旦断裂，调试成本极高。

OpenAI Swarm 正是为解决这些问题而生。它不是一个完整的 AI 框架，而是一个轻量级的智能体编排工具，让你能像搭积木一样组合多个智能体，每个智能体负责独立任务，通过函数调用传递上下文，并内置故障恢复机制。

技术方案：Swarm 的核心设计哲学

Swarm 的设计理念可以用三个词概括：轻量、函数驱动、动态路由。

轻量：Swarm 不依赖复杂的状态管理或消息队列，它只是一个 Python 库，通过 Agent 和 Swarm 两个核心类实现编排。
函数驱动：每个智能体通过注册函数（functions）来定义能力。当智能体需要执行任务时，它会调用这些函数，就像调用普通 Python 函数一样。
动态路由：Swarm 支持智能体之间的上下文传递。一个智能体的输出可以自动作为另一个智能体的输入，无需手动拼接。

在我们的 Demo 中，三个智能体分工如下：

Analyst：分析任务数据，返回分析结果。
Reporter：基于分析结果生成报告。
Notifier：发送通知。

每个智能体注册一个函数，Swarm 负责按顺序调用它们，并在失败时自动重试。

核心实现解析：从代码到原理

1. 安装与初始化

1	pip install openai-swarm

Swarm 依赖 OpenAI 的 API，所以你需要设置 OPENAI_API_KEY 环境变量。不过在我们的 Demo 中，所有操作都是模拟的，不实际调用 GPT。

2. 定义智能体函数

每个智能体对应一个函数，函数签名遵循“输入字符串，输出字符串”的约定。这保证了智能体之间的数据流是线性的。

import json
import random
from swarm import Swarm, Agent

# 模拟故障
def simulate_failure() -> bool:
    return random.random() < 0.3  # 30% 概率失败

# 智能体函数：分析数据
def analyze_data(task: str) -> str:
    if simulate_failure():
        raise Exception("Data analysis failed due to timeout")
    return f"Analyzed {task}: found 3 key insights."

# 智能体函数：生成报告
def generate_report(analysis: str) -> str:
    if simulate_failure():
        raise Exception("Report generation failed due to missing template")
    return f"Generated report based on: {analysis}"

# 智能体函数：发送通知
def send_notification(report: str) -> str:
    if simulate_failure():
        raise Exception("Notification failed due to network error")
    return f"Notification sent with report: {report}"

注意每个函数的输入参数名（task、analysis、report）不是随意取的。Swarm 通过参数名匹配来自动传递上下文。当 generate_report 被调用时，Swarm 会查找前一步的输出，并将其赋值给 analysis 参数。

3. 创建智能体并注册函数

# 创建智能体
analyst_agent = Agent(
    name="Analyst",
    instructions="You are a data analyst. Analyze the given task and return insights.",
    functions=[analyze_data]
)

reporter_agent = Agent(
    name="Reporter",
    instructions="You are a report generator. Create a report based on the analysis.",
    functions=[generate_report]
)

notifier_agent = Agent(
    name="Notifier",
    instructions="You are a notification sender. Send the report to stakeholders.",
    functions=[send_notification]
)

每个 Agent 需要三个参数：

name：智能体名称，用于日志和调试。
instructions：系统提示词，描述智能体的职责。
functions：智能体可以调用的函数列表。

4. 定义任务队列和流水线

# 模拟任务队列
TASKS = [
    "Analyze quarterly sales data",
    "Generate customer segmentation report",
    "Summarize recent user feedback",
    "Draft email to stakeholders",
    "Check inventory levels"
]

# 定义流水线：智能体顺序
PIPELINE = [
    ("Analyst", analyze_data),
    ("Reporter", generate_report),
    ("Notifier", send_notification)
]

这里我们显式定义了流水线的执行顺序。Swarm 本身不强制顺序，但通过函数调用链可以隐式实现。为了清晰展示，我们手动指定了步骤。

5. 执行流水线并处理故障

def run_pipeline(task: str, max_retries: int = 2):
    print(f"\n=== Processing task: {task} ===")
    current_context = task
    
    for agent_name, func in PIPELINE:
        retries = 0
        success = False
        
        while retries <= max_retries and not success:
            try:
                # 调用智能体函数
                result = func(current_context)
                print(f"[{agent_name}] {result}")
                current_context = result  # 更新上下文
                success = True
            except Exception as e:
                retries += 1
                if retries <= max_retries:
                    print(f"[{agent_name}] Failed (attempt {retries}/{max_retries}): {e}")
                else:
                    print(f"[{agent_name}] Failed after {max_retries} retries: {e}")
                    return None
        
        if not success:
            return None
    
    print(f"[Result] Pipeline completed successfully!")
    return current_context

# 运行所有任务
client = Swarm()
for task in TASKS:
    result = run_pipeline(task)
    if result:
        print(f"Final output: {result}\n")

关键点：

上下文传递：current_context 变量保存上一个函数的输出，作为下一个函数的输入。
重试机制：每个步骤最多重试 2 次，每次失败后打印日志。
故障隔离：一个步骤的失败不会影响其他任务的执行。

6. 完整代码

将以上代码整合到 main.py 中，即可运行。

运行效果：故障恢复演示

执行 python main.py，你会看到类似输出：

=== Processing task: Analyze quarterly sales data ===
[Analyst] Analyzed Analyze quarterly sales data: found 3 key insights.
[Reporter] Failed (attempt 1/2): Report generation failed due to missing template
[Reporter] Generated report based on: Analyzed Analyze quarterly sales data: found 3 key insights.
[Notifier] Notification sent with report: Generated report based on: Analyzed Analyze quarterly sales data: found 3 key insights.
[Result] Pipeline completed successfully!

=== Processing task: Generate customer segmentation report ===
[Analyst] Failed (attempt 1/2): Data analysis failed due to timeout
[Analyst] Analyzed Generate customer segmentation report: found 3 key insights.
[Reporter] Generated report based on: Analyzed Generate customer segmentation report: found 3 key insights.
[Notifier] Failed (attempt 1/2): Notification failed due to network error
[Notifier] Failed (attempt 2/2): Notification failed due to network error
[Result] Pipeline failed at Notifier

可以看到：

第一个任务中，Reporter 第一次失败，重试后成功。
第二个任务中，Analyst 第一次失败，重试成功；Notifier 两次都失败，任务最终失败，但其他任务不受影响。

这种故障恢复能力在实际生产环境中至关重要。想象一下，如果发送通知的 API 暂时不可用，系统不会直接崩溃，而是重试几次，可能就恢复正常了。

总结与展望

通过这个 Demo，我们看到了 OpenAI Swarm 的核心价值：

多智能体协同：通过函数注册和上下文传递，智能体可以像微服务一样独立工作。
动态任务分配：Swarm 根据函数签名自动匹配参数，无需手动拼接。
故障恢复：内置重试机制，提升系统鲁棒性。

但 Swarm 也有其局限性：

线性流水线：当前实现是顺序执行的，不支持并行或条件分支。
无状态：每个任务独立执行，无法持久化状态。
依赖 OpenAI API：虽然 Demo 是模拟的，但实际使用需要 API 调用。

展望未来，Swarm 可以扩展的方向包括：

支持异步执行和消息队列。
引入状态管理，支持长时间运行的任务。
集成更多 AI 模型（如本地模型）。

对于有经验的开发者来说，Swarm 是一个很好的起点。你可以基于它构建更复杂的智能体系统，比如客服机器人、自动化报告生成、智能监控告警等。它的轻量特性让你能快速原型验证，再根据需求定制化扩展。

现在，打开你的终端，安装 Swarm，开始构建你的第一个多智能体流水线吧！