356 lines
16 KiB
Python
356 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
memory-system-self-upgrade.py — 记忆系统自我升级（方案A统一架构）
|
||
每天凌晨 4:00 由 cron 触发
|
||
|
||
职责（L7 统一层优先）：
|
||
1. L7 llm_context.json v2：验证 9 字段完整，distill_status 检查
|
||
2. 织忆（L1/L2）：健康度 + pattern 节点增长 + 索引
|
||
3. Soulful（L5/L6）：过期 cares 清理 + distilled_rules 补充 + 心迹去重
|
||
4. 统一写入优化报告到飞书
|
||
"""
|
||
import json, os, sys, subprocess
|
||
from datetime import datetime, timezone, timedelta
|
||
|
||
HERMES = os.path.expanduser("~/.hermes")
|
||
ZHIYI_URL = "http://127.0.0.1:7821"
|
||
ZHIYI_KEY = "zhiyi-dev-key-2026"
|
||
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/446db983-e392-4d2c-bfb8-f9060e5df3ad"
|
||
REPORT = [] # 收集升级报告
|
||
|
||
def log(msg):
|
||
ts = datetime.now().strftime("%H:%M:%S")
|
||
print(f"[{ts}] {msg}")
|
||
|
||
def feishu_alert(title, content):
|
||
try:
|
||
payload = json.dumps({
|
||
"msg_type": "interactive",
|
||
"card": {
|
||
"header": {"title": {"tag": "plain_text", "content": title}, "template": "red"},
|
||
"elements": [{"tag": "markdown", "content": content}]
|
||
}
|
||
}).encode("utf-8")
|
||
req = urllib.request.Request(FEISHU_WEBHOOK, data=payload,
|
||
headers={"Content-Type": "application/json"}, method="POST")
|
||
with urllib.request.urlopen(req, timeout=10): pass
|
||
except Exception as e:
|
||
log(f"飞书通知失败: {e}")
|
||
|
||
import urllib.request
|
||
|
||
# ── 1. 织忆自我检查 ──────────────────────────────────────────────────────────
|
||
def upgrade_zhiyi():
|
||
log("=== 织忆自检 ===")
|
||
actions = []
|
||
|
||
# Health check
|
||
try:
|
||
req = urllib.request.Request(
|
||
f"{ZHIYI_URL}/api/v1/health",
|
||
headers={"X-API-Key": ZHIYI_KEY},
|
||
method="GET"
|
||
)
|
||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||
health = json.loads(resp.read().decode())
|
||
log(f" health: {health.get('status')}")
|
||
except Exception as e:
|
||
REPORT.append(f"❌ 织忆健康检查失败: {e}")
|
||
return actions
|
||
|
||
# Stats
|
||
try:
|
||
req = urllib.request.Request(
|
||
f"{ZHIYI_URL}/api/v1/stats",
|
||
headers={"X-API-Key": ZHIYI_KEY},
|
||
method="GET"
|
||
)
|
||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||
stats = json.loads(resp.read().decode())
|
||
episodes = stats.get("total_episodes", 0)
|
||
memories = stats.get("total_memories", 0)
|
||
tombstones = stats.get("tombstones", 0)
|
||
log(f" episodes={episodes} memories={memories} tombstones={tombstones}")
|
||
actions.append(f"episodes={episodes} memories={memories} tombstones={tombstones}")
|
||
|
||
# Tombstone 增长检测（和昨天比）
|
||
state_file = f"{HERMES}/.hermes/watchdog/zhiyi_stats.json"
|
||
os.makedirs(os.path.dirname(state_file), exist_ok=True)
|
||
prev = {}
|
||
if os.path.exists(state_file):
|
||
prev = json.load(open(state_file))
|
||
prev_tomb = prev.get("tombstones", 0)
|
||
prev_mem = prev.get("memories", 0)
|
||
if prev_tomb:
|
||
delta = tombstones - prev_tomb
|
||
if delta > 50:
|
||
REPORT.append(f"⚠️ 织忆 tombstone 单日增长 {delta}，注意清理")
|
||
# 保存当天快照
|
||
json.dump({"tombstones": tombstones, "memories": memories, "episodes": episodes}, open(state_file, "w"))
|
||
except Exception as e:
|
||
REPORT.append(f"❌ 织忆 stats 获取失败: {e}")
|
||
|
||
# Metrics 健康度
|
||
try:
|
||
req = urllib.request.Request(
|
||
f"{ZHIYI_URL}/api/v1/metrics",
|
||
headers={"X-API-Key": ZHIYI_KEY},
|
||
method="GET"
|
||
)
|
||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||
metrics = json.loads(resp.read().decode())
|
||
recall_hit = metrics.get("recall_hit_rate", 0)
|
||
gap_closures = metrics.get("gap_closure_rate", 0)
|
||
log(f" recall_hit={recall_hit:.2f} gap_closure={gap_closures:.2f}")
|
||
if recall_hit and recall_hit < 0.6:
|
||
REPORT.append(f"⚠️ 织忆 recall_hit={recall_hit:.2f} 偏低，建议补充相关记忆")
|
||
except Exception as e:
|
||
log(f" metrics API 不存在或失败(非致命): {e}")
|
||
|
||
# P2 离线整合（LightMem UPDATE_PROMPT）：每日自动合并相似记忆
|
||
# POST /api/v1/consolidate/memory — 找出相似记忆对 → LLM 三选一(update/delete/ignore)
|
||
try:
|
||
req = urllib.request.Request(
|
||
f"{ZHIYI_URL}/api/v1/consolidate/memory",
|
||
data=json.dumps({"namespace": "hermes-main", "limit": 50}).encode("utf-8"),
|
||
headers={"X-API-Key": ZHIYI_KEY, "Content-Type": "application/json"},
|
||
method="POST"
|
||
)
|
||
with urllib.request.urlopen(req, timeout=120) as resp:
|
||
cons = json.loads(resp.read().decode())
|
||
log(f" consolidate: processed={cons.get('processed',0)} updated={cons.get('updated',0)} deleted={cons.get('deleted',0)} ignored={cons.get('ignored',0)}")
|
||
if cons.get("processed", 0) > 0:
|
||
actions.append(f"🧹 P2 记忆整合: 处理 {cons.get('processed')} 对相似记忆 (更新 {cons.get('updated')} / 删除 {cons.get('deleted')} / 忽略 {cons.get('ignored')})")
|
||
except Exception as e:
|
||
log(f" consolidate 失败(非致命): {e}")
|
||
|
||
return actions
|
||
|
||
# ── 2. Soulful 自我升级 ─────────────────────────────────────────────────────
|
||
def upgrade_soulful():
|
||
log("=== Soulful 自检 ===")
|
||
actions = []
|
||
soulful_dir = f"{HERMES}/soulful"
|
||
|
||
# 2a. 清理 30 天以上的过期/done cares
|
||
cq_path = f"{soulful_dir}/cares-queue.json"
|
||
removed = 0
|
||
if os.path.exists(cq_path):
|
||
try:
|
||
d = json.load(open(cq_path))
|
||
today = datetime.now().date()
|
||
new_cares = []
|
||
for c in d.get("cares", []):
|
||
due = c.get("follow_up_date", "")
|
||
status = c.get("status", "pending")
|
||
if status == "done":
|
||
# done 项保留 7 天后删
|
||
if due:
|
||
try:
|
||
due_date = datetime.strptime(due, "%Y-%m-%d").date()
|
||
if (today - due_date).days > 7:
|
||
removed += 1
|
||
continue
|
||
except: pass
|
||
elif due:
|
||
try:
|
||
due_date = datetime.strptime(due, "%Y-%m-%d").date()
|
||
if (today - due_date).days > 30:
|
||
removed += 1 # 30 天前已过期
|
||
continue
|
||
except: pass
|
||
new_cares.append(c)
|
||
if removed > 0:
|
||
d["cares"] = new_cares
|
||
json.dump(d, open(cq_path, "w"), ensure_ascii=False, indent=2)
|
||
actions.append(f"清理 {removed} 条过期/旧 cares")
|
||
log(f" 清理 {removed} 条过期 cares")
|
||
except Exception as e:
|
||
REPORT.append(f"❌ Soulful cares 清理失败: {e}")
|
||
|
||
# 2b. 心迹去重
|
||
ht_path = f"{soulful_dir}/heart-traces.jsonl"
|
||
if os.path.exists(ht_path):
|
||
lines = open(ht_path, encoding="utf-8").readlines()
|
||
seen = set()
|
||
new_lines = []
|
||
dupes = 0
|
||
for line in lines:
|
||
if not line.strip():
|
||
continue
|
||
try:
|
||
e = json.loads(line)
|
||
key = e.get("content", "")[:60] # 按前60字去重
|
||
if key in seen:
|
||
dupes += 1
|
||
continue
|
||
seen.add(key)
|
||
new_lines.append(line)
|
||
except: new_lines.append(line)
|
||
if dupes > 0:
|
||
with open(ht_path, "w", encoding="utf-8") as f:
|
||
f.writelines(new_lines)
|
||
actions.append(f"心迹去重移除 {dupes} 条")
|
||
log(f" 心迹去重: 移除 {dupes} 条")
|
||
|
||
# 2c. 画像字段检查
|
||
profile_path = f"{soulful_dir}/user-profile.json"
|
||
if os.path.exists(profile_path):
|
||
profile = json.load(open(profile_path, encoding="utf-8"))
|
||
empty_fields = [k for k, v in profile.items() if not v or (isinstance(v, dict) and not any(v.values()))]
|
||
if empty_fields:
|
||
actions.append(f"画像 {len(empty_fields)} 个空字段待补充")
|
||
log(f" 画像空字段: {empty_fields}")
|
||
|
||
log(f" Soulful 升级动作: {actions or '无需操作'}")
|
||
return actions
|
||
|
||
# ── Phase 5: Consolidation 引擎 ─────────────────────────────────────────────
|
||
def _consolidate_memories():
|
||
"""
|
||
YantrikDB-style consolidation：
|
||
1. 合并相似记忆（LLM判断重复>80%）
|
||
2. 挖掘跨域模式（journal里跨category的关联）
|
||
3. 生成主动触发器写入 triggers.jsonl
|
||
"""
|
||
import urllib.request as ureq, re, hashlib
|
||
triggers = []
|
||
log("=== Phase 5: Consolidation ===")
|
||
|
||
# 1. 检测织忆重复记忆（通过 time_decay_recall 结果）
|
||
try:
|
||
# 获取织忆 recall 结果，找时间接近+内容相似的
|
||
req_body = json.dumps({"query": "牧尘 工作 项目 决策", "top_k": 10,
|
||
"agent_id": "hermes-a06", "use_rerank": True}).encode()
|
||
req = ureq.Request(f"{ZHIYI_URL}/api/v1/recall", data=req_body,
|
||
headers={"X-API-Key": ZHIYI_KEY, "Content-Type": "application/json"},
|
||
method="POST")
|
||
with ureq.urlopen(req, timeout=8) as resp:
|
||
results = json.loads(resp.read().decode()).get("results", [])
|
||
# 检查时间接近的记忆对（5天内，内容相似度阈值通过 LLM 检测）
|
||
dupes = 0
|
||
for i, r1 in enumerate(results):
|
||
for r2 in results[i+1:]:
|
||
# 简单：内容前50字符相同 → 重复
|
||
if r1.get("content","")[:50] == r2.get("content","")[:50]:
|
||
dupes += 1
|
||
if dupes > 0:
|
||
triggers.append({"type": "dedup_trigger", "severity": "info",
|
||
"content": f"织忆发现 {dupes} 对重复记忆，建议去重",
|
||
"detected_at": datetime.now().isoformat()})
|
||
except Exception as e:
|
||
log(f" Consolidation step1 failed: {e}")
|
||
|
||
# 2. 读 journal，挖掘高频行为模式
|
||
journal_path = f"{HERMES}/daemon/journal.jsonl"
|
||
if os.path.exists(journal_path):
|
||
try:
|
||
lines = open(journal_path).readlines()
|
||
entries = [json.loads(l) for l in lines[-50:] if json.loads(l).get("type") != "startup"]
|
||
# 统计 type 出现频率
|
||
from collections import Counter
|
||
type_counts = Counter(e.get("type","") for e in entries)
|
||
if type_counts:
|
||
top_type, top_count = type_counts.most_common(1)[0]
|
||
if top_count >= 5:
|
||
triggers.append({"type": "pattern_trigger", "severity": "info",
|
||
"content": f"最近50条日志中 '{top_type}' 出现 {top_count} 次，频率较高",
|
||
"detected_at": datetime.now().isoformat()})
|
||
except Exception as e:
|
||
log(f" Consolidation step2 failed: {e}")
|
||
|
||
# 3. 写触发器到文件
|
||
trigger_path = f"{HERMES}/daemon/triggers.jsonl"
|
||
existing = []
|
||
if os.path.exists(trigger_path):
|
||
try:
|
||
existing = [json.loads(l) for l in open(trigger_path) if l.strip()]
|
||
except: pass
|
||
new_triggers = [t for t in triggers
|
||
if not any(ex.get("content") == t.get("content") for ex in existing)]
|
||
if new_triggers:
|
||
existing.extend(new_triggers)
|
||
open(trigger_path, "w").writelines(json.dumps(t, ensure_ascii=False) + "\n" for t in existing[-20:])
|
||
log(f" 新增 {len(new_triggers)} 个触发器，总 {len(existing)} 个")
|
||
# 飞书通知
|
||
try:
|
||
lines = [f"• [{t['type']}] {t['content']}" for t in new_triggers[:5]]
|
||
if lines:
|
||
feishu_alert("🧠 记忆系统触发器提醒",
|
||
"\n".join(lines) + f"\n\n（最近 {len(existing)} 个触发器待处理）")
|
||
except: pass
|
||
else:
|
||
log(" 无新触发器")
|
||
|
||
# 返回可读字符串（与其它子系统 actions 契约一致：list[str]）。
|
||
# 2026-09-07 修复：此前直接返回 dict 列表，Phase5 一旦产生新触发器，
|
||
# 汇总报告 '; '.join(acts) 就会 TypeError（sequence item 0: expected str instance, dict found）。
|
||
return [f"[{t.get('type', 'trigger')}] {t.get('content', '')}" for t in triggers]
|
||
|
||
|
||
# ── 主流程 ──────────────────────────────────────────────────────────────────
|
||
if __name__ == "__main__":
|
||
log("记忆系统自我升级开始")
|
||
|
||
# ── L7 统一层：llm_context.json v2 验证 ──────────────────
|
||
try:
|
||
ctx_path = f"{HERMES}/llm_context.json"
|
||
ctx = json.load(open(ctx_path))
|
||
required = ["updated_at", "uptime_minutes", "daemon_status", "distill_status",
|
||
"user_profile", "scenes", "observations", "cares", "recent_moments"]
|
||
missing = [f for f in required if f not in ctx]
|
||
if missing:
|
||
REPORT.append(f"❌ llm_context.json 缺少字段: {missing}")
|
||
else:
|
||
log(f" L7 统一层: ✅ {len(required)} 字段完整")
|
||
st = ctx.get("observations", {})
|
||
if isinstance(st, list):
|
||
log(f" L7 short_term: {len(st)} 条")
|
||
elif isinstance(st, dict):
|
||
log(f" L7 short_term: {len(st)} 组")
|
||
else:
|
||
log(f" L7 short_term: {type(st).__name__}")
|
||
# 检查 distill_status
|
||
ds = ctx.get("distill_status", "unknown")
|
||
log(f" L7 distill_status: {ds}")
|
||
except Exception as e:
|
||
REPORT.append(f"❌ llm_context.json 读取失败: {e}")
|
||
all_actions = {}
|
||
all_actions["织忆"] = upgrade_zhiyi()
|
||
all_actions["Soulful"] = upgrade_soulful()
|
||
all_actions["Consolidation"] = _consolidate_memories()
|
||
|
||
# 汇总报告
|
||
summary = []
|
||
for sys_, acts in all_actions.items():
|
||
if acts:
|
||
summary.append(f"**{sys_}**: {'; '.join(acts)}")
|
||
|
||
if REPORT:
|
||
title = "🔴 记忆系统升级异常"
|
||
content = "\n\n".join(REPORT)
|
||
feishu_alert(title, content)
|
||
print("异常已通知飞书")
|
||
|
||
# ── 每日人话报告（2026-09-05 改造：不再只输出调试日志，输出可读日报）──
|
||
# 组装自进化/记忆质量日报，正常也输出（cron deliver 到飞书群即成为"记忆日报"）
|
||
print("\n📊 记忆系统日报 " + datetime.now().strftime("%Y-%m-%d"))
|
||
print("=" * 40)
|
||
for sys_, acts in all_actions.items():
|
||
if acts:
|
||
print(f"• {sys_}: {'; '.join(acts)}")
|
||
else:
|
||
print(f"• {sys_}: 无需动作")
|
||
if not REPORT and not summary:
|
||
print("• 状态: 三套记忆系统健康，自进化引擎运行中")
|
||
# 自进化指标行（若织忆自检收集到）
|
||
if "织忆" in all_actions and all_actions["织忆"]:
|
||
zacts = all_actions["织忆"]
|
||
zhits = [a for a in zacts if a.startswith("recall_hit") or a.startswith("episodes=")]
|
||
if zhits:
|
||
print("• 织忆数据: " + " | ".join(zhits))
|
||
print("=" * 40)
|
||
if REPORT:
|
||
print(f"⚠️ 发现 {len(REPORT)} 项异常（已飞书告警）")
|
||
else:
|
||
print("✅ 无异常") |