164 lines
5.9 KiB
Python
164 lines
5.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
OmniRoute 观测脚本 — 持续采集第二网关健康/延迟/成本/路由数据
|
||
每 6h 由 cron 触发（no_agent 模式）：
|
||
- 正常时静默（数据追加到 JSONL 观测日志）
|
||
- 异常时输出报警（cron 会自动推送）
|
||
数据用途：OmniRoute 是否值得全面替代 NewAPI 的评估依据
|
||
"""
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import time
|
||
import requests
|
||
from datetime import datetime, timezone
|
||
|
||
API = "http://127.0.0.1:3001/v1"
|
||
KEY = "local-test-key"
|
||
STATE_DIR = os.path.expanduser("~/.hermes/omniroute-observe")
|
||
LOG = os.path.join(STATE_DIR, "observations.jsonl")
|
||
STATE_FILE = os.path.join(STATE_DIR, "state.json")
|
||
|
||
# 测试用的模型组合 — 覆盖不同路由策略
|
||
TEST_MODELS = ["auto/chat", "auto/best-free", "auto/coding", "auto/best-reasoning"]
|
||
|
||
# 已切换到 OmniRoute 的 cron 任务（观测其运行状态）
|
||
TRACKED_CRONS = {
|
||
"6061a782b772": "股票投研周学习",
|
||
"b46f060eb16b": "每日复盘",
|
||
}
|
||
|
||
|
||
def api_call(model, max_tokens=20):
|
||
"""发一次真实请求，返回 (ok, latency_ms, routed_model, cost, provider)
|
||
注意：① 必须用 requests（urllib 与 OmniRoute 不兼容，连响应头都收不到）
|
||
② 显式 stream:false 会挂起，用默认流式 + iter_lines 遇 [DONE] 退出"""
|
||
t0 = time.time()
|
||
try:
|
||
r = requests.post(f"{API}/chat/completions",
|
||
json={"model": model,
|
||
"messages": [{"role": "user", "content": "ping"}],
|
||
"max_tokens": max_tokens},
|
||
headers={"Authorization": f"Bearer {KEY}"},
|
||
timeout=45, stream=True)
|
||
# 读 SSE 流到 [DONE]（不读完连接就释放）
|
||
routed = "unknown"
|
||
for line in r.iter_lines():
|
||
s = line.decode("utf-8", "ignore") if isinstance(line, bytes) else (line or "")
|
||
s = s.strip()
|
||
if s.startswith("data:") and "[DONE]" in s:
|
||
break
|
||
if s.startswith("data:"):
|
||
try:
|
||
j = json.loads(s[5:].strip())
|
||
if j.get("model"):
|
||
routed = j["model"]
|
||
except Exception:
|
||
pass
|
||
latency = round((time.time() - t0) * 1000)
|
||
cost = r.headers.get("x-omniroute-response-cost", "0")
|
||
provider = r.headers.get("x-omniroute-provider", "")
|
||
return r.status_code == 200, latency, routed, cost, provider
|
||
except requests.exceptions.RequestException as e:
|
||
return False, round((time.time() - t0) * 1000), str(e)[:60], "0", ""
|
||
|
||
|
||
def check_systemd():
|
||
try:
|
||
r = subprocess.run(["systemctl", "--user", "is-active", "omniroute.service"],
|
||
capture_output=True, text=True, timeout=10)
|
||
return r.stdout.strip()
|
||
except Exception:
|
||
return "unknown"
|
||
|
||
|
||
def check_cron_status():
|
||
"""读取 hermes cron 状态（通过 DB/文件，简单方式：检查输出目录最新文件）"""
|
||
out = {}
|
||
for jid, name in TRACKED_CRONS.items():
|
||
d = os.path.expanduser(f"~/.hermes/cron/output/{jid}")
|
||
try:
|
||
files = sorted(os.listdir(d), reverse=True) if os.path.isdir(d) else []
|
||
latest = files[0] if files else None
|
||
out[name] = {"latest_output": latest, "has_output": bool(latest)}
|
||
except Exception as e:
|
||
out[name] = {"error": str(e)}
|
||
return out
|
||
|
||
|
||
def main():
|
||
os.makedirs(STATE_DIR, exist_ok=True)
|
||
ts = datetime.now(timezone.utc).isoformat()
|
||
svc = check_systemd()
|
||
|
||
# 服务健康
|
||
models_ok = False
|
||
model_count = 0
|
||
try:
|
||
r = requests.get(f"{API}/models", timeout=15)
|
||
models_ok = r.status_code == 200
|
||
model_count = len(r.json().get("data", []))
|
||
except Exception:
|
||
models_ok = False
|
||
|
||
# 逐模型测试
|
||
tests = []
|
||
failures = 0
|
||
for m in TEST_MODELS:
|
||
ok, lat, routed, cost, provider = api_call(m)
|
||
tests.append({"model": m, "ok": ok, "latency_ms": lat,
|
||
"routed_to": routed, "cost": cost, "provider": provider})
|
||
if not ok:
|
||
failures += 1
|
||
|
||
# cron 状态
|
||
cron_status = check_cron_status()
|
||
|
||
obs = {
|
||
"ts": ts,
|
||
"service": svc,
|
||
"models_api_ok": models_ok,
|
||
"model_count": model_count,
|
||
"tests": tests,
|
||
"success_count": len(tests) - failures,
|
||
"total_tests": len(tests),
|
||
"cron_status": cron_status,
|
||
}
|
||
|
||
# 追加观测日志
|
||
with open(LOG, "a") as f:
|
||
f.write(json.dumps(obs, ensure_ascii=False) + "\n")
|
||
|
||
# 更新状态文件（供周度汇总用）
|
||
with open(STATE_FILE, "w") as f:
|
||
json.dump(obs, f, ensure_ascii=False, indent=1)
|
||
|
||
# 异常检测：服务挂 OR 模型 API 挂 OR 超过一半测试失败 OR 连续3次失败
|
||
consecutive_file = os.path.join(STATE_DIR, "consecutive_failures")
|
||
if svc != "active" or not models_ok or failures > len(tests) // 2:
|
||
# 连续失败计数
|
||
n = 0
|
||
if os.path.exists(consecutive_file):
|
||
n = int(open(consecutive_file).read().strip() or "0")
|
||
n += 1
|
||
open(consecutive_file, "w").write(str(n))
|
||
|
||
# 输出报警（no_agent 模式非空 stdout 会推送）
|
||
print(f"🚨 OmniRoute 异常 (第{n}次连续)")
|
||
print(f" systemd: {svc} | models API: {'OK' if models_ok else 'FAIL'} ({model_count})")
|
||
for t in tests:
|
||
mark = "✅" if t["ok"] else "❌"
|
||
print(f" {mark} {t['model']}: {t['routed_to']} | {t['latency_ms']}ms | cost={t['cost']}")
|
||
if n >= 3:
|
||
print("⚠️ 连续3次异常 — 建议人工检查 omniroute.service 或回滚 cron 到 newapi-local")
|
||
else:
|
||
# 正常：清连续失败计数，静默（不输出）
|
||
if os.path.exists(consecutive_file):
|
||
os.remove(consecutive_file)
|
||
|
||
# 常规模式下静默退出（正常时不打印任何东西）
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|