本地任务拆分实战：L0硬规则+L1模型两级流水线

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

1.1 从一次线上事故说起

去年年底我接手了一个内部工单系统的自动化改造，需求本身不复杂：把用户提交的自然语言工单，自动拆成结构化的子任务，再分派给对应的处理人。一开始我图省事，直接调了一个云端大模型的接口，把整段工单文本丢进去，让它一次性输出拆分结果。测试阶段一切正常，上线第三天出了事——那天网络抖动，接口超时率飙到 40%，积压的工单直接把队列堵死，值班同事半夜打电话过来，我爬起来手动补数据补到凌晨四点。

那次之后我就下定决心，把任务拆分这条链路搬到本地来做。原因有三个：第一，工单内容涉及内部业务信息，走云端接口本身就有合规风险；第二，拆分逻辑其实有大量固定模式，比如"帮我改一下登录页面的按钮颜色，顺便把文案也更新一下"这种，完全可以用规则命中，没必要每次都调模型；第三，本地推理虽然单次慢一点，但胜在稳定可控，不会因为外部服务抖动就把整条链路拖垮。

这就是"L0硬规则前置 + L1模型兜底"两级流水线的由来。L0 层用确定性规则处理高频、格式固定的输入，L1 层用本地模型处理规则覆盖不到的复杂语义。两级串联，既保证了吞吐，又保住了准确率。

1.2 这套方案到底解决什么问题

说白了，它解决的是**"用规则太死、用模型太贵"**这个经典矛盾。纯规则方案的问题是泛化能力差，用户换个说法就命中不了；纯模型方案的问题是成本和延迟都不可控，尤其是本地部署场景下，模型推理本身就吃资源，如果每个请求都走模型，GPU 排队能排到你怀疑人生。

两级流水线的核心思路是分流：让简单请求走快车道，复杂请求走慢车道。我实测下来，在一个日均 8000 条工单的场景里，L0 层能拦下大约 62% 的请求，剩下 38% 才进 L1。这意味着模型的实际负载只有全量请求的三分之一多一点，单卡就能扛住，不需要堆硬件。

适合参考这套方案的人，我总结了几类：一是做企业内部自动化工具、对数据不出域有要求的开发者；二是本地部署了模型但发现推理压力大、想优化吞吐的同行；三是刚接触任务拆分这个概念、想找个能落地的入门架构的新手。如果你属于第三类，建议先把 L0 层吃透，规则引擎玩明白了，再上模型也不迟。

1.3 两级流水线的整体架构长什么样

架构本身不复杂，我用文字描述一下数据流向。请求进来之后，先经过一个预处理模块，做文本清洗、编码统一、长度截断这些杂活。清洗完的文本送进 L0 规则引擎，引擎里挂了一组按优先级排序的规则，每条规则包含匹配条件和对应的拆分模板。命中规则就直接产出结果，打上source: L0的标记；没命中就转交 L1。

L1 层是一个本地部署的模型服务，接收原始文本，输出结构化的拆分结果，打上source: L1的标记。两层的结果最终汇入同一个输出格式，下游消费方不需要关心这个结果是哪一层产出的。

这里有个设计细节值得说一下：L0 和 L1 的输出格式必须严格一致。我一开始偷懒，L0 输出的字段名和 L1 不一样，结果下游解析代码里写了一堆 if-else 判断来源，维护起来极其痛苦。后来统一成同一套 schema，下游代码干净了一大截。

2. L0 硬规则层的设计与实现

2.1 规则怎么定：从真实语料里挖模式

规则不是拍脑袋想出来的，得从真实数据里挖。我的做法是：先收集最近一个月的工单文本，大概两万条，做一轮词频统计和句式聚类。具体操作上，我用 jieba 做分词，把高频动词和名词组合提取出来，再人工过一遍，归纳出几类典型模式。

挖下来发现，工单文本其实高度模板化，主要集中在这几类：

前四类加起来 62%，正好对应我前面说的 L0 拦截率。这个数字不是巧合，是规则覆盖能力的直接体现。你要做的第一件事，就是拿自己的数据跑一遍这个统计，看看你的 L0 理论上能覆盖多少。如果低于 40%，说明你的数据太发散，得重新考虑架构。

2.2 规则引擎的选型：为什么我没用现成的

市面上规则引擎不少，Drools、Easy Rules 这些我都试过。最后我选择自己写一个轻量级的，原因很实际：现成引擎功能太全，配置复杂，学习成本高，而我需要的只是"按优先级匹配 + 输出模板"这两个功能。为了这两个功能引入一个几万行的依赖，不划算。

我自己写的引擎核心就三个类：Rule（规则定义）、RuleEngine（匹配调度）、TemplateRenderer（模板渲染）。规则用 YAML 配置，方便非开发同事也能改。一条规则的配置长这样：

