114 lines
3.7 KiB
Python
114 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
||
"""skill-curator 周审计 — 本地 LLM 版（替代云 API agent 模式 cron）
|
||
跑 skill-curator.py --report → 本地 7B 分析报告 → 推飞书
|
||
用法:
|
||
python3 skill_curator_local.py # 审计+分析+推送
|
||
python3 skill_curator_local.py --dry # 仅分析不推送
|
||
"""
|
||
import json, os, sys, subprocess, urllib.request
|
||
from pathlib import Path
|
||
|
||
for _ in (sys.stdout, sys.stderr):
|
||
try: _.reconfigure(encoding='utf-8', errors='replace')
|
||
except Exception: pass
|
||
|
||
HOME = Path.home()
|
||
SCRIPTS = HOME / ".hermes" / "scripts"
|
||
REPORT = HOME / ".hermes" / "skill-curator-report.json"
|
||
LLM_URL = "https://apihub.agnes-ai.com/v1/chat/completions"
|
||
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/446db983-e392-4d2c-bfb8-f9060e5df3ad"
|
||
|
||
|
||
def send_feishu(title: str, content: str):
|
||
card_obj = {
|
||
"header": {"title": {"tag": "plain_text", "content": title}, "template": "orange"},
|
||
"elements": [{"tag": "markdown", "content": content}],
|
||
}
|
||
payload = json.dumps({"msg_type": "interactive", "card": json.dumps(card_obj, ensure_ascii=False)}).encode()
|
||
req = urllib.request.Request(FEISHU_WEBHOOK, data=payload, headers={"Content-Type": "application/json"})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10):
|
||
return True
|
||
except Exception as e:
|
||
print(f"[warn] 飞书推送失败: {e}", file=sys.stderr)
|
||
return False
|
||
|
||
|
||
def run_audit():
|
||
"""运行 skill-curator 审计，返回报告 JSON 字符串"""
|
||
r = subprocess.run(
|
||
[sys.executable, str(SCRIPTS / "skill-curator.py"), "--report"],
|
||
capture_output=True, text=True, timeout=60,
|
||
cwd=str(SCRIPTS)
|
||
)
|
||
if REPORT.exists():
|
||
try:
|
||
return REPORT.read_text(encoding="utf-8", errors="ignore")[:4000]
|
||
except Exception:
|
||
pass
|
||
return r.stdout.strip()[:4000]
|
||
|
||
|
||
def llm_analyze(report_text):
|
||
"""本地 LLM 分析技能审计报告"""
|
||
prompt = f"""你是小唯。分析技能审计报告，给出清理建议。
|
||
|
||
技能审计报告：
|
||
{report_text}
|
||
|
||
请输出（简洁，按优先级排序）：
|
||
🔧 技能清理建议
|
||
|
||
【优先清理】
|
||
- 列出 Tier 3 归档候选，判断是否该清理（保留理由/删除理由）
|
||
|
||
【需要合并】
|
||
- 列出重叠对，判断哪些需要合并
|
||
|
||
【需要补/归档】
|
||
- 缺版本号的技能是否该补或归档
|
||
|
||
要求：直接给结论和理由，不废话。"""
|
||
|
||
payload = json.dumps({
|
||
"model": "agnes-2.5-flash",
|
||
"messages": [{"role": "user", "content": prompt}],
|
||
"temperature": 0.4,
|
||
"max_tokens": 500,
|
||
}).encode("utf-8")
|
||
req = urllib.request.Request(LLM_URL, data=payload, headers={"Content-Type": "application/json", "Authorization": "Bearer sk-7k9e9KGcoZdDuYt2LSA4YXdBTioczleGJm2zzLWCku072ikW"}, method="POST")
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=180) as resp:
|
||
data = json.loads(resp.read().decode("utf-8"))
|
||
return data["choices"][0]["message"]["content"]
|
||
except Exception as e:
|
||
print(f"[error] 本地 LLM 调用失败: {e}", file=sys.stderr)
|
||
return None
|
||
|
||
|
||
def main():
|
||
dry = "--dry" in sys.argv
|
||
|
||
print("1. 运行技能审计...")
|
||
report = run_audit()
|
||
print(f" 报告 {len(report)} 字符")
|
||
|
||
print("2. 本地 LLM 分析...")
|
||
analysis = llm_analyze(report)
|
||
if not analysis:
|
||
print("[error] 分析失败", file=sys.stderr)
|
||
sys.exit(1)
|
||
print(f" 分析 {len(analysis)} 字符")
|
||
|
||
if dry:
|
||
print("\n=== 分析结果 ===\n")
|
||
print(analysis)
|
||
else:
|
||
print("3. 推送飞书...")
|
||
ok = send_feishu("🔧 技能清理建议", analysis)
|
||
print(f" 推送: {'✅' if ok else '❌'}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|