名称： agent-protocol
描述： 智能体间通信协议。使技能能够通过事件进行通信、构建工作流链，并在无需人工干预的情况下进行编排。
版本： 1.0.0

智能体间通信协议

Clawdbot 技能与智能体的基础通信层。

使您的智能体能够相互通信、构建自动化工作流，并在无需人工干预的情况下编排复杂的多步骤任务。

愿景

研究智能体找到文章
    ↓ 发布 "research.found" 事件
摘要智能体订阅事件
    ↓ 生成摘要
    ↓ 发布 "summary.ready" 事件
通知智能体订阅
    ↓ 发布到 Telegram/Discord

架构

1. 事件总线 (基于文件的消息传递)

• 智能体将事件发布到 ~/.clawdbot/events/
• 事件是经过模式验证的 JSON 文件
• 持久化、可调试、可审计
• 自动清理已处理的事件

2. 工作流引擎

• 使用 JSON 或 YAML 定义管道
• 基于事件数据的条件路由
• 错误处理、重试、回退机制
• 集成 Cron 用于计划执行

3. 共享上下文

• 智能体读写共享内存空间
• 在工作流步骤间传递上下文
• 跨智能体调用的状态持久化

4. 智能体注册表

• 发现可用的智能体/技能
• 能力通告
• 权限管理

核心概念

事件

事件是通信的基本单位：

{
  "event_id": "evt_20260128_001",
  "event_type": "research.article_found",
  "timestamp": "2026-01-28T23:00:00Z",
  "source_agent": "research-agent",
  "payload": {
    "title": "ETH 2.0 升级完成",
    "url": "https://example.com/article",
    "importance": 9,
    "summary": "以太坊重大升级..."
  },
  "metadata": {
    "session_id": "main",
    "requires_action": true
  }
}

工作流

工作流定义了智能体如何响应事件：

{
  "workflow_id": "research-to-telegram",
  "name": "研究 → 摘要 → 通知",
  "trigger": {
    "event_type": "research.article_found",
    "conditions": {
      "payload.importance": { "gte": 7 }
    }
  },
  "steps": [
    {
      "agent": "summary-agent",
      "action": "summarize",
      "input": "{{payload}}",
      "output_event": "summary.ready"
    },
    {
      "agent": "notification-agent",
      "action": "notify",
      "input": "{{previous.summary}}",
      "channels": ["telegram"]
    }
  ]
}

快速开始

1. 安装

cd /root/clawd/skills/agent-protocol
python3 scripts/setup.py

2. 启动事件总线

python3 scripts/event_bus.py start

3. 发布您的第一个事件

python3 scripts/publish.py \
  --type "test.hello" \
  --source "my-agent" \
  --payload '{"message": "Hello, world!"}'

4. 订阅事件

python3 scripts/subscribe.py \
  --types "test.hello" \
  --handler "./my_handler.py"

5. 定义一个工作流

cp examples/simple-workflow.json config/workflows/my-workflow.json
python3 scripts/workflow_engine.py --validate

事件类型 (约定)

标准事件类型

• research.article_found - 研究智能体找到相关内容
• research.topic_suggested - 建议新的研究主题
• summary.ready - 摘要已生成
• analytics.insight - 个人分析洞察
• sports.goal_scored - 体育比分事件
• sports.match_started - 比赛开始
• notification.sent - 通知已发送
• workflow.started - 工作流执行开始
• workflow.completed - 工作流完成
• workflow.failed - 工作流失败

事件命名约定

<领域>.<动作_过去式>
- 使用小写字母和下划线
- 领域：广泛类别 (research, sports, notification)
- 动作：发生了什么 (article_found, goal_scored)

工作流示例

示例 1: 研究 → 通知

{
  "workflow_id": "eth-news-alert",
  "trigger": {
    "event_type": "research.article_found",
    "conditions": {
      "payload.keywords": { "contains": ["ethereum", "ETH"] },
      "payload.importance": { "gte": 8 }
    }
  },
  "steps": [
    {
      "agent": "notification-agent",
      "action": "send_telegram",
      "input": {
        "message": "🚨 重要 ETH 新闻！\n{{payload.title}}\n{{payload.url}}"
      }
    }
  ]
}

示例 2: 体育进球 → TTS 播报