- id: rule_001 priority: 100 pattern: "(修改|调整|改一下).*(颜色|样式|配色)" template: | { "action": "modify_style", "target": "{target}", "attribute": "color" } enabled: true

priority越大越先匹配，pattern是正则，template是输出模板。匹配到之后，用正则的捕获组填充模板占位符。这套东西加起来不到 300 行代码，但覆盖了我 90% 的规则需求。

提示：正则写规则有个坑，中文的贪婪匹配很容易吃掉不该吃的内容。建议在关键位置用非贪婪模式.*?，并且给每条规则写单元测试，用真实语料验证。

2.3 规则优先级与冲突处理

规则一多，冲突就来了。比如"修改登录页按钮颜色"这句话，可能同时命中"修改样式"和"修改页面元素"两条规则。这时候优先级就起作用了，数字大的先匹配，匹配到就返回，不再往下走。

但光靠优先级不够，因为有些规则是互斥的，有些是可以叠加的。我的处理方式是给规则加一个exclusive字段，标记这条规则是否独占。独占规则命中后直接返回，非独占规则命中后继续往下匹配，最后合并结果。

def match(self, text): results = [] for rule in sorted(self.rules, key=lambda r: -r.priority): if rule.pattern.search(text): results.append(rule.render(text)) if rule.exclusive: break return results

这段逻辑很简单，但实际跑起来效果很好。我踩过的坑是：非独占规则的结果合并顺序很重要。一开始我用列表追加，结果下游拿到的字段顺序是乱的。后来改成按 priority 排序后再合并，问题解决。

2.4 规则的可维护性：让非开发也能改

规则这东西，业务变化快，今天加一条明天改一条。如果每次都要开发介入，效率太低。我的做法是把规则配置抽成独立的 YAML 文件，放在一个共享目录里，业务同事改完提交，走一个简单的审核流程就能生效。

为了防止改错，我加了一个规则自检脚本，每次配置变更后自动跑一遍：检查正则语法是否合法、模板占位符是否和捕获组数量匹配、优先级是否有重复。这个脚本帮我拦下了好几次低级错误，强烈建议你也加一个。

import re, yaml def validate_rule(rule): try: pattern = re.compile(rule['pattern']) except re.error as e: return f"正则语法错误: {e}" groups = pattern.groups placeholders = re.findall(r'\{(\w+)\}', rule['template']) if len(placeholders) > groups: return f"占位符数量({len(placeholders)})超过捕获组数量({groups})" return None

3. L1 模型兜底层的落地细节

3.1 本地模型选型：不是越大越好

L1 层的核心是本地模型。选型的时候我对比了几个维度：参数量、推理速度、显存占用、中文能力。最后选了一个 7B 级别的指令微调模型，量化到 4bit 之后，显存占用大概 5GB，单张消费级显卡就能跑。

为什么不选更大的？因为任务拆分这个场景，对模型的指令遵循能力要求高于知识储备。7B 模型在指令微调之后，处理结构化输出完全够用。我实测过 13B 和 7B 在同一个测试集上的表现，准确率差距不到 3 个百分点，但推理速度差了将近一倍。在吞吐优先的场景下，7B 是更划算的选择。

部署方式上，我用的是 Ollama，一条命令就能拉起来，省去了自己配环境的麻烦。启动命令很简单：

ollama run qwen2.5:7b-instruct-q4_K_M

跑起来之后，它会暴露一个本地 HTTP 接口，默认在 11434 端口。Python 侧用 requests 直接调就行，不需要额外的 SDK。

3.2 提示词设计：结构化输出的关键

模型能不能稳定输出结构化结果，全看提示词怎么写。我试过好几种写法，最后稳定下来的版本包含四个部分：角色设定、任务描述、输出格式、示例。

角色设定要具体，不能只说"你是一个助手"，要说"你是一个任务拆分专家，负责把用户的自然语言需求拆成结构化的子任务列表"。任务描述要明确边界，告诉它什么该拆什么不该拆。输出格式用 JSON Schema 描述，越详细越好。示例给两到三个，覆盖典型场景。

PROMPT_TEMPLATE = """你是一个任务拆分专家。请把用户输入拆分成子任务列表。 要求： 1. 每个子任务包含 action（动作）、target（对象）、detail（细节）三个字段 2. 如果无法拆分，返回包含单个子任务的列表 3. 只输出 JSON，不要输出任何解释文字 输出格式： {{"tasks": [{{"action": "...", "target": "...", "detail": "..."}}]}} 示例输入：改一下登录页按钮颜色，顺便更新文案 示例输出：{{"tasks": [{{"action": "modify", "target": "登录页按钮", "detail": "颜色"}}, {{"action": "update", "target": "文案", "detail": ""}}]}} 用户输入：{user_input} """

