多 Agent 协作：用技能链构建 AI 团队

从单兵作战到团队协作

单个 AI Agent 的能力存在天花板——它可能擅长规划但不擅长执行，或者擅长编码但不擅长测试。这就像一个人无法同时成为顶级的架构师、程序员和测试工程师。多 Agent 协作（Multi-Agent Collaboration）正是为解决这个问题而生：将复杂任务拆解为多个子任务，由不同专长的 Agent 分别负责，通过协作完成整体目标。

多 Agent 协作的核心思想借鉴了人类团队的工作模式：有人负责规划分解（项目经理），有人负责具体执行（工程师），有人负责质量把关（审核员）。每个 Agent 专注于自己的领域，通过标准化的通信协议交换信息，最终交付高质量的结果。

多 Agent 协作的三种模式

1. 顺序流水线（Sequential Pipeline）：Agent 按固定顺序执行，每个 Agent 的输出是下一个 Agent 的输入。适用于流程明确的任务，如文档处理流水线。

2. 层级调度（Hierarchical）：一个「管理者」Agent 负责任务分配和结果汇总，多个「执行者」Agent 并行工作。适用于可并行分解的任务。

3. 辩论协作（Debate）：多个 Agent 独立生成方案，然后互相评审、辩论，最终达成共识。适用于需要多角度分析的决策任务。

本文重点介绍层级调度模式，这也是「多 Agent 协作链」技能链采用的模式。

技能链拆解：多 Agent 协作链

我们以「多 Agent 协作链」为例，展示层级调度模式的节点编排：

节点 1：任务规划（sp-188）——「规划者」Agent 负责接收用户需求，将复杂任务拆解为可执行的子任务，并分配给合适的执行者。这个节点的质量直接决定了整个协作链的效率。

节点 2：代码执行（sp-187）——「执行者」Agent 根据规划者分配的任务，具体完成编码工作。多个执行者可以并行工作，各自负责不同的模块。

节点 3：质量审核（sp-195）——「审核者」Agent 对执行者的产出进行质量检查。审核者需要独立于执行者，以客观标准评估代码质量、逻辑正确性和安全性。

节点 4：结果汇总（sp-122）——规划者再次介入，收集所有执行者的产出和审核者的反馈，整合为最终交付物。如果审核不通过，规划者会重新分配任务回到执行节点。

实战：多 Agent 协作框架

以下代码展示了层级调度式多 Agent 协作的实现：

import json
from openai import OpenAI
from concurrent.futures import ThreadPoolExecutor

client = OpenAI(
    api_key="your-deepseek-api-key",
    base_url="https://api.deepseek.com"
)

class AgentRole:
    """定义 Agent 角色"""
    def __init__(self, name, system_prompt):
        self.name = name
        self.system_prompt = system_prompt

    def execute(self, task):
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": task}
            ]
        )
        return response.choices[0].message.content

# 定义角色
PLANNER = AgentRole("规划者", """你是一位资深技术架构师。
你的职责：
1. 分析用户需求，将其拆解为 2-4 个可并行执行的子任务
2. 为每个子任务指定执行者类型（前端/后端/测试/文档）
3. 定义子任务之间的依赖关系

输出格式（JSON）：
{
  "subtasks": [
    {"id": 1, "type": "类型", "description": "任务描述", "depends_on": []}
  ]
}""")

EXECUTOR = AgentRole("执行者", """你是一位全栈开发工程师。
根据任务描述，产出高质量的代码实现。
包含：
1. 完整的代码实现
2. 关键设计决策说明
3. 潜在风险点""")

REVIEWER = AgentRole("审核者", """你是一位严格的代码审核专家。
审核标准：
1. 代码正确性（逻辑是否有误）
2. 安全性（是否有注入、泄露等风险）
3. 性能（是否有明显瓶颈）
4. 可维护性（代码是否清晰易懂）

输出格式（JSON）：
{
  "approved": true/false,
  "score": 0-100,
  "issues": ["问题描述"],
  "suggestions": ["改进建议"]
}""")

class MultiAgentSystem:
    def __init__(self, max_iterations=5):
        self.max_iterations = max_iterations

    def run(self, user_request):
        print(f"用户需求：{user_request}\n")

        # 阶段 1：规划
        print("=== 阶段 1：任务规划 ===")
        plan = PLANNER.execute(f"请为以下需求制定执行计划：\n{user_request}")
        subtasks = json.loads(plan)["subtasks"]
        print(f"拆解为 {len(subtasks)} 个子任务\n")

        # 阶段 2：并行执行
        print("=== 阶段 2：并行执行 ===")
        results = {}
        with ThreadPoolExecutor(max_workers=4) as executor:
            futures = {}
            for task in subtasks:
                future = executor.submit(
                    EXECUTOR.execute,
                    f"子任务 {task['id']}：{task['description']}"
                )
                futures[future] = task['id']

            for future in futures:
                task_id = futures[future]
                results[task_id] = future.result()
                print(f"子任务 {task_id} 执行完成")

        # 阶段 3：审核
        print("\n=== 阶段 3：质量审核 ===")
        all_approved = True
        for task_id, result in results.items():
            review = REVIEWER.execute(f"请审核以下代码：\n{result}")
            review_data = json.loads(review)
            status = "✓ 通过" if review_data["approved"] else "✗ 需修改"
            print(f"子任务 {task_id}：{status}（评分：{review_data['score']}）")
            if not review_data["approved"]:
                all_approved = False

        # 阶段 4：汇总
        print("\n=== 阶段 4：结果汇总 ===")
        summary_prompt = f"""请汇总以下多 Agent 协作的结果：

原始需求：{user_request}
子任务执行结果：
{json.dumps(results, ensure_ascii=False, indent=2)}

请生成最终交付报告，包含：
1. 项目概述
2. 各模块说明
3. 集成方案
4. 注意事项"""

        final_report = PLANNER.execute(summary_prompt)
        return final_report

# 使用示例
system = MultiAgentSystem(max_iterations=5)
report = system.run("开发一个简单的待办事项 Web 应用，支持增删改查和状态标记")
print(f"\n最终报告：\n{report}")

多 Agent 协作的挑战与对策

通信开销：Agent 之间的信息传递会产生大量 token 消耗。解决方案：使用结构化的通信协议（如 JSON Schema），减少冗余信息；对中间结果进行摘要压缩。

一致性问题：不同 Agent 可能产生冲突的输出。解决方案：设置明确的接口契约，定义每个 Agent 的输入输出格式；引入「仲裁者」角色解决冲突。

错误传播：上游 Agent 的错误会传播到下游。解决方案：每个节点加入验证步骤，审核不通过则回退到上一节点重新执行。

成本控制：多 Agent 协作的 token 消耗是单 Agent 的数倍。解决方案：根据任务复杂度动态选择 Agent 数量；简单任务使用单 Agent，复杂任务才启用多 Agent。

多 Agent 协作的未来

随着 AI 能力的提升，多 Agent 协作正在从实验走向生产。未来的 Agent 团队将更加灵活——Agent 可以动态加入或退出，根据任务需求自动调整团队规模。Agent 之间甚至可以互相「雇佣」，形成去中心化的 AI 协作网络。掌握多 Agent 协作，就是掌握了未来 AI 应用开发的核心竞争力。