用免费模型自动分类 10 万行日志，每小时只读关键错误

用免费模型自动分类 10 万行日志，每小时只读关键错误 | 前端进阶之旅

Full translation

完整中文译文

我的 side project 日志文件在某个周二晚上膨胀到了 40MB。

我已经三周没打开过它了。错误淹没在框架噪音里，而噪音正在占上风。我需要一个分类系统。我有一台免费的服务器和 MonkeyCode 提供的免费模型端点。我还有一个固执的信念——这事能成。

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

一个 cron 任务。一个 Python 脚本。一个模型端点。零美元。

服务器每小时收集日志行。模型将每行分类为 critical、warning 或 noise。我只读 critical 的那些。

import subprocess

LOG_PATH = "/var/log/myapp/app.log"

def tail_logs(n=200):
    result = subprocess.run(
        ["tail", "-n", str(n), LOG_PATH],
        capture_output=True, text=True
    )
    return result.stdout.strip().split("\n")

没有花哨的日志收集器。只需要 tail。

import json
import urllib.request

def classify(lines):
    prompt = (
        "Classify each log line as critical, warning, or noise. "
        "Return a JSON array of objects with 'line' and 'level' keys.\n\n"
        + "\n".join(f"{i}: {line}" for i, line in enumerate(lines))
    )
    payload = json.dumps({
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0
    }).encode()
    req = urllib.request.Request(MODEL_URL, data=payload, headers={
        "Authorization": f"Bearer {TOKEN}",
        "Content-Type": "application/json"
    })
    with urllib.request.urlopen(req) as resp:
        return json.loads(resp.read())["choices"][0]["message"]["content"]

Temperature 设为零。我要的是一个分类器，不是诗人。

Step 3: Parse the response

这里开始变得棘手。模型有时会把 JSON 包在 markdown 围栏里。

def parse_response(text):
    text = text.strip()
    if text.startswith("```json"):
        text = text.split("\n", 1)[1].rsplit("```", 1)[0]
    return json.loads(text)

三行代码，帮我省下了一小时的调试时间。

Deploy: one cron line

0 * * * * cd /opt/log-triage && python3 triage.py >> triage.log 2>&1

这就是整个部署。免费服务器每小时运行这个任务。在告警触发之前，我根本不用去想它。

我用这个管道跑了 10,000 条真实的日志行，这些日志来自我 side project 三周积累的内容。

模型找到了 47 条 MemoryError，我之前标记为"稍后调查"然后就忘了。它还找到了 12 次静默的数据库连接断开。它正确识别出了 318 行框架噪音。

模型漏掉了 5 条 critical 行。标记了 7 条 false positive。

有一条漏报很痛。一次数据库迁移在凌晨 3 点失败了。日志行写的是 FATAL: relation does not exist。模型把它判定为 warning。

为什么？因为它被 40 行"5 秒后重试"的噪音包围了。模型看到了一个模式，就假设这是常规操作。

我在模型输出之上加了一层规则。

HARD_FAILURES = ["FATAL", "PANIC", "relation does not exist", "Segmentation fault"]

def escalate(line, level):
    if any(token in line for token in HARD_FAILURES):
        return "critical"
    return level

规则捕获模型遗漏的东西。模型捕获规则无法表达的东西。

模型处理每批 200 行需要 8 到 15 秒。对于一个每小时任务来说这没问题。对于实时日志流来说这就太糟了。

谁不应该用这个

有 on-call 轮班的团队。如果已经有人在读日志了，这个管道就是多余的。

任何需要确定性严重级别的人。temperature: 0 的模型仍然不是规则引擎。

有严格数据主权要求的组织。日志是数据。把它们发送到第三方端点有法律风险。

Zero. 免费服务器运行 cron 任务。免费模型端点处理批次。1000 万 token 的额度覆盖了整个实验还绰绰有余。

我花了两个小时写脚本，一个小时审查输出。加起来三个小时，换来的是再也不用手工读原始日志文件了。

如果重来我会怎么做

我会加一个去重步骤。同一个错误重复了 40 次应该产生一条告警，而不是四十条。

我还会加一个每周摘要。模型把本周的 critical 事件总结成五行报告。那是下一个迭代。

你搭过日志分类管道吗？你的模型漏掉了什么 regex 本来能捕获的东西？