714 lines
28 KiB
Python
Executable File
714 lines
28 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
NewAPI 模型健康巡检（快速版）
|
||
每 6h 运行，测试关键模型的响应状态
|
||
输出: ~/.hermes/model-health.json
|
||
"""
|
||
|
||
import json
|
||
import yaml
|
||
import time
|
||
import urllib.request
|
||
import urllib.error
|
||
import os
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
|
||
API = "http://127.0.0.1:3000/v1"
|
||
KEY = "0ExNiLblJvIWBDpkS50fwOBw4MmqLyKdHJK5iQtlw9dOMWBP"
|
||
OUTPUT = os.path.expanduser("~/.hermes/model-health.json")
|
||
|
||
# ============ 配置自愈 ============
|
||
CONFIG_PATH = os.path.expanduser("~/.hermes/config.yaml")
|
||
# 配置中声明的模型 — 巡检会交叉验证
|
||
CONFIG_DECLARED_MODELS = [
|
||
"openai/gpt-oss-120b",
|
||
"nvidia/llama-3.3-nemotron-super-49b-v1",
|
||
"mistralai/mistral-medium-3.5-128b",
|
||
"nvidia/nemotron-3-super-120b-a12b",
|
||
]
|
||
# 候选池 — 配置里死了就从这里替补
|
||
CANDIDATE_POOL = [
|
||
"openai/gpt-oss-120b", # 120B ⭐ 质量第一
|
||
"nvidia/nemotron-3-super-120b-a12b", # 120B MoE 推理强
|
||
"mistralai/mistral-nemotron", # ~12B 品质均衡
|
||
"nvidia/nvidia-nemotron-nano-9b-v2", # 9B 备用
|
||
"meta/llama-3.1-8b-instruct", # 8B 极速响应
|
||
"nvidia/nemotron-mini-4b-instruct", # 4B 兜底
|
||
]
|
||
|
||
# 已知忽略的模型（系统/不支持/垃圾，永远不测也不自动加入）
|
||
KNOWN_IGNORE = {
|
||
"gpt-4o", "gpt-4o-mini", "gpt-4o-audio-preview", "gpt-4o-mini-audio-preview",
|
||
"gpt-4o-search-preview", "gpt-4o-mini-search-preview",
|
||
"o1", "o3-mini",
|
||
"dall-e-3", "dall-e-2",
|
||
"tts-1", "tts-1-hd",
|
||
"whisper-1",
|
||
"text-embedding", "text-moderation",
|
||
"comfyui", "sd-", "stable-diffusion",
|
||
"deepseek-v4-pro", "deepseek-v4-pro-",
|
||
"deepseek-ai/deepseek-v4-pro",
|
||
}
|
||
# 已知死模型（不重复测试，直接标记 dead）
|
||
KNOWN_DEAD = {
|
||
"minimaxai/minimax-m2.7",
|
||
"stepfun-ai/step-3.5-flash",
|
||
"qwen/qwen3.5-122b-a10b",
|
||
"mistralai/mistral-medium-3.5-128b",
|
||
}
|
||
|
||
# 已知付费模型（绝不用免费额度测试，也不加入免费配置）
|
||
KNOWN_PAID = {
|
||
"deepseek-ai/deepseek-v4-pro",
|
||
}
|
||
|
||
# OpenClaw 配置中的模型 — 也会巡检和自愈
|
||
OPENCLAW_MODELS = [
|
||
"minimaxai/minimax-m2.7",
|
||
"stepfun-ai/step-3.5-flash",
|
||
"qwen/qwen3.5-122b-a10b",
|
||
"mistralai/devstral-2-123b-instruct-2512",
|
||
"moonshotai/kimi-k2-instruct",
|
||
"nvidia/llama-3.3-nemotron-super-49b-v1.5",
|
||
"qwen/qwen3-next-80b-a3b-thinking",
|
||
]
|
||
|
||
ALL_MODELS = [m for m in (
|
||
CONFIG_DECLARED_MODELS + [m for m in CANDIDATE_POOL if m not in CONFIG_DECLARED_MODELS]
|
||
+ [m for m in OPENCLAW_MODELS if m not in CONFIG_DECLARED_MODELS and m not in CANDIDATE_POOL]
|
||
) if m not in KNOWN_PAID]
|
||
|
||
HEADERS = {
|
||
"Authorization": f"Bearer {KEY}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
PROMPT = "你好"
|
||
|
||
# ============ 质量探针 ============
|
||
# 固定测试题，自动评分（0-100）
|
||
PROBE_QUESTIONS = [
|
||
{
|
||
"question": "如果所有 A 是 B，所有 B 是 C，那么所有 A 是 C 吗？请只回答是或不是。",
|
||
"check": lambda resp: "是" in resp,
|
||
"weight": 25,
|
||
},
|
||
{
|
||
"question": "1.8 和 1.11 哪个大？请只回答数字。",
|
||
"check": lambda resp: "1.8" in resp,
|
||
"weight": 25,
|
||
},
|
||
{
|
||
"question": "中国的首都是哪个城市？请只回答城市名。",
|
||
"check": lambda resp: "北京" in resp,
|
||
"weight": 25,
|
||
},
|
||
{
|
||
"question": "用 Python 写一行反转列表的代码，列表是 [1,2,3]。请只输出代码，不要解释。",
|
||
"check": lambda resp: "[::-1]" in resp or ".reverse()" in resp or "reversed(" in resp,
|
||
"weight": 25,
|
||
},
|
||
]
|
||
|
||
|
||
def _extract_param_b(model: str) -> float:
|
||
"""从模型名提取参数量（B），如 120b→120, 8b→8, 4b→4"""
|
||
import re
|
||
m = re.search(r'(\d+)[bB]', model)
|
||
if m:
|
||
return float(m.group(1))
|
||
# fallback: 用已知映射
|
||
KNOWN = {
|
||
"nemotron-3-super": 120,
|
||
"nemotron-super": 49,
|
||
"nemotron-nano": 9,
|
||
"nemotron-mini": 4,
|
||
"mistral-nemotron": 12,
|
||
"gpt-oss": 120,
|
||
}
|
||
for key, val in KNOWN.items():
|
||
if key in model.lower():
|
||
return val
|
||
return 7.0 # 默认 7B
|
||
|
||
|
||
def _family_score(model: str) -> float:
|
||
"""家族声誉评分 0-100"""
|
||
ml = model.lower()
|
||
if "openai" in ml or "gpt" in ml:
|
||
return 95
|
||
if "nvidia" in ml or "nemotron" in ml:
|
||
return 80
|
||
if "mistral" in ml:
|
||
return 75
|
||
if "meta" in ml or "llama" in ml:
|
||
return 70
|
||
if "minimax" in ml:
|
||
return 65
|
||
if "qwen" in ml:
|
||
return 70
|
||
return 60
|
||
|
||
|
||
def _param_score(param_b: float) -> float:
|
||
"""参数量级分：log2缩放，120b→100, 49b→85, 8b→55, 4b→40"""
|
||
import math
|
||
return min(round(math.log2(param_b) * 14.5), 100)
|
||
|
||
|
||
def _speed_score(latency_ms: int, fastest_latency: int) -> float:
|
||
"""速度分：相对最快模型的延迟比例"""
|
||
if fastest_latency <= 0 or latency_ms <= 0:
|
||
return 50
|
||
ratio = fastest_latency / latency_ms
|
||
return min(round(ratio * 100), 100)
|
||
|
||
|
||
def _run_quality_probe(model: str) -> dict:
|
||
"""运行质量探针，返回探针分和详细结果"""
|
||
probe_results = []
|
||
total = 0
|
||
for q in PROBE_QUESTIONS:
|
||
payload = json.dumps({
|
||
"model": model,
|
||
"messages": [{"role": "user", "content": q["question"]}],
|
||
"max_tokens": 200,
|
||
"temperature": 0.1,
|
||
}).encode()
|
||
req = urllib.request.Request(
|
||
f"{API}/chat/completions",
|
||
data=payload,
|
||
headers=HEADERS,
|
||
method="POST",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=20) as resp:
|
||
body = json.loads(resp.read())
|
||
msg = body.get("choices", [{}])[0].get("message", {}) or {}
|
||
# 有些模型把实际回答放 reasoning_content（gpt-oss-120b 等）
|
||
content = msg.get("content") or msg.get("reasoning_content") or msg.get("reasoning") or ""
|
||
passed = 1 if q["check"](content) else 0
|
||
score = q["weight"] if passed else 0
|
||
total += score
|
||
probe_results.append({
|
||
"question": q["question"][:40],
|
||
"passed": passed,
|
||
"score": score,
|
||
})
|
||
except Exception:
|
||
probe_results.append({
|
||
"question": q["question"][:40],
|
||
"passed": 0,
|
||
"score": 0,
|
||
})
|
||
return {"probe_score": total, "probe_detail": probe_results}
|
||
|
||
|
||
def _discover_new_models() -> list:
|
||
"""从 NewAPI 发现当前可用模型，返回最看好的 N 个新模型（限制数量避免超时）"""
|
||
req = urllib.request.Request(f"{API}/models", headers=HEADERS, method="GET")
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||
body = json.loads(resp.read())
|
||
except Exception:
|
||
return []
|
||
all_remote = [m["id"] for m in body.get("data", [])]
|
||
|
||
known = set(ALL_MODELS) | KNOWN_IGNORE | KNOWN_DEAD | KNOWN_PAID
|
||
# 只挑 chat 模型
|
||
candidates = []
|
||
for m in all_remote:
|
||
if m in known:
|
||
continue
|
||
if any(kw in m.lower() for kw in ["instruct", "gpt", "llama", "nemotron", "mistral",
|
||
"qwen", "minimax", "deepseek", "yi-", "glm",
|
||
"gemma", "phi", "falcon", "command", "dbrx",
|
||
"mixtral", "solar", "aya", "c4ai", "kimi",
|
||
"stockmark", "zamba"]):
|
||
candidates.append(m)
|
||
|
||
# 按潜力排序：优先大参数量 + 知名家族
|
||
def _priority(m: str) -> int:
|
||
score = 0
|
||
# 参数量越大越优先
|
||
import re
|
||
nums = re.findall(r'(\d+)[bB]', m)
|
||
if nums:
|
||
score += int(nums[0])
|
||
# 知名家族加分
|
||
for fam, pts in [("openai", 50), ("deepseek", 40), ("meta/llama", 35),
|
||
("nvidia/nemotron", 30), ("mistral", 25), ("google/gemma", 20),
|
||
("qwen", 20), ("minimax", 15)]:
|
||
if fam in m.lower():
|
||
score += pts
|
||
break
|
||
return -score # 降序
|
||
|
||
candidates.sort(key=_priority)
|
||
MAX_NEW_PER_RUN = 5
|
||
return candidates[:MAX_NEW_PER_RUN]
|
||
|
||
|
||
def test_model(model: str, fastest_latency: int = None) -> dict:
|
||
"""测试单个模型 2 次，返回汇总"""
|
||
trials = []
|
||
|
||
for t in range(2):
|
||
payload = json.dumps({
|
||
"model": model,
|
||
"messages": [{"role": "user", "content": PROMPT}],
|
||
"max_tokens": 20,
|
||
}).encode()
|
||
|
||
req = urllib.request.Request(
|
||
f"{API}/chat/completions",
|
||
data=payload,
|
||
headers=HEADERS,
|
||
method="POST",
|
||
)
|
||
|
||
start = time.time()
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=15) as resp:
|
||
body = json.loads(resp.read())
|
||
except urllib.error.HTTPError as e:
|
||
trials.append({"status": "fail", "error": f"HTTP_{e.code}", "latency_ms": round((time.time() - start) * 1000)})
|
||
continue
|
||
except Exception as e:
|
||
trials.append({"status": "fail", "error": str(e)[:60], "latency_ms": round((time.time() - start) * 1000)})
|
||
continue
|
||
|
||
elapsed = round((time.time() - start) * 1000)
|
||
|
||
try:
|
||
choice = body["choices"][0]
|
||
msg = choice.get("message", {})
|
||
content = msg.get("content", "") or ""
|
||
finish = choice.get("finish_reason", "")
|
||
usage = body.get("usage", {})
|
||
|
||
# ttft 从 nvext 取，没有就估计
|
||
ttft = body.get("nvext", {}).get("timing", {}).get("ttft_ms", -1)
|
||
if ttft < 0:
|
||
ttft = round(elapsed * 0.3)
|
||
|
||
trials.append({
|
||
"status": "ok",
|
||
"latency_ms": elapsed,
|
||
"ttft_ms": ttft,
|
||
"has_content": 1 if content.strip() else 0,
|
||
"completion_tokens": usage.get("completion_tokens", 0),
|
||
"finish_reason": finish,
|
||
})
|
||
except (KeyError, IndexError, json.JSONDecodeError) as e:
|
||
trials.append({"status": "fail", "error": f"parse: {e}", "latency_ms": elapsed})
|
||
|
||
# 汇总
|
||
ok_count = sum(1 for t in trials if t["status"] == "ok")
|
||
fail_count = 2 - ok_count
|
||
|
||
if ok_count == 2:
|
||
stability = "stable"
|
||
elif ok_count == 1:
|
||
stability = "unstable"
|
||
else:
|
||
stability = "dead"
|
||
|
||
ok_trials = [t for t in trials if t["status"] == "ok"]
|
||
avg_latency = round(sum(t["latency_ms"] for t in ok_trials) / len(ok_trials)) if ok_trials else 0
|
||
avg_ttft = round(sum(t.get("ttft_ms", 0) for t in ok_trials) / len(ok_trials)) if ok_trials else -1
|
||
|
||
last_ok = ok_trials[-1] if ok_trials else trials[-1]
|
||
last_finish = last_ok.get("finish_reason", "error")
|
||
|
||
# 质量探针（仅稳定模型）
|
||
probe = _run_quality_probe(model) if stability == "stable" else {"probe_score": 0, "probe_detail": []}
|
||
|
||
# 综合排名分
|
||
param_b = _extract_param_b(model)
|
||
ps = _param_score(param_b)
|
||
fs = _family_score(model)
|
||
ss = _speed_score(avg_latency, fastest_latency) if fastest_latency and avg_latency > 0 else 50
|
||
stab_s = 100 if stability == "stable" else (50 if stability == "unstable" else 0)
|
||
probe_s = probe["probe_score"]
|
||
|
||
rank_score = round(
|
||
probe_s * 0.35 + ps * 0.25 + fs * 0.20 + stab_s * 0.10 + ss * 0.10
|
||
)
|
||
|
||
return {
|
||
"model": model,
|
||
"tests": 2,
|
||
"success": ok_count,
|
||
"failure": fail_count,
|
||
"avg_latency_ms": avg_latency,
|
||
"avg_ttft_ms": avg_ttft,
|
||
"stability": stability,
|
||
"last_status": "ok" if ok_count > 0 else "fail",
|
||
"last_finish": last_finish,
|
||
"probe_score": probe_s,
|
||
"probe_detail": probe["probe_detail"],
|
||
"rank_score": rank_score,
|
||
"param_b": param_b,
|
||
"family_score": fs,
|
||
"param_score": ps,
|
||
}
|
||
|
||
|
||
def main():
|
||
timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
results = []
|
||
deadline = time.time() + 300 # 5分钟全局超时（探针增加了耗时）
|
||
|
||
# ============ 自动发现新模型 ============
|
||
new_models = _discover_new_models()
|
||
if new_models:
|
||
print(f"🔍 发现 {len(new_models)} 个新模型: {', '.join(new_models)}")
|
||
# 加入测试列表
|
||
for m in new_models:
|
||
if m not in ALL_MODELS:
|
||
# 动态扩展 ALL_MODELS（用 list 可变性）
|
||
ALL_MODELS.append(m)
|
||
sys.stdout.flush()
|
||
|
||
for model in ALL_MODELS:
|
||
if time.time() > deadline:
|
||
print(f"⏰ 全局超时，跳过剩余模型")
|
||
break
|
||
|
||
# 先跑测试获取延迟数据，传递给 test_model 用于速度分
|
||
entry = test_model(model)
|
||
results.append(entry)
|
||
|
||
icon = "✅" if entry["stability"] == "stable" else ("⚠️" if entry["stability"] == "unstable" else "❌")
|
||
rank = entry.get("rank_score", 0)
|
||
probe = entry.get("probe_score", 0)
|
||
print(f"{icon} {model:45s} {entry['avg_latency_ms']:>6}ms | {entry['success']}/2 ok | 排名分:{rank:>3} | 探针:{probe}")
|
||
sys.stdout.flush()
|
||
|
||
# 重新计算速度分：确定最快稳定模型的延迟
|
||
stable_models = [r for r in results if r["stability"] == "stable"]
|
||
fastest_latency = min((r["avg_latency_ms"] for r in stable_models if r["avg_latency_ms"] > 0), default=0)
|
||
# 用最快延迟重新计算所有模型的速度分 + 排名分
|
||
for r in results:
|
||
if r["avg_latency_ms"] > 0 and fastest_latency > 0:
|
||
ss = _speed_score(r["avg_latency_ms"], fastest_latency)
|
||
else:
|
||
ss = 50
|
||
stab_s = 100 if r["stability"] == "stable" else (50 if r["stability"] == "unstable" else 0)
|
||
r["rank_score"] = round(
|
||
r.get("probe_score", 0) * 0.35
|
||
+ r.get("param_score", 50) * 0.25
|
||
+ r.get("family_score", 60) * 0.20
|
||
+ stab_s * 0.10
|
||
+ ss * 0.10
|
||
)
|
||
|
||
# 汇总
|
||
healthy = sum(1 for r in results if r["stability"] == "stable")
|
||
flaky = sum(1 for r in results if r["stability"] == "unstable")
|
||
dead = sum(1 for r in results if r["stability"] == "dead")
|
||
|
||
# 按 rank_score 降序排列（质量优先）
|
||
stable_sorted = sorted(stable_models, key=lambda x: x["rank_score"], reverse=True)
|
||
fastest_by_latency = sorted(stable_models, key=lambda x: x["avg_latency_ms"])
|
||
|
||
# 质量排名（全量，含探针分）
|
||
all_ranked = sorted(
|
||
[r for r in results if r["stability"] in ("stable", "unstable")],
|
||
key=lambda x: x["rank_score"], reverse=True
|
||
)
|
||
|
||
summary = {
|
||
"timestamp": timestamp,
|
||
"total_models": len(results),
|
||
"stable": healthy,
|
||
"unstable": flaky,
|
||
"dead": dead,
|
||
"fastest_stable": [m["model"] for m in fastest_by_latency[:5]],
|
||
"quality_ranking": [m["model"] for m in stable_sorted], # 按质量排
|
||
"recommendations": {
|
||
"by_quality": [m["model"] for m in stable_sorted],
|
||
"by_speed": [m["model"] for m in fastest_by_latency],
|
||
"priorities": {
|
||
"首选质量": stable_sorted[:1] if stable_sorted else [],
|
||
"日常推荐": stable_sorted[:3] if len(stable_sorted) >= 3 else stable_sorted,
|
||
"快速响应": fastest_by_latency[:3] if len(fastest_by_latency) >= 3 else fastest_by_latency,
|
||
},
|
||
},
|
||
"models": results,
|
||
}
|
||
|
||
os.makedirs(os.path.dirname(OUTPUT), exist_ok=True)
|
||
with open(OUTPUT + ".new", "w") as f:
|
||
json.dump(summary, f, indent=2, ensure_ascii=False)
|
||
os.replace(OUTPUT + ".new", OUTPUT)
|
||
|
||
# ============ 自愈：检测到死的模型自动替换 ============
|
||
|
||
def _heal_config(config_path: str, declared: list, label: str) -> bool:
|
||
"""修复一个配置文件的模型列表，返回是否修改"""
|
||
dead_in = [r for r in results if r["model"] in declared and r["stability"] == "dead"]
|
||
if not dead_in:
|
||
return False
|
||
|
||
print(f"\n🔧 [{label}] 检测到 {len(dead_in)} 个模型已死亡，正在自愈...")
|
||
with open(config_path) as f:
|
||
cfg = yaml.safe_load(f)
|
||
|
||
current_models = cfg["providers"]["newapi-local"]["models"]
|
||
changed = False
|
||
|
||
for dead in dead_in:
|
||
if dead["model"] not in current_models:
|
||
continue
|
||
replacement = None
|
||
# 按质量排名选最优替补（高 rank_score 优先）
|
||
ranked_candidates = sorted(
|
||
[r for r in results if r["model"] in CANDIDATE_POOL
|
||
and r["model"] not in current_models
|
||
and r["stability"] != "dead"],
|
||
key=lambda x: x["rank_score"], reverse=True
|
||
)
|
||
if ranked_candidates:
|
||
replacement = ranked_candidates[0]["model"]
|
||
if not replacement:
|
||
print(f" ❌ [{label}] {dead['model']} 已死，但无可用替补")
|
||
continue
|
||
|
||
idx = current_models.index(dead["model"])
|
||
current_models[idx] = replacement
|
||
changed = True
|
||
print(f" ✅ [{label}] {dead['model']} → {replacement}")
|
||
|
||
if cfg["providers"]["newapi-local"].get("default_model") == dead["model"]:
|
||
cfg["providers"]["newapi-local"]["default_model"] = replacement
|
||
print(f" default_model 同步更新为 {replacement}")
|
||
|
||
if cfg.get("model", {}).get("default") == dead["model"]:
|
||
cfg["model"]["default"] = replacement
|
||
print(f" model.default 同步更新为 {replacement}")
|
||
|
||
if changed:
|
||
with open(config_path, "w") as f:
|
||
yaml.dump(cfg, f, default_flow_style=False, allow_unicode=True, sort_keys=False)
|
||
print(f" ✅ [{label}] config.yaml 已更新")
|
||
return changed
|
||
|
||
# 修复主配置
|
||
_heal_config(CONFIG_PATH, CONFIG_DECLARED_MODELS, "主配置")
|
||
|
||
# 修复 prof-b 分身配置
|
||
PROF_B_PATH = os.path.expanduser("~/.hermes-prof-b/config.yaml")
|
||
if os.path.exists(PROF_B_PATH):
|
||
_heal_config(PROF_B_PATH, CONFIG_DECLARED_MODELS, "prof-b")
|
||
|
||
# 修复 OpenClaw 配置（JSON 格式）
|
||
|
||
def _heal_openclaw():
|
||
oc_path = os.path.expanduser("~/.openclaw/openclaw.json")
|
||
if not os.path.exists(oc_path):
|
||
return
|
||
with open(oc_path) as f:
|
||
cfg = json.load(f)
|
||
changed = False
|
||
|
||
# --- 1. 修复 models.providers.minimax.models 列表 ---
|
||
models_list = cfg.get("models", {}).get("providers", {}).get("minimax", {}).get("models", [])
|
||
if models_list:
|
||
for entry in models_list:
|
||
mid = entry.get("id", "")
|
||
# 移除付费模型
|
||
if mid in KNOWN_PAID:
|
||
print(f" 🗑️ [OpenClaw] 移除付费模型: {mid}")
|
||
models_list.remove(entry)
|
||
changed = True
|
||
continue
|
||
# 替换死模型
|
||
dead_result = next((r for r in results if r["model"] == mid and r["stability"] == "dead"), None)
|
||
if not dead_result:
|
||
continue
|
||
ranked_candidates = sorted(
|
||
[r for r in results if r["model"] in CANDIDATE_POOL
|
||
and r["stability"] != "dead"
|
||
and not any(m.get("id") == r["model"] for m in models_list)],
|
||
key=lambda x: x["rank_score"], reverse=True
|
||
)
|
||
replacement = ranked_candidates[0]["model"] if ranked_candidates else None
|
||
if not replacement:
|
||
print(f" ❌ [OpenClaw] {mid} 已死，但无可用替补")
|
||
continue
|
||
entry["id"] = replacement
|
||
entry["name"] = replacement.split("/")[-1].replace("-", " ").title()
|
||
changed = True
|
||
print(f" ✅ [OpenClaw model] {mid} → {replacement}")
|
||
if changed:
|
||
cfg["models"]["providers"]["minimax"]["models"] = models_list
|
||
|
||
# --- 2. 修复 agents.list[*].model.primary ---
|
||
agents_list = cfg.get("agents", {}).get("list", [])
|
||
for agent in agents_list:
|
||
primary = agent.get("model", {}).get("primary", "")
|
||
if not primary:
|
||
continue
|
||
# primary 格式: "minimax/minimaxai/minimax-m2.7"
|
||
# 实际模型 ID 是最后两段: "minimaxai/minimax-m2.7"
|
||
parts = primary.split("/")
|
||
raw_model = "/".join(parts[-2:]) if len(parts) >= 2 else parts[-1]
|
||
dead_result = next((r for r in results if r["model"] == raw_model and r["stability"] == "dead"), None)
|
||
if not dead_result:
|
||
continue
|
||
# 找替补
|
||
ranked_candidates = sorted(
|
||
[r for r in results if r["stability"] == "stable"],
|
||
key=lambda x: x["rank_score"], reverse=True
|
||
)
|
||
replacement = ranked_candidates[0]["model"] if ranked_candidates else None
|
||
if not replacement:
|
||
print(f" ❌ [OpenClaw agent] {agent.get('workspace','?')} primary={raw_model} 已死，无替补")
|
||
continue
|
||
# 保持前缀格式: "minimax/<model-id>"
|
||
prefix = primary.split("/")[0] + "/"
|
||
agent["model"]["primary"] = f"{prefix}{replacement}"
|
||
changed = True
|
||
print(f" ✅ [OpenClaw agent] {raw_model} → {replacement}")
|
||
|
||
# --- 3. 修复 agents.defaults.compaction.model ---
|
||
defaults = cfg.get("agents", {}).get("defaults", {})
|
||
comp_model = defaults.get("compaction", {}).get("model", "")
|
||
if comp_model:
|
||
dead_result = next((r for r in results if r["model"] == comp_model and r["stability"] == "dead"), None)
|
||
if dead_result:
|
||
ranked_candidates = sorted(
|
||
[r for r in results if r["stability"] == "stable" and r["rank_score"] > 50],
|
||
key=lambda x: x["rank_score"], reverse=True
|
||
)
|
||
replacement = ranked_candidates[0]["model"] if ranked_candidates else None
|
||
if replacement:
|
||
defaults["compaction"]["model"] = replacement
|
||
changed = True
|
||
print(f" ✅ [OpenClaw compaction] {comp_model} → {replacement}")
|
||
|
||
if changed:
|
||
with open(oc_path, "w") as f:
|
||
json.dump(cfg, f, indent=2, ensure_ascii=False)
|
||
print(f" ✅ [OpenClaw] openclaw.json 全面修复完成")
|
||
|
||
_heal_openclaw()
|
||
|
||
# ============ 主动升级：新模型排名更高则自动替换 ============
|
||
|
||
def _auto_promote_config(config_path: str, label: str, n_keep: int = 4) -> bool:
|
||
"""检查配置中模型列表，如果发现有排名更高的新模型，替换掉最差的"""
|
||
if not os.path.exists(config_path):
|
||
return False
|
||
# 新发现的稳定模型（不在 CANDIDATE_POOL 里的）
|
||
new_stable = [r for r in results if r["model"] not in CANDIDATE_POOL
|
||
and r["stability"] == "stable" and r["rank_score"] > 0
|
||
and r["model"] not in KNOWN_IGNORE]
|
||
if not new_stable:
|
||
return False
|
||
# 读取当前配置的模型列表
|
||
with open(config_path) as f:
|
||
cfg = yaml.safe_load(f)
|
||
current_models = cfg.get("providers", {}).get("newapi-local", {}).get("models", [])
|
||
if not current_models:
|
||
return False
|
||
# 对每个新模型，检查是否超过当前配置中的最低排名模型
|
||
changed = False
|
||
for new_r in sorted(new_stable, key=lambda x: x["rank_score"], reverse=True):
|
||
# 如果已经存在，跳过
|
||
if new_r["model"] in current_models:
|
||
continue
|
||
# 找当前配置中排名最低的
|
||
ranked_current = sorted(
|
||
[r for r in results if r["model"] in current_models and r["stability"] != "dead"],
|
||
key=lambda x: x["rank_score"]
|
||
)
|
||
if not ranked_current or len(current_models) < n_keep:
|
||
continue
|
||
worst = ranked_current[0]
|
||
if new_r["rank_score"] > worst["rank_score"]:
|
||
idx = current_models.index(worst["model"])
|
||
print(f" ⬆️ [{label}] 升级: {worst['model']}({worst['rank_score']}分) → {new_r['model']}({new_r['rank_score']}分)")
|
||
current_models[idx] = new_r["model"]
|
||
changed = True
|
||
if changed:
|
||
cfg["providers"]["newapi-local"]["models"] = current_models
|
||
with open(config_path, "w") as f:
|
||
yaml.dump(cfg, f, default_flow_style=False, allow_unicode=True, sort_keys=False)
|
||
print(f" ✅ [{label}] 主动升级完成")
|
||
return changed
|
||
|
||
_auto_promote_config(CONFIG_PATH, "主配置")
|
||
_auto_promote_config(os.path.expanduser("~/.hermes-prof-b/config.yaml"), "prof-b")
|
||
|
||
# ============ 修复脚本中硬编码的模型名 ============
|
||
|
||
def _heal_hardcoded_models():
|
||
"""扫描并修复 Python 脚本中硬编码的模型名"""
|
||
# 当前首选模型（质量第一的稳定模型）
|
||
top_stable = [r for r in results if r["stability"] == "stable"]
|
||
if not top_stable:
|
||
return
|
||
top_stable.sort(key=lambda x: x["rank_score"], reverse=True)
|
||
best_model = top_stable[0]["model"]
|
||
|
||
# 如果首选没变，跳过
|
||
if best_model == "openai/gpt-oss-120b":
|
||
return # 当前首选就是 gpt-oss-120b，不用动
|
||
|
||
# 需要修复的文件和替换模式
|
||
fixes = [
|
||
# daemon.py — 3 个模型常量
|
||
("daemon.py", 'FAST_MODEL = "openai/gpt-oss-120b"',
|
||
f'FAST_MODEL = "{best_model}"'),
|
||
("daemon.py", 'DEEP_MODEL = "openai/gpt-oss-120b"',
|
||
f'DEEP_MODEL = "{best_model}"'),
|
||
("daemon.py", 'COMPACTION_MODEL = "openai/gpt-oss-120b"',
|
||
f'COMPACTION_MODEL = "{best_model}"'),
|
||
# daemon.py 中硬编码的 API 调用
|
||
("daemon.py", '"model": "openai/gpt-oss-120b"',
|
||
f'"model": "{best_model}"'),
|
||
# wiki_curator.py
|
||
('wiki_curator.py', 'LLM_MODEL = "openai/gpt-oss-120b"',
|
||
f'LLM_MODEL = "{best_model}"'),
|
||
# cangjie_distill.py
|
||
('cangjie_distill.py', 'model="openai/gpt-oss-120b"',
|
||
f'model="{best_model}"'),
|
||
]
|
||
|
||
scripts_dir = os.path.expanduser("~/.hermes/scripts")
|
||
changed = False
|
||
for filename, old_str, new_str in fixes:
|
||
filepath = os.path.join(scripts_dir, filename)
|
||
if not os.path.exists(filepath):
|
||
continue
|
||
with open(filepath) as f:
|
||
content = f.read()
|
||
if old_str not in content:
|
||
continue
|
||
content = content.replace(old_str, new_str)
|
||
with open(filepath, "w") as f:
|
||
f.write(content)
|
||
print(f" 🔧 [{filename}] {old_str.split(chr(34))[1]} → {best_model}")
|
||
changed = True
|
||
|
||
if changed:
|
||
print(f" ✅ 硬编码模型已全部更新为 {best_model}")
|
||
|
||
_heal_hardcoded_models()
|
||
|
||
print(f"\n{'='*50}")
|
||
print(f"巡检完成: {healthy}个稳定 / {flaky}个不稳定 / {dead}个死 (共{len(results)}个)")
|
||
if stable_sorted:
|
||
quality_list = ', '.join(summary['recommendations']['by_quality'])
|
||
print(f"质量排名: {quality_list}")
|
||
print(f"首选: {summary['recommendations']['priorities']['首选质量']}")
|
||
print(f"日常推荐: {summary['recommendations']['priorities']['日常推荐']}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|