本地装好AI大模型之后，大家最容易踩的坑不是模型跑不起来，而是你太信任它了。尤其是当你准备做一个正经工具，比如把散落在下载目录、桌面、项目临时文件夹里的几百个文件自动整理归档，直接写个prompt让本地AI全权处理，结果往往是：跑得又慢又飘，偶尔还会把文件扔错地方。我自己也是在踩过几次坑之后，才把一套“本地AI任务拆分”的两级流水线稳定下来——先用L0硬规则前置把一切能枚举的机械工作做掉，再用L1模型兜底处理规则覆盖不到的长尾语义判断。这套思路不挑显卡、不依赖云端，配合本地部署的轻量模型和Python脚本就能落地。这篇文章适合已经把Ollama这类本地模型跑通的开发者，也适合所有想用本地AI做文档整理、企业文件归档、代码库分类等批处理任务的人。

1. 为什么需要两级流水线：本地AI不是万能胶

1.1 本地模型处理批量任务的现实成本

很多人把AI想成万能胶，以为本地部署一个7B模型就能替代人力处理一切。但真实体验完全不同。以一个消费级显卡为例，比如Titan RTX这样24GB显存的卡，7B模型单次推理生成100到200个token，通常需要几秒；如果面对的是五百个文件，每个让模型判断一次分类，光这一轮就要跑二十分钟到半小时。换成8GB显存的卡，跑7B模型可能还要更吃力，只能降低量化级别或者干脆用CPU推理，速度更难看。

更麻烦的是模型输出有随机性。同一个文件，同一套prompt，跑两遍可能给出不同的分类结果，有时候格式还会漂移，让你写好的解析脚本直接崩掉。上下文窗口也有限，没法把一个大目录的所有文件信息一次性塞进去。这些瓶颈决定了一个道理：在本地环境，模型推理是非常贵的资源，必须省着用。这里说的“贵”不只是电费，而是时间和显存，是你在做批处理时最缺的资源。

1.2 全交给模型为什么又慢又飘

假设你真让模型处理一切，遇到的问题基本可以归纳成三类。

慢。每个文件都要走一次推理，几百个文件就意味着几百次调用。即便每次只要5秒，500个文件也要40分钟，这还是在模型推理稳定、不超时的理想情况下。如果某个文件触发了长输出，比如模型开始解释自己的判断依据，生成几百个token，单次耗时直接翻倍。

飘。模型对同一个输入的输出是不稳定的。你今天跑，它把一份合同归到“财务”，明天跑，可能归到“合同”，后天甚至归到“文档”。如果是流水线，这种不确定意味着每次运行结果都不一样，没法给同事、给自己一个交代。

不可测。规则可以写单元测试，但模型行为很难穷举验证。你要测它的“智能”，就得准备一大堆样例，成本远高于写几条if else。当然模型的价值在于处理规则覆盖不了的情况，但前提是只让它处理少数长尾，而不是让它处理所有决策。

所以两级流水线的核心思路是：把能枚举的、高概率出现的分支，全部交给L0硬规则；只有规则无法判断、需要语义理解的少数情况，才轮到L1模型兜底。这很像一个小工坊的团队分工：工具箱旁边贴着一张标准作业流程，能按清单解决的活儿绝不麻烦老师傅；只有流程表上没有写到的情况，才轮到老师傅凭经验拍板。老师傅时间有限、工钱也贵，让他处理所有琐事是对资源的浪费。

1.3 两级流水线到底解决了什么问题

• 速度：L0规则就是几次字符串匹配和字典查找，对本地机器来说基本是零成本，几百个文件瞬间完成。
• 成本：进入L1的文件数量大幅下降，模型推理次数可能从几百次降到几十次，整体耗时减少一个量级。
• 可复现：规则是确定性的，同样的输入永远得到同样的结果，流水线可以放心反复跑。
• 可测试：每条规则都可以单独写测试，出了问题能定位到具体某条规则。
• 可审计：规则和模型各自留下决策记录，回滚和复盘都有据可查。

这套思路适用范围很广：文档自动整理、文件批量归类、日志路由、代码仓库分类、批量重命名、自动打标签，都可以用同一个框架。接下来我以“本地文档自动整理”为例，把两级的每一级拆开细讲。

