6.1 智能体间通信协议设计

在多智能体系统中，通信协议 是系统的神经网络。它决定了智能体之间如何交换信息、协调行动，以及在冲突时如何达成共识。

本节探讨多智能体通信的主流模式——从自由的自然语言交互到严格的结构化协议，帮助你为不同场景选择合适的通信策略。

6.1.1 为什么通信协议很重要

在单智能体系统里，很多“沟通问题”只会表现为回答不稳定；但在多智能体系统里，沟通不良会被放大为重复劳动、互相覆盖、甚至死锁。通信协议的价值在于把“可交流”变成“可协作”：让信息具备明确的语义、可追踪的状态与可执行的交接。

6.1.2 无协议的混乱

想象三个智能体在没有规则的情况下协作：

• 智能体 A: "我觉得应该先搜索一下。"

• 智能体 B: "好的，我来搜索。" (开始搜索)

• 智能体 C: "我也来搜索不同的关键词。" (重复搜索)

• 智能体 A: "等等，我刚才已经改变主意了..." (信息滞后)

结果：重复劳动、信息混乱、Token 浪费。

6.1.3 协议的价值

一个良好的通信协议应该解决：

问题	解决方案
语义歧义	结构化消息格式
乱序混乱	顺序控制机制
信息冗余	共享状态管理
死锁	超时和回退机制

6.1.4 自然语言通信

这是智能体最"原生"的交流方式——就像两个人类在 Slack 上聊天。

6.1.5 工作方式

具体示例如下：

智能体 A（研究员）： 
"我刚搜索了关于 Agentic AI 的最新论文，发现有三个重要趋势..."

智能体 B（写作者）：
"收到！我根据你的发现开始写第一段了。能把论文链接发我吗？"

智能体 A（研究员）：
"好的，链接在这里：[…]. 另外我还发现了一个有意思的案例..."

优点

1. 通用性强：不需要预定义 Schema，任何信息都能表达。

2. 可解释性好：人类可以直接阅读对话日志，方便调试。

3. 灵活应变：面对意外情况可以自由表达。

缺点

1. 语义歧义：

   • A 说 "检查代码"，B 可能理解为 "静态分析" 或 "运行测试"。

2. 废话多：

   • LLM 天生礼貌："非常感谢您的分享！这真的太有帮助了..."

   • 这些客套话是纯粹的 Token 浪费。

3. 解析困难：

   • 从自然语言中提取关键信息需要额外的 LLM 调用。

6.1.6 适用场景

• 创意性任务（头脑风暴、内容创作）

• 探索性对话

• 需要高度灵活性的场景

6.1.7 结构化通信

为了解决自然语言的缺陷，工程界引入了"合约"——智能体之间通过 JSON、Protobuf 等结构化格式交换信息。

6.1.8 JSON 消息协议

设计原则：在系统提示词中强制要求输出格式。

SYSTEM_PROMPT = """
你是产品经理智能体 (PM Agent)。当你需要向开发团队传递需求时，必须使用以下 JSON 格式：

{
  "message_type": "task_assignment",
  "from": "你的角色名",
  "to": "目标角色名",
  "payload": {
    "task_id": "唯一任务ID",
    "description": "任务描述",
    "priority": "high/medium/low",
    "deadline": "ISO格式时间",
    "constraints": ["约束条件列表"]
  }
}

严禁使用自由文本。
"""

实际消息示例：

{
  "message_type": "task_assignment",
  "from": "PM_Agent",
  "to": "Developer_Agent",
  "payload": {
    "task_id": "TASK-101",
    "description": "实现用户登录功能，包括 OAuth 集成",
    "priority": "high",
    "deadline": "<date-time>",
    "constraints": [
      "按钮颜色必须为蓝色",
      "支持常见 OAuth 提供方"
    ]
  }
}

6.1.9 消息类型枚举

定义有限的消息类型，避免歧义：

from enum import Enum

class MessageType(Enum):
    TASK_ASSIGNMENT = "task_assignment"    # 分配任务
    TASK_COMPLETE = "task_complete"        # 任务完成
    QUERY = "query"                        # 提问
    RESPONSE = "response"                  # 回复
    ERROR = "error"                        # 错误报告
    STATUS_UPDATE = "status_update"        # 状态更新
    APPROVAL_REQUEST = "approval_request"  # 请求批准
    APPROVAL_RESPONSE = "approval_response" # 批准结果

6.1.10 状态机驱动的对话流

可以用状态机严格控制对话流转：

from langgraph.graph import StateGraph

# 定义状态

class ConversationState(Enum):
    PLANNING = "planning"
    CODING = "coding"
    REVIEW = "review"
    TESTING = "testing"
    DONE = "done"

