130 lines
4.3 KiB
Python
Executable File
130 lines
4.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
NewAPI 观测脚本 — 持续采集主网关健康/延迟/路由数据
|
||
每 6h 由 cron 触发（no_agent 模式）：
|
||
- 正常时静默（数据追加到 JSONL 观测日志）
|
||
- 异常时输出报警（cron 会自动推送）
|
||
数据用途：监测 NewAPI + 9 个 NIM key 池健康度
|
||
替代 omniroute-observe.py（OmniRoute 已于 2026-09-02 关停）
|
||
"""
|
||
import json
|
||
import os
|
||
import time
|
||
import requests
|
||
from datetime import datetime, timezone
|
||
|
||
API = "http://127.0.0.1:3000/v1"
|
||
# api-test-token 是 NewAPI 唯一启用 token (user_id=1)
|
||
KEY = "0ExNiLblJvIWBDpkS50fwOBw4MmqLyKdHJK5iQtlw9dOMWBP"
|
||
STATE_DIR = os.path.expanduser("~/.hermes/newapi-observe")
|
||
LOG = os.path.join(STATE_DIR, "observations.jsonl")
|
||
STATE_FILE = os.path.join(STATE_DIR, "state.json")
|
||
|
||
# 测试用的模型组合 — 覆盖 NewAPI 真实 channel
|
||
# - sensenova-free: 主 channel (sensenova deepseek-v4-flash)
|
||
# - minimaxai/minimax-m3: NIM-k1 (NVIDIA 集成, 9 个 key 池)
|
||
# - google/gemma-4-31b-it: NIM 池可用模型
|
||
TEST_MODELS = [
|
||
"deepseek-v4-flash", # sensenova 通道
|
||
"minimaxai/minimax-m3", # NIM-k1 (已知可用)
|
||
"google/gemma-4-31b-it", # NIM 池备选
|
||
]
|
||
|
||
|
||
def api_call(model, max_tokens=20):
|
||
"""发一次真实请求，返回 (ok, latency_ms, model, content_len)"""
|
||
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=30, stream=False)
|
||
elapsed = round((time.time() - t0) * 1000)
|
||
if r.status_code == 200:
|
||
try:
|
||
data = r.json()
|
||
m = data.get("model", model)
|
||
choices = data.get("choices", [])
|
||
content_len = len(choices[0].get("message", {}).get("content", "")) if choices else 0
|
||
return True, elapsed, m, content_len
|
||
except Exception:
|
||
return True, elapsed, model, 0
|
||
else:
|
||
return False, elapsed, f"HTTP {r.status_code}", 0
|
||
except Exception as e:
|
||
return False, round((time.time() - t0) * 1000), f"EXC: {str(e)[:50]}", 0
|
||
|
||
|
||
def check_service():
|
||
"""检查 NewAPI 服务 + models API 是否健康"""
|
||
try:
|
||
r = requests.get(f"{API}/models",
|
||
headers={"Authorization": f"Bearer {KEY}"},
|
||
timeout=10)
|
||
if r.status_code == 200:
|
||
models = r.json().get("data", [])
|
||
return True, len(models)
|
||
return False, 0
|
||
except Exception:
|
||
return False, 0
|
||
|
||
|
||
def main():
|
||
# 确保 state 目录存在
|
||
os.makedirs(STATE_DIR, exist_ok=True)
|
||
|
||
# 1. 服务健康
|
||
svc_ok, model_count = check_service()
|
||
if not svc_ok:
|
||
print(f"🚨 NewAPI 服务不可达或 models API 失败")
|
||
return
|
||
|
||
# 2. 测试每个模型
|
||
results = []
|
||
for m in TEST_MODELS:
|
||
ok, lat, routed, content_len = api_call(m)
|
||
results.append({
|
||
"model": m,
|
||
"ok": ok,
|
||
"latency_ms": lat,
|
||
"routed_to": routed,
|
||
"content_len": content_len
|
||
})
|
||
|
||
# 3. 判断是否有异常
|
||
fails = [r for r in results if not r["ok"]]
|
||
slow = [r for r in results if r["ok"] and r["latency_ms"] > 8000]
|
||
|
||
# 4. 追加 JSONL 观测记录
|
||
obs = {
|
||
"ts": datetime.now(timezone.utc).isoformat(),
|
||
"service": "ok" if svc_ok else "fail",
|
||
"model_count": model_count,
|
||
"tests": results
|
||
}
|
||
with open(LOG, "a") as f:
|
||
f.write(json.dumps(obs) + "\n")
|
||
|
||
# 5. 写 state.json（供其他脚本查询）
|
||
with open(STATE_FILE, "w") as f:
|
||
json.dump(obs, f, indent=2)
|
||
|
||
# 6. 报警逻辑
|
||
if fails:
|
||
msgs = [f"❌ {r['model']}: {r['routed_to']} ({r['latency_ms']}ms)" for r in fails]
|
||
print(f"🚨 NewAPI 异常 ({len(fails)}/{len(results)} 失败):")
|
||
for m in msgs:
|
||
print(m)
|
||
elif slow:
|
||
msgs = [f"⚠️ {r['model']}: {r['latency_ms']}ms" for r in slow]
|
||
print(f"⚠️ NewAPI 慢响应 ({len(slow)}/{len(results)} > 8s):")
|
||
for m in msgs:
|
||
print(m)
|
||
# else: 静默（健康）
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|