第3章分工与协调¶

学习目标¶

掌握多 Agent 系统中资源分配与任务委派的核心策略
理解层级化（Hierarchical）协调与动态角色分配的实现方式
能够设计冲突解决机制，避免 Agent 间推诿与死锁
使用 Python 实现基于主管 Agent 的任务分配与协调
识别分工协调中的常见反模式并提出优化方案

问题导入¶

某研发团队用三个 Agent（前端、后端、测试）协作交付功能。初始设计让"产品经理 Agent"把任务平均分给三者，结果前端 Agent 任务堆积、后端 Agent 空闲、测试 Agent 在代码没写完时就开始测试，反复报错。更糟的是，前后端 Agent 对"谁负责接口联调"产生分歧，互相推诿。问题根源在于：分工不是简单均分，而是要根据能力、负载、依赖关系动态分配；协调也不是发任务就完事，还要解决冲突与依赖。本章讲解如何科学分工、高效协调。

核心概念¶

1. 资源分配策略¶

多 Agent 系统中的"资源"包括：算力（LLM 调用配额）、工具（API 限流）、注意力（哪些任务优先）。常见分配策略：

策略	描述	适用场景
平均分配	任务均分给所有 Agent	Agent 能力相当、任务粒度均匀
能力匹配	按 Agent 专长分配	Agent 各有所长（如法务/财务）
负载均衡	分给当前最空闲的 Agent	同类 Agent 多副本场景
优先级调度	高优先级任务先执行	有 SLA 差异的业务
拍卖/竞标	Agent 自愿"认领"任务	去中心化、自主性强

2. 任务委派机制¶

任务委派（Delegation）指一个 Agent 把部分子任务交给另一个 Agent 完成。常见模式：

主管委派（Manager Delegation）：主管 Agent 拆解任务后分派给执行 Agent，回收结果汇总。CrewAI 的 Hierarchical 流程即此类。
Handoff 转交：Agent A 判断任务更适合 Agent B，把控制权连同上下文一起转交。OpenAI Agents SDK 的 Handoff 是典型。
递归委派：主管把大任务拆成中任务，中任务负责人再拆成小任务，形成多级树状委派。

委派的关键约束： - 明确委派的输入输出契约（避免接收方不知道要做什么）。 - 设置最大委派深度（防止无限递归）。 - 保留委派链路（谁委派给谁），便于追溯与回滚。

3. 层级化（Hierarchical）协调¶

层级化协调模拟人类组织架构：

总监 Agent（战略层）
├── 经理 Agent A（战术层）
│   ├── 执行 Agent A1
│   └── 执行 Agent A2
└── 经理 Agent B（战术层）
    ├── 执行 Agent B1
    └── 执行 Agent B2

优点：分而治之、可控性好、单点故障可隔离到子树。
缺点：层级过深会引入延迟与信息失真；高层 Agent 成为瓶颈。
适用：大型任务分解、多团队协作、需要分级审批的场景。

4. 动态角色分配¶

静态角色（启动时固定）简单但僵化。动态角色分配让 Agent 在运行时根据任务与状态调整角色：

基于能力声明：每个 Agent 声明自己的能力标签（如 ["python", "testing"]），主管按标签匹配。
基于历史表现：记录每个 Agent 完成同类任务的成功率与耗时，优先分给表现好的。
基于当前负载：优先分给队列最短的 Agent。
竞标机制：主管发布任务，Agent 自评契合度"投标"，主管择优录取。

5. 冲突解决机制¶

Agent 协作中常见冲突：

冲突类型	表现	解决方式
职责冲突	两 Agent 都认为自己该做某事	明确职责边界 + 主管裁决
依赖冲突	B 等 A 的输出，但 A 卡住	设置超时 + 降级方案
结果冲突	两 Agent 给出矛盾结论	引入评审 Agent 投票（见第4章）
资源冲突	多 Agent 抢占同一工具/限流	资源锁 + 排队
死循环	A 转交给 B，B 又转回 A	最大轮次 + 主管强制终止

通用冲突解决原则：①预防优于解决（明确契约与边界）；②设置超时与最大轮次兜底；③引入第三方裁决（主管或评审 Agent）；④记录冲突日志用于改进。

6. 协同算法示例¶

合同网协议（Contract Net Protocol, CNP）：经典的去中心化任务分配算法。 1. 主管 Agent 发布任务通告（announcement）。 2. 各执行 Agent 评估自身能力，提交标书（bid）。 3. 主管评估标书，选出最合适的 Agent 并"授标"（award）。 4. 中标 Agent 执行任务并返回结果。

CNP 适合异构 Agent 池、任务难以预先分配的场景。

