多 Agent 协作教程：角色分工、任务流和失败重试

这篇教程适合你吗

你已经搭过单个 AI Agent（可以看 AI Agent 开发教程），但发现复杂任务一个 Agent 搞不定——需要搜索、分析、写作多种能力，一个 Agent 的 prompt 越写越长，效果越来越差。多 Agent 协作就是解决方案。

前置知识：了解 AI Agent 和 Function Calling 基本概念。 最终产物：一个包含角色分工、任务编排和失败重试的多 Agent 协作系统。

为什么需要多 Agent

单 Agent 处理复杂任务时的问题：

• prompt 膨胀：塞太多指令，模型反而什么都做不好
• 职责不清：一个 Agent 要兼顾搜索、分析、写作，容易出错
• 难以优化：改一个能力的 prompt 可能影响其他能力

多 Agent 的思路：把复杂任务拆分给不同角色。

基础架构：共享状态 + 角色分工

from dataclasses import dataclass, field
from typing import Any
@dataclass
class TaskState:
    """共享状态：所有 Agent 读写这个对象"""
    query: str = ""
    search_results: list = field(default_factory=list)
    analysis: str = ""
    draft: str = ""
    final_output: str = ""
    errors: list = field(default_factory=list)
    current_step: str = "pending"

定义 Agent 基类

from openai import OpenAI
client = OpenAI()
class Agent:
    def __init__(self, name: str, system_prompt: str, model: str = "gpt-4o"):
        self.name = name
        self.system_prompt = system_prompt
        self.model = model
    def run(self, state: TaskState) -> TaskState:
        """子类实现具体逻辑"""
        raise NotImplementedError
    def ask(self, user_message: str) -> str:
        """调用 LLM"""
        response = client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": user_message}
            ]
        )
        return response.choices[0].message.content

定义具体 Agent

搜索 Agent

class SearchAgent(Agent):
    def __init__(self):
        super().__init__(
            name="搜索助手",
            system_prompt="你是一个搜索助手，负责根据用户问题生成搜索关键词和分析搜索结果。"
        )
    def run(self, state: TaskState) -> TaskState:
        # 生成搜索关键词
        keywords = self.ask(f"用户问题：{state.query}\n\n请生成 3 个搜索关键词，每行一个。")
        state.search_results = [f"搜索结果: {kw}" for kw in keywords.strip().split("\n")]
        state.current_step = "search_done"
        return state

分析 Agent

class AnalysisAgent(Agent):
    def __init__(self):
        super().__init__(
            name="分析师",
            system_prompt="你是一个分析师，负责从搜索结果中提取关键信息并形成分析结论。"
        )
    def run(self, state: TaskState) -> TaskState:
        results_text = "\n".join(state.search_results)
        state.analysis = self.ask(
            f"用户问题：{state.query}\n\n搜索结果：\n{results_text}\n\n请分析这些信息，给出关键结论。"
        )
        state.current_step = "analysis_done"
        return state

写作 Agent

class WritingAgent(Agent):
    def __init__(self):
        super().__init__(
            name="写作者",
            system_prompt="你是一个写作者，负责根据分析结论撰写清晰、有结构的回答。"
        )
    def run(self, state: TaskState) -> TaskState:
        state.final_output = self.ask(
            f"用户问题：{state.query}\n\n分析结论：\n{state.analysis}\n\n请撰写一篇结构清晰的回答。"
        )
        state.current_step = "writing_done"
        return state

任务编排

定义 Agent 之间的执行顺序：

class Pipeline:
    def __init__(self, agents: list):
        self.agents = agents
    def run(self, state: TaskState) -> TaskState:
        for agent in self.agents:
            try:
                state = agent.run(state)
                print(f"[{agent.name}] 完成，当前步骤: {state.current_step}")
            except Exception as e:
                state.errors.append(f"{agent.name} 失败: {str(e)}")
                print(f"[{agent.name}] 错误: {e}")
        return state
# 组装流水线
pipeline = Pipeline([
    SearchAgent(),
    AnalysisAgent(),
    WritingAgent()
])
# 运行
state = TaskState(query="AI Agent 和传统 AI 有什么区别？")
result = pipeline.run(state)
print(result.final_output)

失败重试机制

给 Agent 加上重试能力：

import time
class RetryableAgent(Agent):
    def __init__(self, name, system_prompt, max_retries: int = 3, kwargs):

        super().__init__(name, system_prompt, kwargs)
        self.max_retries = max_retries
    def run_with_retry(self, state: TaskState) -> TaskState:
        for attempt in range(self.max_retries):
            try:
                return self.run(state)
            except Exception as e:
                if attempt == self.max_retries - 1:
                    state.errors.append(f"{self.name} 重试 {self.max_retries} 次后仍失败: {e}")
                    return state
                print(f"[{self.name}] 第 {attempt+1} 次失败，{2  attempt} 秒后重试...")

                time.sleep(2  attempt)  # 指数退避
        return state

条件分支

根据任务状态决定下一步：

class ConditionalPipeline:
    def __init__(self):
        self.agents = {}
    def register(self, step_name: str, agent: Agent, condition=None):
        self.agents[step_name] = {"agent": agent, "condition": condition}
    def run(self, state: TaskState) -> TaskState:
        for step_name, config in self.agents.items():
            condition = config["condition"]
            if condition and not condition(state):
                print(f"[{step_name}] 跳过（条件不满足）")
                continue
            state = config["agent"].run(state)
        return state
# 使用
pipeline = ConditionalPipeline()
pipeline.register("search", SearchAgent())
pipeline.register("analysis", AnalysisAgent())
pipeline.register("writing", WritingAgent(), condition=lambda s: s.search_results)

实际应用：研究报告生成

# 组装一个研究报告生成系统
report_pipeline = Pipeline([
    SearchAgent(),      # 搜索相关信息
    AnalysisAgent(),    # 分析搜索结果
    WritingAgent()      # 撰写报告
])
state = TaskState(query="2026 年 AI Agent 发展趋势")
result = report_pipeline.run(state)
print(result.final_output)

常见问题与排错

Agent 之间传递信息丢失

原因：Agent 没有正确读写共享状态。 解决：确保每个 Agent 都从 state 中读取输入，把输出写回 state，而不是只返回字符串。

整体速度太慢

原因：串行执行，每个 Agent 都要等前一个完成。 解决：对于没有依赖关系的 Agent，可以用并行执行。例如搜索 Agent 和意图识别 Agent 可以同时运行。

某个 Agent 频繁失败

原因：该 Agent 的 prompt 不够清晰，或输入质量太差。 解决：优化该 Agent 的 system_prompt；检查上游 Agent 的输出质量；增加重试次数。

进阶学习

• 想了解 Agent 概念？看 AI Agent 是什么
• 想了解工作流？看 AI 工作流是什么
• 想了解工具调用？看 Function Calling 教程
• 想了解 Agent 记忆？看 AI Agent 记忆管理教程

总结

• 多 Agent 通过角色分工处理复杂任务
• 共享状态是最简单的 Agent 通信方式
• 任务编排定义执行顺序和条件分支
• 失败重试用指数退避策略保证稳定性
• 先用简单 pipeline 跑通，再逐步增加复杂度

---

本文最后更新于 2026-07-27。代码示例基于当前常见版本，请根据最新 API 文档调整。