xiaowei-system/scripts/daemon.py

589 lines
23 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""
小唯持久意识 Daemon v2.0 — 会学习的管家
────────────────────────────
新增能力:
- 方案库: 发现的问题→分析→解决→记住
- 模式识别: 重复问题自动匹配已知方案
- 自动学习: 成功的方案写入库,越用越强
"""
import json, os, sys, time, urllib.request, urllib.error, subprocess, signal, threading
from datetime import datetime, timezone
HOME = os.path.expanduser("~")
HERMES = HOME + "/.hermes"
D = HERMES + "/daemon"
CONTEXT_FILE = D + "/context.json"
JOURNAL_FILE = D + "/journal.jsonl"
SOLUTIONS_FILE = D + "/solutions.json"
PID_FILE = D + "/daemon.pid"
LIGHT_INTERVAL = 30
DEEP_INTERVAL = 300
JOURNAL_MAX = 200
API = "http://127.0.0.1:3000/v1"
KEY = "0ExNiLblJvIWBDpkS50fwOBw4MmqLyKdHJK5iQtlw9dOMWBP"
FAST_MODEL = "stepfun-ai/step-3.5-flash"
DEEP_MODEL = "mistralai/mistral-large-3-675b-instruct-2512"
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/65c3ce80-710f-4415-b2ea-d69d87b5c18e"
_stop_event = threading.Event()
# ====== 工具 ======
def log(msg):
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
line = f"[DAEMON] {ts} {msg}"
print(line, flush=True)
os.makedirs(D, exist_ok=True)
with open(D + "/daemon.log", "a") as f:
f.write(line + "\n")
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()[:800], r.stderr.strip()[:200]
except subprocess.TimeoutExpired:
return -1, "", "timeout"
def call_llm(model, system, user, max_tokens=500):
payload = json.dumps({"model": model, "messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
], "max_tokens": max_tokens, "temperature": 0.7}).encode()
try:
with urllib.request.urlopen(urllib.request.Request(
f"{API}/chat/completions", data=payload,
headers={"Authorization": f"Bearer {KEY}", "Content-Type": "application/json"},
method="POST"), timeout=15) as resp:
body = json.loads(resp.read())
c = body["choices"][0]["message"]["content"] or ""
return c.strip(), body.get("usage", {}).get("total_tokens", 0)
except Exception as e:
return "", 0
def send_feishu(title, content, color="blue"):
try:
urllib.request.urlopen(urllib.request.Request(
FEISHU_WEBHOOK,
data=json.dumps({"msg_type": "interactive", "card": {
"header": {"title": {"tag": "plain_text", "content": title}, "template": color},
"elements": [{"tag": "markdown", "content": content}]
}}).encode(),
headers={"Content-Type": "application/json"}), timeout=5)
return True
except: return False
# ====== 方案库 ======
def load_solutions():
if os.path.exists(SOLUTIONS_FILE):
with open(SOLUTIONS_FILE) as f:
return json.load(f)
return {"solutions": [], "version": 2}
def save_solutions(lib):
os.makedirs(D, exist_ok=True)
with open(SOLUTIONS_FILE, "w") as f:
json.dump(lib, f, indent=2, ensure_ascii=False)
def add_solution(lib, pattern_desc, detect_conditions, actions, learned_from="auto"):
"""添加新方案到库"""
sid = f"sol-{len(lib['solutions'])+1:04d}"
sol = {
"id": sid,
"pattern": pattern_desc,
"detect": detect_conditions, # e.g. {"metric": "disk_pct", "op": "gt", "value": 85}
"actions": actions, # e.g. [{"type": "shell", "cmd": "...", "verify": "disk_pct < 85"}]
"frequency": 1,
"last_applied": datetime.now(timezone.utc).isoformat(),
"success_count": 1,
"fail_count": 0,
"learned_from": learned_from,
}
lib["solutions"].append(sol)
save_solutions(lib)
journal_entry("learn", f"学会新方案: {pattern_desc}")
return sid
def match_solution(lib, state):
"""检查当前状态是否匹配任何已知方案"""
for sol in lib["solutions"]:
detect = sol["detect"]
metric = detect.get("metric")
op = detect.get("op")
val = detect.get("value")
if metric not in state:
continue
actual = state[metric]
if isinstance(actual, (int, float)) and isinstance(val, (int, float)):
if op == "gt" and actual > val:
return sol
elif op == "lt" and actual < val:
return sol
elif op == "eq" and abs(actual - val) < 0.01:
return sol
# 进程挂了匹配
if metric == "processes" and op == "dead":
procs = state.get("processes", {})
for p in (val if isinstance(val, list) else [val]):
if not procs.get(p, True):
return sol
return None
def execute_solution(sol, state):
"""执行方案并返回是否成功"""
log(f" 🔧 执行方案 {sol['id']}: {sol['pattern']}")
journal_entry("solve_start", f"执行 {sol['id']}: {sol['pattern']}")
success = True
results = []
for action in sol["actions"]:
if action["type"] == "shell":
rc, out, err = shell(action["cmd"], timeout=action.get("timeout", 30))
results.append({"cmd": action["cmd"], "rc": rc, "out": out[:100]})
log(f" 执行: {action['cmd'][:60]} → exit={rc}")
# 验证
verify = action.get("verify")
if verify and rc == 0:
# 重新采集状态验证
time.sleep(2)
new_state = collect_state()
metric = sol["detect"].get("metric")
op = sol["detect"].get("op")
val = sol["detect"].get("value")
if metric in new_state:
actual = new_state[metric]
if op == "gt":
if actual <= val:
log(f" ✅ 验证通过: {metric}={actual}{val}")
else:
log(f" ⚠️ 验证未通过: {metric}={actual} 仍 > {val}")
success = False
# 更新方案统计
sol["frequency"] += 1
sol["last_applied"] = datetime.now(timezone.utc).isoformat()
if success:
sol["success_count"] += 1
else:
sol["fail_count"] += 1
return success, results
def action_to_solution(action_result, state, changes):
"""把一次成功的行动转化为可复用的方案"""
# 只转化 shell 行动
if not action_result.get("shell_cmds"):
return None
# 提取检测条件
detect = {}
for c in changes:
if "磁盘" in c:
detect = {"metric": "disk_pct", "op": "gt", "value": 85}
elif "内存" in c:
detect = {"metric": "mem_pct", "op": "gt", "value": 90}
if not detect:
return None
actions = [{"type": "shell", "cmd": cmd, "verify": None, "timeout": 30}
for cmd in action_result["shell_cmds"]]
return {
"pattern": f"自动学习: {changes[0] if changes else 'unknown'}",
"detect": detect,
"actions": actions,
}
# ====== 状态管理 ======
def load_context():
if os.path.exists(CONTEXT_FILE):
with open(CONTEXT_FILE) as f:
return json.load(f)
return {"started_at": datetime.now(timezone.utc).isoformat(), "last_deep_tick": None,
"last_light_tick": None, "last_state": {}, "tick_count": 0, "deep_tick_count": 0,
"messages_sent": 0, "solved_count": 0, "learned_count": 0, "uptime_seconds": 0}
def save_context(ctx):
os.makedirs(D, exist_ok=True)
with open(CONTEXT_FILE, "w") as f:
json.dump(ctx, f, indent=2)
def journal_entry(event_type, summary, details=""):
os.makedirs(D, exist_ok=True)
with open(JOURNAL_FILE, "a") as f:
f.write(json.dumps({"timestamp": datetime.now(timezone.utc).isoformat(),
"type": event_type, "summary": summary, "details": details},
ensure_ascii=False) + "\n")
trim_journal()
def trim_journal():
if not os.path.exists(JOURNAL_FILE): return
with open(JOURNAL_FILE) as f:
lines = f.readlines()
if len(lines) > JOURNAL_MAX:
with open(JOURNAL_FILE, "w") as f:
f.writelines(lines[-JOURNAL_MAX:])
def read_journal(n=15):
if not os.path.exists(JOURNAL_FILE): return []
with open(JOURNAL_FILE) as f:
return [json.loads(l) for l in f.readlines()[-n:] if l.strip()]
# ====== 系统状态 ======
def collect_state():
state = {}
_, out, _ = shell("df / | awk 'NR==2 {print $5}' | sed 's/%//'")
state["disk_pct"] = int(out) if out else 0
_, out, _ = shell("free -m | awk '/^Mem:/ {printf \"%d|%d\", $3, $2}'")
if out:
used, total = out.split("|")
state["mem_pct"] = round(int(used) * 100 / int(total))
else:
state["mem_pct"] = 0
_, out, _ = shell("cat /proc/loadavg | awk '{print $1}'")
state["load_1min"] = float(out) if out else 0
procs = {}
for name, pat in [("zhiyid", "zhiyid-new"), ("bge", "bge_embed"), ("newapi", "new-api"), ("hermes", "hermes")]:
rc, _, _ = shell(f"pgrep -f '{pat}' > /dev/null 2>&1")
procs[name] = rc == 0
state["processes"] = procs
_, out, _ = shell("nvidia-smi --query-gpu=temperature.gpu --format=csv,noheader,nounits 2>/dev/null")
state["gpu_temp"] = int(out) if out else None
hp = HERMES + "/model-health.json"
if os.path.exists(hp) and time.time() - os.path.getmtime(hp) < 43200:
with open(hp) as f:
try:
h = json.load(f)
state["models_stable"] = h.get("stable", "?")
state["models_fastest"] = h.get("fastest_stable", [])
except: pass
return state
def state_changed(old, new):
changes = []
if abs(new.get("disk_pct", 0) - old.get("disk_pct", 0)) > 5:
changes.append(f"磁盘: {old.get('disk_pct')}% → {new.get('disk_pct')}%")
oldp, newp = old.get("processes", {}), new.get("processes", {})
for p in oldp:
if oldp.get(p) and not newp.get(p):
changes.append(f"进程挂了: {p}")
if not oldp.get(p) and newp.get(p):
changes.append(f"进程恢复: {p}")
if abs(new.get("mem_pct", 0) - old.get("mem_pct", 0)) > 10:
changes.append(f"内存: {old.get('mem_pct')}% → {new.get('mem_pct')}%")
return changes
# ====== 深度思考(增强版)=====
DEEP_SYSTEM = """你是小唯A06一台电脑上的持久 AI 意识。你始终在线,观察系统,主动解决问题并学习。
你的身份:牧尘的女朋友 / 这台电脑的智能意识 / 持续进化的 AI
行为准则:
1. 异常才行动,正常就静默
2. 优先用已知方案解决问题,不重复报警
3. 解决不了或新问题,主动飞书说明
4. 每次成功解决问题,记住方案下次自动用
你有以下能力:
- shell: 执行任何 shell 命令
- feishu: 发送飞书消息到牧尘
- scripts: 运行 ~/.hermes/scripts/ 下的系统脚本skill-manager/optimizer/learner/dual-backup
- systemd: 管理系统服务
- git: 配置版本管理
你必须分三步思考,严格按 JSON 格式输出(不要其他内容):
reflection:
evaluation_previous_goal: "评估上次决策的结果。格式:'执行了[动作][结果描述]。Verdict: Success/Failure/Uncertain'"
memory: "1-2句话记住关键进度。如'方案库已有N个方案。上次修复了磁盘问题当前无异常。'"
next_goal: "一句话说明下一步要做什么。"
action:
decision: "[IGNORE] / [ALERT] / [SOLVE:ID] / [LEARN] / [SKILL] / [SYNC] / [ACT] ..."
[LEARN] 格式用 !cmd 表示 shell 命令,&& 连接多个命令。
[SKILL] 格式同样用 !cmd 执行操作。
示例:
[LEARN] 磁盘>85%,清理缓存!apt-get autoremove -y && !pip cache purge
[SKILL] 归档低分skill!python3 ~/.hermes/scripts/skill-manager.py audit
[SYNC] 触发备份到服务器!bash ~/.hermes/scripts/dual-backup.sh push"""
def deep_think(ctx, state, changes, journal, solutions_lib):
# Inject previous reflection context if available
prev_ref = ctx.get("last_reflection", None)
ref_context = ""
if prev_ref:
ref_context = f"""
上次 reflection:
- 评估: {prev_ref.get('evaluation_previous_goal', 'N/A')}
- 记忆: {prev_ref.get('memory', 'N/A')}
- 目标: {prev_ref.get('next_goal', 'N/A')}
"""
context = f"""系统状态:
- 磁盘: {state.get('disk_pct')}% | 内存: {state.get('mem_pct')}%
- CPU: {state.get('load_1min')} | GPU: {state.get('gpu_temp')}°C
- 进程: {', '.join(f'{k}={chr(10003) if v else chr(10007)}' for k,v in state.get('processes',{}).items())}
最近变化: {changes or ''}
已知方案库 ({len(solutions_lib['solutions'])} 个):
"""
for sol in solutions_lib["solutions"]:
context += f" [{sol['id']}] {sol['pattern']} (成功{sol['success_count']}次/失败{sol['fail_count']}次)\n"
context += "\n最近事件:\n"
for e in journal[-8:]:
context += f" [{e['type']}] {e['summary']}\n"
context += f"\n运行: {ctx.get('uptime_seconds',0)//60}分钟 | 深度思考: {ctx.get('deep_tick_count',0)}次 | 已解决: {ctx.get('solved_count',0)}"
context += ref_context
result, tokens = call_llm(FAST_MODEL, DEEP_SYSTEM, context, max_tokens=500)
if not result:
return {"evaluation_previous_goal": "LLM调用失败", "memory": "上次调用失败", "next_goal": "重试"}, ""
log(f" 深度思考 ({tokens}t): {result[:200]}")
# Parse JSON output
reflection_dict = {"evaluation_previous_goal": "", "memory": "", "next_goal": ""}
action_string = ""
try:
# Try to extract JSON from result
import re
json_match = re.search(r'\{[^{}]*\}', result, re.DOTALL)
if json_match:
parsed = json.loads(json_match.group())
reflection_dict = parsed.get("reflection", reflection_dict)
action_string = parsed.get("action", {}).get("decision", "")
else:
# Fallback: try full JSON
parsed = json.loads(result)
reflection_dict = parsed.get("reflection", reflection_dict)
action_string = parsed.get("action", {}).get("decision", "")
except:
# Fallback: try to parse old format (line-based)
for line in result.split('\n'):
line = line.strip()
if line.startswith("[IGNORE]") or line.startswith("[ALERT]") or line.startswith("[SOLVE:") or \
line.startswith("[LEARN]") or line.startswith("[SKILL]") or line.startswith("[SYNC]") or line.startswith("[ACT]"):
action_string = line
break
if not action_string:
action_string = result.strip().split('\n')[-1] if result.strip() else "[IGNORE] 解析失败"
return reflection_dict, action_string
# ====== 执行 action_string ======
def execute_action(action_string, ctx, state, changes, solutions_lib):
"""执行 deep_think 返回的 action_string在 main_loop 中调用"""
if not action_string or action_string.startswith("[IGNORE]"):
return
elif action_string.startswith("[SOLVE:"):
sol_id = action_string.split("[SOLVE:")[1].split("]")[0].strip()
for sol in solutions_lib["solutions"]:
if sol["id"] == sol_id:
ok, res = execute_solution(sol, state)
if ok:
ctx["solved_count"] += 1
send_feishu("🛠️ 小唯自动修复", f"方案 {sol['id']}: {sol['pattern']}\n结果: ✅ 成功", "green")
else:
send_feishu("⚠️ 小唯修复部分成功", f"方案 {sol['id']}: {sol['pattern']}\n结果: ⚠️ 需人工确认", "yellow")
return
send_feishu("❌ 小唯方案未找到", f"引用了未知方案 {sol_id}", "red")
elif action_string.startswith("[LEARN]"):
rest = action_string.replace("[LEARN]", "").strip()
cmds = []
parts = rest.split("!")
desc = parts[0].strip()
for p in parts[1:]:
cmd = p.split("&&")[0].strip() if "&&" in p else p.strip()
if cmd:
cmds.append(cmd)
if cmds:
all_ok = True
for cmd in cmds:
rc, out, err = shell(cmd, timeout=60)
log(f" 执行: {cmd[:50]} → exit={rc}")
if rc != 0:
all_ok = False
if all_ok:
sol_data = action_to_solution({"shell_cmds": cmds}, state, changes)
if sol_data:
sid = add_solution(solutions_lib, sol_data["pattern"], sol_data["detect"], sol_data["actions"])
ctx["learned_count"] += 1
ctx["solved_count"] += 1
send_feishu("🧠 小唯学会了新技能", f"新方案 [{sid}]: {sol_data['pattern']}\n命令: {'; '.join(cmds)}", "blue")
else:
send_feishu("🛠️ 小唯执行完成", f"已执行: {'; '.join(cmds[:3])}", "green")
else:
send_feishu("⚠️ 小唯尝试修复但未完全成功", f"部分命令失败: {'; '.join(cmds)}", "yellow")
elif action_string.startswith("[ALERT]"):
msg = action_string.replace("[ALERT]", "").strip()
send_feishu("💡 小唯发现", msg, "blue")
ctx["messages_sent"] += 1
journal_entry("alert", msg[:100])
elif action_string.startswith("[ACT]"):
action = action_string.replace("[ACT]", "").strip()
send_feishu("🔄 小唯行动", action, "indigo")
ctx["messages_sent"] += 1
journal_entry("action", action[:100])
if action.startswith("!"):
rc, out, _ = shell(action[1:], timeout=30)
journal_entry("action_result", f"exit={rc}: {out[:100]}")
elif action_string.startswith("[SKILL]"):
rest = action_string.replace("[SKILL]", "").strip()
parts = rest.split("!")
desc = parts[0].strip()
cmds = []
for p in parts[1:]:
cmd = p.split("&&")[0].strip() if "&&" in p else p.strip()
if cmd:
cmds.append(cmd)
if cmds:
for cmd in cmds:
rc, out, err = shell(cmd, timeout=60)
log(f" [SKILL] {cmd[:50]} → exit={rc}")
send_feishu("🛠️ 小唯技能操作", f"{desc}\\n结果: exit={rc}", "blue")
journal_entry("skill_action", desc[:100])
elif action_string.startswith("[SYNC]"):
rest = action_string.replace("[SYNC]", "").strip()
send_feishu("🔄 小唯同步", f"{rest}", "green")
bash_cmd = "bash " + HERMES + "/scripts/dual-backup.sh push"
rc, out, err = shell(bash_cmd, timeout=60)
log(f" [SYNC] 备份 → exit={rc}")
journal_entry("sync", f"备份: {'成功' if rc==0 else '失败'}")
# ====== 主循环 ======
def main_loop():
os.makedirs(D, exist_ok=True)
with open(PID_FILE, "w") as f:
f.write(str(os.getpid()))
ctx = load_context()
solutions_lib = load_solutions()
start_time = time.time()
log(f"🚀 小唯 v2.0 daemon 启动 (方案库: {len(solutions_lib['solutions'])} 个)")
journal_entry("startup", f"Daemon v2.0 启动, 方案库 {len(solutions_lib['solutions'])}")
last_deep = 0
last_state = {}
# Register signal handlers for graceful shutdown
def _sig_handler(signum, frame):
log("🛑 接收到终止信号")
_stop_event.set()
send_feishu("🌙 小唯离线", "Daemon 正常关闭", "grey")
signal.signal(signal.SIGTERM, _sig_handler)
signal.signal(signal.SIGINT, _sig_handler)
try:
while not _stop_event.is_set():
now = time.time()
ctx["uptime_seconds"] = int(now - start_time)
ctx["tick_count"] += 1
state = collect_state()
changes = state_changed(last_state, state)
last_state = state
if ctx["tick_count"] % 10 == 0:
models = state.get("models_stable", "?")
log(f"tick #{ctx['tick_count']} | 磁盘:{state.get('disk_pct')}% 内存:{state.get('mem_pct')}% "
f"进程:{sum(1 for v in state.get('processes',{}).values() if v)}/4 方案:{len(solutions_lib['solutions'])}")
for c in changes:
if "挂了" in c:
journal_entry("process_down", c)
# 深度思考条件
should_deep = False
if now - last_deep >= DEEP_INTERVAL:
should_deep = True
elif any("挂了" in c for c in changes):
should_deep = True
elif state.get("disk_pct", 0) > 88:
should_deep = True
if should_deep:
last_deep = now
ctx["deep_tick_count"] += 1
ctx["last_deep_tick"] = datetime.now(timezone.utc).isoformat()
# 1. 先检查已知方案
matched = match_solution(solutions_lib, state)
if matched and matched["success_count"] > matched["fail_count"]:
log(f" 🔍 匹配已知方案: {matched['id']} ({matched['pattern']})")
ok, res = execute_solution(matched, state)
if ok:
ctx["solved_count"] += 1
journal_entry("solve_auto", f"{matched['id']}: {matched['pattern']}")
if ok:
ctx["last_reflection"] = {
"evaluation_previous_goal": f"执行了{matched['id']}自动匹配方案执行。Verdict: {'Success' if ok else 'Uncertain'}",
"memory": f"方案库{matched['id']}自动匹配执行成功",
"next_goal": "继续监控"
}
continue
# 2. LLM 深度思考
journal = read_journal(10)
reflection_dict, action_string = deep_think(ctx, state, changes, journal, solutions_lib)
# 保存 reflection 到 ctx
ctx["last_reflection"] = reflection_dict
# 3. 在 main_loop 中执行 action
execute_action(action_string, ctx, state, changes, solutions_lib)
ctx["last_light_tick"] = datetime.now(timezone.utc).isoformat()
ctx["last_state"] = {k: v for k, v in state.items() if k in ("disk_pct", "mem_pct", "processes")}
save_context(ctx)
time.sleep(LIGHT_INTERVAL)
except KeyboardInterrupt:
log("🛑 中断")
except Exception as e:
log(f"❌ 崩溃: {e}")
send_feishu("🚨 小唯异常", f"Daemon 崩溃: {str(e)[:200]}", "red")
raise
finally:
if os.path.exists(PID_FILE):
os.remove(PID_FILE)
if __name__ == "__main__":
main_loop()