原理图¶

flowchart TB
    User[用户任务] --> Manager[主管 Agent\n任务分解]

    Manager -->|能力匹配| A1[前端 Agent]
    Manager -->|负载均衡| A2[后端 Agent]
    Manager -->|优先级| A3[测试 Agent]

    A1 -->|依赖等待| Wait[依赖检查]
    A2 -->|产出接口契约| Wait
    Wait -->|就绪| A3

    A1 -.冲突.-> A2
    A1 -.冲突.-> Manager
    A2 -.冲突.-> Manager
    Manager -->|裁决| Resolve[冲突解决]

    A3 --> Review[评审/汇总]
    Review --> Final[最终交付]

    subgraph 动态角色分配
        D1[能力标签库] --> D2[匹配引擎]
        D2 --> D3[动态指派]
    end

图：多 Agent 分工协调流程，含能力匹配、依赖检查、冲突解决与动态角色分配

示例代码¶

实现一个基于能力匹配与负载均衡的主管 Agent 任务分配器。

import time
from dataclasses import dataclass, field
from collections import defaultdict
from typing import Callable


@dataclass
class WorkerAgent:
    """执行 Agent，声明能力标签并维护任务队列。"""
    agent_id: str
    capabilities: list  # 能力标签，如 ["python", "frontend"]
    queue_size: int = 0
    success_rate: float = 1.0  # 历史成功率
    handler: Callable = None

    def match_score(self, required_caps: list) -> float:
        """计算能力匹配度：匹配标签数 / 所需标签数 * 成功率 / (1+负载)。"""
        if not required_caps:
            score = 1.0
        else:
            matched = len(set(required_caps) & set(self.capabilities))
            score = matched / len(required_caps)
        # 负载惩罚：队列越长得分越低
        load_penalty = 1.0 / (1 + self.queue_size)
        return score * self.success_rate * load_penalty

    def execute(self, task):
        if self.handler:
            self.queue_size += 1
            try:
                result = self.handler(task)
                return result
            finally:
                self.queue_size -= 1


@dataclass
class Task:
    task_id: str
    description: str
    required_caps: list  # 所需能力标签
    depends_on: list = field(default_factory=list)  # 依赖的前置任务 ID


class ManagerAgent:
    """主管 Agent：负责任务分解、分配、依赖检查与冲突解决。"""

    def __init__(self, workers: list, max_rounds: int = 10):
        self.workers = {w.agent_id: w for w in workers}
        self.max_rounds = max_rounds
        self.results = {}
        self.conflict_log = []

    def assign(self, task: Task) -> str:
        """按能力匹配 + 负载均衡选出最合适的 Worker。"""
        # 检查依赖是否已完成
        for dep in task.depends_on:
            if dep not in self.results:
                raise RuntimeError(f"任务 {task.task_id} 依赖 {dep} 未完成")

        # 计算每个 Worker 的得分，选最高
        scored = [(w.agent_id, w.match_score(task.required_caps))
                  for w in self.workers.values()]
        scored.sort(key=lambda x: x[1], reverse=True)
        winner_id = scored[0][0]
        print(f"[主管] 任务 {task.task_id} → {winner_id} "
              f"(得分 {scored[0][1]:.2f})")
        return winner_id

    def delegate(self, task: Task) -> object:
        """委派任务并回收结果。"""
        winner = self.workers[self.assign(task)]
        try:
            result = winner.execute(task)
            self.results[task.task_id] = result
            return result
        except Exception as e:
            self.conflict_log.append({"task": task.task_id, "error": str(e)})
            # 降级：换一个 Worker 重试
            for w in self.workers.values():
                if w.agent_id != winner.agent_id and w.match_score(task.required_caps) > 0:
                    print(f"[主管] 降级重试：{task.task_id} → {w.agent_id}")
                    result = w.execute(task)
                    self.results[task.task_id] = result
                    return result
            raise

    def resolve_conflict(self, task_id, proposals):
        """多 Agent 给出矛盾结论时的裁决：选成功率最高 Agent 的方案。"""
        proposals.sort(key=lambda p: self.workers[p["agent_id"]].success_rate,
                       reverse=True)
        winner = proposals[0]
        self.conflict_log.append({"task": task_id, "type": "result_conflict",
                                  "resolved": winner["agent_id"]})
        return winner["result"]


