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: 提取 keyvalue 形式的配置项 }, { 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, jsonpayload, timeout60) 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(rjson\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 串起来整体流程是这样的输入内容先做标准化预处理。进入 L0 规则匹配。如果 L0 命中且置信度高于阈值直接采纳结果。如果 L0 未命中或置信度低于阈值转入 L1。L1 调用本地模型做语义拆分。合并 L0 和 L1 的结果做去重和排序。输出最终任务列表。这个流程里阈值设定是个关键参数。我设的是 0.85。高于 0.85 的直接采纳低于的转 L1。这个值可以根据你的实际数据调原则是宁可多转 L1也不要让低置信度的规则结果污染输出。4.2 批量处理与并发控制本地模型推理是串行的如果 L1 待处理内容多流水线就会堵在这里。我的做法是批量提交 有限并发。Ollama 本身支持并发请求但并发数太高会导致显存溢出。我实测下来16G 显存的机器并发数设 2 到 3 比较稳。再高就会开始排队反而变慢。from concurrent.futures import ThreadPoolExecutor def process_batch(items, max_workers2): results [] with ThreadPoolExecutor(max_workersmax_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, threshold0.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 常见问题速查表问题现象可能原因排查方法解决方案L0 命中率低规则太窄或输入格式乱抽样看未命中内容加预处理、加弱规则L1 输出格式错提示词约束不够看原始输出加强格式约束、加示例流水线整体慢L1 并发不足看各阶段耗时调并发数、用量化模型结果重复多去重阈值太低看重复项相似度提高阈值、加归一化模型卡死显存不足看显存占用降并发、换小模型注意本地部署的显存是硬约束不要为了追求速度把并发调太高显存溢出导致的崩溃比慢更麻烦。6. 一些实操心得这套流水线我跑了小半年最大的体会是不要迷信模型。很多人一上来就想用模型解决所有问题结果就是又慢又不稳。实际上大部分任务拆分场景里规则能覆盖的比例远超你的想象。先把规则做扎实模型只做兜底整体效果和成本都会好很多。另一个心得是日志要打全。L0 命中了什么、L1 输出了什么、解析失败了什么都要记下来。这些日志是你后续优化的唯一依据。我每周会花半小时看日志调整规则和提示词流水线的准确率从最初的 70% 慢慢爬到了 92%。最后分享一个小技巧L1 的提示词里加上“如果输入内容已经是结构化任务直接原样返回”。这样能避免模型对已经拆好的内容做二次拆分减少不必要的算力消耗。这个细节很小但实际跑起来能省不少事。
