昨天 AI 圈最大的新闻，DeepSeek 宕机了近 5 小时，好多人的应用跟着一起挂。上一期我们刚把 RAG 检索引擎接上大模型，做成了命令行里能对话的问答机器人。机器人跑起来只是第一步，上线之后它要是挂了，第一个挨骂的是你。监控这种东西，平时觉得多余，出事那天能救命。

今天解决的就一件事：接口挂了，你要比用户先知道。问答机器人依赖两样东西：自己的服务进程，大模型的 API。任何一边抽风，用户看到的就是转圈圈。靠手动刷页面不现实，得有个脚本替你盯着，7x24 小时那种。

环境简单到离谱。Python 3.8 以上就行，今天全用标准库，一个第三方包都不装。threading 起假服务，http.server 造抽风接口，urllib 发探测请求，statistics 算平均延迟。零依赖，复制粘贴就能跑。

第一步，先造一个会抽风的假服务，方便本地演示。
下面是整个监控脚本的工作流程：

第一步，先造一个会抽风的假服务，方便本地演示。
真实接口的毛病无非三种：直接 500，慢到像超时，干脆连不上。这个假服务每第 5 次请求返回 500，每第 7 次请求睡 3 秒，剩下的正常返回。昨天 DeepSeek 那种宕机，监控脚本要的就是在这种时候第一时间喊出来。

import threading, time
from http.server import BaseHTTPRequestHandler, HTTPServer

class FlakyHandler(BaseHTTPRequestHandler):
    n = 0
    def _reply(self, code, body):
        try:  # 客户端超时断开时不刷屏报错
            self.send_response(code)
            self.end_headers()
            self.wfile.write(body)
        except BrokenPipeError:
            pass
    def do_GET(self):
        FlakyHandler.n += 1
        i = FlakyHandler.n
        if i % 5 == 0:      # 每第 5 次请求直接挂
            self._reply(500, b"boom")
        elif i % 7 == 0:    # 每第 7 次请求卡 3 秒
            time.sleep(3)
            self._reply(200, b"slow ok")
        else:
            self._reply(200, b"ok")
    def log_message(self, *a):
        pass  # 关掉默认访问日志，清静一点

def start_fake_service(port=8901):
    srv = HTTPServer(("127.0.0.1", port), FlakyHandler)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv

第二步写探测函数。一次探测只干三件事：发请求，记延迟，判生死。timeout 必须显式设置，这里设 2 秒。不设置的话，一个慢请求能把整个监控线程卡死，监控自己先挂了。返回一个字典，ok、延迟毫秒、错误类型都带着，后面统计用得上。

import urllib.request

def probe(url, timeout=2.0):
    t0 = time.perf_counter()
    try:
        with urllib.request.urlopen(url, timeout=timeout) as r:
            ok = (r.status == 200)
            err = None if ok else f"http {r.status}"
    except Exception as e:  # 超时、连接拒绝都算失败
        ok, err = False, f"{type(e).__name__}"
    latency_ms = round((time.perf_counter() - t0) * 1000, 1)
    return {"ok": ok, "latency_ms": latency_ms, "error": err}

第三步是监控主循环。用一个长度 10 的滑动窗口，只看最近 10 次，算可用率和平均延迟。连续失败 3 次就告警一次，告警完把计数清零，不然你的终端会被刷屏。单次延迟超过 1500ms 也吱一声，慢和挂一样要命。

import statistics
from collections import deque

def monitor(url, rounds=20, interval=1.0, fail_streak_alert=3, slow_ms=1500):
    window = deque(maxlen=10)  # 最近 10 次的滑动窗口
    streak = 0                 # 连续失败计数
    for i in range(1, rounds + 1):
        r = probe(url)
        window.append(r)
        streak = streak + 1 if not r["ok"] else 0
        oks = [x for x in window if x["ok"]]
        avail = len(oks) / len(window) * 100
        lat = [x["latency_ms"] for x in oks]
        avg = round(statistics.mean(lat), 1) if lat else 0.0
        flag = "OK " if r["ok"] else "FAIL"
        print(f"[{i:02d}] {flag} 延迟={r['latency_ms']}ms "
              f"近{len(window)}次可用率={avail:.0f}% 平均延迟={avg}ms"
              + (f" 错误={r['error']}" if r["error"] else ""))
        if streak >= fail_streak_alert:
            print(f"  >>> 告警：已连续失败 {streak} 次，服务可能挂了！")
            streak = 0  # 告警后清零，避免刷屏
        elif r["ok"] and r["latency_ms"] > slow_ms:
            print(f"  >>> 注意：本次延迟 {r['latency_ms']}ms 超过 {slow_ms}ms")
        time.sleep(interval)

if __name__ == "__main__":
    start_fake_service()
    monitor("http://127.0.0.1:8901")

跑起来，输出长这样，下面是真实运行结果，不是编的：

[05] FAIL 延迟=1.4ms 近5次可用率=80% 平均延迟=10.1ms 错误=HTTPError
[07] FAIL 延迟=2002.2ms 近7次可用率=71% 平均延迟=8.3ms 错误=TimeoutError
[10] FAIL 延迟=1.3ms 近10次可用率=70% 平均延迟=6.7ms 错误=HTTPError
[14] FAIL 延迟=2003.3ms 近10次可用率=60% 平均延迟=4.0ms 错误=TimeoutError

第 5 行和第 10 行，500 错误被抓出来，错误类型是 HTTPError。第 7 行和第 14 行，延迟直接飙到 2000ms，错误类型是 TimeoutError，这就是慢请求。可用率在 60% 到 70% 之间晃。你的机器人要是这个状态，用户已经在骂了，而你在第 5 行的时候就已经知道了。早 5 分钟知道和晚 5 分钟知道，区别就是一次 P0 故障和一次"用户反馈"。

三个坑，都是我亲手踩过的。坑一，urllib 不设 timeout，默认等到天荒地老，监控脚本自己先卡死，等于没监控。坑二，只看状态码 200 不够。宕机之前一般先有一阵延迟飙升，慢请求必须单独告警，只盯状态码会漏掉一半问题。坑三，告警不去重。一次故障刷一百条消息，手机震到没电，关键信息反而被淹没。连续失败计数加告警后清零，这两行代码不能省。

这篇建议收藏。下次你的服务半夜抽风，翻出这 80 行代码改个 URL 就能用上，至少省两小时排查时间。下期 Python 实战第 4 期，我们把告警接到微信和钉钉上，手机直接收到消息，关注我别错过。讨论时间：你遇到过最离谱的一次服务挂掉是什么情况？评论区聊聊。

Cyrus