252 lines
8.2 KiB
Python
252 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
小唯主动学习体系 — 长期学习计划
|
||
================================
|
||
四个维度:
|
||
1. 现有工作优化 (optimizer / skill-manager)
|
||
2. 新工具/新技能探索 (skill-scan)
|
||
3. 业务领域知识积累 (operation/product/growth research)
|
||
4. 系统智能化 (self-evolve / learner — 已有)
|
||
|
||
用法:
|
||
python3 proactive_learning.py report → 产出自检报告
|
||
python3 proactive_learning.py learn → 执行全部学习维度的检查
|
||
python3 proactive_learning.py status → 查看四个维度的当前状态
|
||
"""
|
||
|
||
import json, os, sys, subprocess
|
||
from datetime import datetime, timezone
|
||
|
||
HOME = os.path.expanduser("~")
|
||
HERMES = HOME + "/.hermes"
|
||
D = HERMES + "/daemon"
|
||
LEARN_DIR = HERMES + "/proactive_learning"
|
||
STATE_FILE = LEARN_DIR + "/state.json"
|
||
os.makedirs(LEARN_DIR, exist_ok=True)
|
||
|
||
|
||
def log(msg):
|
||
ts = datetime.now().strftime("%H:%M:%S")
|
||
print(f"[PROLEARN] {ts} {msg}", flush=True)
|
||
|
||
|
||
def load_state():
|
||
if os.path.exists(STATE_FILE):
|
||
with open(STATE_FILE) as f:
|
||
return json.load(f)
|
||
return {
|
||
"dimensions": {
|
||
"existing_work": {"last_run": None, "status": "pending", "findings": []},
|
||
"new_tools": {"last_run": None, "status": "pending", "findings": []},
|
||
"domain_knowledge": {"last_run": None, "status": "pending", "topics": []},
|
||
"system_smart": {"last_run": None, "status": "ok"},
|
||
},
|
||
"cycles": 0,
|
||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||
}
|
||
|
||
|
||
def save_state(state):
|
||
with open(STATE_FILE, "w") as f:
|
||
json.dump(state, f, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def shell(cmd, timeout=15):
|
||
try:
|
||
r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout)
|
||
return r.returncode, r.stdout.strip()[:500], r.stderr.strip()[:200]
|
||
except subprocess.TimeoutExpired:
|
||
return -1, "", "timeout"
|
||
|
||
|
||
# ===================== 维度1: 现有工作优化 =====================
|
||
|
||
def check_existing_work(state):
|
||
"""检查 skill 健康分 / optimizer 报告 / daemon 错误日志"""
|
||
dim = state["dimensions"]["existing_work"]
|
||
dim["status"] = "running"
|
||
|
||
findings = []
|
||
|
||
# 1. Skill 健康分
|
||
rc, out, err = shell("python3 ~/.hermes/scripts/skill-manager.py scan --quiet", timeout=30)
|
||
if rc == 0:
|
||
findings.append("✅ skill-manager 扫描正常")
|
||
else:
|
||
findings.append(f"⚠️ skill-manager 异常: {err or out}")
|
||
|
||
# 2. Optimizer 报告
|
||
opt_report = HERMES + "/optimization-report.json"
|
||
if os.path.exists(opt_report):
|
||
mtime = datetime.fromtimestamp(os.path.getmtime(opt_report), tz=timezone.utc)
|
||
age_h = (datetime.now(timezone.utc) - mtime).total_seconds() / 3600
|
||
findings.append(f"📊 optimizer 报告 {age_h:.1f}h 前更新")
|
||
else:
|
||
findings.append("ℹ️ optimizer 报告尚无历史数据")
|
||
|
||
# 3. Daemon journal 近期错误
|
||
journal = D + "/journal.jsonl"
|
||
recent_errors = []
|
||
if os.path.exists(journal):
|
||
lines = open(journal).readlines()
|
||
for line in lines[-100:]:
|
||
try:
|
||
entry = json.loads(line)
|
||
if entry.get("type") == "error" or "error" in entry.get("message", "").lower():
|
||
recent_errors.append(entry.get("message", "")[:80])
|
||
except:
|
||
pass
|
||
if recent_errors:
|
||
findings.append(f"⚠️ daemon journal 近100条中 {len(recent_errors)} 条错误")
|
||
else:
|
||
findings.append("✅ daemon journal 无错误")
|
||
|
||
dim["findings"] = findings
|
||
dim["last_run"] = datetime.now(timezone.utc).isoformat()
|
||
dim["status"] = "ok"
|
||
return findings
|
||
|
||
|
||
# ===================== 维度2: 新工具/新技能探索 =====================
|
||
|
||
def check_new_tools(state):
|
||
"""扫描 skills 目录，识别低分/缺失领域，提出新技能建议"""
|
||
dim = state["dimensions"]["new_tools"]
|
||
dim["status"] = "running"
|
||
|
||
findings = []
|
||
|
||
# 读取 skill-manager 评分
|
||
rc, out, err = shell("python3 ~/.hermes/scripts/skill-manager.py scan", timeout=30)
|
||
skill_json = HERMES + "/skill-health.json"
|
||
|
||
low_score_skills = []
|
||
if os.path.exists(skill_json):
|
||
data = json.load(open(skill_json))
|
||
for cat in data.get("categories", {}).values():
|
||
for skill in cat.get("skills", []):
|
||
if skill.get("score", 10) < 6:
|
||
low_score_skills.append(f"{skill['name']}({skill['score']})")
|
||
|
||
if low_score_skills:
|
||
findings.append(f"📉 低分技能({len(low_score_skills)}): {', '.join(low_score_skills[:5])}")
|
||
else:
|
||
findings.append("✅ 技能评分无明显短板")
|
||
|
||
# 扫描 skills 目录，列出最近添加
|
||
skills_dir = HERMES + "/skills"
|
||
all_skills = []
|
||
if os.path.exists(skills_dir):
|
||
for root, dirs, files in os.walk(skills_dir):
|
||
for f in files:
|
||
if f == "SKILL.md":
|
||
skill_name = os.path.basename(root)
|
||
mtime = os.path.getmtime(os.path.join(root, f))
|
||
all_skills.append((skill_name, mtime))
|
||
|
||
all_skills.sort(key=lambda x: x[1], reverse=True)
|
||
recent = all_skills[:5]
|
||
findings.append(f"📦 共 {len(all_skills)} 个 skill，最近: {', '.join([s[0] for s in recent])}")
|
||
|
||
dim["findings"] = findings
|
||
dim["last_run"] = datetime.now(timezone.utc).isoformat()
|
||
dim["status"] = "ok"
|
||
return findings
|
||
|
||
|
||
# ===================== 维度3: 业务领域知识积累 =====================
|
||
|
||
def check_domain_knowledge(state):
|
||
"""检查领域知识积累状态"""
|
||
dim = state["dimensions"]["domain_knowledge"]
|
||
dim["status"] = "running"
|
||
|
||
findings = []
|
||
topics_file = LEARN_DIR + "/topics.json"
|
||
|
||
topics = []
|
||
if os.path.exists(topics_file):
|
||
topics = json.load(open(topics_file)).get("topics", [])
|
||
|
||
if not topics:
|
||
findings.append("ℹ️ 尚未定义学习主题（请告诉我想深入的方向）")
|
||
else:
|
||
for t in topics:
|
||
status = t.get("status", "pending")
|
||
last = t.get("last_research", "从未")
|
||
findings.append(f"📚 {t['name']}: {status}（上次: {last}）")
|
||
|
||
dim["topics"] = topics
|
||
dim["last_run"] = datetime.now(timezone.utc).isoformat()
|
||
dim["status"] = "ok"
|
||
return findings
|
||
|
||
|
||
# ===================== 维度4: 系统智能化 =====================
|
||
|
||
def check_system_smart(state):
|
||
"""检查 self-evolve / learner 运行状态"""
|
||
dim = state["dimensions"]["system_smart"]
|
||
dim["status"] = "running"
|
||
|
||
findings = []
|
||
|
||
# self-evolve 最近运行
|
||
rc, out, _ = shell("ls -t ~/.hermes/*.log 2>/dev/null | head -1", timeout=5)
|
||
if rc == 0 and out:
|
||
findings.append(f"📝 最近日志: {os.path.basename(out)}")
|
||
else:
|
||
findings.append("ℹ️ 暂无运行日志")
|
||
|
||
dim["findings"] = findings
|
||
dim["last_run"] = datetime.now(timezone.utc).isoformat()
|
||
dim["status"] = "ok"
|
||
return findings
|
||
|
||
|
||
# ===================== 统一执行 =====================
|
||
|
||
def run_all_checks():
|
||
state = load_state()
|
||
state["cycles"] += 1
|
||
|
||
results = {}
|
||
results["existing_work"] = check_existing_work(state)
|
||
results["new_tools"] = check_new_tools(state)
|
||
results["domain_knowledge"] = check_domain_knowledge(state)
|
||
results["system_smart"] = check_system_smart(state)
|
||
|
||
save_state(state)
|
||
|
||
# 输出报告
|
||
report = f"**小唯主动学习报告** #{state['cycles']}\n"
|
||
report += f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n"
|
||
|
||
labels = {
|
||
"existing_work": "🔧 现有工作优化",
|
||
"new_tools": "🛠️ 新工具/新技能",
|
||
"domain_knowledge": "📖 业务领域知识",
|
||
"system_smart": "🧠 系统智能化",
|
||
}
|
||
|
||
for key, label in labels.items():
|
||
report += f"**{label}**\n"
|
||
for f in results[key]:
|
||
report += f" {f}\n"
|
||
report += "\n"
|
||
|
||
return report
|
||
|
||
|
||
if __name__ == "__main__":
|
||
cmd = sys.argv[1] if len(sys.argv) > 1 else "report"
|
||
|
||
if cmd == "report":
|
||
print(run_all_checks())
|
||
elif cmd == "status":
|
||
state = load_state()
|
||
print(json.dumps(state["dimensions"], ensure_ascii=False, indent=2))
|
||
elif cmd == "learn":
|
||
print(run_all_checks())
|
||
else:
|
||
print(f"用法: proactive_learning.py [report|status|learn]") |