5.2 协作协议与策略

5.2 协作协议与策略 — Agent智能体开发实战

本节导读：5.1节解决了智能体之间"怎么说得上话"的问题，本节回答"按什么规则协作"——任务委派协议、协商机制、四种经典协作模式及其工程实现，让一群智能体从"能通信"升级为"有秩序地干活"。

学习目标

• 理解协作协议在多智能体系统中的定位与设计要素

• 掌握任务委派（Contract Net）与结果交接（Handoff）两大核心协议

• 学会按任务形态选择协作模式（流水线/并行/分组/监督者）

• 能用Python实现一个可扩展的协作协议框架

核心概念

从通信到协作：协议的价值

通信机制（5.1节）只保证消息可达；协作协议在此基础上规定谁在何时向谁请求什么、对方如何应答、失败如何处理。没有协议的多智能体系统就像没有议事规则的会议室——人人都在说话，决策却永远出不来。

协议的四个设计要素：

两大核心协议

Contract Net（合同网协议）——经典的任务分配协议：

委派者：广播任务公告（ANNOUNCE）
承接者：评估自身能力与负载，投标（BID）或沉默
委派者：择优授予（AWARD），其余落选
承接者：执行并回报结果（REPORT）
​

适合多个能力相近的智能体竞争任务的场景，天然实现负载均衡。

Handoff（交接协议）——现代Agent框架的主流模式：

当前智能体：判断任务超出职责 → 生成交接包（上下文+任务描述+已完成的结论）
目标智能体：接收交接包 → 继续执行 → 结束后可交还
​

适合职责分明、串行流转的场景（如客服→技术支持→退款专员），OpenAI Swarm、LangGraph均以此为核心抽象。

四种协作模式

工程实践的第一课：能用简单模式就不用复杂模式。监督者模式最通用也最贵，很多团队上来就用它，结果为一个两步任务支付了十倍token。

分步实战

步骤 1：定义协议消息与数据结构

from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Optional
import time, uuid

class ProtocolState(Enum):
    ANNOUNCED = "announced"     # 已公告
    BIDDING = "bidding"         # 竞标中
    AWARDED = "awarded"         # 已授予
    RUNNING = "running"         # 执行中
    DONE = "done"               # 已完成
    FAILED = "failed"           # 失败

@dataclass
class TaskContract:
    """任务合同：Contract Net的载体"""
    task_id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])
    description: str = ""
    required_skills: list = field(default_factory=list)
    deadline: float = 0.0
    state: ProtocolState = ProtocolState.ANNOUNCED
    contractor: Optional[str] = None      # 最终承接者
    result: Any = None
    attempts: int = 0                     # 已尝试次数（异常路径用）
    MAX_RETRIES: int = 2

@dataclass
class HandoffPackage:
    """交接包：Handoff协议的载体"""
    from_agent: str = ""
    to_agent: str = ""
    task: str = ""
    context: dict = field(default_factory=dict)   # 已收集的事实
    conclusions: list = field(default_factory=list)  # 已得出的结论
    reason: str = ""        # 为什么交接（审计用）
    ts: float = field(default_factory=time.time)
​

步骤 2：实现Contract Net任务分配

class ContractNetCoordinator:
    """合同网协调者：管理公告→竞标→授予全流程"""

    def __init__(self):
        self.contracts: dict[str, TaskContract] = {}
        self.agents = []                      # 注册的承接者

    def register(self, agent):
        self.agents.append(agent)

    def announce(self, description: str, skills: list) -> TaskContract:
        contract = TaskContract(description=description, required_skills=skills,
                                deadline=time.time() + 300)
        self.contracts[contract.task_id] = contract
        # 广播公告，收集投标
        bids = []
        for agent in self.agents:
            bid = agent.evaluate(contract)    # 各自评估能力与负载
            if bid is not None:
                bids.append(bid)
        return self._award(contract, bids)

    def _award(self, contract, bids):
        if not bids:                          # 无人应标 → 异常路径
            contract.state = ProtocolState.FAILED
            raise RuntimeError(f"任务{contract.task_id}无人承接，需人工介入")
        # 简化策略：技能匹配优先，其次负载最低
        best = min(bids, key=lambda b: (-b.skill_match, b.current_load))
        contract.contractor = best.agent_id
        contract.state = ProtocolState.AWARDED
        return contract

    def report(self, task_id: str, success: bool, result=None):
        c = self.contracts[task_id]
        if success:
            c.state, c.result = ProtocolState.DONE, result
        else:
            c.attempts += 1
            if c.attempts <= c.MAX_RETRIES:   # 重试路径：换人再招标
                c.state = ProtocolState.ANNOUNCED
                return self.announce(c.description, c.required_skills)
            c.state = ProtocolState.FAILED    # 重试耗尽 → 上报
            raise RuntimeError(f"任务{task_id}重试{c.attempts}次仍失败")
​

