开头
如果你做过本地AI相关的小工具，一定有过这种体验：脑子里想的是"让大模型帮我搞定一切"，拿到需求之后，只要是有文本的地方，第一反应就是往提示词里塞。可一旦真把任务落到本地的Llama、Qwen上，第二次、第三次跑同一份数据的时候，你会开始怀疑人生——结果不稳定、延迟忽高忽低、GPU风扇像飞机起飞，而且很多本来一句话就能判断的事情，大模型非要兜一大圈才给出一个似是而非的回答。
我最近在做一个本地文档自动整理的小项目，目标是让一个文件夹里的散乱文件自动归类并生成摘要。最开始把全部逻辑都压在本地模型上，结果非常不理想。后来推倒重来，改成一套"L0硬规则前置 + L1模型兜底"的两级流水线，才算真正跑顺。说白了就是：凡是能用确定规则判断的，绝不让模型上场；模型只做规则解决不了的那部分。
这篇文章把整条流水线的设计思路、代码骨架、模型选型、性能实测和踩坑记录都拆开讲一遍。适合正在做本地AI任务分发、想控制资源开销，或者被LLM结果随机性折磨过的人参考。
1. 为什么要把"笨办法"放在大模型前面——L0硬规则层的定位思考
先说一个反直觉的结论：在很多AI落地场景里，硬规则比模型更值得优先使用。
1.1 本地模型的硬伤：速度、成本与随机性
本地部署大模型的好处大家都知道：隐私不出本机，没有API费用，断网也能跑。但真正用起来之后，三个问题会非常明显。
第一是速度。我用Ollama跑Qwen2.5 7B的Q4量化版，在NVIDIA RTX 3060上，一个几十字的短文本推理大概需要300~800毫秒。如果任务复杂一点、提示词长一点，一次调用动辄1~2秒。而一条正则表达式匹配同一个文本，耗时在微秒级。这中间差了至少三个数量级。当任务量是上千个文件、几万个条目的时候，"所有事情都交给模型"意味着你只能坐在那里等。
第二是随机性。LLM本质是采样，温度不为零的话，同一个输入跑两次可能给出不同结果。哪怕把temperature调到0，很多模型在结构化输出上仍然会偶发格式漂移。这在"给文档分类"这种任务上尤其致命——昨天模型把这篇文章判为"技术文档"，今天它改判为"个人笔记"，而你根本不知道它依据什么。
第三是资源占用。本地模型推理不仅吃显存，还在持续拉高功耗和温度。如果你在一台既要写代码、又要跑渲染、还要做数据处理的机器上长期跑模型，其他工作的体验会被明显拖累。用规则先拦掉大部分简单任务，模型只在少数时候被调用，整个系统的资源负载曲线会平滑很多。
1.2 80/20法则在任务拆分里同样成立
我把需要处理的文档随机抽了500个，做了一次"如果用规则能处理多少"的测试，结果很意外：其中约62%的文件可以仅靠扩展名、路径关键词、文件名模式这些规则就完成正确分类。再叠加正文前50行里的正则匹配规则，这个比例提升到了78%。
这里说的"规则能处理"，不是指任务本身简单，而是指我们其实有大量不需要语义理解的特征可用。比如一份文件的文件名是"2025_财报_汇总.xlsx"，那它属于"财务"类别是明摆着的事，不需要大模型来读一遍表格内容再判断。一个PDF文件路径中包含"tax_return"关键词，归到税务归档也很自然。遗憾的是，很多人一上来就把这些显然特征忽略掉了，把模型当成唯一的判断器。
这正是L0硬规则层存在的意义：它不是要替代模型，而是要在能力边界清晰的前提下，拦截掉确定性场景，让模型把算力花在真正需要理解的地方。
1.3 两级流水线的整体架构
整个流水线可以这样理解：
任务入队 -> 预处理(清洗文本/提取元数据) -> L0规则判断 -> 命中规则 -> 直接出结果，结束后处理 -> 未命中 -> 交给L1模型 -> 模型结构化输出 -> 结果校验 -> 出结果L0是前置闸门，L1是兜底通道。两者不是并列关系，而是顺序关系。L0的结果不仅决定"走哪条路"，还可以作为L1的上下文输入。比如L0识别出文件类型是"发票"，L1只需要在"发票"这个范围内做摘要和金额提取，任务难度会大幅下降。
这个架构还有一个隐性优势：可解释性。规则命中的结果，你可以明确说出"是根据哪条规则得出的结论"，这在审计、归档、合规场景里非常重要。而模型结果天然不透明，能少用就少用。
2. L0层实战设计：用确定性规则拦下80%的低难度任务
2.1 规则层的三层结构
L0规则层不是简单写一堆正则就完事，它需要分层组织。我在项目里把规则分了三个层级：
- 元数据规则：不看内容，只看文件名、扩展名、路径、修改时间、文件大小。比如
*.pdf默认进"PDF文档"，*.jpg进"图片"，路径含/contracts/进"合同"。 - 文本特征规则：看内容前若干行的模式。比如正文正则匹配到"发票号码"、"税额"字段，判定为发票；匹配到"此致\n敬礼"判断为正式函件。
- 统计规则：基于词频、长度、格式特征做分数累计。比如一份Markdown文件如果标题层级多、代码块占比高，很可能是技术文档；如果行文流水账且日期变体多，更可能是日志笔记。
这三层规则要按"代价从小到大"的顺序执行。先查扩展名，再读文件头，最后才是全文正则。尽量在低成本阶段把任务拦下来，避免为了判断一个文件类型而把整份几兆的文本全部读进内存。
2.2 规则引擎的代码骨架
我给L0层写了一个简洁的规则引擎，核心思路是"注册表 + 优先级队列"。每条规则有独立的匹配函数和权重，命中后能输出原因。
# l0_engine.py from dataclasses import dataclass from typing import Callable, Optional @dataclass class Rule: name: str priority: int # 数值越小越先执行 apply: Callable[[dict], Optional[dict]] # 入参是任务上下文，返回值是补充信息或None class L0Engine: def __init__(self): self._rules = [] def register(self, rule: Rule): self._rules.append(rule) self._rules.sort(key=lambda r: r.priority) def execute(self, ctx: dict) -> dict: result = {"stage": "L0", "matched": False, "reason": "", "meta": {}} for rule in self._rules: try: extra = rule.apply(ctx) if extra: result["matched"] = True result["reason"] = f"rule:{rule.name}" result["meta"].update(extra) break except Exception as e: # 规则出错不能拖垮主流程，记录后继续下一条 ctx.setdefault("warnings", []).append(f"{rule.name}: {e}") return result每条规则本身是独立的纯函数，输入是任务上下文，输出是补充信息或None。这样做的好处是：规则之间没有隐式耦合，想加一条新规则只需要写一个函数然后注册，不用改主流程。
2.3 典型规则实例
下面是文件分类场景里几条最有价值的规则，列个表格方便对照：
| 规则名称 | 层级 | 匹配逻辑 | 判定结果 |
|---|---|---|---|
| ext_pdf | 元数据 | 扩展名为.pdf | 文档类型=PDF |
| path_contract | 元数据 | 路径包含contract或合同 | 业务线=合同 |
| invoice_header | 文本特征 | 前100字匹配`发票号码 | 税额` |
| meeting_minutes | 文本特征 | 匹配`会议纪要 | 参会人 |
| code_ratio | 统计 | 代码块字符占比>30% | 文档类型=技术文档 |
| diary_style | 统计 | 日期变体密度高且段落短 | 文档类型=日记/日志 |
一条规则命中之后，流水线并不会直接信任结果，还需要做一次"结果置信度判断"。比如只有ext_pdf命中，说明知识来源太单薄——任意一个PDF都可能被分到"PDF文档"这一类，我给它一个偏低的置信度；但如果invoice_header和path_contract同时命中，置信度就非常高，L1层甚至不需要再调用模型。
2.4 规则结果的可观测性
L0层另一个经常被忽略的点是日志。规则判定不透明的系统，调试起来就是噩梦。我给每条规则都保留了"命中/未命中"的完整记录，包括输入文本摘要、匹配到的关键词位点、规则执行耗时。
这样做的直接收益是：当某个文件被分错类别时，你能很快定位到是哪条规则误判，然后决定是修正规则还是把这条规则标记为"不适用于某些场景"。这个能力在后面踩坑排查的时候帮了大忙。
3. L1模型兜底层选型与部署：Ollama + 量化模型怎么配
3.1 模型选型思路
L1层的任务不是处理全部请求，而是处理L0无法规则化的部分。对这部分任务，需要的是"结构化输出能力+指令遵循能力"，而不是"最聪明的模型"。我实测对比了几个本地可跑的模型：
| 模型 | 显存占用(Q4量化) | 平均单次推理耗时(短文本) | 结构化输出稳定性 | 我的评价 |
|---|---|---|---|---|
| Qwen2.5 7B | 约6.2GB | 0.4~0.8秒 | 高 | 首选，中文与英文都不错 |
| Llama 3.1 8B | 约6.8GB | 0.5~1.0秒 | 中 | 英文优秀，中文偶尔格式漂移 |
| Phi-3 mini 3.8B | 约3.5GB | 0.2~0.5秒 | 中高 | 轻量，规则失败后短任务够用 |
| Qwen2.5 14B | 约11GB | 1.2~2.5秒 | 很高 | 最优，但显存压力大 |
如果机器是16GB显存，我建议7B起步。如果只有8GB显存，3B~4B也是可以接受的——毕竟L1只处理少量复杂任务，模型品质下降带来的影响有限。显存不够但又想跑大模型的话，退路是使用更低的量化等级，比如Q3_K_S，但效果下降比较明显，我一般不推荐。
3.2 Ollama部署流程
Ollama是目前本地跑模型最省事的工具之一，几条命令就能架好。以Linux + NVIDIA为例：
# 安装Ollama curl -fsSL https://ollama.com/install.sh | sh # 拉取模型（Qwen2.5 7B Q4量化版） ollama pull qwen2.5:7b # 测试调用 ollama run qwen2.5:7b "用一句话总结：机器学习是一种监督学习方法"Ollama默认会把模型跑在GPU上，如果检测不到NVIDIA驱动会回退到CPU。CPU跑7B模型的速度惨不忍睹，强烈建议在调用之前检查一下是否真的用上了GPU：
nvidia-smi # 看进程列表里是否有ollama的进程占用显存如果Ollama没有走GPU，常见的排查点：NVIDIA驱动版本太低、CUDA运行库缺失、Ollama服务需要重启。
3.3 通过Python调用Ollama做结构化输出
L1层的核心诉求是拿到稳定、可解析的结构化结果。我封装了一个调用函数，专门让模型输出JSON格式，并在外层做格式校验和重试。
# l1_model.py import json import requests OLLAMA_URL = "http://localhost:11434/api/generate" def call_llm(prompt: str, model: str = "qwen2.5:7b", retries: int = 2) -> dict: payload = { "model": model, "prompt": prompt, "stream": False, "temperature": 0.0, "format": "json", # 让Ollama尽量输出JSON "options": { "num_ctx": 4096, "num_predict": 512 } } for attempt in range(retries + 1): resp = requests.post(OLLAMA_URL, json=payload, timeout=60) resp.raise_for_status() text = resp.json().get("response", "") try: return json.loads(text) except json.JSONDecodeError: # 偶尔模型会输出多余的前缀或suffix，尝试修正 cleaned = text.strip() start = cleaned.find("{") end = cleaned.rfind("}") if start >= 0 and end > start: return json.loads(cleaned[start:end+1]) if attempt < retries: continue return {"error": "invalid_json", "raw": text} return {"error": "failed"}提示词模板我固定成"指令 + 输出格式 + 有限的候选枚举"，这样模型更难跑偏：
SYSTEM_PROMPT = """你是文档分类助手。以下输入来自本地文件。 请根据内容分类，输出JSON： {"category": "tech|finance|person|other", "summary": "最多50字的中文摘要", "confidence": 0-1} 只输出JSON，不要解释。""" def classify_by_llm(text: str) -> dict: prompt = f"{SYSTEM_PROMPT}\n\n文件内容片段：\n{text[:2000]}" return call_llm(prompt)3.4 为什么"兜底"而不是"全部交给模型"
这是我踩过的一个大坑。最初版本我试图让LLM做所有分类，理由是"模型理解力更强"。结果发现不仅慢，而且对类别的判定标准不稳定。比如同一份文件，第一次跑它分到"finance"，第二次同样的输入它分到"business_paper"——这只是因为我的类别设计有重叠，模型的决策边界本身就是模糊的。
改成"只有L0没命中才交给L1"之后，模型只处理那些真正需要语义理解的样本，比如一篇没有明显格式特征的博客文章、一份手写扫描件转出来的乱序文本、一套跨语言混合的会议记录。这些场景里模型的价值才能最大化体现。同时因为样本量小了，模型输出偶尔不稳定带来的返工成本也大幅下降。
4. 两级流水线的编排细节：任务队列、超时与失败降级
4.1 任务生命周期与状态机
一个任务会经历如下生命周期：
PENDING->PREPROCESSING->L0_DETECT->L1_FALLBACK(可选) ->POST_PROCESS->DONE/FAILED
如果L0命中，直接跳过L1_FALLBACK；否则进入L1。每进入一个阶段，任务状态都写日志，方便事后回放和统计。
# pipeline.py from enum import Enum class TaskState(str, Enum): PENDING = "pending" PREPROCESSING = "preprocessing" L0_DETECT = "l0_detect" L1_FALLBACK = "l1_fallback" POST_PROCESS = "post_process" DONE = "done" FAILED = "failed"4.2 并发与排队控制
本地模型服务通常不擅长高并发。Ollama默认会对多个请求做排队，但如果同时发10个请求，显存会被中间状态占满，可能导致OOM或者推理速度急剧下降。我给L1层的调用加了一个信号量，限制最多同时2个推理任务：
import asyncio from functools import wraps LLM_SEMAPHORE = asyncio.Semaphore(2) async def limited_llm_call(text: str): async with LLM_SEMAPHORE: loop = asyncio.get_event_loop() return await loop.run_in_executor(None, classify_by_llm, text)对于L0层，因为都是本地规则判断，没有外部资源争抢，可以用线程池并行跑，比如8个线程同时处理文件，瓶颈主要在磁盘IO和文本解析上。
4.3 超时熔断与反向兜底
一条容易被忽略的原则：L1模型调用必须有超时控制，而且超时后的行为不能是"直接报错"，要降级成"用更保守的规则做兜底"。
比如某个任务进入L1后30秒还没返回，我就自动标记该任务走"L0弱规则结果"——即使L0之前没有完全命中，但L0的中间特征（比如文件扩展名、路径关键词）可以给出一个低置信度的候选类别。这个结果虽然不如模型准确，但总比任务死掉好。处理完超时任务后，把原始文件路径记录下来，等模型恢复后可以重新跑一轮补偿。
反向兜底也很重要。如果L0层出现异常（比如读文件超限、正则引擎报错），不能直接让任务失败，而是降级为"跳过L0、直接进入L1"。这两套通道互为保险，整条流水线才不容易全挂。
4.4 上下文传递的设计
L0的结果不是用完就丢。我的做法是把L0产生的所有中间信息都塞进任务上下文，L1在需要的时候可以读取。
class TaskContext: def __init__(self, filepath: str): self.filepath = filepath self.mtime = None self.file_size = 0 self.ext = "" self.text_sample = "" self.l0_result = None # 记录L0命中的规则和meta self.l1_result = None # 记录L1模型输出 self.final_result = None这样L1的提示词可以动态构造。比如L0发现文本里存在"税号"关键词但没完全命中分类规则，L1就可以这样提示："文件内容包含税务相关关键词，但格式不标准，请判断它是否属于财务类文档。"模型拿到这个先验信息后，准确率明显提升。
4.5 审计日志与统计看板
所有任务最后都落一条ES或SQLite记录：哪个阶段命中的、每条规则耗时多少、是否走了L1、模型输出置信度是多少。这些数据不只是拿来查问题，还能反向指导L0规则迭代——如果发现某类任务大量走L1，说明规则没覆盖到，应该去提取新的特征补规则。
5. 一个完整的落地案例：自动化文档整理助手
5.1 项目需求定义
我在本机有一个~/inbox目录，平时截图、PDF、Word、Markdown笔记、发票扫描件全都往里丢。需求很简单：丢进去一个文件，自动按照类别移到~/archive/{category}下，并给每个文件生成一份摘要放到同目录的_index.json。
这个场景特别适合两级流水线，因为文件来源五花八门，但其中很大一部分有明确的规则特征。
5.2 目录与文件结构
项目分成几个模块，基本上和流水线的阶段一一对应：
doc_organizer/ ├── main.py # 入口，监听inbox目录变化 ├── pipeline.py # 流水线编排 ├── l0_engine.py # 规则引擎 ├── rules.py # 具体规则定义 ├── l1_model.py # LLM调用封装 ├── preprocess.py # 文本抽取（PDF/Word/图片OCR等） ├── postprocess.py # 文件移动、JSON索引更新 └── config.yaml # 类别定义、阈值、模型名5.3 核心流程代码全解
监听目录我用的是watchdog库，有新文件落盘就触发任务：
# main.py import yaml from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler from pipeline import process_file class InboxHandler(FileSystemEventHandler): def on_created(self, event): if event.is_directory: return process_file(event.src_path) if __name__ == "__main__": observer = Observer() observer.schedule(InboxHandler(), path="./inbox", recursive=False) observer.start() observer.join()处理单个文件的核心逻辑很直接：
# pipeline.py def process_file(filepath: str): ctx = TaskContext(filepath) # 阶段1：预处理，抽取文本与元数据 preprocess(ctx) # 阶段2：L0规则判断 ctx.l0_result = l0_engine.execute(ctx) # 阶段3：如果L0未命中，走L1 if not ctx.l0_result["matched"]: ctx.l1_result = classify_by_llm(ctx.text_sample) final_category = decide_category(ctx) else: final_category = normalize_output(ctx.l0_result) # 阶段4：后处理，移动文件、更新索引 postprocess(ctx, final_category)decide_category负责把L1的JSON输出映射到预定义的目标目录。如果JSON里confidence低于0.7，我选择不移动文件，只把它标为"未分类"，等人工复核，避免乱移文件。
5.4 两种典型路径的效果对比
拿一个名为2025-03-15_产品需求评审.md的文件举例。L0层的路径正则看到文件名里的产品需求，同时正文里匹配到用户故事、验收标准、优先级等词，多个规则叠加命中，直接判定为"产品文档"，置信度0.91。全程没有调用模型，耗时约12毫秒。
再看一份scan_001.pdf——文件名毫无特征，L0所有规则都没命中。进入L1，把OCR后的文本（大约1500字）拼成提示词发给Qwen2.5 7B。模型输出：
{"category": "finance", "summary": "增值税发票，含税金额1230元，开票日期2025-03-10", "confidence": 0.85}最终文件被移动到~/archive/finance/，摘要写进索引。这个流程耗时约1.2秒，虽然比L0慢，但它在没有规则可循的情况下依然给出了正确结果——这就是L1的价值。
5.5 输出的索引结构
每次处理完成，更新_index.json：
{ "scan_001.pdf": { "category": "finance", "summary": "增值税发票，含税金额1230元", "confidence": 0.85, "handler": "l1", "ts": "2025-03-15T14:32:10" } }记录handler字段特别重要，后续统计哪些类型任务依赖模型、应该沉淀成规则，就看这个字段。
6. 实测中的性能数据与踩坑记录
6.1 性能对比：L0快车道 vs L1模型兜底
用500个真实文件做测试，统计两种路径的耗时分布：
| 路径 | 平均耗时 | 95分位耗时 | 最慢单次 | 正确率 |
|---|---|---|---|---|
| L0命中（约78%的任务） | 14ms | 35ms | 120ms | 93.2% |
| L1兜底（约22%的任务） | 1.8s | 3.5s | 8.7s | 89.4% |
整体平均耗时大约420ms，如果全走模型则平均耗时约1.9秒。两级流水线让整批任务耗时下降了约78%。正确率方面，L1略低一些，但主要是集中在"类别本身模糊"的样本上，人工复核后可接受。
6.2 坑1：正则误伤与全角/半角符号
最典型的翻车案例是发票识别规则。我最初的正则匹配的是发票号码：，结果不少PDF用的是全角冒号：或者中间有多个空格。规则匹配率一下子就掉下去。最后统一做了一层文本清洗，把所有全角符号转成半角，同时把多余空白压缩，再喂给规则。这层预处理对L0和L1都有效，我建议放在最前面。
# preprocess.py import re def normalize_text(text: str) -> str: # 全角转半角 half = [] for ch in text: code = ord(ch) if code == 0x3000: code = 0x20 elif 0xFF01 <= code <= 0xFF5E: code -= 0xFEE0 half.append(chr(code)) text = "".join(half) # 压缩空白 text = re.sub(r"\s+", " ", text) return text.strip()6.3 坑2：Ollama并发请求导致排队雪崩
刚开始L1调用没有做并发控制。某个文件夹一下涌入几百个文件，L1层同时发出大量请求，Ollama把所有请求塞进队列，结果导致整体吞吐量下降，单个任务的最长耗时飙到十几秒。
加了Semaphore(2)之后问题解决。更稳妥的做法是维护一个FIFO任务队列，L1 worker固定2个，按顺序消费，既保证公平又防止OOM。规则层虽然快，也要注意磁盘IO峰值——预览500个PDF时一次性读入内存，内存直接吃满2GB，后来改成流式读取，只读前64KB内容。
6.4 坑3：JSON输出解析不稳
Ollama虽然配合了"format": "json"，但小模型偶尔还是会输出多余的解释性句子。最常见的是在JSON前面加一句"根据文件内容，分类结果如下："，或者在JSON后面补一句"如果有疑问请追问"。
我前面贴的call_llm函数里已经做了"提取第一个{到最后一个}"的兜底处理，但这里要强调一点：rfind("}")存在风险。如果模型在JSON的字符串值里又提到一个}，比如摘要内容里包含表情符号或代码片段，rfind会截错位置。更稳的方案是靠json.JSONDecoder().raw_decode去定位合法的JSON末尾，或者用ollama官方SDK的format配合严格schema校验。
import json def extract_json(text: str): # 逐位置尝试解析，找到合法JSON的终止位置 for i in range(len(text)): if text[i] == "{": decoder = json.JSONDecoder() try: obj, end = decoder.raw_decode(text[i:]) return obj except json.JSONDecodeError: continue raise ValueError("no json found")6.5 坑4：上下文长度设太小导致模型忽略关键信息
默认num_ctx是2048，对于中等长度的文档，2000字可能就把所有token吃掉了。模型拿到截断的文本之后，分类结果容易变得离谱。
我一开始没注意这个参数，有份合同全文4000多字，结果模型只看了一半，漏掉了关键的违约条款关键词，把"租赁合同"分成了"办公用品说明"。后来把num_ctx提升到8192，同时提示词里明确告诉模型"如果文本被截断，优先根据开头段落判断"。注意num_ctx越大，KV Cache占用越多，显存不够的话要配合更小的batch size或者更低量化等级。
6.6 优化后的综合效果
经过这几轮调优，最终流水线跑500份文件的完整时间是2分17秒。78%的文件走了L0快车道，22%走了L1兜底。L1部分因为加了并发限制和超时保护，没有出现一个任务拖死全局的情况。错误案例里最大的来源是名称相似但类型不同的文件，比如一份名为"税务说明"的个人笔记被分到了财务类——这种语义边界模糊的情况，即使是全模型方案也不一定能处理干净。
一点个人体会
这套L0+L1架构做下来，我最大的感受是：本地AI场景里，模型不是万能的，规则也不是老土的技术。它们各有各的不可替代性。规则稳定、快、便宜、可解释，但缺少语义理解；模型灵活、聪明、能处理开放问题，但慢、不稳定、要资源。把两者的特长组合成"规则优先、模型兜底"的流水线，既保住了大部分任务的确定性，也让模型只在最需要它的地方发光。
如果你也在做类似本地AI任务拆分的项目，我强烈建议先用一天的日志数据做一次"规则覆盖率模拟"。你会发现大多数任务根本不需要模型上场。剩下的那一小部分，放着让大模型慢慢算，体验会好非常多。
