本地大模型做安全日志分析：Ollama + 规则粗筛 + LLM 研判完整实现

为什么是本地模型，而不是直接调云端 API

把服务器日志发给云端大模型，听起来很省事，但对安全场景是个悖论：你在用「数据外泄」的方式解决「数据安全」问题。

访问日志里有什么？内网 IP 段、真实用户 ID、API 路径、偶尔还有误打进 URL 的 token。这些内容一旦离开你的网络，就脱离了你的控制。所以日志分析这个场景，本地模型不是「更便宜的替代品」，而是唯一合理的选择。

但本地模型有个硬约束：7B 级别的模型，上下文和推理能力都有限，不可能把几百万行日志直接丢进去。所以正确做法不是「让 AI 读日志」，而是——

用规则做粗筛，把 0.1% 的可疑对象挑出来，再让模型做研判。模型只负责最后一步：判断「这堆证据说明什么」。

一、架构：模型只做最后 5%

access.log
   │
   ├─ ① 解析 + 结构化          （Python 正则，毫秒级）
   ├─ ② 规则打分 + 候选筛选      （阈值，无模型）
   ├─ ③ 证据打包               （每个 IP 一段结构化摘要）
   ├─ ④ LLM 研判               （Ollama，只处理几十条）
   └─ ⑤ 输出 JSON + 人工复核

这个设计的核心取舍：召回靠规则，解释靠模型。规则便宜、确定、可审计；模型擅长的是「把零散证据串成一个结论」，而不擅长「在百万行里找异常」。

如果你反过来做（先让模型扫全量），结果一定是：慢、贵、上下文截断、漏报严重。

二、环境：Ollama 与模型选择

# 安装后拉模型
ollama pull qwen2.5:7b
ollama list
ollama run qwen2.5:7b "你好"

硬件门槛

模型规模	量化	显存需求	纯 CPU 表现	适用
7B	Q4_K_M	约 5 GB	可用，单条 5–15 s	个人/小团队首选
14B	Q4_K_M	约 9 GB	偏慢	判定质量更好，需 12G 以上显卡
32B	Q4_K_M	约 20 GB	不实用	有显卡的团队

模型怎么选

模型	特点	适合
`qwen2.5:7b`	中文强、指令遵循好、JSON 输出稳定	推荐起点
`llama3.1:8b`	英文强，中文一般	英文日志环境
`deepseek-r1:7b`	有推理链，但慢且啰嗦	复杂判定，不追求速度
`qwen2.5:14b`	判定更细，误报更低	有显卡时优先

不要用带 “coder” 或对话专用的模型做研判，指令遵循会明显变差。

三、第一步：解析与规则粗筛

假设是标准 Nginx combined 格式。先做结构化统计，这一步完全不用模型：

import collections
import re

LINE = re.compile(
    r'(?P<ip>\d+\.\d+\.\d+\.\d+) .*? '
    r'"(?P<method>\w+) (?P<path>\S+) [^"]*" '
    r'(?P<status>\d{3}) .*? "(?P<ua>[^"]*)"'
)


def scan(path):
    stats = collections.defaultdict(
        lambda: {"n": 0, "e4": 0, "e5": 0, "paths": collections.Counter(), "ua": set()}
    )
    with open(path, encoding="utf-8", errors="replace") as f:
        for line in f:
            m = LINE.search(line)
            if not m:
                continue
            ip = m["ip"]
            code = int(m["status"])
            s = stats[ip]
            s["n"] += 1
            if code in (401, 403, 429):
                s["e4"] += 1
            if code >= 500:
                s["e5"] += 1
            s["paths"][m["path"][:80]] += 1
            s["ua"].add(m["ua"][:100])
    return stats


def pick(stats, limit=30):
    """规则打分，只挑出值得让模型看的对象"""
    scored = []
    for ip, s in stats.items():
        score = s["e4"] * 2 + s["e5"] * 3
        if s["n"] >= 300:
            score += 5          # 高频访问
        if len(s["paths"]) > 50:
            score += 5          # 路径发散，疑似扫描
        if s["e4"] >= 10:
            score += 10         # 大量认证失败
        if score > 0:
            scored.append((score, ip, s))
    scored.sort(key=lambda x: -x[0])
    return scored[:limit]

这段代码的价值不在「聪明」，而在稳定：同样的输入永远得到同样的候选集，方便复现和审计。

四、第二步：证据打包

给模型的输入必须是结构化、有限、无歧义的。不要直接把原始日志行贴进去——模型会被格式噪声带偏。