2. L0硬规则前置：把能枚举的事全做掉

2.1 怎么设计L0规则：从文件特征出发

L0规则的输入不是文件路径本身，而是从文件里提取出来的一组特征。我习惯先写一个特征提取函数，把后续规则要用的信息一次性拿全。

from pathlib import Path import re def extract_features(path: Path) -> dict: stat = path.stat() return { "path": path, "ext": path.suffix.lower(), "name": path.stem, "size": stat.st_size, "mtime": stat.st_mtime, # 尝试从文件名里抓日期，如 2024-01-30 或 20240130 "date_hit": re.search( r"(20\d{2})[-_]?(\d{1,2})[-_]?(\d{1,2})", path.stem ), "is_hidden": path.name.startswith("."), }

这个函数不依赖任何第三方库，逻辑也很简单。重点在于，后面所有规则都只看这个dict，不直接碰路径，这样规则模块可以单独测试。传入文件路径，返回扩展名、文件名、大小、修改时间和日期匹配结果，规则引擎在这些特征基础上做判断。

还有一种更可靠的特征是文件签名，也就是magic bytes。靠扩展名判断文件类型，经常会被改过后缀的文件骗过。比如一个文件叫“资料.pdf”，实际内容是一张图片，扩展名是pdf但文件头是JFIF。真要处理不可信的文件，建议用Python的python-magic或第三方库解析文件头，我在后面问题排查一节再展开。

2.2 规则的优先级与动作分离

规则设计的第一步是列一张表，把你能想到的常见分支全部写下来，同时给每条规则标注优先级。优先级的顺序很重要，因为有些文件名会同时命中多条规则，这时候要按从上到下的顺序执行，类似防火墙规则。

一个设计原则是：一条规则只负责一个分支，粒度要适中。如果你把“图片且含日期且大小大于1MB”写进一条规则，等于把三个逻辑耦合在一起，将来想调整其中一个条件就麻烦。宁可让规则短小一些，用优先级串联起来，也不要写成一个巨大的if else嵌套地狱。

另外，没有对应动作的规则不要写。规则的目的不是做判断，而是产生可执行的动作。如果某个判断结果确定不了动作，那它就不该是L0规则，应该交给L1或人工处理。

2.3 硬规则的“硬”到底体现在哪里

规则引擎写完，大概长这样。

def match_l0(feats: dict): name = feats["name"] ext = feats["ext"] if feats["is_hidden"] or ext in TEMP_EXTS: return Rule("ignore") if ext in IMAGE_EXTS: return Rule("image") if ext in DOC_EXTS: return Rule("doc") if "发票" in name or "合同" in name: return Rule("finance") if feats.get("date_hit"): return Rule("by_date") return None # 弃权，交给L1

我强调“硬”的含义：L0的判定优先级最高，模型结果不能覆盖L0已经确定的动作。这不是对模型能力的不信任，而是对可复现性的坚持。L0规则是我们反复验证过的逻辑，它在已知分支上的正确率可以做到接近100%；模型则是概率输出，偶尔会幻觉。如果一个文件已经被规则命中，就没必要再让模型判断一次，更不应该让模型推翻这个判断。

那“硬”是不是意味着完全不看模型意见？也不全是。如果一条规则命中，但模型给出了不同的理由，你可以把模型结果记到日志里，当作下次调整规则的依据，但当前这次执行仍然以规则为准。规则负责给出确定的动作，模型负责补充视角，两者不混在同一决策层。

还有一个小细节：如果多条规则发生冲突，比如某个文件扩展名是图片，文件名又包含“合同”，按优先级应该进图片库，但财务规则排在后面。实践中我会把这类双重命中设为“搁置区”，而不是盲目按某一条压过去。在文档自动整理场景里，把少数拿不准的放进待人工目录，成本比误分类后到处找文件低得多。

2.4 一个容易被忽略的问题：规则要监控

规则不是一次写好就能一劳永逸。文件命名习惯会变，业务场景会变，曾经常出现的分支可能慢慢消失。所以我在跑流水线时会顺手给每条规则计数，记录命中次数和误判率。比如每条规则命中后，人工复核时会回填一个“对/错”标记，定期统计。

规则命中率极低，说明它覆盖的场景已经很少，可以考虑删除；误判率高，说明特征太宽泛，需要收紧条件。我见过不少人把规则越写越长，最后变成一坨没人敢动的意大利面，就是因为从不回头看统计。给规则加版本号、记录每一条规则的命中率，是让这套系统可持续迭代的关键。

3. L1模型兜底：只为长尾语义付成本

3.1 什么任务才值得交给本地AI

L0弃权后，文件就会进入L1。但不是说只要L0没认出，就一定要调模型。建模前先想清楚，这类文件值不值得用模型判断。

适合进L1的文件通常有这些特征：文件名完全无法从扩展名和关键词上判断，比如“新建文件夹 (7).docx”；文件命名看似有关键词但存在歧义，比如“Q3计划final2”到底是方案还是报告；或者同一批文件里混杂了多种类型，靠规则分类会漏掉不少。反过来说，如果一个文件能被“2024年12月会议纪要.docx”这类名字中的日期和“纪要”关键词准确命中，那它就应该直接被L0规则带走，根本不该出现在模型面前。

每次调用模型都有成本，所以L1的入口应该尽可能窄。我实际操作时会再加一道门槛：如果文件大小超过某个阈值，比如100MB，即使规则没命中，也不让模型读内容来判断，而是直接归到“大文件待人工”目录。模型不能读内容，只靠文件名猜，对大文件的误判率特别高，不如别浪费算力。

3.2 用Ollama + Python实现轻量调用

调用层我用Ollama的HTTP接口，因为它简单、跨平台，不需要自己处理模型加载细节。一个最小实现是这样的。

import requests import re def l1_classify(feats: dict) -> dict: prompt = f""" 请判断这个文件应该归入哪个类别，只输出JSON。 候选类别：{", ".join(CATEGORIES)} 文件扩展名：{feats["ext"]} 文件名：{feats["name"]} 文件大小：{feats["size"]} 字节 最后修改时间：{feats["mtime"]} 输出格式：{{"category": "...", "reason": "一句简短理由"}} """ resp = requests.post( "http://localhost:11434/api/generate", json={ "model": "qwen2.5:3b", "prompt": prompt, "stream": False, "options": { "temperature": 0.1, "num_predict": 128, "seed": 42 } }, timeout=120, ) raw = resp.json()["response"] return parse_json_object(raw)

这里有几个参数值得解释。模型选的是3B小模型，不是本地最大的模型。我自己的判断标准是：语义分类、标题归一这类任务，3B和7B在准确率上差距不大，但推理时间可能差两三倍；如果机器显存不大，3B还能避免频繁换入换出。你真用着大显存卡，比如Titan RTX，想去试7B甚至14B完全可以，但分类任务上不必迷信大模型，尤其当你还有别的任务要跑。

temperature设成0.1，是希望输出尽量稳定。分类任务不需要创造性，温度高了只会让同样的输入产生不同结果。num_predict限制到128，因为只需要一个类别加一句理由，长篇解释只会拖慢推理。seed设成固定值，可以让Ollama在相同输入下尽可能可复现，虽然不能保证绝对一致，但能减少一部分随机性。

3.3 模型输出校验：L1兜底的第二层

模型结果不代表可以无条件信任。模型偶尔会在JSON外面多包一层解释，偶尔会输出一个不在候选类别里的奇怪答案，偶尔干脆输出一堆废话。所以我在接入模型时会加一个校验函数。

def validate_output(raw, allowed): # 如果已经是dict，直接检查category if isinstance(raw, dict) and raw.get("category") in allowed: return raw # 兼容字符串里夹带JSON的情况 match = re.search(r'"category"\s*:\s*"(.*?)"', raw) if match and match.group(1) in allowed: return {"category": match.group(1), "reason": "regex_fallback"} # 都解析不出来就认定无效 return None

这个函数只承认两类结果：一是完整解析出的JSON且category合法，二是通过正则兜底抓到的category且合法。除此之外全部返回None。返回None的文件不会按模型输出移动，而是统一进入待人工目录。

这就是L1“兜底”的第二层含义：模型本身也在被校验。L0兜底了规则覆盖不到的空白，校验兜底了模型的不可靠。你可以在流水线里给模型一次重试机会，比如校验失败后换一个更精简的prompt再试一次，但不要无限重试。我实测下来，模型输出不合法时，换prompt重试的成功率能增加一些，但第二次还不合法，第三次往往也是白搭，不如直接送人工目录。

3.4 Prompt工程注意事项与配置经验

很多第一次做L1调用的人，prompt写得太开放。比如只说“请你帮我分类这个文件”，模型就会自由发挥，要么输出一段话，要么给你解释半天，反正不是你要的结构。我自己的经验是：一定要给约束。

给候选类别。让模型在固定集合里选，而不是自由生成类别。类别多了，比如超过20个，模型容易混淆，可以拆成两级，先判断大类，再判断子类，但那样调用次数也会翻倍，总体不一定划算。

给输出格式。明确告诉它“只输出JSON”，并且给一个具体的结构。如果担心模型不理解JSON，可以在prompt里带一个few-shot例子，比如：

示例输入：文件名“2024年Q3费用报销.pdf” 示例输出：{"category": "finance", "reason": "文件名包含费用报销"}

一次只处理一个文件。我曾经试过把十几个文件塞进一个prompt让模型逐行分类，确实省了推理次数，但模型的格式偶尔会乱，某个文件被漏掉，还要额外写解析逻辑来对齐行号。一次一个文件虽然慢一点，但稳定性和可调试性都强得多，对于本地批处理场景来说，稳定性比那点时间重要。

别让模型解释太多。reason字段虽然有用，但长度应该限制在几十个token内，否则模型会越写越啰嗦，单次推理时间明显变长。

4. 两级流水线实战：从杂乱下载目录到规整归档

4.1 整体流程：七步走

把L0和L1串起来后，一个可运行的流水线流程长这样。

1. 扫描入口目录，先滤掉隐藏文件、临时文件、目录本身。
2. 对每个文件调用extract_features，提取特征。
3. 用match_l0做第一轮匹配。
4. 命中规则且动作不是ignore，就执行动作并写日志。
5. 没命中规则，组装特征和候选类别，调用l1_classify。
6. 对模型输出做validate_output，通过则执行动作，不通过则移入待人工目录。
7. 记录每一条处理日志，包含文件原名、特征、规则或模型决策、目标路径。

这套流程看着简单，但每个环节都有坑，我后面第5章会细讲。整体上你要保证的是：任何文件都只可能走向三条路——被规则处理、被模型处理、进入待人工区，不会出现“算法觉得该处理但没人知道它去哪了”的黑洞。

4.2 核心主循环：一段可扩展的骨架代码

下面这段代码略去了完整异常处理和日志细节，但控制流是完整可用的。

def run_pipeline(root_dir): tmp_dir = root_dir / "_pending" tmp_dir.mkdir(exist_ok=True) for path in collect_candidates(root_dir): feats = extract_features(path) try: rule = match_l0(feats) model_res = None if rule and rule.action != "ignore": move_to_tmp(path, rule.action, feats, tmp_dir) elif rule is None: model_res = l1_classify(feats) if validate_output(model_res, CATEGORIES): move_to_tmp(path, model_res["category"], feats, tmp_dir) else: move_to_quarantine(path, feats) except Exception as e: log_exception(path, e) log_decision(path, feats, rule, model_res)

这个主循环有几个设计点。move_to_tmp是先把文件复制到一个临时区，而不是直接移动到最终目录。因为最终归档目录通常有严格的命名规则，万一日后发现分错了，从临时区调整比从最终目录翻出来容易得多。collect_candidates也要做成生成器，边扫描边处理，避免几百个文件一次性占满内存。

4.3 实测效果：一组有代表性的数据

我用同一个目录做过对比实验，入口是218个文件，包括文档、图片、压缩包、代码文件和一些命名特别随意的文件。机器是普通8GB显存消费级显卡，模型用3B小模型。

纯模型方案就是每个文件都调一次模型，有的文件因为输出格式不合法，还要重试，所以总时长被拉得很长。两级流水线里，L0直接处理了196个文件，只有22个文件因为命名太随意而进入L1。总时长从40分钟压到6分钟，模型调用次数只有原来的十分之一，这个趋势在不同机器和不同任务上是稳定的。

当然不同配置下绝对数字会变。你要是用Titan RTX这种大显存卡，纯模型方案可能也就10分钟出头，但两级流水线依然有明显优势，因为规则匹配是微秒级，模型调用再快也不如不调用快。

4.4 工程化落地：幂等、安全、可审计

如果只是自己跑一次实验，上面代码够用了。但要让这套流水线持续运转，还要补三件事。

幂等。同一批文件重复跑，不能把已经归档的文件又移动一遍。我的做法是计算文件内容的SHA-256，存入本地状态库，每次处理前先检查是否已经处理过。如果文件内容没变，即使文件名变了，也跳过。否则一旦你手动调整了某个文件，下次跑流水线它可能被再次移动，制造混乱。

安全。默认移动而不是删除。移动前先复制到临时区，确认目标目录稳定后再执行真正移动。如果之后想加“清理重复文件”的功能，也依然遵循这个原则：先挪到独立的回收目录，保留三十天再清空。

可审计。日志不能只记“成功”或“失败”。我会记录文件特征、L0规则、L1模型响应、目标路径、时间戳。出了错能快速定位是规则问题还是模型问题，模型问题改prompt，规则问题改配置。没有日志，这套系统就是一个黑盒，出了问题只能全部人工重查，完全没有积累价值。

4.5 这套思路不限于文档整理

两级流水线只是思想，不是某个具体工具。我拿它做过代码仓库重构的辅助任务：先用正则扫描using、类名、命名空间这些确定性信息，把文件预分类；命名空间混乱、语义不清晰的部分才交给本地模型判断归属。这比让模型从头到尾读一遍所有代码文件，再决定哪里该搬哪里该合并要快得多，也安全得多。

同样可以扩展到日志自动路由。线上日志先按关键字规则过滤，命中“ERROR”“WARN”等固定模式就直接走告警逻辑；那些关键字没覆盖到的异常文本，才让模型判断意图，决定是继续追踪还是归档。还有个人知识库自动打标签、邮件附件归档、影视资源整理，本质都是同一套：规则负责大流量，模型负责小尾巴。

5. 常见问题与排查技巧实录

5.1 高频问题速查表

5.2 几条值得写下来的实战心得

规则不是越细越好，而是命中率越高越好。你的真正目标是减少进入L1的数量，因为每次模型调用都有成本。与其把规则写得无比复杂，不如先观察高频文件长什么样，把高频分支覆盖住，剩下的长尾交给模型。

给模型看的特征一定要够多。我早期只传文件名，模型经常被“新建文本文档 (7).txt”这种名字搞得不知所措，分类结果明显差。后来把扩展名、路径、大小、修改时间一起传进去，模型的判断准确率高了一大截。模型不是神，它也靠信息做判断，你给的线索越足，它猜得越准。

日志一定要记模型输出和最终动作。出了错，你能立刻判断是规则问题还是模型问题。我见过不少流水线项目，日志只记“处理成功”，真出问题的时候，根本无从追溯。决策理由写清楚，比多写一万行注释都有用。

还有一个很实用的习惯：先跑dry-run。所有动作都只写日志，不做实际移动，先看一遍它打算怎么处理这批文件。确认无误后再正式跑。这个习惯成本极低，但能拦住几乎所有低级错误，尤其适合刚写完规则、还没验证过的阶段。

最后分享一个我自己的体会。早期我做过一个本地文件自动归档工具，一开始迷信大模型，让一个14B模型直接接管所有分类判断，结果同事说有时候它会把合同归到图片里，而且一跑就是半小时。后来我把L0规则提到前面，模型只处理大概三成文件，误判率反而降下来了，因为被规则命中过的文件有明确的证据链，模型只负责那些规则说不清的少数情况。到最后你会发现，本地AI真正有生产力的地方，不是让它一个人把所有事都干完，而是把它放进一个知道自己在干什么的流水线里。先把L0做好，让模型去补长尾，你手上的本地部署才算真正变成了生产工具，而不是一个时髦但拖后腿的demo。