# 使用示例
if __name__ == "__main__":
    def handle_frontend(task): return f"前端代码: {task.description}"
    def handle_backend(task): return f"后端API: {task.description}"
    def handle_test(task): return f"测试用例: {task.description}"

    workers = [
        WorkerAgent("fe-001", ["frontend", "html"], handler=handle_frontend),
        WorkerAgent("be-001", ["python", "backend"], handler=handle_backend),
        WorkerAgent("qa-001", ["testing", "python"], handler=handle_test),
    ]

    manager = ManagerAgent(workers)

    t1 = Task("T1", "实现登录页面", ["frontend"])
    t2 = Task("T2", "实现登录接口", ["python", "backend"])
    t3 = Task("T3", "编写登录测试", ["testing"], depends_on=["T1", "T2"])

    r1 = manager.delegate(t1)
    r2 = manager.delegate(t2)
    r3 = manager.delegate(t3)
    print("结果：", r1, r2, r3)

企业案例¶

案例 1：软件研发流水线的层级协调¶

某科技公司用 LangGraph 构建"总监→经理→执行"三级 Agent 体系。总监 Agent 把"上线一个功能"拆为前端、后端、测试三个子任务，分派给三位经理 Agent；经理 Agent 再细分为编码、自测、提 PR 等子任务给执行 Agent。设置最大委派深度为 3 层，避免无限递归。流水线让 5 人团队的交付效率提升 40%。

案例 2：客服中心的动态角色分配¶

某客服中心部署 20 个同类 Agent，初期静态分配用户导致部分 Agent 过载、部分空闲。引入动态分配后，新会话按"当前队列最短 + 历史成功率最高"加权选择 Agent。改造后平均响应时间下降 35%，Agent 利用率从 60% 提升到 85%。

案例 3：合同网协议在物流调度中的应用¶

某物流平台用 CNP 算法调度配送 Agent。调度中心发布"取件任务"，周边配送 Agent 根据距离与当前负载投标，调度中心择优授标。相比固定派单，CNP 让配送 Agent 自主认领，整体配送时长缩短 18%。

常见问题¶

Q1：如何避免 Agent 间互相推诿？ A：①明确每个 Agent 的职责边界与能力标签；②由主管 Agent 强制裁决归属；③设置最大转交轮次（如 3 次）；④记录转交链路用于事后分析。

Q2：层级化协调的层级多深合适？ A：通常 2-3 层为宜。过深会引入延迟、信息失真与调试困难。可通过"扁平化 + 小组自治"在可控性与效率间取得平衡。

Q3：动态角色分配会不会让系统难以预测？ A：会牺牲部分可预测性，但能提升利用率与适应性。建议保留分配日志、设置白名单（敏感任务只给指定 Agent）、关键场景用静态分配保可控。

Q4：依赖冲突怎么处理？ A：①任务建模时显式声明 depends_on；②调度器检查依赖就绪后再分派；③为前置任务设超时，超时后降级（用默认值或人工介入）。

面试题¶

1. 设计一个支持动态角色分配的多 Agent 调度系统。 答题要点：Agent 声明能力标签；调度器按"能力匹配 × 负载均衡 × 历史成功率"打分选 Agent；记录分配日志；设置最大重试与降级方案；关键任务用白名单保可控。

2. 简述合同网协议（CNP）的流程及其适用场景。 参考答案：主管发布任务通告→执行 Agent 评估能力并投标→主管评估标书授标→中标 Agent 执行并返回结果。适用于异构 Agent 池、任务难以预先分配、Agent 自主性强的去中心化场景。

3. 多 Agent 系统常见的冲突类型有哪些？如何预防与解决？ 参考答案：职责冲突（明确边界+主管裁决）、依赖冲突（显式声明依赖+超时降级）、结果冲突（评审投票）、资源冲突（资源锁+排队）、死循环（最大轮次+强制终止）。预防优于解决：契约清晰、边界明确、超时兜底。

练习题¶

分工实操：用本章的 ManagerAgent 实现一个"内容生产"流水线（选题→调研→撰写→校对），为每个任务定义所需能力标签并分配给合适的 Agent。
冲突设计：构造一个"前后端 Agent 对接口契约有分歧"的冲突场景，编写主管 Agent 的裁决逻辑（提示：可引入契约模板或评审 Agent）。
思考题：如果一个任务的依赖任务长时间不完成，系统应如何处理？设计一个超时降级方案。

小结¶

分工与协调是多智能体系统的"管理艺术"。能力匹配、负载均衡、优先级调度是核心分配策略；层级化协调适合大型任务分解；动态角色分配提升利用率但需以可观测性为代价。冲突不可避免，预防优于解决——明确契约、设置超时、引入裁决、记录链路是工程落地的四大要点。下一章将讲解如何通过投票与共识机制让多 Agent 做出更可靠的决策。延伸阅读：合同网协议 CNP、论文《Hierarchical Multi-Agent Systems》。

第3章 分工与协调¶