def evidence(ip, s, top_paths=8):
    return {
        "ip": ip,
        "requests": s["n"],
        "auth_failures": s["e4"],
        "server_errors": s["e5"],
        "unique_paths": len(s["paths"]),
        "top_paths": [p for p, _ in s["paths"].most_common(top_paths)],
        "user_agents": sorted(s["ua"])[:3],
    }

一份好的证据包应该满足三件事：字段名自解释、数值可核对、长度可控。这样模型不需要「猜」，判定才稳定。

五、Prompt 的三个原则

这是整篇文章里最容易被忽略、但影响最大的部分。

原则 1：明确禁止编造

模型非常擅长「补全合理但不存在」的细节，比如给一个根本没访问过的路径编出攻击意图。必须写死：

只依据给出的字段判断。
不要推测未提供的字段（例如地理位置、历史行为、是否为已知恶意 IP）。
如果证据不足以判断，verdict 填 "unknown"，不要强行给结论。

原则 2：把分类收敛到少数几类

让模型自由发挥会产生几十种说法，无法统计。给定枚举值：

"verdict": "attack" | "suspicious" | "normal" | "unknown"
"category": "bruteforce" | "scanner" | "injection" | "crawler" | "error" | "none"

原则 3：强制 JSON，temperature 设 0

Ollama 支持 format: "json"，能让输出可直接解析。再加上 temperature: 0，大幅降低同样输入得出不同结论的概率。

六、完整脚本

import json
import sys

import requests

MODEL = "qwen2.5:7b"
URL = "http://127.0.0.1:11434/api/chat"

SYSTEM = """你是网络安全分析师，负责研判 Web 访问日志中的可疑来源。
规则：
1. 只依据用户提供的字段判断，严禁编造未提供的字段。
2. 证据不足时 verdict 用 "unknown"，不要强行下结论。
3. 输出必须是 JSON，不要有任何额外文字。
字段：ip, verdict(attack|suspicious|normal|unknown),
category(bruteforce|scanner|injection|crawler|error|none),
confidence(0-1 的小数), reason(50 字以内中文)"""


def judge(ev):
    resp = requests.post(URL, json={
        "model": MODEL,
        "messages": [
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": json.dumps(ev, ensure_ascii=False)},
        ],
        "format": "json",
        "stream": False,
        "options": {"temperature": 0, "num_ctx": 4096},
    }, timeout=120)
    resp.raise_for_status()
    raw = resp.json()["message"]["content"]
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        return {"ip": ev["ip"], "verdict": "error", "reason": raw[:80]}


def main(log_path):
    stats = scan(log_path)
    cands = pick(stats)
    print("候选 %d 个" % len(cands))

    results = []
    for score, ip, s in cands:
        ev = evidence(ip, s)
        r = judge(ev)
        r["score"] = score
        results.append(r)
        print("[%s] %-10s %s" % (r.get("verdict"), ip, r.get("reason", "")))

    with open("verdicts.json", "w", encoding="utf-8") as f:
        json.dump(results, f, ensure_ascii=False, indent=2)
    print("已写入 verdicts.json")


if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "access.log")

依赖只有 requests。跑起来：

pip install requests
python log_triage.py access.log

输出的 verdicts.json 可以直接丢进任何看板，或者再写一步自动封禁。

七、实际输出长什么样

[
  {
    "ip": "203.0.113.45",
    "verdict": "attack",
    "category": "scanner",
    "confidence": 0.86,
    "reason": "路径发散 217 个且集中 404/403，UA 为扫描器特征，判定为目录扫描",
    "score": 38
  },
  {
    "ip": "198.51.100.7",
    "verdict": "suspicious",
    "category": "bruteforce",
    "confidence": 0.71,
    "reason": "登录接口 401 达 143 次且时间集中，疑似撞库，但缺少失败账户多样性",
    "score": 31
  }
]

注意 reason 里出现了「但缺少…」——这正是模型的价值：它给出判断依据，而不是只给一个标签。人工复核时，看这行字比看分数有用得多。

八、怎么知道它准不准

这一节是本文和大多数「AI 文章」的区别：不测就不知道能不能用。

从历史日志里抽 50–100 个候选，人工标注一遍（正常 / 可疑 / 攻击）。
跑脚本，和人工标注对比，算三个数：
- Precision：判定为 attack 的里，多少是真的
- Recall：真正攻击里，被标出来的比例
- 稳定性：同一批数据跑 3 次，结果一致率
如果 Precision < 70%，先别接自动封禁，只做排序提示。

经验上，7B 模型在「取证辅助」这个定位上够用，在「自动处置」上不够。它最大的贡献是把人工翻日志的时间从两小时压到二十分钟，而不是替你做决定。