这里有个细节：示例的质量直接决定输出质量。我一开始给的示例太简单，模型遇到复杂输入就开始自由发挥。后来把示例换成带条件约束的复杂案例，输出稳定性明显提升。

3.3 输出解析与容错

模型输出 JSON 这件事，理想很丰满，现实很骨感。即使提示词写得再好，偶尔还是会冒出多余的解释文字、markdown 代码块标记、或者干脆 JSON 格式错误。所以解析层必须做容错。

我的解析逻辑分三步：先尝试直接json.loads；失败了就用正则提取第一个完整的 JSON 对象；再失败就返回一个降级结果，标记source: L1_failed，让下游决定怎么处理。

import json, re def parse_model_output(text): try: return json.loads(text) except json.JSONDecodeError: pass match = re.search(r'\{.*\}', text, re.DOTALL) if match: try: return json.loads(match.group()) except json.JSONDecodeError: pass return {"tasks": [], "error": "parse_failed"}

这个降级机制救过我好几次。有一次模型升级后输出格式变了，解析全挂，但因为降级逻辑在，系统没崩，只是准确率掉了，给了我缓冲时间修。

3.4 推理性能优化：批处理与缓存

本地模型推理，性能是绕不开的话题。我做了两件事来优化：批处理和缓存。

批处理是指把多个请求攒一批一起送进模型。Ollama 本身支持 batch，但需要你在请求时指定。我设的批大小是 8，实测下来吞吐能提升 2 倍多。不过批大小不是越大越好，太大显存扛不住，而且单次延迟会变高。8 是我在 5GB 显存下的甜点值。

缓存是指对相同或相似的输入直接返回历史结果。我用了一个简单的 LRU 缓存，key 是输入文本的哈希，value 是拆分结果。命中率大概 15%，虽然不高，但省下来的推理时间很可观。

from functools import lru_cache @lru_cache(maxsize=1000) def split_with_model(text): # 调用模型推理 ...

注意：缓存要设过期时间，不然业务规则变了，缓存还返回旧结果，会出问题。我设的是 24 小时。

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

4.1 请求分发逻辑

两级串联的核心是分发逻辑。请求进来先走 L0，命中就返回，没命中转 L1。听起来简单，但实际写的时候有几个细节要注意。

第一，L0 的匹配要有超时保护。正则匹配虽然快，但如果规则写得不好，遇到超长文本可能会卡住。我加了一个 50ms 的超时，超时就当没命中，转 L1。

第二，L1 的调用要异步。模型推理慢，如果同步等，会把整个请求线程堵死。我用了一个简单的线程池，L1 调用丢进池子里异步执行，主线程轮询结果。

第三，结果要统一格式。前面说过，L0 和 L1 的输出 schema 必须一致。我定义了一个TaskResult类，两层都往这个类里填，下游只认这个类。

class TaskResult: def __init__(self, tasks, source, latency): self.tasks = tasks self.source = source # "L0" or "L1" self.latency = latency

4.2 降级与熔断策略

本地部署虽然稳定，但也不是铁板一块。模型服务可能因为显存不足崩掉，规则文件可能因为误改加载失败。这些情况都要有降级方案。

我的降级策略是分层的：L1 挂了，就返回 L0 的结果，哪怕 L0 没命中，也返回一个空结果加错误标记，而不是直接抛异常。L0 挂了，就直接全量走 L1。两层都挂了，返回一个兜底结果，标记source: fallback，同时触发告警。

熔断用的是经典的滑动窗口算法，统计最近 100 次调用的失败率，超过 30% 就熔断 30 秒，期间所有请求直接走降级路径。这个机制在模型服务重启的时候特别有用，避免了大量请求堆积。

4.3 监控指标与日志

没有监控的流水线就是黑盒。我埋了几个关键指标：L0 命中率、L1 平均延迟、L1 失败率、端到端延迟 P99。这些指标用 Prometheus 采集，Grafana 展示。

日志方面，每个请求都记一条结构化日志，包含请求 ID、输入文本哈希、走的哪一层、耗时、结果状态。出问题的时候，拿请求 ID 一查就能定位。

import logging, json logger = logging.getLogger("pipeline") def log_request(req_id, text, source, latency, status): logger.info(json.dumps({ "req_id": req_id, "text_hash": hash(text), "source": source, "latency_ms": latency, "status": status }))

这套监控上线之后，我发现了一个之前没注意到的问题：L1 的延迟在每天下午三点左右会有一个尖峰。查下来是因为那个时间段有批量任务在跑，抢了 GPU 资源。后来把批量任务挪到凌晨，问题解决。

5. 实操中踩过的坑与排查技巧

5.1 规则误命中：一个真实的案例

有一次业务同事反馈，说"删除用户"这个工单被拆成了"修改用户信息"。查下来发现，L0 里有一条规则是(修改|调整|改).*(用户)，而"删除用户"里的"除"字被正则的.*吃掉了，导致误命中。

