1. 为什么要在本地做任务拆分

1.1 从一次真实的需求说起

去年年底我接手了一个内部工具链的改造项目，核心诉求很朴素：把一堆格式混乱的本地文档、日志、配置片段，自动整理成结构化的任务清单。听起来像是调个接口就能搞定的事，但真正动手才发现，问题根本不在模型能力上，而在任务拆分的稳定性上。

我试过直接把整段文本丢给本地部署的大模型，让它一次性输出拆分结果。结果呢？同一个输入，今天跑出来是五条任务，明天跑出来是三条，字段名还时不时变一下。对于需要下游系统消费的结构化数据来说，这种不确定性是致命的。后来我换了个思路：能用规则判断的，绝不交给模型。这就是 L0 硬规则前置 + L1 模型兜底这套两级流水线的由来。

这套方案解决的核心问题是：在本地 AI 部署环境下，如何用最低的算力成本，获得最稳定的任务拆分结果。它适合所有需要在本地跑 AI 任务、又对输出稳定性有要求的场景，比如文档自动整理、日志归类、配置项提取、工单预处理等等。不管你是刚接触本地部署的新手，还是已经在用 Ollama 跑模型的老手，这套思路都能直接抄作业。

1.2 两级流水线的核心设计逻辑

先说清楚这两级分别是什么。

L0 硬规则层，指的是用确定性的代码逻辑做第一道过滤和拆分。比如按固定分隔符切分、按正则匹配提取、按关键词命中分类。这一层的输出是百分之百可预测的，同样的输入永远得到同样的输出。它的优势是快、稳、零算力消耗，缺点是只能处理模式固定的内容。

L1 模型兜底层，指的是当 L0 无法处理或处理结果置信度不足时，才把剩余内容交给本地大模型做语义级拆分。这一层负责处理那些规则覆盖不到的、需要理解语义的模糊场景。它的优势是灵活、能处理自然语言，缺点是慢、吃算力、输出有波动。

两级串起来，就形成了一条流水线：先规则后模型，规则能搞定的绝不麻烦模型，规则搞不定的才让模型上。这个顺序不能反，反了就等于放弃了稳定性优势。

为什么这么设计？我给你算笔账。假设你有 1000 条待拆分内容，其中 70% 是格式相对固定的（比如日志行、配置块），30% 是自由文本。如果全部走模型，1000 次推理，按本地 7B 模型每次 2 秒算，就是 2000 秒。如果 L0 先过滤掉 700 条，只剩 300 条走模型，那就是 600 秒。算力省了 70%，而且那 700 条的输出稳定性是 100%。这笔账怎么算都划算。

2. L0 硬规则层的实操细节

2.1 规则设计的三条原则

L0 层看起来简单，就是写几个正则和判断嘛。但我踩过的坑告诉我，规则设计如果没想清楚，后面维护起来就是灾难。我总结了三条原则，你可以直接拿去用。

第一条：规则要可组合，不要写死。不要把“如果包含 A 且包含 B 且不包含 C 就归类为 D”这种逻辑硬编码在一个函数里。应该把每个判断条件拆成独立的规则单元，用配置的方式组合。这样后面加规则、改规则都不用动核心代码。

第二条：每条规则必须带置信度标记。规则命中不代表一定对。比如你用“错误”这个关键词去匹配日志级别，那“错误率下降”这种描述也会被误命中。所以每条规则输出时，要带上一个置信度分数。高置信度的直接采纳，低置信度的转给 L1 复核。

第三条：规则要能解释自己。每条规则命中时，要记录是哪条规则命中的、命中了什么内容。这样出问题时能快速定位，也方便后面做规则效果分析。

下面是我实际项目里用的规则配置结构，用 Python 字典表示：

RULES = [ { "id": "rule_log_level", "pattern": r"\[(ERROR|WARN|INFO|DEBUG)\]", "action": "extract_level", "confidence": 0.95, "description": "从方括号中提取日志级别" }, { "id": "rule_config_block", "pattern": r"^(\w+)\s*=\s*(.+)$", "action": "extract_kv", "confidence": 0.90, "description": "提取 key=value 形式的配置项" }, { "id": "rule_task_marker", "pattern": r"^[-*]\s+(.+)$", "action": "extract_task", "confidence": 0.85, "description": "提取列表项作为任务" } ]

这个结构的好处是，加新规则就是往列表里加一项，不用改任何处理逻辑。置信度低于阈值的，自动流转到 L1。

2.2 规则命中率低怎么办

实际跑起来，你可能会发现 L0 的命中率没想象中高。我第一版规则只覆盖了 40% 的内容，剩下 60% 全涌到 L1，流水线等于白搭。后来我做了两件事把命中率拉到了 75%。

第一件事：做输入预处理。很多内容格式不统一，是因为源头就有问题。比如有的日志行前面有空格，有的没有；有的用中文冒号，有的用英文冒号。在进 L0 之前，先做一轮标准化：去首尾空白、统一标点、合并连续空行。这一步能把规则命中率提升 15% 左右。

