#!/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("✅ 无异常")