{
  "workflow_id": "goal-announcement",
  "trigger": {
    "event_type": "sports.goal_scored",
    "conditions": {
      "payload.team": { "eq": "Barcelona" }
    }
  },
  "steps": [
    {
      "agent": "tts-agent",
      "action": "announce",
      "input": {
        "text": "巴塞罗那进球了！{{payload.scorer}} 得分！当前比分 {{payload.score}}"
      }
    }
  ]
}

示例 3: 每日分析 → 研究主题

{
  "workflow_id": "analytics-to-research",
  "trigger": {
    "event_type": "analytics.daily_report",
    "schedule": "0 9 * * *"
  },
  "steps": [
    {
      "agent": "analytics-agent",
      "action": "generate_insights",
      "output_event": "analytics.insights_ready"
    },
    {
      "agent": "research-agent",
      "action": "suggest_topics",
      "input": "{{previous.insights}}",
      "conditions": {
        "previous.insights.count": { "gte": 3 }
      }
    }
  ]
}

命令

事件总线

# 启动事件总线守护进程
python3 scripts/event_bus.py start

# 检查状态
python3 scripts/event_bus.py status

# 停止
python3 scripts/event_bus.py stop

# 查看最近事件
python3 scripts/event_bus.py tail --count 20

发布事件

# 发布事件 (JSON 负载)
python3 scripts/publish.py \
  --type "research.article_found" \
  --source "research-agent" \
  --payload '{"title": "文章", "url": "..."}'

# 从文件发布
python3 scripts/publish.py --file event.json

# 带优先级发布
python3 scripts/publish.py \
  --type "alert.urgent" \
  --priority high \
  --payload '{"message": "紧急警报！"}'

订阅事件

# 订阅事件类型
python3 scripts/subscribe.py \
  --types "research.*,sports.goal_scored" \
  --handler "./handlers/my_handler.py"

# 带过滤器订阅
python3 scripts/subscribe.py \
  --types "research.*" \
  --filter '{"importance": {"gte": 8}}' \
  --handler "./handlers/important_only.py"

# 列出活跃订阅
python3 scripts/subscribe.py --list

工作流管理

# 验证工作流
python3 scripts/workflow_engine.py --validate

# 运行工作流引擎 (处理工作流)
python3 scripts/workflow_engine.py --run

# 测试特定工作流
python3 scripts/workflow_engine.py --test eth-news-alert

# 列出工作流
python3 scripts/workflow_engine.py --list

# 启用/禁用工作流
python3 scripts/workflow_engine.py --enable research-to-telegram
python3 scripts/workflow_engine.py --disable research-to-telegram

智能体注册表

# 注册您的智能体
python3 scripts/registry.py register \
  --name "my-agent" \
  --capabilities "summarize,notify" \
  --events "research.article_found"

# 列出可用智能体
python3 scripts/registry.py list

# 按能力查询智能体
python3 scripts/registry.py query --capability "summarize"

与现有技能集成

体育比分集成

修改 sports-ticker/scripts/live_monitor.py 以发布事件：

from agent_protocol import publish_event

# 检测到进球后：
publish_event(
    event_type="sports.goal_scored",
    source="sports-ticker",
    payload={
        "team": team_name,
        "scorer": player_name,
        "opponent": opponent,
        "score": f"{home_score}-{away_score}",
        "minute": clock
    }
)

研究智能体集成

from agent_protocol import publish_event

# 找到文章后：
publish_event(
    event_type="research.article_found",
    source="research-agent",
    payload={
        "title": article_title,
        "url": article_url,
        "importance": calculate_importance(article),
        "summary": snippet
    }
)

个人分析集成

from agent_protocol import publish_event

# 每日洞察：
publish_event(
    event_type="analytics.insight",
    source="personal-analytics",
    payload={
        "type": "productivity",
        "insight": "您本周的专注时间增加了 20%",
        "recommendations": ["在早上安排深度工作"]
    }
)

安全与权限

权限模型

{
  "agent": "research-agent",
  "permissions": {
    "publish": ["research.*"],
    "subscribe": ["summary.*", "notification.*"],
    "workflows": ["research-to-telegram"]
  }
}

沙箱机制

• 智能体只能发布到其指定的事件类型
• 订阅需要显式权限
• 工作流在执行前经过验证

配置

主配置: config/protocol.json

