Agent集群通讯系统 v2

基于SQLite嘅轻量级消息队列系统，用于Agent间可靠通讯。

特性

• ✅ 持久化存储 — 消息存储喺SQLite数据库，重启唔会丢失
• ✅ 消息确认 — 发送消息后等待对方确认
• ✅ 自动重试 — 处理失败自动重试（最多3次）
• ✅ 死信队列 — 超过重试次数嘅消息转为死信
• ✅ 状态同步 — Agent之间共享状态
• ✅ 统一时间 — 所有消息包含统一时间信息
• ✅ 并发处理 — 支持同时处理多条消息
• ✅ 跨平台 — 支持Windows/Linux/macOS

快速开始

1. 安装

# 通过SkillHub安装
skillhub install agent-cluster-comm-v2

# 或者手动安装
cp -r agent-cluster-comm-v2 ~/.hermes/skills/

2. 配置

复制配置模板：

cp ~/.hermes/skills/agent-cluster-comm-v2/config.yaml.example ~/.hermes/skills/agent-cluster-comm-v2/config.yaml

编辑配置文件：

# 消息队列配置
queue:
  db_path: "agent_messages.db"
  poll_interval: 2
  max_retries: 3
  batch_size: 5

# Agent配置
agents:
  - name: "agent1"
    enabled: true
  - name: "agent2"
    enabled: true

3. 使用

发送消息

from sqlite_message_queue import SQLiteMessageQueue

queue = SQLiteMessageQueue("agent_messages.db")
queue.send("agent1", "agent2", "你好！")
queue.close()

接收消息

from sqlite_message_queue import SQLiteMessageQueue

queue = SQLiteMessageQueue("agent_messages.db")
messages = queue.poll("agent2", limit=10)

for msg in messages:
    print(f"来自 {msg.from_agent}: {msg.content}")
    queue.ack(msg.id, success=True)

queue.close()

启动监听器

python start.py agent1

API参考

SQLiteMessageQueue

初始化

queue = SQLiteMessageQueue(db_path: str)

发送消息

msg_id = queue.send(
    from_agent: str,
    to_agent: str,
    content: str,
    msg_type: str = "text"
) -> str

轮询消息

messages = queue.poll(
    agent_name: str,
    limit: int = 10
) -> List[Message]

确认消息

queue.ack(
    message_id: str,
    success: bool = True,
    error_message: str = None
)

获取统计

stats = queue.get_stats()

重置卡住消息

count = queue.reset_stuck_messages(timeout_seconds: int = 300)

配置说明

queue配置

agents配置

示例

查看 examples/ 目录获取更多示例：

• basic_usage.py — 基本使用示例

故障排查

消息积压

stats = queue.get_stats()
if stats['pending_count'] > 100:
    print("消息积压！")
    queue.reset_stuck_messages(timeout_seconds=300)

处理失败

stats = queue.get_stats()
if stats['failed_count'] > 0:
    print(f"有 {stats['failed_count']} 条失败消息")

死信消息

stats = queue.get_stats()
if stats['dead_count'] > 0:
    print(f"有 {stats['dead_count']} 条死信消息")

最佳实践

1. 消息内容 — 保持简洁，避免过长
2. 确认机制 — 处理完消息后及时确认
3. 错误处理 — 处理失败时提供错误信息
4. 心跳监控 — 定期检查心跳文件
5. 定期清理 — 定期清理已完成消息

更新日志

v2.0.0 (2026-06-08)

• 初始版本
• 支持SQLite持久化
• 支持消息确认
• 支持自动重试
• 支持死信队列
• 支持状态同步
• 支持统一时间
• 支持并发处理
• 支持跨平台