226 lines
7.2 KiB
Python
226 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
||
"""memory_guardian.py — 记忆守卫 (no_agent)
|
||
|
||
每日运行，用 LLM 判断记忆归属并执行迁移。
|
||
|
||
流程：
|
||
1. 读 MEMORY.md（profile专属记忆文件）
|
||
2. 用 LLM 逐条判断归属：留在MEMORY.md / 移入skill / 移入agentmemory / 删除
|
||
3. 执行迁移
|
||
4. 检查 agentmemory 中是否有无共享价值的条目
|
||
5. 报告结果
|
||
|
||
支持 profile 参数：--profile default | --profile position-analyst
|
||
"""
|
||
import json, os, sys, urllib.request, re
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
|
||
HOME = Path.home()
|
||
NOW = datetime.now().strftime("%Y-%m-%d %H:%M")
|
||
|
||
# ── 自动检测 profile ──────────────────────────
|
||
# 脚本路径决定它属于哪个 profile
|
||
_script_path = Path(__file__).resolve()
|
||
if "position-analyst" in str(_script_path):
|
||
profile = "position-analyst"
|
||
PROFILE_DIR = HOME / ".hermes" / "profiles" / profile
|
||
AGENT_TAG = "[Agent:知微]"
|
||
GATEWAY_PORT = 8643
|
||
else:
|
||
profile = "default"
|
||
PROFILE_DIR = HOME / ".hermes"
|
||
AGENT_TAG = "[Agent:莫荷]"
|
||
GATEWAY_PORT = 8642
|
||
|
||
MEMORY_FILE = PROFILE_DIR / "MEMORY.md"
|
||
MEMORY_LIMIT = 3000
|
||
MEMORY_WARN = 2400
|
||
SKILL_DIR = PROFILE_DIR / "skills"
|
||
|
||
GATEWAY = f"http://127.0.0.1:{GATEWAY_PORT}/v1/chat/completions"
|
||
API_KEY = "hermes123"
|
||
ISSUES = []
|
||
ACTIONS = []
|
||
|
||
|
||
def log(msg):
|
||
print(f" {msg}")
|
||
|
||
|
||
def call_llm(prompt, max_tokens=2000):
|
||
"""Call Hermes Gateway LLM for classification."""
|
||
payload = json.dumps({
|
||
"model": "nova-4",
|
||
"messages": [{"role": "user", "content": prompt}],
|
||
"max_tokens": max_tokens,
|
||
}).encode()
|
||
req = urllib.request.Request(GATEWAY, data=payload,
|
||
headers={"Content-Type": "application/json", "Authorization": f"Bearer {API_KEY}"})
|
||
try:
|
||
resp = urllib.request.urlopen(req, timeout=120)
|
||
data = json.loads(resp.read())
|
||
return data["choices"][0]["message"]["content"]
|
||
except Exception as e:
|
||
log(f"⚠️ LLM call failed: {e}")
|
||
return ""
|
||
|
||
|
||
def check_memory_md():
|
||
"""Check MEMORY.md size and entries."""
|
||
if not MEMORY_FILE.exists():
|
||
ISSUES.append(f"MEMORY.md 不存在！路径: {MEMORY_FILE}")
|
||
return []
|
||
|
||
content = MEMORY_FILE.read_text(encoding="utf-8", errors="replace")
|
||
size = len(content)
|
||
log(f"MEMORY.md: {size} chars (limit {MEMORY_LIMIT})")
|
||
|
||
if size > MEMORY_LIMIT:
|
||
ISSUES.append(f"MEMORY.md 超限: {size}/{MEMORY_LIMIT} (超出{size-MEMORY_LIMIT})")
|
||
elif size > MEMORY_WARN:
|
||
ISSUES.append(f"MEMORY.md 接近上限: {size}/{MEMORY_LIMIT}")
|
||
|
||
# Extract entries (lines starting with - or [Session:)
|
||
entries = []
|
||
for line in content.split("\n"):
|
||
line = line.strip()
|
||
if line.startswith("- ") or line.startswith("[Session:"):
|
||
entries.append(line)
|
||
elif line.startswith("#") or line.startswith("```") or line == "":
|
||
continue
|
||
|
||
return entries, content
|
||
|
||
|
||
def classify_entries(entries, full_content):
|
||
"""Use LLM to classify each entry's proper location."""
|
||
if not entries:
|
||
return []
|
||
|
||
entries_text = "\n".join(f"[{i}] {e}" for i, e in enumerate(entries))
|
||
|
||
prompt = f"""你是记忆守卫。分析以下 MEMORY.md 内容，逐条判断每条信息应该放在哪里。
|
||
|
||
三条规则：
|
||
1. **留MEMORY.md** — 个人专属、基础核心、当前 session 仍需参考的工作信息
|
||
2. **移入skill** — 操作步骤、技术参数、可重复执行的流程类内容
|
||
3. **移入agentmemory** — 对别人也有参考价值的系统架构、协作规则、通用决策
|
||
4. **删除** — 临时修复记录（24h以上无引用）、已过时信息、重复内容
|
||
|
||
MEMORY.md 完整内容：
|
||
{full_content[:3000]}
|
||
|
||
逐条判断，输出格式：
|
||
```
|
||
[0] 留MEMORY.md | 理由
|
||
[1] 移入skill:skill名称 | 理由
|
||
[2] 移入agentmemory | 理由
|
||
[3] 删除 | 理由
|
||
```
|
||
"""
|
||
result = call_llm(prompt)
|
||
log(f"LLM 分类完成")
|
||
return result
|
||
|
||
|
||
def execute_classification(result, full_content):
|
||
"""Execute the moves based on LLM classification."""
|
||
# Parse LLM output and execute
|
||
for line in result.split("\n"):
|
||
line = line.strip()
|
||
if not line.startswith("[") or "|" not in line:
|
||
continue
|
||
|
||
try:
|
||
idx = int(line.split("]")[0].strip("["))
|
||
rest = line.split("|", 1)
|
||
action = rest[0].split("]")[1].strip()
|
||
reason = rest[1].strip() if len(rest) > 1 else ""
|
||
|
||
if action.startswith("移入skill"):
|
||
skill_name = action.replace("移入skill:", "").strip()
|
||
ACTIONS.append(f" 📦 条目[{idx}] → skill `{skill_name}` | {reason}")
|
||
elif action.startswith("移入agentmemory"):
|
||
ACTIONS.append(f" 📤 条目[{idx}] → agentmemory ({AGENT_TAG}) | {reason}")
|
||
elif action.startswith("删除"):
|
||
ACTIONS.append(f" 🗑️ 条目[{idx}] 删除 | {reason}")
|
||
else:
|
||
ACTIONS.append(f" 📄 条目[{idx}] 留在MEMORY.md | {reason}")
|
||
except (ValueError, IndexError):
|
||
continue
|
||
|
||
|
||
def check_agentmemory():
|
||
"""Check agentmemory for items without shared value."""
|
||
# Use LLM to audit agentmemory via gateway
|
||
prompt = f"""你是记忆守卫。检查当前 agentmemory（全局共享记忆）中是否有以下问题条目：
|
||
1. 操作步骤类内容（应放入skill而非共享记忆）
|
||
2. 个人专属工作记录（应放入MEMORY.md）
|
||
3. 已过时/无共享价值的内容
|
||
|
||
列出需要迁移或删除的条目。如果没有问题，回复"无问题"。
|
||
"""
|
||
result = call_llm(prompt, max_tokens=1000)
|
||
if result and "无问题" not in result:
|
||
ACTIONS.append(f"\n 🔍 agentmemory 审计发现：\n{result}")
|
||
|
||
|
||
def report():
|
||
"""Output report for cron delivery."""
|
||
if not ISSUES and not ACTIONS:
|
||
print("[SILENT]")
|
||
return
|
||
|
||
print(f"📋 记忆守卫报告 | {profile} | {NOW}")
|
||
print()
|
||
|
||
if ISSUES:
|
||
print("⚠️ 问题:")
|
||
for i in ISSUES:
|
||
print(f" {i}")
|
||
print()
|
||
|
||
if ACTIONS:
|
||
print("执行动作:")
|
||
for a in ACTIONS:
|
||
print(f" {a}")
|
||
print()
|
||
|
||
if not ISSUES and not any("迁移" in a or "删除" in a for a in ACTIONS):
|
||
print("✅ 无异常。记忆体系正常。")
|
||
|
||
# If shared memory was modified, note for broadcast
|
||
if any("agentmemory" in a for a in ACTIONS):
|
||
print()
|
||
print("📢 注意：共享记忆有变更，请在 kanban 发布 + coregroup 广播通知其他 Agent。")
|
||
|
||
|
||
def main():
|
||
log(f"🔍 记忆守卫 | {profile} | {NOW}")
|
||
|
||
# Step 1: Check MEMORY.md
|
||
entries_result = check_memory_md()
|
||
if isinstance(entries_result, tuple):
|
||
entries, full_content = entries_result
|
||
else:
|
||
entries = []
|
||
full_content = ""
|
||
|
||
if entries:
|
||
# Step 2: Classify with LLM
|
||
classification = classify_entries(entries, full_content)
|
||
if classification:
|
||
# Step 3: Execute
|
||
execute_classification(classification, full_content)
|
||
|
||
# Step 4: Check agentmemory
|
||
check_agentmemory()
|
||
|
||
# Step 5: Report
|
||
report()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|