{
  "event_bus": {
    "storage_path": "~/.clawdbot/events",
    "retention_days": 7,
    "max_event_size_kb": 512
  },
  "workflow_engine": {
    "enabled": true,
    "poll_interval_seconds": 30,
    "max_concurrent_workflows": 5
  },
  "registry": {
    "agents_path": "~/.clawdbot/agents/registry.json"
  },
  "security": {
    "require_permissions": true,
    "audit_log": true
  }
}

高级功能

1. 条件路由

{
  "steps": [
    {
      "condition": {
        "payload.importance": { "gte": 9 }
      },
      "then": { "agent": "urgent-notifier" },
      "else": { "agent": "standard-notifier" }
    }
  ]
}

2. 并行执行

{
  "steps": [
    {
      "parallel": [
        { "agent": "telegram-notifier" },
        { "agent": "discord-notifier" },
        { "agent": "email-notifier" }
      ]
    }
  ]
}

3. 错误处理

{
  "steps": [
    {
      "agent": "external-api",
      "retry": {
        "max_attempts": 3,
        "backoff_seconds": 5
      },
      "on_error": {
        "agent": "error-logger",
        "continue": true
      }
    }
  ]
}

4. 计划工作流

{
  "trigger": {
    "schedule": "0 9 * * *",
    "event_type": "cron.daily_run"
  }
}

监控与调试

事件日志

所有事件都记录到 ~/.clawdbot/events/log/

# 查看事件日志
tail -f ~/.clawdbot/events/log/events.log

# 搜索事件
python3 scripts/query.py --type "research.*" --since "1 hour ago"

工作流执行日志

# 查看工作流执行历史
python3 scripts/workflow_engine.py --history

# 检查失败的工作流
python3 scripts/workflow_engine.py --inspect <workflow_id>

指标

# 显示事件统计信息
python3 scripts/metrics.py

# 输出示例：
# 已发布事件总数: 1,234
# 事件类型: 15
# 活跃订阅数: 8
# 已执行工作流: 456
# 平均工作流持续时间: 2.3s

最佳实践

1. 事件设计
   - 保持负载小而专注
   - 为处理程序包含足够的上下文
   - 使用一致的命名约定

2. 工作流设计
   - 保持工作流简单且专注
   - 使用描述性名称
   - 启用前彻底测试

3. 错误处理
   - 始终定义错误处理程序
   - 记录错误以便调试
   - 对瞬时故障使用重试机制

4. 性能
   - 避免高频事件
   - 定期清理旧事件
   - 监控工作流执行时间

5. 安全
   - 验证事件负载
   - 使用权限系统
   - 审计敏感操作

Python API

from agent_protocol import (
    publish_event,
    subscribe,
    create_workflow,
    register_agent
)

# 发布事件
publish_event(
    event_type="my.event",
    source="my-agent",
    payload={"key": "value"}
)

# 订阅事件
@subscribe(["research.*"])
def handle_research(event):
    print(f"收到研究事件: {event['payload']}")

# 以编程方式创建工作流
workflow = create_workflow(
    workflow_id="my-workflow",
    trigger={"event_type": "my.trigger"},
    steps=[
        {"agent": "processor", "action": "process"}
    ]
)

# 注册智能体
register_agent(
    name="my-agent",
    capabilities=["process", "notify"],
    event_types=["my.event"]
)

JavaScript API

const { publishEvent, subscribe, createWorkflow } = require('./scripts/protocol.js');

// 发布事件
await publishEvent({
  eventType: 'my.event',
  source: 'my-agent',
  payload: { key: 'value' }
});

// 订阅
subscribe(['research.*'], (event) => {
  console.log('收到事件:', event);
});

// 创建工作流
await createWorkflow({
  workflowId: 'my-workflow',
  trigger: { eventType: 'my.trigger' },
  steps: [
    { agent: 'processor', action: 'process' }
  ]
});

路线图

• [ ] 可视化工作流构建器 (Web UI)
• [ ] WebSocket 支持实时事件
• [ ] 跨实例事件中继 (多机器人网络)
• [ ] AI 驱动的工作流建议
• [ ] 事件重放和调试工具
• [ ] 性能分析
• [ ] 事件的 GraphQL 查询 API

贡献

此技能是 Clawdbot 核心基础设施的一部分。欢迎贡献！

许可证

MIT

由 Robby 使用 🦎 构建