智能体通信协议
在多智能体系统中，Agent之间的通信是实现协作与协调的基础。通信协议定义了Agent之间交换信息的格式、语义和规则，决定了整个系统能否高效、可靠地运作。本文将深入探讨智能体通信的核心协议和模式。
一、FIPA-ACL：智能体通信语言标准
FIPA-ACL（Foundation for Intelligent Physical Agents - Agent Communication Language）是由FIPA组织制定的智能体通信语言标准。它是最经典的Agent通信协议，定义了结构化的消息格式和交互协议。
FIPA-ACL消息的核心组成要素包括：
- 言语行为（Illocutionary Force）：消息的意图类型，如请求（request）、通知（inform）、提议（propose）、拒绝（refuse）等。
- 发送者（Sender）：消息的发起Agent。
- 接收者（Receiver）：消息的目标Agent。
- 内容（Content）：消息的具体内容，通常用逻辑语言（如SL表达式）描述。
- 对话标识（Conversation ID）：用于关联同一对话中的多条消息。
- 本体（Ontology）：消息中使用的术语和概念的定义域。
# FIPA-ACL 消息格式示例
message = {
"performative": "request", # 言语行为：请求
"sender": "agent_planner", # 发送者
"receiver": "agent_executor", # 接收者
"content": '(action: deploy_service\n'
' (params:\n'
' (service: "web_app")\n'
' (replicas: 3)\n'
' (port: 8080)))',
"conversation_id": "conv_20260101_001",
"ontology": "deployment_ops",
"protocol": "fipa-request", # 交互协议
"language": "SL", # 内容语言
"reply_with": "inform", # 期望的回复类型
}
# 现代Agent框架中的简化版本
class ACLMessage:
"""简化的ACL消息实现"""
PERFORMATIVES = [
"request", "inform", "propose",
"accept", "reject", "query",
"subscribe", "cancel"
]
def __init__(self, performative, sender, receiver, content):
self.performative = performative
self.sender = sender
self.receiver = receiver
self.content = content
self.conversation_id = uuid4().hex[:8]
self.timestamp = datetime.now()
self.in_reply_to = None
def to_dict(self):
return {
"type": self.performative,
"from": self.sender,
"to": self.receiver,
"data": self.content,
"conv_id": self.conversation_id,
"time": self.timestamp.isoformat()
}FIPA-ACL 在现代 Agent 中的演变
虽然严格的FIPA-ACL在现代LLM Agent中使用较少，但其核心思想——结构化的消息格式和明确的言语行为分类——仍然深刻影响着Agent通信设计。现代框架通常使用JSON格式的消息，但保留了performative分类和对话管理的概念。
二、消息传递模式
Agent之间的消息传递有多种模式，每种模式适用于不同的通信需求：
1. 直接消息传递（Direct Message Passing）
Agent之间直接发送消息，发送者需要知道接收者的地址。这是最简单的通信模式，适合Agent数量较少且拓扑结构固定的场景。
2. 请求-响应（Request-Reply）
一个Agent发送请求，等待另一个Agent的响应。这种同步模式适合需要确认执行结果的场景。
3. 异步消息（Asynchronous Messaging）
通过消息队列实现异步通信。发送者将消息放入队列后立即返回，接收者在方便时从队列中取出消息处理。这种模式解耦了发送者和接收者，提高了系统的可扩展性。
class MessageBus:
"""Agent间消息总线"""
def __init__(self):
self.channels = {}
self.subscriptions = {}
self.history = []
def send(self, message: ACLMessage):
"""发送消息到指定Agent"""
self.history.append(message)
receiver = message.receiver
if receiver in self.subscriptions:
for callback in self.subscriptions[receiver]:
callback(message)
def broadcast(self, message: ACLMessage):
"""广播消息给所有Agent"""
self.history.append(message)
for agent_id, callbacks in self.subscriptions.items():
if agent_id != message.sender:
for callback in callbacks:
callback(message)
def subscribe(self, agent_id, callback):
"""Agent订阅消息"""
if agent_id not in self.subscriptions:
self.subscriptions[agent_id] = []
self.subscriptions[agent_id].append(callback)
def query_history(self, conversation_id=None, sender=None):
"""查询消息历史"""
results = self.history
if conversation_id:
results = [m for m in results
if m.conversation_id == conversation_id]
if sender:
results = [m for m in results if m.sender == sender]
return results三、黑板系统
黑板系统（Blackboard System）是一种间接通信机制。Agent不直接互相发送消息，而是通过一个共享的"黑板"来交换信息。每个Agent监控黑板上的变化，当发现与自己相关的信息时，就进行相应的处理并更新黑板。
黑板系统由三个核心组件构成：
- 黑板（Blackboard）：共享的数据结构，存储所有Agent可见的信息。通常按层次组织，从原始数据到高级抽象。
- 知识源（Knowledge Source）：各个专家Agent，每个知识源负责特定领域的问题解决。
- 控制器（Controller）：调度组件，决定下一个应该由哪个知识源来处理黑板上的信息。
class Blackboard:
"""黑板系统实现"""
def __init__(self):
self.data = {} # 黑板数据
self.knowledge_sources = [] # 知识源（Agent）
self.lock = threading.Lock()
def write(self, key, value, source_agent):
"""写入黑板数据"""
with self.lock:
self.data[key] = {
"value": value,
"source": source_agent,
"timestamp": datetime.now()
}
# 通知所有知识源黑板已更新
self._notify_all(key)
def read(self, key):
"""读取黑板数据"""
return self.data.get(key)
def register_knowledge_source(self, ks):
"""注册知识源"""
self.knowledge_sources.append(ks)
def _notify_all(self, changed_key):
"""通知所有知识源"""
for ks in self.knowledge_sources:
if ks.is_interested(changed_key, self.data):
ks.contribute(self.data)
def solve(self, problem):
"""协作求解"""
self.write("problem", problem, "system")
while not self.is_solved():
# 控制器选择最合适的知识源
ks = self.select_next_ks()
if ks:
ks.process(self.data)
else:
break # 没有知识源可以贡献
return self.read("solution")四、发布-订阅模式
发布-订阅（Publish-Subscribe，Pub/Sub）模式是Agent通信中最灵活的模式之一。Agent可以发布特定主题的消息，也可以订阅感兴趣的主题。发布者和订阅者完全解耦，不需要知道对方的存在。
Pub/Sub模式在Agent系统中的优势：
- 松耦合：发布者和订阅者不需要直接引用对方，降低了系统复杂度。
- 可扩展：新增Agent只需订阅感兴趣的主题，不需要修改现有Agent。
- 灵活过滤：基于主题的过滤机制让Agent只接收相关信息，减少信息过载。
- 动态适应：Agent可以在运行时动态订阅或取消订阅主题。
class AgentPubSub:
"""Agent发布-订阅系统"""
def __init__(self):
self.topics = {} # 主题 -> 订阅者列表
self.middlewares = [] # 消息中间件
def publish(self, topic, message, publisher_id):
"""发布消息到主题"""
# 经过中间件处理
for mw in self.middlewares:
message = mw.process(message)
subscribers = self.topics.get(topic, [])
delivery_count = 0
for subscriber in subscribers:
if subscriber.matches_filter(message):
subscriber.receive(
topic=topic,
message=message,
publisher=publisher_id
)
delivery_count += 1
return delivery_count
def subscribe(self, topic, agent, filter_fn=None):
"""Agent订阅主题"""
if topic not in self.topics:
self.topics[topic] = []
subscription = Subscription(
agent=agent,
filter=filter_fn or (lambda msg: True)
)
self.topics[topic].append(subscription)
def unsubscribe(self, topic, agent):
"""取消订阅"""
if topic in self.topics:
self.topics[topic] = [
s for s in self.topics[topic]
if s.agent.id != agent.id
]
# 使用示例
bus = AgentPubSub()
# 监控Agent订阅所有告警
bus.subscribe("alerts.*", monitor_agent)
# 数据分析Agent订阅特定数据更新
bus.subscribe("data.sales.daily", analytics_agent,
filter_fn=lambda m: m.get("region") == "CN")
# 部署Agent发布事件
bus.publish("alerts.critical", {
"service": "web_app",
"issue": "high_latency",
"value": "2500ms"
}, publisher_id="monitor_agent")通信模式选择指南
选择合适的通信模式对系统性能至关重要：直接消息传递适合Agent数量少且关系固定的场景；消息总线适合中等规模的协作系统；黑板系统适合需要共享工作进度的协作任务；发布-订阅适合松耦合的事件驱动系统。在OpenClaw中，可以混合使用多种通信模式来满足不同Agent的通信需求。
五、通信协议的安全与可靠性
在生产环境中，Agent通信还需要考虑安全和可靠性问题：
- 消息认证：确保消息来自合法的Agent，防止冒充攻击。
- 消息加密：对敏感信息进行加密传输，防止中间人窃听。
- 消息确认：接收者确认收到消息，发送者在超时后重发。
- 消息排序：保证消息按发送顺序被接收和处理。
- 幂等处理：重复消息不应导致重复执行。
智能体通信协议是多智能体系统的神经系统。选择合适的通信机制，直接影响系统的效率、可靠性和可扩展性。随着Agent系统规模的不断增长，通信协议的设计将变得越来越重要。