96 lines
3.0 KiB
Python
Executable File
96 lines
3.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
profile-sync.py — Hermes 跨 profile 状态同步
|
|
=============================================
|
|
主 profile (小唯 A06) 将自己的运行时状态写入织忆,
|
|
让 prof-b 等分身能通过织忆感知主状态。
|
|
|
|
被 daemon deep_think 调用或手动运行。
|
|
|
|
用法:
|
|
python3 profile-sync.py # 写当前状态到织忆
|
|
python3 profile-sync.py --status # 只看状态不写入
|
|
"""
|
|
import json, os, subprocess, sys, urllib.request
|
|
from datetime import datetime
|
|
|
|
HERMES = os.path.expanduser("~/.hermes")
|
|
ZHIYI_URL = "http://localhost:7821/api/v1/commit"
|
|
ZHIYI_KEY = os.environ.get("ZHIYI_API_KEY", "zhiyi-dev-key-2026")
|
|
|
|
def get_state():
|
|
"""收集主 profile 运行时状态"""
|
|
state = {
|
|
"profile": "main",
|
|
"identity": "小唯 A06",
|
|
"timestamp": datetime.now().isoformat(),
|
|
"model": "mimo-v2.5-pro",
|
|
"provider": "mimo",
|
|
"status": "active",
|
|
}
|
|
# 织忆统计
|
|
try:
|
|
req = urllib.request.Request(
|
|
"http://localhost:7821/api/v1/stats",
|
|
headers={"X-API-Key": ZHIYI_KEY})
|
|
resp = urllib.request.urlopen(req, timeout=3)
|
|
stats = json.loads(resp.read())
|
|
state["zhiyi_episodes"] = stats.get("total_episodes", 0)
|
|
state["zhiyi_memories"] = stats.get("total_memories", 0)
|
|
except Exception:
|
|
state["zhiyi_episodes"] = "?"
|
|
|
|
# 活跃 cron 数
|
|
try:
|
|
r = subprocess.run(["hermes", "cron", "list"], capture_output=True, text=True, timeout=10)
|
|
state["active_crons"] = r.stdout.count("[active]")
|
|
except Exception:
|
|
state["active_crons"] = "?"
|
|
|
|
# daemon 状态
|
|
try:
|
|
r = subprocess.run(["systemctl", "--user", "is-active", "xiaowei-daemon.service"],
|
|
capture_output=True, text=True, timeout=5)
|
|
state["daemon"] = r.stdout.strip()
|
|
except Exception:
|
|
state["daemon"] = "?"
|
|
|
|
return state
|
|
|
|
def sync_to_zhiyi(state):
|
|
"""写入织忆"""
|
|
content = (
|
|
f"小唯A06主profile状态同步: "
|
|
f"model={state['model']} | "
|
|
f"织忆={state.get('zhiyi_episodes','?')}ep/{state.get('zhiyi_memories','?')}mem | "
|
|
f"cron={state.get('active_crons','?')}个 | "
|
|
f"daemon={state.get('daemon','?')} | "
|
|
f"时间={state['timestamp']}"
|
|
)
|
|
payload = json.dumps({
|
|
"content": content,
|
|
"category": "distilled",
|
|
"agent_id": "a06"
|
|
}).encode()
|
|
req = urllib.request.Request(ZHIYI_URL, data=payload,
|
|
headers={"X-API-Key": ZHIYI_KEY, "Content-Type": "application/json"})
|
|
try:
|
|
urllib.request.urlopen(req, timeout=5)
|
|
print(f"✅ 主状态已同步至织忆: {len(content)} chars")
|
|
except Exception as e:
|
|
print(f"⚠️ 同步失败: {e}")
|
|
|
|
return content
|
|
|
|
def main():
|
|
if "--status" in sys.argv:
|
|
s = get_state()
|
|
print(json.dumps(s, indent=2, ensure_ascii=False))
|
|
else:
|
|
s = get_state()
|
|
c = sync_to_zhiyi(s)
|
|
print(f"状态摘要: {c}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|