531 lines
19 KiB
Python
Executable File
531 lines
19 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
自我优化系统 — 采集、分析、推荐
|
||
===================================
|
||
用法:
|
||
optimizer.py collect → 从各数据源采集指标
|
||
optimizer.py analyze → 分析瓶颈和模式
|
||
optimizer.py recommend → 输出优化建议
|
||
optimizer.py report → 完整报告（收集+分析+推荐）
|
||
"""
|
||
|
||
import json, os, re, sys, time
|
||
from datetime import datetime, timezone, timedelta
|
||
from collections import defaultdict
|
||
|
||
HOME = os.path.expanduser("~")
|
||
HERMES = HOME + "/.hermes"
|
||
D = HERMES + "/daemon"
|
||
REPORT_FILE = HERMES + "/optimization-report.json"
|
||
|
||
# ====== 采集器 ======
|
||
|
||
def collect_daemon_metrics():
|
||
"""从 daemon 日志中提取指标"""
|
||
metrics = {
|
||
"uptime_minutes": 0,
|
||
"total_ticks": 0,
|
||
"deep_thoughts": 0,
|
||
"alerts_sent": 0,
|
||
"solutions_applied": 0,
|
||
"learned_solutions": 0,
|
||
"errors": 0,
|
||
"crashes": 0,
|
||
"model_calls": 0,
|
||
"model_tokens": 0,
|
||
}
|
||
|
||
# 从 context.json 读取
|
||
ctx_file = D + "/context.json"
|
||
if os.path.exists(ctx_file):
|
||
with open(ctx_file) as f:
|
||
ctx = json.load(f)
|
||
metrics["uptime_minutes"] = ctx.get("uptime_seconds", 0) // 60
|
||
metrics["total_ticks"] = ctx.get("tick_count", 0)
|
||
metrics["deep_thoughts"] = ctx.get("deep_tick_count", 0)
|
||
metrics["alerts_sent"] = ctx.get("messages_sent", 0)
|
||
metrics["solutions_applied"] = ctx.get("solved_count", 0)
|
||
metrics["learned_solutions"] = ctx.get("learned_count", 0)
|
||
|
||
# 从 daemon.log 提取崩溃和错误
|
||
# ⚠️ 2026-09-06 修复：
|
||
# 1) 只统计日志尾部 2MB（≈最近1-2天），避免把历史(如 07-25 NVML 6145 次崩溃)当"当前崩溃"；
|
||
# 2) 旧代码用子串 count("Error"/"error"/"失败") 会把 deep_think 的 LLM 输出文本
|
||
# (如 soulful 行"沮丧情绪（失败）") 也数进去 → 报出 60295 个假错误。
|
||
# 改为逐行分类：只统计 [DAEMON] 时间戳前缀行（daemon 自己的 log() 写的行），
|
||
# 跳过 LLM 输出续行；崩溃=行首"❌ 崩溃"，错误=⚠️/[call_llm] 失败标记行。
|
||
log_file = D + "/daemon.log"
|
||
if os.path.exists(log_file):
|
||
with open(log_file, "rb") as f:
|
||
f.seek(0, 2) # 到末尾
|
||
size = f.tell()
|
||
f.seek(max(0, size - 2 * 1024 * 1024)) # 尾部 2MB
|
||
content = f.read().decode("utf-8", errors="ignore")
|
||
line_re = re.compile(r"^\[DAEMON\] \d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} (.*)$")
|
||
call_re = re.compile(r"^\[DAEMON\] \d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\s+深度思考 \((\d+)t\)")
|
||
err_words = ("失败", "异常", "错误", "failed", "Failed", "超时", "timeout", "Timeout")
|
||
crashes = errors = model_calls = model_tokens = 0
|
||
for line in content.splitlines():
|
||
m = line_re.match(line)
|
||
if not m:
|
||
continue # LLM 输出续行等非 daemon 行，跳过
|
||
cm = call_re.match(line)
|
||
if cm:
|
||
model_calls += 1
|
||
model_tokens += int(cm.group(1))
|
||
continue
|
||
msg = m.group(1).lstrip()
|
||
if msg.startswith("❌ 崩溃"):
|
||
crashes += 1
|
||
elif (msg.startswith(("⚠️", "❌")) or msg.startswith("[call_llm]")) \
|
||
and any(w in msg for w in err_words):
|
||
errors += 1
|
||
elif "Traceback" in msg or re.search(r"\b(Error|ERROR|Exception)\b", msg):
|
||
errors += 1
|
||
metrics["crashes"] = crashes
|
||
metrics["errors"] = errors
|
||
# 模型调用次数 + token 数（只从"深度思考 (Nt)"调用行取，N=该次调用 tokens）
|
||
metrics["model_calls"] = model_calls
|
||
metrics["model_tokens"] = model_tokens
|
||
|
||
return metrics
|
||
|
||
|
||
def collect_cron_metrics():
|
||
"""从 cron job 状态采集指标"""
|
||
metrics = {
|
||
"total_jobs": 0,
|
||
"ok_jobs": 0,
|
||
"failed_jobs": 0,
|
||
"no_agent_jobs": 0,
|
||
"agent_jobs": 0,
|
||
}
|
||
|
||
# 从 cron/jobs.json 读取
|
||
# ⚠️ 2026-09-06 修复：jobs.json 顶层是 dict {"jobs": [57个任务], "updated_at": ...}，
|
||
# 旧代码把 dict.values() 当任务列表 → total_jobs 只数到顶层键个数(2)，
|
||
# 57 个真实任务全部漏计（报告显示 "2任务 0成功"）。
|
||
jobs_file = HERMES + "/cron/jobs.json"
|
||
if os.path.exists(jobs_file):
|
||
with open(jobs_file) as f:
|
||
try:
|
||
data = json.load(f)
|
||
if isinstance(data, dict):
|
||
jobs = data.get("jobs", [])
|
||
# 兜底: 兼容 {job_id: job} 形态
|
||
if not isinstance(jobs, list) and jobs:
|
||
jobs = list(jobs.values())
|
||
elif isinstance(data, list):
|
||
jobs = data
|
||
else:
|
||
jobs = []
|
||
|
||
metrics["total_jobs"] = len(jobs)
|
||
for j in jobs:
|
||
if isinstance(j, dict):
|
||
status = j.get("last_status") or ""
|
||
if status == "ok":
|
||
metrics["ok_jobs"] += 1
|
||
elif status in ("error", "failed"):
|
||
metrics["failed_jobs"] += 1
|
||
if j.get("no_agent"):
|
||
metrics["no_agent_jobs"] += 1
|
||
else:
|
||
metrics["agent_jobs"] += 1
|
||
except:
|
||
pass
|
||
|
||
return metrics
|
||
|
||
|
||
def collect_skill_metrics():
|
||
"""从技能健康报告采集指标"""
|
||
metrics = {
|
||
"total_skills": 0,
|
||
"avg_score": 0,
|
||
"grade_distribution": {},
|
||
"needs_attention": 0,
|
||
"categories": 0,
|
||
}
|
||
|
||
health_file = HERMES + "/skill-health.json"
|
||
if os.path.exists(health_file):
|
||
with open(health_file) as f:
|
||
try:
|
||
report = json.load(f)
|
||
s = report.get("summary", {})
|
||
metrics["total_skills"] = s.get("active", 0)
|
||
metrics["avg_score"] = s.get("avg_score", 0)
|
||
metrics["grade_distribution"] = s.get("grades", {})
|
||
metrics["needs_attention"] = s.get("needs_attention", 0)
|
||
metrics["categories"] = s.get("categories", 0)
|
||
except:
|
||
pass
|
||
|
||
return metrics
|
||
|
||
|
||
def collect_model_metrics():
|
||
"""从模型健康报告采集指标"""
|
||
metrics = {
|
||
"total_models_tested": 0,
|
||
"stable_models": 0,
|
||
"unstable_models": 0,
|
||
"dead_models": 0,
|
||
"fastest_model": "",
|
||
"fastest_latency": 0,
|
||
}
|
||
|
||
health_file = HERMES + "/model-health.json"
|
||
if os.path.exists(health_file):
|
||
with open(health_file) as f:
|
||
try:
|
||
data = json.load(f)
|
||
metrics["total_models_tested"] = data.get("total_models", 0)
|
||
metrics["stable_models"] = data.get("stable", 0)
|
||
metrics["unstable_models"] = data.get("unstable", 0)
|
||
metrics["dead_models"] = data.get("dead", 0)
|
||
fastest = data.get("fastest_stable", [])
|
||
metrics["fastest_model"] = fastest[0] if fastest else ""
|
||
except:
|
||
pass
|
||
|
||
return metrics
|
||
|
||
|
||
def collect_disk_trend():
|
||
"""采集磁盘趋势（过去N次看门狗记录）"""
|
||
# 从看门狗日志提取
|
||
watchdog_log = HERMES + "/watchdog"
|
||
snapshots = []
|
||
|
||
# 从 daemon context 读取最后状态
|
||
ctx_file = D + "/context.json"
|
||
if os.path.exists(ctx_file):
|
||
with open(ctx_file) as f:
|
||
ctx = json.load(f)
|
||
last = ctx.get("last_state", {})
|
||
snapshots.append({
|
||
"disk_pct": last.get("disk_pct", 0),
|
||
"mem_pct": last.get("mem_pct", 0),
|
||
"timestamp": ctx.get("last_light_tick", ""),
|
||
})
|
||
|
||
# 从 daemon journal 中提取历史磁盘数据
|
||
journal_file = D + "/journal.jsonl"
|
||
if os.path.exists(journal_file):
|
||
with open(journal_file) as f:
|
||
for line in f:
|
||
try:
|
||
entry = json.loads(line)
|
||
if entry.get("type") == "startup":
|
||
snapshots.append({
|
||
"timestamp": entry.get("timestamp", ""),
|
||
"event": "startup",
|
||
})
|
||
except:
|
||
pass
|
||
|
||
return snapshots
|
||
|
||
|
||
# ====== 分析引擎 ======
|
||
|
||
def analyze_bottlenecks(metrics):
|
||
"""分析性能瓶颈"""
|
||
bottlenecks = []
|
||
|
||
# 模型瓶颈
|
||
model = metrics.get("model", {})
|
||
if model.get("stable_models", 0) < 2:
|
||
bottlenecks.append({
|
||
"area": "model",
|
||
"severity": "high",
|
||
"desc": f"可用模型不足 (稳定{model.get('stable_models',0)}/{model.get('total_models_tested',0)})",
|
||
"suggestion": "扩大模型测试范围，或检查 NewAPI 状态",
|
||
})
|
||
if model.get("dead_models", 0) > 10:
|
||
bottlenecks.append({
|
||
"area": "model",
|
||
"severity": "info",
|
||
"desc": f"大量模型不可用 ({model.get('dead_models',0)}个dead)",
|
||
"suggestion": "可能 NewAPI 后端负载高，特定时段再测",
|
||
})
|
||
|
||
# 技能瓶颈
|
||
skill = metrics.get("skill", {})
|
||
if skill.get("avg_score", 10) < 5:
|
||
bottlenecks.append({
|
||
"area": "skill",
|
||
"severity": "medium",
|
||
"desc": f"技能平均质量偏低 ({skill.get('avg_score',0)}/10)",
|
||
"suggestion": "运行 skill-manager.py fix 修复元数据，归档低分技能",
|
||
})
|
||
if skill.get("needs_attention", 0) > 30:
|
||
bottlenecks.append({
|
||
"area": "skill",
|
||
"severity": "low",
|
||
"desc": f"{skill.get('needs_attention',0)}个技能需关注",
|
||
"suggestion": "逐步清理或升级这些技能",
|
||
})
|
||
|
||
# Daemon 瓶颈
|
||
daemon = metrics.get("daemon", {})
|
||
if daemon.get("crashes", 0) > 0:
|
||
bottlenecks.append({
|
||
"area": "daemon",
|
||
"severity": "high",
|
||
"desc": f"Daemon 崩溃 {daemon.get('crashes',0)} 次",
|
||
"suggestion": "检查 daemon.log 定位崩溃原因",
|
||
})
|
||
if daemon.get("errors", 0) > 5:
|
||
bottlenecks.append({
|
||
"area": "daemon",
|
||
"severity": "medium",
|
||
"desc": f"Daemon 有 {daemon.get('errors',0)} 个错误",
|
||
"suggestion": "审查 daemon 日志中的错误模式",
|
||
})
|
||
|
||
# 效率分析
|
||
if daemon.get("deep_thoughts", 0) > 0 and daemon.get("model_tokens", 0) > 0:
|
||
avg_tokens = daemon["model_tokens"] / daemon["deep_thoughts"]
|
||
if avg_tokens > 1000:
|
||
bottlenecks.append({
|
||
"area": "efficiency",
|
||
"severity": "info",
|
||
"desc": f"深度思考平均 {int(avg_tokens)} tokens/次（偏高）",
|
||
"suggestion": "考虑精简深度思考的 system prompt",
|
||
})
|
||
|
||
return bottlenecks
|
||
|
||
|
||
def analyze_trends(daemon_metrics, model_metrics):
|
||
"""分析趋势"""
|
||
trends = []
|
||
|
||
# 如果 daemon 运行超过 1h，检查稳定性
|
||
if daemon_metrics.get("uptime_minutes", 0) > 60:
|
||
error_rate = daemon_metrics.get("errors", 0) / max(daemon_metrics.get("uptime_minutes", 1), 1)
|
||
if error_rate < 0.1:
|
||
trends.append({"type": "positive", "desc": f"Daemon 稳定运行 {daemon_metrics['uptime_minutes']}分钟，错误率低"})
|
||
else:
|
||
trends.append({"type": "negative", "desc": f"Daemon 错误率 {error_rate:.2f}/分钟"})
|
||
|
||
# 模型稳定率
|
||
if model_metrics.get("total_models_tested", 0) > 0:
|
||
stable_rate = model_metrics.get("stable_models", 0) / model_metrics.get("total_models_tested", 1) * 100
|
||
trends.append({"type": "neutral", "desc": f"模型稳定率 {stable_rate:.0f}%（{model_metrics.get('stable_models',0)}/{model_metrics.get('total_models_tested',0)}）"})
|
||
|
||
return trends
|
||
|
||
|
||
# ====== 推荐系统 ======
|
||
|
||
def generate_recommendations(metrics, bottlenecks, trends):
|
||
"""基于分析结果生成可执行建议"""
|
||
recs = []
|
||
|
||
# 从瓶颈推导建议
|
||
for b in bottlenecks:
|
||
recs.append({
|
||
"priority": b["severity"],
|
||
"area": b["area"],
|
||
"action": b["suggestion"],
|
||
"expected_impact": "",
|
||
"effort": "5min" if b["severity"] == "low" else "15min" if b["severity"] == "medium" else "30min",
|
||
})
|
||
|
||
# 从趋势推导建议
|
||
skill = metrics.get("skill", {})
|
||
if skill.get("avg_score", 10) < 6:
|
||
recs.append({
|
||
"priority": "medium",
|
||
"area": "skill",
|
||
"action": "运行 skill-manager.py archive <name> 归档低分技能（D级44个）",
|
||
"expected_impact": "减少技能库噪音，提升检索质量",
|
||
"effort": "20min",
|
||
})
|
||
|
||
model = metrics.get("model", {})
|
||
if model.get("fastest_model"):
|
||
recs.append({
|
||
"priority": "info",
|
||
"area": "model",
|
||
"action": f"考虑将默认模型切换到 {model['fastest_model']}（当前最快稳定模型）",
|
||
"expected_impact": "提升响应速度",
|
||
"effort": "2min",
|
||
})
|
||
|
||
# 提示用户更新
|
||
recs.append({
|
||
"priority": "info",
|
||
"area": "system",
|
||
"action": "运行 optimizer.py collect 定期采集指标，积累数据后分析更准确",
|
||
"expected_impact": "更精准的优化建议",
|
||
"effort": "0min",
|
||
})
|
||
|
||
return recs
|
||
|
||
|
||
# ====== 报告生成 ======
|
||
|
||
def generate_report():
|
||
"""生成完整优化报告"""
|
||
print("📊 采集指标...")
|
||
daemon = collect_daemon_metrics()
|
||
cron = collect_cron_metrics()
|
||
skill = collect_skill_metrics()
|
||
model = collect_model_metrics()
|
||
|
||
metrics = {
|
||
"daemon": daemon,
|
||
"cron": cron,
|
||
"skill": skill,
|
||
"model": model,
|
||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||
}
|
||
|
||
print("🔍 分析瓶颈...")
|
||
bottlenecks = analyze_bottlenecks(metrics)
|
||
trends = analyze_trends(daemon, model)
|
||
|
||
print("💡 生成建议...")
|
||
recommendations = generate_recommendations(metrics, bottlenecks, trends)
|
||
|
||
# 整合报告
|
||
report = {
|
||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||
"metrics": metrics,
|
||
"bottlenecks": bottlenecks,
|
||
"trends": trends,
|
||
"recommendations": recommendations,
|
||
"health_score": calculate_health_score(metrics, bottlenecks),
|
||
}
|
||
|
||
# 保存
|
||
os.makedirs(os.path.dirname(REPORT_FILE), exist_ok=True)
|
||
with open(REPORT_FILE, "w") as f:
|
||
json.dump(report, f, indent=2, ensure_ascii=False)
|
||
|
||
# 打印
|
||
print_report(report)
|
||
|
||
return report
|
||
|
||
|
||
def calculate_health_score(metrics, bottlenecks):
|
||
"""计算系统健康分 (0-100)"""
|
||
score = 100
|
||
|
||
# 减分项
|
||
for b in bottlenecks:
|
||
if b["severity"] == "high":
|
||
score -= 20
|
||
elif b["severity"] == "medium":
|
||
score -= 10
|
||
elif b["severity"] == "low":
|
||
score -= 5
|
||
|
||
# 加分项
|
||
daemon = metrics.get("daemon", {})
|
||
if daemon.get("crashes", 0) == 0:
|
||
score += 5
|
||
if daemon.get("uptime_minutes", 0) > 60:
|
||
score += 5
|
||
|
||
# 技能加分
|
||
skill = metrics.get("skill", {})
|
||
if skill.get("avg_score", 0) > 5:
|
||
score += 5
|
||
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def print_report(report):
|
||
"""打印人类可读报告"""
|
||
m = report["metrics"]
|
||
d = m["daemon"]
|
||
c = m["cron"]
|
||
s = m["skill"]
|
||
mo = m["model"]
|
||
|
||
print(f'\n{"="*50}')
|
||
print(f' 自我优化报告 | 健康分: {report["health_score"]}/100')
|
||
print(f'{"="*50}')
|
||
|
||
print(f'\n📡 Daemon:')
|
||
print(f' 运行 {d["uptime_minutes"]}分钟 | {d["total_ticks"]} ticks | {d["deep_thoughts"]}次思考')
|
||
print(f' 告警 {d["alerts_sent"]}次 | 解决问题 {d["solutions_applied"]}个 | 学会 {d["learned_solutions"]}个')
|
||
print(f' 模型调用 {d["model_calls"]}次 | {d["model_tokens"]} tokens | 错误 {d["errors"]}次')
|
||
|
||
print(f'\n⏰ Cron:')
|
||
print(f' {c["total_jobs"]}个任务 | {c["ok_jobs"]}成功 | {c["failed_jobs"]}失败 | {c["no_agent_jobs"]}个no_agent')
|
||
|
||
print(f'\n🛠️ 技能:')
|
||
print(f' {s["total_skills"]}个活跃 | 均分{s["avg_score"]}/10 | {s["needs_attention"]}个需关注')
|
||
g = s.get("grade_distribution", {})
|
||
print(f' 分布: A={g.get("A",0)} B={g.get("B",0)} C={g.get("C",0)} D={g.get("D",0)}')
|
||
|
||
print(f'\n🤖 模型:')
|
||
print(f' 测试{mo["total_models_tested"]}个 | 稳定{mo["stable_models"]}个 | 最快: {mo["fastest_model"]}')
|
||
|
||
if report["bottlenecks"]:
|
||
print(f'\n⚠️ 瓶颈 ({len(report["bottlenecks"])}个):')
|
||
for b in sorted(report["bottlenecks"], key=lambda x: {"high": 0, "medium": 1, "low": 2, "info": 3}[x["severity"]]):
|
||
icon = {"high": "🔴", "medium": "🟡", "low": "🟢", "info": "ℹ️"}[b["severity"]]
|
||
print(f' {icon} [{b["area"]}] {b["desc"]}')
|
||
print(f' → {b["suggestion"]}')
|
||
|
||
if report["trends"]:
|
||
print(f'\n📈 趋势:')
|
||
for t in report["trends"]:
|
||
icon = {"positive": "✅", "negative": "📉", "neutral": "➡️"}[t["type"]]
|
||
print(f' {icon} {t["desc"]}')
|
||
|
||
if report["recommendations"]:
|
||
print(f'\n💡 建议 ({len(report["recommendations"])}条):')
|
||
for r in sorted(report["recommendations"], key=lambda x: {"high": 0, "medium": 1, "low": 2, "info": 3}[x["priority"]]):
|
||
icon = {"high": "🔴", "medium": "🟡", "low": "🟢", "info": "ℹ️"}[r["priority"]]
|
||
print(f' {icon} [{r["area"]}] {r["action"]} ({r["effort"]})')
|
||
|
||
print(f'\n{"="*50}\n')
|
||
|
||
|
||
if __name__ == "__main__":
|
||
cmd = sys.argv[1] if len(sys.argv) > 1 else "report"
|
||
|
||
if cmd == "collect":
|
||
print(json.dumps({
|
||
"daemon": collect_daemon_metrics(),
|
||
"cron": collect_cron_metrics(),
|
||
"skill": collect_skill_metrics(),
|
||
"model": collect_model_metrics(),
|
||
}, indent=2, ensure_ascii=False))
|
||
elif cmd == "analyze":
|
||
metrics = {
|
||
"daemon": collect_daemon_metrics(),
|
||
"cron": collect_cron_metrics(),
|
||
"skill": collect_skill_metrics(),
|
||
"model": collect_model_metrics(),
|
||
}
|
||
bottlenecks = analyze_bottlenecks(metrics)
|
||
for b in bottlenecks:
|
||
print(f"[{b['severity']}] {b['area']}: {b['desc']}")
|
||
print(f" → {b['suggestion']}")
|
||
elif cmd == "recommend":
|
||
metrics = {
|
||
"daemon": collect_daemon_metrics(),
|
||
"cron": collect_cron_metrics(),
|
||
"skill": collect_skill_metrics(),
|
||
"model": collect_model_metrics(),
|
||
}
|
||
bottlenecks = analyze_bottlenecks(metrics)
|
||
trends = analyze_trends(metrics["daemon"], metrics["model"])
|
||
recs = generate_recommendations(metrics, bottlenecks, trends)
|
||
for r in recs:
|
||
print(f"[{r['priority']}] [{r['area']}] {r['action']}")
|
||
else:
|
||
generate_report()
|