192 lines
6.4 KiB
Python
Executable File
192 lines
6.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
openclaw-bridge.py — Hermes↔OpenClaw 智能桥接
|
||
==============================================
|
||
让 Hermes（小唯 A06）能够：
|
||
1. 向 OpenClaw 指定 agent 发送消息
|
||
2. 读取 OpenClaw agent 的回复
|
||
3. 在 织忆 中记录跨系统通信
|
||
|
||
用法:
|
||
# 向 A02 铁锋发消息
|
||
python3 openclaw-bridge.py send a02 "检查服务器磁盘空间"
|
||
|
||
# 向 A01 小雪发消息
|
||
python3 openclaw-bridge.py send main "帮我整理今天的记忆"
|
||
|
||
# 读取指定 agent 最近的回复
|
||
python3 openclaw-bridge.py read main --limit 5
|
||
|
||
# 查看所有 agent 健康状态
|
||
python3 openclaw-bridge.py status
|
||
|
||
# 列出 织忆 中的跨系统记录
|
||
python3 openclaw-bridge.py history --limit 10
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import urllib.request
|
||
|
||
OPENCLAW = "/home/muc/nodejs/node-v24.16.0-linux-x64/bin/openclaw"
|
||
ZHIYI_URL = "http://127.0.0.1:7821/api/v1/commit"
|
||
ZHIYI_RECALL = "http://127.0.0.1:7821/api/v1/recall"
|
||
ZHIYI_KEY = os.environ.get("ZHIYI_API_KEY", "zhiyi-dev-key-2026")
|
||
|
||
AGENTS = {
|
||
"main": "A01小雪",
|
||
"a02": "A02铁锋",
|
||
"a03": "A03墨文",
|
||
"a04": "A04卫安",
|
||
"a05": "A05蓝图",
|
||
}
|
||
|
||
|
||
def shell_list(cmd_list, timeout=30):
|
||
"""执行命令（列表参数，防注入）"""
|
||
try:
|
||
r = subprocess.run(cmd_list, capture_output=True, text=True, timeout=timeout)
|
||
return r.stdout, r.stderr, r.returncode
|
||
except subprocess.TimeoutExpired:
|
||
return "", "timeout", -1
|
||
|
||
|
||
def zhiyi_log(action, agent, content):
|
||
"""记录跨系统通信到织忆，失败时输出日志
|
||
|
||
2026-09-07 质量门槛(t_9316bf56): status/健康检查类动作不写织忆(731条刷屏重复根因)，
|
||
短内容(<20字)也不写——碎片无记忆价值，污染 recall 与蒸馏质量。
|
||
"""
|
||
if action == "status" or content is None:
|
||
return # 健康检查无记忆价值，写=刷屏(08-17/18 曾产生 731 条完全重复)
|
||
if len(str(content).strip()) < 20:
|
||
return # 碎片过滤
|
||
agent_name = AGENTS.get(agent, agent)
|
||
payload = json.dumps({
|
||
"content": f"小唯↔{agent_name} 桥接: {action} | {content[:200]}",
|
||
"category": "distilled",
|
||
"agent_id": "a06"
|
||
}).encode()
|
||
try:
|
||
req = urllib.request.Request(ZHIYI_URL, data=payload,
|
||
headers={"X-API-Key": ZHIYI_KEY, "Content-Type": "application/json"})
|
||
urllib.request.urlopen(req, timeout=5)
|
||
except Exception as e:
|
||
print(f"[bridge] ⚠ 织忆日志写入失败: {e}", file=__import__('sys').stderr)
|
||
|
||
|
||
def cmd_send(agent, message):
|
||
"""向 OpenClaw agent 发送消息"""
|
||
agent_name = AGENTS.get(agent, agent)
|
||
print(f"📤 → {agent_name}: {message[:80]}...")
|
||
|
||
# 用列表参数替代 shell 字符串拼接
|
||
stdout, stderr, rc = shell_list(
|
||
[OPENCLAW, "chat", "--session", f"session://{agent}", "--input", message],
|
||
timeout=60
|
||
)
|
||
|
||
if rc == 0:
|
||
print(f"✅ 已发送至 {agent_name}")
|
||
zhiyi_log("send", agent, f"发送消息: {message[:200]}")
|
||
else:
|
||
print(f"⚠️ CLI 发送失败({stderr[:50]}), 走织忆中转")
|
||
zhiyi_log("send_queued", agent, f"排队消息: {message[:200]}")
|
||
print(f"ℹ️ 消息已存织忆, {agent_name} 下次读取织忆时可获取")
|
||
return rc
|
||
|
||
|
||
def cmd_read(agent, limit=5):
|
||
"""读取 agent 最近的会话"""
|
||
agent_name = AGENTS.get(agent, agent)
|
||
print(f"📖 ← {agent_name} (最近{limit}条)")
|
||
|
||
stdout, stderr, rc = shell_list(
|
||
[OPENCLAW, "session", "list", "--agent", agent, "--limit", str(limit)],
|
||
timeout=15
|
||
)
|
||
if rc == 0 and stdout.strip():
|
||
print(stdout[:2000])
|
||
else:
|
||
if stderr.strip():
|
||
print(f"ℹ️ 诊断: {stderr.strip()[:100]}")
|
||
print(f"ℹ️ {agent_name} 暂无新输出")
|
||
return rc
|
||
|
||
|
||
def cmd_status():
|
||
"""查看所有 agent 健康状态"""
|
||
print("🔍 OpenClaw Agent 健康检查")
|
||
print("=" * 40)
|
||
for agent_id, name in AGENTS.items():
|
||
# 用 openclaw agent (不加 status) 检查是否能路由
|
||
stdout, stderr, rc = shell_list(
|
||
[OPENCLAW, "agent", "--agent", agent_id, "--help"],
|
||
timeout=10
|
||
)
|
||
stderr_lower = stderr.lower()
|
||
if rc == 0 or "registered" in stderr_lower or "plugins" in stderr_lower:
|
||
print(f" {name} ({agent_id}): ✅ 已配置")
|
||
else:
|
||
err = stderr.strip()[:40] if stderr.strip() else f"exit={rc}"
|
||
print(f" {name} ({agent_id}): ❌ {err}")
|
||
print("=" * 40)
|
||
zhiyi_log("status", "all", "健康检查完成")
|
||
|
||
|
||
def cmd_history(limit=10):
|
||
"""从织忆读取跨系统通信历史"""
|
||
payload = json.dumps({"query": "小唯↔ 桥接", "top_k": limit}).encode()
|
||
req = urllib.request.Request(ZHIYI_RECALL, data=payload,
|
||
headers={"X-API-Key": ZHIYI_KEY, "Content-Type": "application/json"})
|
||
try:
|
||
resp = urllib.request.urlopen(req, timeout=10)
|
||
data = json.loads(resp.read())
|
||
results = data.get("results", [])
|
||
print(f"📋 桥接历史 (最近{len(results)}条):")
|
||
for r in results:
|
||
print(f" • {r.get('content','')[:120]}")
|
||
except Exception as e:
|
||
print(f"⚠️ 读取织忆失败: {e}", file=__import__('sys').stderr)
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Hermes↔OpenClaw 桥接")
|
||
sub = parser.add_subparsers(dest="command")
|
||
|
||
p_send = sub.add_parser("send", help="向 agent 发消息")
|
||
p_send.add_argument("agent", choices=list(AGENTS.keys()) + ["all"])
|
||
p_send.add_argument("message", help="消息内容")
|
||
|
||
p_read = sub.add_parser("read", help="读取 agent 输出")
|
||
p_read.add_argument("agent", choices=list(AGENTS.keys()))
|
||
p_read.add_argument("--limit", type=int, default=5)
|
||
|
||
sub.add_parser("status", help="健康检查")
|
||
|
||
p_hist = sub.add_parser("history", help="通信历史")
|
||
p_hist.add_argument("--limit", type=int, default=10)
|
||
|
||
args = parser.parse_args()
|
||
|
||
if args.command == "send":
|
||
if args.agent == "all":
|
||
for a in AGENTS:
|
||
cmd_send(a, args.message)
|
||
else:
|
||
cmd_send(args.agent, args.message)
|
||
elif args.command == "read":
|
||
cmd_read(args.agent, args.limit)
|
||
elif args.command == "status":
|
||
cmd_status()
|
||
elif args.command == "history":
|
||
cmd_history(args.limit)
|
||
else:
|
||
parser.print_help()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|