#!/usr/bin/env python3 # -*- coding: utf-8 -*- """作业12 · 步骤3：推理健康检查（失败重试 3 次 + 告警文案） 用法（207）： /usr/bin/python3 /opt/workbuddy/scripts_cursor/作业12_推理延迟压测与健康检查/hw12_03_健康检查.py /usr/bin/python3 .../hw12_03_健康检查.py --via-higress /usr/bin/python3 .../hw12_03_健康检查.py --fail-demo # 故意指向坏地址，演示告警 Windows 也可直接对本机可达的 207 执行： python scripts_cursor/作业12_推理延迟压测与健康检查/hw12_03_健康检查.py """ from __future__ import annotations import argparse import json import os import sys import time import urllib.error import urllib.request from datetime import datetime from pathlib import Path DEFAULT_LITELLM = "http://192.168.110.207:4000/v1/chat/completions" DEFAULT_HIGRESS = "http://192.168.110.207:18080/llm/v1/chat/completions" DEFAULT_KEY = "sk-litellm-bocheng-207" DEFAULT_MODEL = "qwen3.6-27b" MAX_RETRIES = 3 RETRY_SLEEP_S = 2.0 TIMEOUT_S = 60.0 def _out_dir() -> Path: env = os.environ.get("HW12_OUT") if env: p = Path(env) p.mkdir(parents=True, exist_ok=True) return p # 优先仓库 data/；仅当 Linux 上真实存在 /opt/workbuddy 时才用服务器路径 repo = Path(__file__).resolve().parents[2] / "data" / "homework12" candidates: list[Path] = [repo] opt = Path("/opt/workbuddy/data/homework12") if os.name != "nt" and Path("/opt/workbuddy").is_dir(): candidates.insert(0, opt) for cand in candidates: try: cand.mkdir(parents=True, exist_ok=True) return cand except OSError: continue p = Path.cwd() / "data" / "homework12" p.mkdir(parents=True, exist_ok=True) return p def build_alert( *, url: str, attempts: int, last_error: str, latencies: list[float], ) -> str: """告警文案：给值班同学可直接贴群的一段话。""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") lat_txt = ", ".join(f"{x:.3f}s" for x in latencies) if latencies else "无" return ( "【推理健康检查告警】\n" f"时间：{now}\n" f"目标：{url}\n" f"结果：连续失败（已重试 {attempts}/{MAX_RETRIES} 次仍未成功）\n" f"末次错误：{last_error}\n" f"各次耗时：{lat_txt}\n" "建议：\n" "1) 查 LiteLLM/vLLM 进程与 GPU 显存（作业5 健康 Agent / nvidia-smi）\n" "2) 查是否限流/鉴权失败（HTTP 401/429）\n" "3) 高峰请降并发或扩副本；恢复后重跑本脚本确认\n" ) def one_probe(url: str, key: str, model: str) -> tuple[bool, float, str, dict | None]: payload = { "model": model, "messages": [{"role": "user", "content": "健康检查：只回复 ok"}], "max_tokens": 8, "stream": False, "temperature": 0, } data = json.dumps(payload, ensure_ascii=False).encode("utf-8") req = urllib.request.Request( url, data=data, headers={ "Authorization": f"Bearer {key}", "Content-Type": "application/json", }, method="POST", ) t0 = time.perf_counter() try: with urllib.request.urlopen(req, timeout=TIMEOUT_S) as resp: body = resp.read().decode("utf-8", errors="replace") elapsed = time.perf_counter() - t0 code = getattr(resp, "status", 200) if code != 200: return False, elapsed, f"HTTP {code}: {body[:200]}", None obj = json.loads(body) content = ( ((obj.get("choices") or [{}])[0].get("message") or {}).get("content") or "" ) if not str(content).strip(): return False, elapsed, "HTTP 200 但 content 为空", obj return True, elapsed, content.strip()[:80], obj except urllib.error.HTTPError as e: elapsed = time.perf_counter() - t0 try: detail = e.read().decode("utf-8", errors="replace")[:200] except Exception: detail = str(e) return False, elapsed, f"HTTPError {e.code}: {detail}", None except Exception as e: elapsed = time.perf_counter() - t0 return False, elapsed, f"{type(e).__name__}: {e}", None def run_health_check(url: str, key: str, model: str) -> int: out = _out_dir() stamp = datetime.now().strftime("%Y%m%d_%H%M%S") report_path = out / f"health_{stamp}.json" alert_path = out / "alerts.jsonl" attempts_log: list[dict] = [] latencies: list[float] = [] last_error = "" ok = False snippet = "" print(f"[INFO] 健康检查 URL={url}") print(f"[INFO] 最多重试 {MAX_RETRIES} 次，间隔 {RETRY_SLEEP_S}s") for i in range(1, MAX_RETRIES + 1): success, elapsed, msg, raw = one_probe(url, key, model) latencies.append(elapsed) attempts_log.append( { "attempt": i, "ok": success, "latency_s": round(elapsed, 4), "message": msg, } ) if success: ok = True snippet = msg print(f"[OK ] 第 {i}/{MAX_RETRIES} 次成功 latency={elapsed:.3f}s reply={msg!r}") break last_error = msg print(f"[FAIL] 第 {i}/{MAX_RETRIES} 次失败 latency={elapsed:.3f}s err={msg}") if i < MAX_RETRIES: time.sleep(RETRY_SLEEP_S) report = { "ok": ok, "url": url, "model": model, "attempts": attempts_log, "finished_at": datetime.now().isoformat(timespec="seconds"), } if ok: report["status"] = "healthy" report["reply_snippet"] = snippet print("[PASS] 推理接口健康") rc = 0 else: alert = build_alert( url=url, attempts=MAX_RETRIES, last_error=last_error, latencies=latencies, ) report["status"] = "unhealthy" report["alert_text"] = alert alert_path.parent.mkdir(parents=True, exist_ok=True) with alert_path.open("a", encoding="utf-8") as f: f.write( json.dumps( { "ts": datetime.now().isoformat(timespec="seconds"), "url": url, "last_error": last_error, "alert_text": alert, }, ensure_ascii=False, ) + "\n" ) print() print(alert) print(f"[ALERT] 已追加写入 {alert_path}") rc = 2 report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") print(f"[INFO] 报告 {report_path}") return rc def main() -> int: ap = argparse.ArgumentParser(description="作业12 推理健康检查") ap.add_argument("--via-higress", action="store_true", help="经 Higress /llm 入口") ap.add_argument("--url", default="", help="自定义 chat completions URL") ap.add_argument("--key", default=os.environ.get("LITELLM_MASTER_KEY", DEFAULT_KEY)) ap.add_argument("--model", default=os.environ.get("LITELLM_CHAT_MODEL", DEFAULT_MODEL)) ap.add_argument( "--fail-demo", action="store_true", help="故意打到不可达地址，演示重试与告警文案", ) args = ap.parse_args() if args.fail_demo: url = "http://127.0.0.1:1/v1/chat/completions" elif args.url: url = args.url elif args.via_higress: url = os.environ.get("HIGRESS_CHAT_URL", DEFAULT_HIGRESS) else: url = os.environ.get("LITELLM_CHAT_URL", DEFAULT_LITELLM) return run_health_check(url, args.key, args.model) if __name__ == "__main__": sys.exit(main())