第二件事：分析未命中内容的模式。我把 L1 处理过的内容抽样出来看，发现很多是“看起来自由，其实有隐含结构”的。比如“明天下午三点前把报告发给张三”这种，表面是自然语言，但时间、动作、对象三个要素是固定的。针对这类，我加了一批弱规则，用更宽松的正则去匹配，置信度设低一点，命中后仍然走 L1 复核，但至少给 L1 提供了结构化提示。

提示：不要追求 L0 命中率 100%，那是不可能的。目标是让 L0 处理掉那些“闭着眼睛都能判断”的内容，把模型的算力留给真正需要理解语义的部分。

2.3 规则层的性能优化

L0 层虽然不耗算力，但如果规则多了、内容量大了，性能也会成为瓶颈。我实测过，1000 条规则对 10 万条内容做匹配，纯 Python 循环要跑将近 30 秒。后来做了两个优化，降到了 3 秒以内。

优化一：规则分组预筛。把规则按首字符或首关键词分组，先用一个简单的哈希判断内容可能命中哪组规则，只对该组规则做详细匹配。比如以“[”开头的内容，只去匹配日志相关的规则组。

优化二：编译正则。Python 的re模块每次调用re.match都会重新解析正则字符串。提前用re.compile编译好，能省不少时间。这个细节很多人忽略，但效果很明显。

import re COMPILED_RULES = [] for rule in RULES: compiled = re.compile(rule["pattern"]) COMPILED_RULES.append({**rule, "compiled": compiled}) def match_rules(text): results = [] for rule in COMPILED_RULES: m = rule["compiled"].search(text) if m: results.append({ "rule_id": rule["id"], "action": rule["action"], "confidence": rule["confidence"], "matched": m.group(0), "groups": m.groups() }) return results

这段代码看着简单，但编译和不编译的差距，在规则数量上去之后非常明显。

3. L1 模型兜底层的落地方法

3.1 本地模型选型与部署

L1 层的关键是选一个合适的本地模型。我的建议是：不要一上来就追求大参数。任务拆分这个场景，对模型的推理能力要求其实不高，7B 到 14B 的模型完全够用。我用的是 Ollama 部署的 Qwen2.5 7B，在 16G 显存的机器上跑得很稳。

选型时重点看三个指标：指令遵循能力、输出格式稳定性、推理速度。指令遵循能力决定了模型能不能按你要求的格式输出；输出格式稳定性决定了你要不要写复杂的后处理；推理速度决定了流水线的整体吞吐。

部署命令很简单，Ollama 装好后一行搞定：

ollama pull qwen2.5:7b ollama serve

然后 Python 侧用 requests 调本地接口：

import requests import json def call_local_model(prompt, system_prompt=""): url = "http://localhost:11434/api/generate" payload = { "model": "qwen2.5:7b", "prompt": prompt, "system": system_prompt, "stream": False, "options": { "temperature": 0.1, "top_p": 0.9, "num_predict": 512 } } resp = requests.post(url, json=payload, timeout=60) return resp.json()["response"]

注意temperature我设的是 0.1，不是 0。设 0 理论上最确定，但实际跑下来有时候会陷入重复输出的死循环。0.1 是个比较稳的值，既有确定性，又不会卡死。

3.2 提示词设计的核心技巧

L1 的提示词设计，直接决定了输出能不能被下游消费。我踩过的最大坑是：提示词写得太“客气”。比如“请帮我分析以下内容并提取任务”，模型就会自由发挥，输出一段散文式的分析。后来我改成强约束格式，效果立竿见影。

我的提示词模板长这样：

你是一个任务拆分引擎。你的唯一职责是将输入内容拆分为结构化任务列表。 规则： 1. 只输出 JSON 数组，不要输出任何其他文字。 2. 每个任务对象包含三个字段：task（任务描述）、priority（优先级，取值 high/medium/low）、source（来源片段）。 3. 如果无法拆分，输出空数组 []。 4. 不要解释，不要总结，不要添加额外字段。 输入内容： {content} 输出：

关键点在于：明确角色、明确规则、明确格式、明确禁止项。特别是“不要输出任何其他文字”这句，能挡掉 90% 的格式问题。

还有一个技巧是给示例。在提示词里放一两个输入输出示例，模型的表现会稳定很多。这叫 few-shot，对本地小模型尤其有效。

3.3 输出解析与容错

即使提示词写得再好，模型偶尔还是会输出带 markdown 代码块包裹的 JSON，或者多一句“好的，以下是结果”。所以 L1 的输出必须做容错解析。

我的做法是写一个健壮的 JSON 提取函数：

import json import re def parse_model_output(text): # 去掉 markdown 代码块标记 text = re.sub(r"```json\s*", "", text) text = re.sub(r"```\s*", "", text) text = text.strip() # 尝试直接解析 try: return json.loads(text) except json.JSONDecodeError: pass # 尝试提取第一个 JSON 数组 match = re.search(r"\[.*\]", text, re.DOTALL) if match: try: return json.loads(match.group(0)) except json.JSONDecodeError: pass # 解析失败，返回空并记录 return []

这个函数能处理大部分格式异常。如果还是解析失败，就把原始输出记到日志里，人工排查。我跑了一个月，解析失败率在 2% 以下，完全可接受。

4. 两级流水线的串联与调度

4.1 流水线的整体架构

把 L0 和 L1 串起来，整体流程是这样的：

1. 输入内容先做标准化预处理。
2. 进入 L0 规则匹配。
3. 如果 L0 命中且置信度高于阈值，直接采纳结果。
4. 如果 L0 未命中或置信度低于阈值，转入 L1。
5. L1 调用本地模型做语义拆分。
6. 合并 L0 和 L1 的结果，做去重和排序。
7. 输出最终任务列表。

这个流程里，阈值设定是个关键参数。我设的是 0.85。高于 0.85 的直接采纳，低于的转 L1。这个值可以根据你的实际数据调，原则是：宁可多转 L1，也不要让低置信度的规则结果污染输出。

4.2 批量处理与并发控制

本地模型推理是串行的，如果 L1 待处理内容多，流水线就会堵在这里。我的做法是批量提交 + 有限并发。

Ollama 本身支持并发请求，但并发数太高会导致显存溢出。我实测下来，16G 显存的机器，并发数设 2 到 3 比较稳。再高就会开始排队，反而变慢。

from concurrent.futures import ThreadPoolExecutor def process_batch(items, max_workers=2): results = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [executor.submit(process_single, item) for item in items] for future in futures: results.append(future.result()) return results

另外，L1 的调用要加超时和重试。本地模型偶尔会因为显存问题卡住，超时设 60 秒，重试一次，还失败就降级为“未拆分”并记录。

4.3 结果合并与去重

L0 和 L1 的结果合并时，最大的问题是重复。比如一条内容既被规则命中了，又被模型拆出了类似的任务。我的去重策略是：基于任务描述做归一化后比对。归一化包括去空格、转小写、去掉标点。如果两条任务的归一化描述相似度超过 0.9，就保留置信度高的那条。

相似度计算用简单的编辑距离就行，不用上 embedding，那个太重了。

def normalize(text): text = text.lower() text = re.sub(r"[^\w\s]", "", text) text = re.sub(r"\s+", " ", text).strip() return text def is_duplicate(a, b, threshold=0.9): from difflib import SequenceMatcher return SequenceMatcher(None, normalize(a), normalize(b)).ratio() > threshold

这个去重逻辑跑下来，能消掉 95% 以上的重复项。

5. 常见问题与排查技巧

5.1 规则误命中怎么排查

规则误命中是最常见的问题。比如你用“错误”匹配日志级别，结果“错误率”也被命中了。排查方法是：给每条规则加一个命中日志，记录命中的原文片段。然后定期抽样看，发现误命中就调整规则。

我的经验是，规则宁窄勿宽。窄规则漏掉的内容，L1 能兜住；宽规则误命中的内容，L1 可兜不住，因为 L0 已经直接采纳了。

5.2 模型输出不稳定怎么调

模型输出不稳定，通常有三个原因：温度太高、提示词约束不够、模型能力不足。按这个顺序排查：先把 temperature 降到 0.1 以下；再检查提示词有没有明确格式要求；如果还不行，考虑换更大的模型或者加 few-shot 示例。

我遇到过一次，模型总是把 priority 字段输出成中文“高/中/低”。后来在提示词里明确写了“priority 取值必须是 high/medium/low 这三个英文单词之一”，问题就解决了。

5.3 流水线吞吐上不去怎么办

吞吐上不去，先看瓶颈在哪一级。如果 L0 慢，优化规则匹配逻辑；如果 L1 慢，看是模型推理慢还是并发不够。模型推理慢的话，可以考虑量化版本，比如 qwen2.5:7b 的 q4 量化版，速度能快一倍，精度损失很小。

还有一个容易被忽略的点：L0 和 L1 之间的数据传递。如果 L0 输出的中间结果很大，序列化反序列化也会耗时。我的做法是 L0 直接输出最终结构，不要传原始文本给 L1，只传需要模型处理的那部分。

5.4 常见问题速查表

注意：本地部署的显存是硬约束，不要为了追求速度把并发调太高，显存溢出导致的崩溃比慢更麻烦。

6. 一些实操心得

这套流水线我跑了小半年，最大的体会是：不要迷信模型。很多人一上来就想用模型解决所有问题，结果就是又慢又不稳。实际上，大部分任务拆分场景里，规则能覆盖的比例远超你的想象。先把规则做扎实，模型只做兜底，整体效果和成本都会好很多。

另一个心得是：日志要打全。L0 命中了什么、L1 输出了什么、解析失败了什么，都要记下来。这些日志是你后续优化的唯一依据。我每周会花半小时看日志，调整规则和提示词，流水线的准确率从最初的 70% 慢慢爬到了 92%。

最后分享一个小技巧：L1 的提示词里加上“如果输入内容已经是结构化任务，直接原样返回”。这样能避免模型对已经拆好的内容做二次拆分，减少不必要的算力消耗。这个细节很小，但实际跑起来能省不少事。