05.2 — 通信与协调
定位: 多 Agent 系统的"神经网络"——理解 Agent 之间如何交换信息、达成共识、解决冲突 面试高频度: ⭐⭐⭐⭐ 考查方式: 设计题（设计 Agent 间通信协议）、对比题（通信模式的优劣）、策略题（冲突解决方案）
一、这是什么？为什么需要它？
是什么
通信 是 Agent 之间交换信息的机制。协调 是多个 Agent 调整行为以达成共同目标的过程。
为什么需要通信与协调？
没有通信的多 Agent = 多个独立工作的单 Agent
Agent A 不知道 Agent B 做了什么
Agent B 重复 Agent A 的工作
结果冲突时无人解决
有通信的多 Agent = 一个协作团队
Agent A 告诉 B "我已完成数据分析"
Agent B 基于 A 的结果继续生成报告
意见不同时通过讨论达成共识核心洞察：通信的质量决定了多 Agent 系统协作的上限。
二、原理拆解
2.1 两种通信范式
① 消息传递（Message Passing）
Agent 直接向特定 Agent 发送消息。
Agent A ──→ [Message] ──→ Agent B
消息结构:
{
"from": "Agent_A",
"to": "Agent_B",
"type": "request" | "response" | "broadcast",
"content": "...",
"metadata": {"timestamp": "...", "priority": "high"}
}| 模式 | 说明 | 适用场景 |
|---|---|---|
| 点对点 | A 直接发给 B | 任务分配、结果反馈 |
| 广播 | A 发给所有 Agent | 状态变更通知 |
| 发布-订阅 | A 发布到主题，订阅者接收 | 事件驱动协作 |
② 共享黑板（Shared Blackboard）
Agent 通过共享空间交换信息，不直接通信。
┌──────────────────────────┐
│ Blackboard │
│ │
│ Task: "写市场报告" │
│ ├── 数据: [A收集完成] │
│ ├── 分析: [B分析中...] │
│ └── 报告: [待生成] │
│ │
│ Decisions: │
│ ├── 数据源：使用最新财报│
│ └── 格式：PDF输出 │
└──────────────────────────┘
▲ ▲ ▲
│ │ │
┌───┘ │ └───┐
▼ ▼ ▼
┌────┐ ┌────┐ ┌────┐
│ A │ │ B │ │ C │
│数据│ │分析│ │报告│
└────┘ └────┘ └────┘2.2 通信模式对比
| 维度 | 消息传递 | 共享黑板 |
|---|---|---|
| 耦合度 | 高（需知道对方） | 低（只读写黑板） |
| 效率 | 高（直接送达） | 中（需要查黑板） |
| 可追溯性 | 好（消息有记录） | 中（变更可能覆盖） |
| 扩展性 | 差（N个Agent需N^2连接） | 好（加Agent不影响） |
| 调试难度 | 中 | 高（谁改了什么） |
| 典型框架 | AutoGen | CrewAI |
2.3 冲突解决机制
多 Agent 必然出现意见分歧，需要冲突解决策略：
| 策略 | 做法 | 适用场景 |
|---|---|---|
| 投票 | 多数决 | 有明确选项时 |
| 仲裁 | 指定仲裁 Agent 做最终决定 | 需要专业判断 |
| 辩论 | 双方给出论据，LLM 判断 | 需要充分讨论 |
| 优先级 | 预设角色优先级 | 有明确层级 |
| 合并 | 取两方方案的共同点 | 双方都有价值 |
python
class ConflictResolver:
"""冲突解决器"""
def resolve(self, agents: list, opinions: dict, strategy: str) -> dict:
if strategy == "vote":
return self._vote(opinions)
elif strategy == "arbitrate":
return self._arbitrate(agents, opinions)
elif strategy == "debate":
return self._debate(agents, opinions)
def _debate(self, agents: list, opinions: dict) -> dict:
"""辩论模式：让双方辩论，LLM 裁判"""
debate_log = []
for agent in agents:
debate_log.append(f"{agent.name}: {opinions[agent.name]}")
verdict = llm.call(f"""
以下 Agent 对同一问题有不同意见：
{'\n'.join(debate_log)}
请分析各方论据，做出最终决定。
要求：
1. 指出各方论据的优缺点
2. 给出最终决定
3. 解释理由
""")
return {"verdict": verdict, "method": "debate"}2.4 通信协议设计
python
# Agent 通信协议
class AgentMessage:
def __init__(self, sender: str, receiver: str = None,
msg_type: str = "message", content: str = None):
self.sender = sender
self.receiver = receiver # None = broadcast
self.msg_type = msg_type # task_assign / result / question / status
self.content = content
self.timestamp = now()
self.msg_id = generate_id()
class CommunicationProtocol:
"""Agent 通信协议"""
def __init__(self):
self.message_queue = []
self.blackboard = {}
def send(self, message: AgentMessage):
self.message_queue.append(message)
def receive(self, agent_name: str) -> list:
"""Agent 收取发给自己的消息"""
inbox = [
msg for msg in self.message_queue
if msg.receiver == agent_name or msg.receiver is None
]
self.message_queue = [
msg for msg in self.message_queue
if msg not in inbox
]
return inbox
def write_to_board(self, key: str, value: any, author: str):
self.blackboard[key] = {
"value": value,
"author": author,
"timestamp": now()
}
def read_from_board(self, key: str) -> any:
entry = self.blackboard.get(key)
return entry["value"] if entry else None三、图解全景
┌─────────────────────────────────────────────────────────┐
│ 多 Agent 通信与协调全景图 │
│ │
│ 通信层 │
│ ┌───────────────────────────────────────────┐ │
│ │ ┌──────────┐ ┌──────────┐ │ │
│ │ │ 消息传递 │ ←──→ │ 共享黑板 │ │ │
│ │ │ 点对点 │ │ 读写 │ │ │
│ │ │ 广播 │ │ 订阅 │ │ │
│ │ │ 发布-订阅 │ │ 通知 │ │ │
│ │ └──────────┘ └──────────┘ │ │
│ └──────────────────┬────────────────────────┘ │
│ │ │
│ 协调层 │ │
│ ┌──────────────────▼────────────────────────┐ │
│ │ │ │
│ │ 冲突解决 → 投票/仲裁/辩论/优先级/合并 │ │
│ │ 共识机制 → 一致同意/多数决/加权决定 │ │
│ │ 同步机制 → 锁/信号量/屏障 │ │
│ │ │ │
│ └──────────────────┬────────────────────────┘ │
│ │ │
│ 应用层 │ │
│ ┌──────────────────▼────────────────────────┐ │
│ │ │ │
│ │ 任务分配 → 谁做什么 │ │
│ │ 进度同步 → 进行到哪了 │ │
│ │ 结果合并 → 怎么组合各自的结果 │ │
│ │ │ │
│ └────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘四、实战验证
python
# 多 Agent 协作：辩论式代码审查
class CodeReviewDebate:
"""多 Agent 辩论式代码审查"""
def __init__(self):
self.comm = CommunicationProtocol()
self.reviewers = [
{"name": "安全专家", "focus": "安全漏洞"},
{"name": "性能专家", "focus": "性能问题"},
{"name": "代码风格", "focus": "可读性和规范"},
]
def review(self, code: str) -> dict:
# Phase 1: 各自审查
for reviewer in self.reviewers:
opinion = llm.call(
f"你是一个{reviewer['name']}，审查以下代码：\n{code}\n"
f"请关注{reviewer['focus']}，列出问题。"
)
self.comm.write_to_board(
f"review_{reviewer['name']}", opinion, reviewer['name']
)
# Phase 2: 讨论冲突
all_opinions = [
self.comm.read_from_board(f"review_{r['name']}")
for r in self.reviewers
]
final_result = llm.call(f"""
以下是三位审查专家的意见：
安全专家：{all_opinions[0]}
性能专家：{all_opinions[1]}
代码风格：{all_opinions[2]}
请整合为一个最终的审查报告，如果有冲突请仲裁解决。
""")
return {"final_review": final_result}
# 测试
debate = CodeReviewDebate()
result = debate.review("""
def process(data):
result = []
for i in range(len(data)):
result.append(data[i] * 2)
return result
""")
print(result["final_review"])五、面试视角
| 追问 | 答案要点 |
|---|---|
| 消息传递 vs 共享黑板的区别？ | 消息传递直接耦合，适合确定性的任务分配。共享黑板解耦，适合灵活协作。选型看：Agent 间是否知道彼此存在、是否需要历史追溯、扩展性要求 |
| 多 Agent 冲突怎么解决？ | 投票（多数决）、仲裁（指定裁判）、辩论（充分讨论后 LLM 判断）、优先级（预设规则）。复杂场景推荐辩论，简单场景推荐仲裁 |
| Agent 间通信的安全考虑？ | 认证（确认消息发送者身份）、授权（Agent 可以读写什么）、加密（敏感信息）、审计（记录所有通信）、防注入（Agent 输出的恶意内容） |
| 如何避免 Agent "吵架"（无限辩论）？ | 设置最大讨论轮次、超时机制（超时自动仲裁）、明确决策权（什么级别的争议由谁决定）、预设优先级规则（某些 Agent 的意见权重更高） |
📚 相关链接
- **多智能体架构** — 通信发生的架构环境
- **角色分配与编排** — 谁和谁通信
- **AutoGen** — AutoGen 的多 Agent 对话
- **CrewAI** — CrewAI 的协作模式
- ← 返回 **多智能体系统索引**