# 定义合法的状态转换

VALID_TRANSITIONS = {
    "planning": ["coding"],
    "coding": ["review"],
    "review": ["coding", "testing"],  # 可以打回重写或进入测试
    "testing": ["coding", "done"],     # 可以打回修复或完成
}

def validate_transition(current: str, next_state: str) -> bool:
    return next_state in VALID_TRANSITIONS.get(current, [])

6.1.11 共享黑板模式

与其让智能体像传声筒一样传递大量信息（导致上下文爆炸），不如共用一块"黑板"。

6.1.12 架构设计

具体示例如下：

图 6-1：共享黑板架构

代码实现

具体示例如下：

from typing import Dict, Any, List
import threading

class SharedBlackboard:
    def __init__(self):
        self._data: Dict[str, Any] = {}
        self._lock = threading.Lock()
        self._subscribers: Dict[str, List[callable]] = {}
    
    def write(self, key: str, value: Any, author: str):
        """写入数据并通知订阅者"""
        with self._lock:
            self._data[key] = {
                "value": value,
                "author": author,
                "timestamp": datetime.now()
            }
        self._notify(key)
    
    def read(self, key: str) -> Any:
        """读取最新数据"""
        with self._lock:
            return self._data.get(key, {}).get("value")
    
    def subscribe(self, key: str, callback: callable):
        """订阅某个 key 的变化"""
        if key not in self._subscribers:
            self._subscribers[key] = []
        self._subscribers[key].append(callback)
    
    def _notify(self, key: str):
        for callback in self._subscribers.get(key, []):
            callback(key, self.read(key))

优势

1. 信息解耦：智能体不需要知道消息来自谁，只关心黑板上的数据。

2. 节省 Token：不需要在消息中重复背景信息。

3. 一致性：所有智能体看到同一个"世界状态"（单一真实来源，Single Source of Truth）。

混合模式：最佳实践

实践中，最佳方案通常是 混合使用——智能体内部用自然语言思考，对外输出用结构化格式。

class HybridAgent:
    def process(self, input_data: dict) -> dict:
        # 1. 内部思考（自然语言思维链 CoT）

        thought = self.llm.generate(f"""
        分析任务：{input_data['task']}
        
        思考过程：
        1. 首先我需要...
        2. 然后...
        3. 最后...
        """)
        
        # 2. 对外输出（结构化 JSON）

        output = self.llm.generate(f"""
        基于思考结果，生成标准输出：
        {thought}
        
        必须输出为 JSON 格式：
        {{"action": "...", "result": "...", "next_step": "..."}}
        """)
        
        return json.loads(output)

6.1.13 智能体间协调的失败模式

即使有了良好的通信协议，多智能体系统仍可能在协调层面失败。实践中，即使有了良好的通信协议，多智能体系统仍可能在协调层面失败。下面列出一些常见模式：

失败模式	描述	典型症状
对话重置	对话历史意外清空或被截断	智能体突然“忘记”之前的讨论
未请求澄清	面对歧义时擅自假设	基于错误假设继续执行
任务偏离	逐渐偏离原始目标	讨论越来越跑题
信息隐瞒	关键信息未传递给队友	其他智能体缺少必要上下文
忽略他人输入	无视队友的建议或反馈	单方面推进决策
推理-行动不匹配	思考过程与实际行动矛盾	说一套做一套

案例：推理-行动不匹配

这是最常见的协调失败之一。智能体在思考过程中表达了正确的意图，但实际输出却与之矛盾：

[智能体思考]: "用户要求创建一个 Wordle 游戏，不使用固定词库..."
[智能体行动]: 生成了包含固定词库的代码

问题诊断：推理-行动不匹配 (FM-2.6)

预防措施

1. 对话历史持久化：使用外部存储保存对话状态，避免 FM-2.1

2. 显式澄清机制：在 Prompt 中要求智能体遇到歧义时必须提问

3. 目标锚定：定期在消息中重申任务目标，防止 FM-2.3

4. 输出验证：增加独立的验证智能体检查推理与行动的一致性

6.1.14 小结

通信模式	适用场景	优点	缺点
自然语言	创意任务	灵活、可读	歧义、冗余
结构化 JSON	工程任务	精确、可解析	需要预定义
状态机	严格流程	可控、可预测	缺乏灵活性
共享黑板	复杂协作	解耦、高效	需要状态管理

核心原则："自然语言思考，结构化交付"。智能体内部可以自由推理，但与队友或系统交互时，务必使用清晰的结构化格式。

---

下一节: 6.2 生成式社会模拟：虚拟小镇解析