Appearance
OpenClaw 多Agent通信协议深度解析：消息传递与数据同步
多Agent协作的核心是通信：消息怎么传、状态怎么同步、任务怎么协调。理解通信协议，才能构建高效的多Agent系统。
通信架构概览
单Agent vs 多Agent通信
| 特性 | 单Agent | 多Agent |
|---|---|---|
| 消息流 | 用户 ↔ Agent | 用户 → 主控 → 子Agent |
| 状态管理 | 简单 | 需要同步 |
| 任务调度 | 直接执行 | 需要协调 |
| 故障处理 | 简单重试 | 需要容错 |
多Agent通信模式
┌─────────────────────────────────────────────────────────────┐
│ 多Agent通信架构 │
└─────────────────────────────────────────────────────────────┘
│
┌─────────────────────┼─────────────────────┐
▼ ▼ ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 同步通信 │ │ 异步通信 │ │ 事件驱动 │
├─────────────┤ ├─────────────┤ ├─────────────┤
│ • 请求响应 │ │ • 消息队列 │ │ • 订阅发布 │
│ • 阻塞等待 │ │ • 非阻塞 │ │ • 事件广播 │
│ • 结果返回 │ │ • 回调处理 │ │ • 解耦消息 │
└─────────────┘ └─────────────┘ └─────────────┘消息传递机制
消息格式标准
json
{
"id": "msg-12345",
"type": "task_request",
"from": "main-agent",
"to": "sub-agent-1",
"timestamp": "2026-04-08T18:45:00Z",
"payload": {
"task": "fetch_data",
"params": {
"url": "https://api.example.com/data"
}
},
"metadata": {
"priority": "high",
"timeout": 30000,
"retry": 3
}
}消息类型
| 类型 | 方向 | 说明 |
|---|---|---|
| task_request | 主控 → 子Agent | 任务请求 |
| task_response | 子Agent → 主控 | 任务结果 |
| status_update | 双向 | 状态更新 |
| error_report | 子Agent → 主控 | 错误报告 |
| heartbeat | 双向 | 心跳检测 |
消息传递示例
javascript
// 主控 Agent 发送任务
const message = {
type: "task_request",
to: "data-agent",
payload: {
task: "fetch",
url: "https://api.example.com"
}
};
const response = await sendAndWait(message, {
timeout: 30000,
retry: 3
});
console.log(response.result);同步通信
直接调用模式
用户请求
│
▼
┌─────────────┐
│ 主控Agent │
└─────────────┘
│
│ sendAndWait()
▼
┌─────────────┐
│ 子Agent A │ ─────────────────────────>
└─────────────┘ 执行任务，返回结果
│
│ 收到结果，继续
▼
┌─────────────┐
│ 主控Agent │
└─────────────┘
│
▼
返回用户结果同步调用代码
javascript
// 主控 Agent 同步调用子Agent
async function handleUserRequest(userMessage) {
// 发送任务给子Agent，等待结果
const response = await spawn({
agent: "data-agent",
message: userMessage,
waitForResponse: true
});
// 处理子Agent的响应
return {
result: response.result,
agent: response.agent
};
}同步通信优缺点
| 优点 | 缺点 |
|---|---|
| 简单直接 | 阻塞等待 |
| 易于调试 | 响应慢 |
| 结果可靠 | 故障影响大 |
异步通信
消息队列模式
用户请求
│
▼
┌─────────────┐
│ 主控Agent │
└─────────────┘
│
│ 发送消息（不等待）
▼
┌─────────────┐
│ 消息队列 │
└─────────────┘
│
▼
┌─────────────┐
│ 子Agent A │ ← 消费消息
└─────────────┘
│
│ 处理完成，发送回调
▼
┌─────────────┐
│ 回调队列 │
└─────────────┘
│
▼
┌─────────────┐
│ 主控Agent │ ← 处理回调
└─────────────┘异步调用代码
javascript
// 主控 Agent 异步调用子Agent
async function handleUserRequest(userMessage) {
// 发送任务，不等待结果
const taskId = await spawn({
agent: "data-agent",
message: userMessage,
waitForResponse: false
});
// 立即返回任务ID
return {
taskId: taskId,
status: "processing"
};
}
// 注册回调处理结果
onTaskComplete("data-agent", (result) => {
console.log("任务完成：", result);
});异步通信配置
json
{
"asyncCommunication": {
"enabled": true,
"messageQueue": {
"type": "redis",
"host": "localhost",
"port": 6379
},
"callbackQueue": {
"type": "redis",
"prefix": "callback:"
},
"maxConcurrent": 10,
"timeout": 60000
}
}事件驱动通信
发布订阅模式
┌─────────────────────────────────────────────────────────────┐
│ 事件总线 │
└─────────────────────────────────────────────────────────────┘
↑ ↑ ↑
│ publish │ publish │ publish
│ │ │
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 主控Agent │ │ 子Agent A │ │ 子Agent B │
└─────────────┘ └─────────────┘ └─────────────┘
│ │ │
│ subscribe │ subscribe │
▼ ▼ ▼
监听事件 监听事件 监听事件事件类型定义
javascript
// 事件类型
const Events = {
// 任务事件
TASK_CREATED: "task.created",
TASK_STARTED: "task.started",
TASK_COMPLETED: "task.completed",
TASK_FAILED: "task.failed",
// 状态事件
AGENT_CONNECTED: "agent.connected",
AGENT_DISCONNECTED: "agent.disconnected",
AGENT_BUSY: "agent.busy",
AGENT_IDLE: "agent.idle",
// 数据事件
DATA_UPDATED: "data.updated",
DATA_SYNCED: "data.synced"
};事件发布订阅代码
javascript
// 主控 Agent 发布事件
eventBus.publish(Events.TASK_CREATED, {
taskId: "task-123",
type: "data_fetch",
params: { url: "https://api.example.com" }
});
// 子Agent 订阅事件
eventBus.subscribe(Events.TASK_CREATED, async (event) => {
console.log("收到任务：", event);
// 处理任务
const result = await processTask(event);
// 发布完成事件
eventBus.publish(Events.TASK_COMPLETED, {
taskId: event.taskId,
result: result
});
});事件驱动配置
json
{
"eventBus": {
"type": "redis",
"channels": {
"task": "openclaw:tasks",
"status": "openclaw:status",
"data": "openclaw:data"
},
"retryPolicy": {
"maxRetries": 3,
"backoffMs": 1000
}
}
}数据同步策略
共享状态管理
javascript
// 共享状态存储
const sharedState = {
// 全局状态
global: {
requestId: "req-123",
userId: "user-456",
startTime: Date.now()
},
// Agent 状态
agents: {
"main-agent": { status: "active", lastUpdate: Date.now() },
"data-agent": { status: "processing", lastUpdate: Date.now() }
},
// 共享数据
data: {
users: [],
results: []
}
};数据同步模式
| 模式 | 说明 | 适用场景 |
|---|---|---|
| 强一致性 | 所有Agent同步更新 | 关键状态 |
| 最终一致性 | 异步同步，最终一致 | 大量数据 |
| 因果一致性 | 保证因果顺序 | 有序操作 |
数据同步代码
javascript
// 状态同步管理器
class StateManager {
constructor() {
this.state = {};
this.subscribers = {};
}
// 更新状态
async update(key, value, options = {}) {
// 乐观更新
this.state[key] = value;
// 通知订阅者
this.notify(key, value);
// 持久化（可选）
if (options.persist) {
await this.persist(key, value);
}
}
// 订阅状态变更
subscribe(key, callback) {
if (!this.subscribers[key]) {
this.subscribers[key] = [];
}
this.subscribers[key].push(callback);
}
// 通知订阅者
notify(key, value) {
const callbacks = this.subscribers[key] || [];
callbacks.forEach(cb => cb(value));
}
}通信工具使用
sessions_spawn
创建新的 Agent 会话：
javascript
// 创建子Agent会话
const session = await sessions_spawn({
agent: "sub-agent",
message: "执行数据抓取任务",
context: {
requestId: "req-123",
parentId: "main-agent"
}
});
// 监听会话状态
session.on("complete", (result) => {
console.log("子Agent完成：", result);
});
session.on("error", (error) => {
console.error("子Agent错误：", error);
});sessions_broadcast
广播消息给所有Agent：
javascript
// 广播状态更新
await sessions_broadcast({
type: "status_update",
payload: {
status: "data_synced",
timestamp: Date.now()
}
});
// 定向广播给特定Agent
await sessions_broadcast({
type: "task_assigned",
to: ["data-agent", "process-agent"],
payload: { task: "batch_process" }
});sessions_send
发送消息给特定Agent：
javascript
// 发送消息给单个Agent
const response = await sessions_send({
to: "data-agent",
message: {
type: "fetch_request",
url: "https://api.example.com/data"
},
waitForResponse: true
});通信最佳实践
1. 消息幂等性
javascript
// 使用消息ID避免重复处理
const processedMessages = new Set();
async function handleMessage(message) {
// 检查是否已处理
if (processedMessages.has(message.id)) {
console.log("消息已处理，跳过");
return;
}
// 处理消息
await processMessage(message);
// 标记为已处理
processedMessages.add(message.id);
}2. 超时与重试
javascript
// 带超时和重试的消息发送
async function sendWithRetry(message, options = {}) {
const { timeout = 30000, retry = 3 } = options;
for (let i = 0; i < retry; i++) {
try {
const result = await Promise.race([
sendMessage(message),
sleep(timeout).then(() => { throw new Error("超时"); })
]);
return result;
} catch (error) {
if (i === retry - 1) throw error;
await sleep(1000 * (i + 1)); // 指数退避
}
}
}3. 错误处理
javascript
// 完善的错误处理
async function handleTask(task) {
try {
const result = await executeTask(task);
// 发送成功事件
eventBus.publish(Events.TASK_COMPLETED, {
taskId: task.id,
result: result
});
return result;
} catch (error) {
// 发送失败事件
eventBus.publish(Events.TASK_FAILED, {
taskId: task.id,
error: error.message
});
// 根据错误类型处理
if (isRetryable(error)) {
await retryTask(task);
} else {
await notifyFailure(task, error);
}
throw error;
}
}实战案例：数据抓取流水线
场景描述
用户请求抓取多个网站数据，主控Agent协调多个子Agent完成：
- data-agent：抓取数据
- process-agent：处理数据
- storage-agent：存储数据
实现代码
javascript
// 主控 Agent
async function orchestrateDataPipeline(urls) {
const taskId = generateTaskId();
// 发布任务开始事件
eventBus.publish(Events.TASK_STARTED, { taskId, urls });
try {
// 阶段1：数据抓取（并行）
const fetchPromises = urls.map(url =>
sessions_spawn({
agent: "data-agent",
message: { task: "fetch", url },
waitForResponse: true
})
);
const fetchResults = await Promise.all(fetchPromises);
// 阶段2：数据处理（串行）
const processResult = await sessions_spawn({
agent: "process-agent",
message: {
task: "process",
data: fetchResults.map(r => r.data)
},
waitForResponse: true
});
// 阶段3：数据存储
const storageResult = await sessions_spawn({
agent: "storage-agent",
message: {
task: "store",
data: processResult.data
},
waitForResponse: true
});
// 发布任务完成事件
eventBus.publish(Events.TASK_COMPLETED, {
taskId,
result: storageResult
});
return storageResult;
} catch (error) {
eventBus.publish(Events.TASK_FAILED, {
taskId,
error: error.message
});
throw error;
}
}总结
| 通信模式 | 适用场景 | 特点 |
|---|---|---|
| 同步通信 | 简单任务、串行流程 | 简单可靠 |
| 异步通信 | 耗时任务、并行处理 | 高效不阻塞 |
| 事件驱动 | 复杂系统、解耦架构 | 灵活扩展 |
通信设计原则：
- 选择合适的通信模式
- 处理好超时与重试
- 保证消息幂等性
- 完善错误处理
- 合理的日志追踪
掌握通信协议是构建高效多Agent系统的关键。根据场景选择合适的通信模式，让Agent之间协作顺畅高效。
