用大模型把每日日志变成故障摘要

我的业余项目大多数时候都运行得很好，而这恰恰就是问题所在。一旦它出了故障，我往往要等到用户反馈，或者查看自己的银行账单时才会发现，因为每天晚上阅读原始日志实在太麻烦了，这件事我坚持到第三周左右就放弃了。

于是，我构建了一个尽可能简单的解决方案：在一台免费的托管服务器上运行夜间任务，拉取当天的日志，让 LLM 将错误分组整理成摘要，然后把一份我真的会阅读的总结推送给我。这篇文章会完整介绍整个配置过程，包括最初失败的那部分。

预算：0 美元。它使用免费的模型访问额度和 MonkeyCode 提供的免费服务器方案运行。声明：本文是 MonkeyCode 产品推广活动的一部分。我在这里选择它，是因为这个任务规模很小、无状态，完全没必要为此付费——不过下面的脚本支持任意兼容 OpenAI 的 endpoint，因此不会被锁定在任何平台上。

耗时：大约 40 分钟，包括我接下来要展示的那次失败。

风险上限：即使摘要有误，我也不会损失什么——原始日志完全不会被改动。相比摘要质量，我更看重这一点。

digest.py，约 55 行，除 requests 外没有其他依赖：

import json, os, sys, requests
from collections import Counter

LOG_DIR = os.environ.get("LOG_DIR", "/var/log/myapp")
API_BASE = os.environ["LLM_API_BASE"]      # any OpenAI-compatible endpoint
API_KEY = os.environ["LLM_API_KEY"]
MODEL = os.environ.get("LLM_MODEL", "default")
MAX_CHARS = 12_000                          # hard cap, see failure below

def today_lines():
    import glob, datetime
    day = datetime.date.today().isoformat()
    lines = []
    for path in glob.glob(f"{LOG_DIR}/*.log"):
        with open(path, errors="replace") as f:
            for line in f:
                if day in line and ("ERROR" in line or "WARN" in line):
                    lines.append(line.strip())
    return lines

def pre_aggregate(lines):
    # Cheap local grouping first: normalize numbers/UUIDs, count patterns.
    import re
    pats = Counter()
    for l in lines:
        norm = re.sub(r"[0-9a-f-]{8,}", "<id>", l)
        norm = re.sub(r"\d+", "<n>", norm)
        pats[norm[:200]] += 1
    return pats.most_common(30)

def summarize(patterns):
    payload = "\n".join(f"{c}x {p}" for p, c in patterns)[:MAX_CHARS]
    r = requests.post(
        f"{API_BASE}/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": MODEL,
            "messages": [
                {"role": "system", "content":
                 "You triage app logs. Output: 1) top 3 issues by count, "
                 "2) anything NEW (count 1-2) that looks actionable, "
                 "3) one line: ignore-worthy noise. Be terse."},
                {"role": "user", "content": payload},
            ],
        },
        timeout=60,
    )
    r.raise_for_status()
    return r.json()["choices"][0]["message"]["content"]

if __name__ == "__main__":
    lines = today_lines()
    if not lines:
        sys.exit(0)  # silent day is a good day
    digest = summarize(pre_aggregate(lines))
    requests.post(os.environ["WEBHOOK_URL"], json={"text": digest}, timeout=10)

免费服务器上的 cron 配置如下：

30 21 * * * LOG_DIR=/var/log/myapp LLM_API_BASE=... LLM_API_KEY=... WEBHOOK_URL=... /usr/bin/python3 /opt/digest.py >> /var/log/digest-cron.log 2>&1

为什么需要预聚合步骤（失败案例）

第一个版本会把原始日志行直接发送给模型。某次糟糕的部署中，我的应用记录了 4,000 次相同的连接错误，只是每次的 request ID 都不一样。模型收到的是一堵被截断的、内容完全重复的噪声墙，随后信心满满地生成了一份毫无用处的总结，将问题归结为“数据库连接”——而真正的问题其实是重试循环配置错误。这个问题只有从该模式的出现次数中才能看出来，但截断操作破坏了这项信息。

解决办法平淡无奇，而且完全在本地完成：对 ID 和数字进行归一化，使用 Counter 计数，然后把出现次数最多的 30 种模式及其计数发送出去。现在，模型看到的是 4000x connect timeout to <id>，而不是 4,000 行日志。这也能将输入大小限制在 MAX_CHARS 以内，而这就是控制成本的全部方案。

坦诚地说说局限性

这是初步排查，不是故障诊断。摘要只会告诉我应该去哪里查看，从不会告诉我应该怎么做。在修改任何东西之前，我都会回到原始日志中核实。

免费套餐是一只金丝雀，而不是坚实的地基。如果某天晚上 endpoint 响应缓慢或受到速率限制，我就收不到摘要，而且也不会察觉。我的回滚标准是：如果一周内摘要悄无声息地失败两次，我就添加一个 dead man's switch ping，也就是在摘要没有送达时发出警报的心跳机制。在那之前，偶尔漏掉一个晚上的摘要是可以接受的，因为日志仍然会保留。

不要通过任何托管模型发送包含用户 PII 的日志。我的日志中只有 request ID 和 stack trace。如果你的情况并非如此，请在 pre_aggregate 中增加一道脱敏处理，或者彻底放弃这种方案。

不适合使用这种方案的人：负责具有 SLA、on-call 轮值或合规要求的系统的人。这只是为独立开发者提供的一层便利工具，本质上仍然建立在 grep 之上，并不是真正的监控。如果你需要告警，那就应该使用真正的告警系统。

目前，摘要没有记忆能力，因此某个反复出现但次数较少的错误，每天晚上都会被标记为“NEW”。用一张很小的 SQLite 表保存已经见过的模式 hash，就能解决这个问题。如果你也构建过类似的东西，为了减少重复噪声，你会为每种模式记录哪个字段——首次出现日期、计数增量，还是其他信息？

如需采取进一步措施，你可以考虑屏蔽此人和/或举报滥用行为。