步骤 3：实现Handoff流转

class HandoffCapableAgent:
    """支持交接的智能体基类"""

    def __init__(self, agent_id: str, skills: list, router):
        self.id = agent_id
        self.skills = skills
        self.router = router                  # 负责投递交接包

    def can_handle(self, task: str) -> bool:
        return any(s in task for s in self.skills)

    def handoff(self, to_agent: str, task: str, context: dict,
                conclusions: list, reason: str):
        pkg = HandoffPackage(from_agent=self.id, to_agent=to_agent,
                             task=task, context=context,
                             conclusions=conclusions, reason=reason)
        return self.router.route(pkg)         # 交给下一个智能体继续
​

步骤 4：用监督者模式组装完整系统

class SupervisorAgent:
    """监督者：接收用户任务 → 拆解 → 选择协议分发 → 汇总"""

    def __init__(self, contract_net, handoff_agents: dict):
        self.net = contract_net               # 竞争性子任务走合同网
        self.specialists = handoff_agents     # 职责性子任务走交接

    def execute(self, user_task: str):
        plan = self.decompose(user_task)      # LLM生成任务拆解
        results = {}
        for sub in plan:
            if sub.competitive:               # 多个通用执行者竞争
                c = self.net.announce(sub.description, sub.skills)
                results[sub.id] = c.result
            else:                             # 按职责交给专家
                agent = self.specialists[sub.role]
                results[sub.id] = agent.run(sub, results)
        return self.synthesize(results)       # 汇总产出最终答案
​

监督者自身的提示词要点：①明确"只能拆解与汇总，不得自己执行"；②每个子任务必须写清验收标准；③限制拆解深度（如≤3层）防止无限递归。

完整示例：一个内容生产管线

任务："撰写一篇关于Serverless的技术周报"。

监督者拆解：
  sub1[竞争型] 搜集本周资讯 → ContractNet在3个检索Agent间招标
  sub2[职责型] 撰写周报草稿 → Handoff给写作Agent（携带sub1结果）
  sub3[职责型] 事实核查 → Handoff给审校Agent（携带sub2草稿）

协议行为：
  sub1无人应标 → FAILED → 降级：使用缓存的上次资讯并打标记
  sub3发现2处事实存疑 → conclusions写入交接包 → 交回写作Agent修订
​

两条协议各司其职：可并行的检索走竞标，有依赖的写作/审校走交接，异常路径都有出口。

常见问题 FAQ

Q1：Contract Net和Handoff怎么选？
一句话：能力可互换用竞标，职责有边界用交接。三个通用"打工人"分活儿用Contract Net；"这事必须财务Agent来办"用Handoff。混合系统两者并存很正常。

Q2：智能体互相交接形成死循环怎么办？
三道保险：①交接包里带 hop_count，超过上限强制终止并上报；②交接目标由规则/监督者裁决而非当前智能体随意指定；③同一任务禁止交回刚交来的来源（防A→B→A乒乓）。

Q3：监督者拆解任务质量差怎么办？
监督者的拆解能力=底层LLM的规划能力+提示词质量。改进顺序：先在提示词里给出拆解模板与验收标准示例；仍不行就引入"反思步"（拆解后先自评一遍再分发）；极端情况退化为人工提供任务图。

Q4：并行子任务的结果怎么合并？
在协议层显式定义合并策略：①同质结果（多路检索）→去重+相关度排序；②异质结果（各写一章）→按模板拼接+统一文风润色；③冲突结果→交给5.3节的冲突解决机制裁决。

Q5：协作深度是不是越深越好？
不是。每层协作都引入延迟、token成本与错误传播概率。经验值：子任务数≤7、协作深度≤3。超过说明任务本身应该被重新建模，而不是堆更多智能体。

最佳实践与避坑

• 避坑一：协议没有异常路径。所有"理想流程"之外的情况（超时、拒标、执行失败）必须显式设计出口，否则一次异常就让整个系统挂起；

• 避坑二：交接包只传任务不传上下文。下游智能体被迫重新收集信息，成本翻倍且结论可能不一致。交接包必须包含已确认的事实与结论；

• 实践：每次协议交互写入结构化日志（任务ID、参与方、状态迁移、耗时），协作系统的可观测性比单体系统更重要；

• 实践：先用最简单的模式做通端到端，再按实测瓶颈增加复杂度——多数系统最终只需要"监督者+两个专家"。

本节小结

本节把协作拆解为"协议×模式"两个正交维度：Contract Net解决竞争性分工，Handoff解决职责性流转；四种模式中按"够用就好"原则选择。协议设计的重心永远在异常路径与上下文传递上。下一节处理协作的必然代价——冲突与一致性。

延伸阅读

• 相关章节：5.1 多智能体通信机制、5.3 冲突解决与一致性、5.4 多智能体系统实现

• LangGraph多智能体文档

• Contract Net Protocol经典论文