这个坑的根源是正则写得太宽泛。修复方式是把动作词限定得更死，用(修改|调整|改一下)而不是(修改|调整|改)，并且在动作词后面加一个边界断言。改完之后，误命中率从 3% 降到了 0.2%。

提示：写规则的时候，一定要用真实语料做回归测试。我现在的做法是维护一个 500 条的测试集，每次改规则都跑一遍，看命中率和准确率的变化。

5.2 模型输出不稳定：温度参数的调节

模型输出时好时坏，是新手最容易遇到的问题。我一开始用的是默认温度 0.7，结果同样的输入，有时候输出两个子任务，有时候输出三个。后来把温度调到 0.1，输出就稳定多了。

温度这个参数，本质是控制采样的随机性。任务拆分这种需要确定性输出的场景，温度越低越好。我实测下来，0.1 到 0.2 之间是比较合适的区间，再低会显得死板，再高就开始飘。

除了温度，top_p和repeat_penalty也值得调。我的配置是temperature=0.1, top_p=0.9, repeat_penalty=1.1，这套参数在我的场景下表现最稳。

5.3 显存不足：量化与显存管理

本地跑模型，显存不足是家常便饭。我的显卡是 12GB 的，跑 7B 模型 4bit 量化后占用 5GB 左右，看起来够用，但实际上跑一段时间就会 OOM。查下来是因为 Ollama 默认会缓存多个模型实例，显存没及时释放。

解决办法有两个：一是设置OLLAMA_MAX_LOADED_MODELS=1，限制同时加载的模型数量；二是定期重启模型服务，我写了个定时任务，每天凌晨重启一次，释放碎片显存。

如果显存实在紧张，可以考虑用更激进的量化，比如 3bit 或者 2bit。但量化越低，模型能力损失越大，需要你自己权衡。我的建议是优先保证 4bit，实在不行再降。

5.4 常见问题速查表

这张表是我从实际运维中总结出来的，基本上覆盖了 80% 的常见问题。遇到问题先查表，能省不少时间。

6. 性能实测与调优记录

6.1 测试环境与数据集

测试环境是一台带 12GB 显存的机器，CPU 是 8 核，内存 32GB。模型用的是 7B 指令微调版，4bit 量化。数据集是从真实工单里抽的 2000 条，覆盖了前面说的五类模式。

测试指标有三个：L0 命中率、L1 平均延迟、端到端 P99 延迟。测试方法是把 2000 条数据灌进流水线，记录每条的处理路径和耗时。

6.2 调优前后的对比

调优前，L0 命中率 58%，L1 平均延迟 1.2 秒，端到端 P99 是 3.5 秒。这个成绩不算差，但 P99 偏高，说明有长尾请求。

调优做了三件事：一是把 L0 的规则重新梳理了一遍，合并了冗余规则，命中率提到 62%；二是给 L1 加了批处理，平均延迟降到 0.8 秒；三是加了缓存，命中缓存的请求直接返回，P99 降到 1.8 秒。

这个提升幅度不算惊艳，但胜在稳定。尤其是 P99 从 3.5 秒降到 1.8 秒，用户体验的改善是肉眼可见的。

6.3 还能怎么继续优化

如果还想继续压榨性能，有几个方向可以试。一是把 L0 的规则引擎换成更高效的数据结构，比如用 Trie 树做前缀匹配，理论上能把匹配耗时再降一半。二是给 L1 加一个更激进的缓存策略，比如用语义相似度做缓存，而不是精确匹配，命中率能提到 30% 以上。三是考虑模型蒸馏，用大模型生成一批标注数据，蒸馏一个小模型专门做任务拆分，推理速度能再快一倍。

不过这些优化都有成本，要不要做取决于你的实际瓶颈在哪。我的建议是先把监控做好，找到真正的瓶颈再动手，别为了优化而优化。

7. 一些个人体会

这套流水线我从去年年底跑到现在，快一年了，中间迭代了七八个版本。最大的体会是：规则和模型不是对立的，而是互补的。很多人一上来就想用模型解决所有问题，结果发现成本和延迟都扛不住；也有人死守规则，结果泛化能力差，维护成本高。两级流水线的价值就在于，让合适的东西做合适的事。

另一个体会是，监控比优化更重要。我见过太多人花大力气调优，结果上线之后连哪里慢都不知道。先把指标埋好，让数据告诉你瓶颈在哪，再动手，效率高得多。

最后分享一个小技巧：L0 的规则不要一次写太多，先从最高频的几条开始，跑一段时间看命中率，再逐步补充。我一开始一口气写了 50 条规则，结果维护起来极其痛苦，后来砍到 20 条核心规则，命中率反而更高了。规则这东西，质量比数量重要。