企业官网建设流程全解析

本地AI任务拆分这件事，我前前后后折腾了大半年，从最初一股脑把整段需求丢给本地模型、结果被它的"自由发挥"坑得怀疑人生，到后来慢慢摸索出一套"硬规则先扛、模型后兜底"的两级流水线，中间踩的坑能写满一个笔记本。今天要聊的这套L0硬规则前置 + L1模型兜底的两级流水线，就是我在真实项目里反复打磨出来的方案。它的核心思路特别朴素：能用确定性代码判断的事情，绝不交给模型去猜；只有规则覆盖不到的模糊地带，才让本地模型上场兜底。这样做的好处是任务拆分的稳定性大幅提升，输出格式几乎不会跑偏，同时本地推理的调用次数被压到最低，对硬件的要求也跟着降下来。

如果你正在本地部署AI大模型、想用它做任务拆分或者文档整理，又或者你手里只有一张消费级显卡、甚至想用Titan RTX这类老卡跑本地推理，那这套流水线大概率能帮到你。它不依赖任何云端服务，全部在本地跑，适合对数据隐私敏感、又希望流程可控的场景。下面我会把整套设计逻辑、代码骨架、参数取舍和踩坑经验全部摊开讲，尽量让刚接触本地AI部署的人也能照着复现。

1. 为什么任务拆分不能全交给本地模型

1.1 本地模型做任务拆分的三个典型翻车现场

先说清楚问题，不然你不会理解为什么要搞两级流水线。我最早的做法很直接：把用户的一段自然语言需求，整段塞给本地部署的模型，让它输出一个JSON格式的任务列表。听起来很美好，实际跑起来问题一堆。

第一个翻车现场是格式漂移。你要求它输出JSON，它前几次确实老老实实输出JSON，跑到第十几次的时候突然给你加一段"好的，以下是我的分析："，或者把数组外面套一层解释文字。本地小模型（7B、13B这个量级）对指令的遵循能力远不如云端大模型，格式约束稍微复杂一点就开始飘。

第二个翻车现场是任务粒度失控。同样一句"帮我整理一下项目文档"，有时候它拆成3个子任务，有时候拆成11个，而且拆出来的粒度完全不均匀——有的任务大到能写一篇论文，有的任务小到"打开文件夹"这种废话。你没法用固定阈值去卡它，因为它的输出本身就不稳定。

第三个翻车现场是幻觉式补全。你明明没提某个需求，它自作主张给你加一个"发送邮件通知相关人员"的任务。本地模型在缺乏足够上下文约束时，会倾向于"脑补"出它认为合理的步骤，这在任务拆分场景里是致命的。

1.2 硬规则和模型各自的能力边界在哪里

踩完这些坑之后我复盘了一下，发现问题的根源在于：我把两类完全不同性质的工作混在一起交给了模型。

任务拆分里其实包含两种判断：一种是确定性的结构判断，比如"这句话里有没有出现时间词""这个需求属于哪一类预设模板""输出必须包含哪几个字段"；另一种是模糊的语义理解，比如"用户这句话背后真正想干什么""这个需求应该拆成几个步骤才合理"。

前者用代码写规则就能100%搞定，而且稳定、可测试、零延迟；后者才是模型真正擅长的地方。我之前的错误就是把确定性的部分也交给了模型，等于让一个擅长模糊判断的家伙去做精确的活，它当然做不好。

所以两级流水线的分工就清晰了：

核心原则：L0能判的绝不上升给L1，L1只处理L0明确标记为"无法确定"的残余部分。

1.3 两级流水线带来的实际收益

改成两级之后，我做了个对比测试，用同一批200条真实需求跑：

• 纯模型方案：格式合规率约78%，任务粒度一致性差，平均每条需求调用模型1次，单条耗时2.3秒（本地13B模型，量化后）。
• 两级流水线：格式合规率100%（因为格式由L0强制约束），任务粒度一致性显著提升，只有约35%的需求需要真正调用L1，单条平均耗时降到0.9秒。

也就是说，超过六成的需求在L0阶段就被规则直接处理掉了，根本没惊动模型。这不仅省了算力，更重要的是把最容易出错的部分用确定性手段锁死了。剩下的三成模糊需求交给模型，因为上下文已经被L0清洗和约束过，模型的表现也稳定得多。

2. L0硬规则层到底该管哪些事

2.1 用关键词表和正则做需求分类

L0的第一件事是给需求打标签。我维护了一张关键词映射表，把常见需求归到几个大类里，比如"文档整理""代码重构""数据处理""信息提取"等。每条需求进来，先过一遍关键词匹配。

# L0 需求分类规则示例 CATEGORY_RULES = { "doc_organize": ["整理", "归档", "分类", "文档", "文件夹"], "code_refactor": ["重构", "优化代码", "拆分函数", "重命名", "代码"], "data_process": ["清洗", "去重", "格式转换", "csv", "excel"], "info_extract": ["提取", "抽取", "找出所有", "汇总"], } def classify_by_rules(text): hits = [] for category, keywords in CATEGORY_RULES.items(): score = sum(1 for kw in keywords if kw in text) if score > 0: hits.append((category, score)) if not hits: return None # 交给L1 hits.sort(key=lambda x: x[1], reverse=True) return hits[0][0]

这段代码的关键在于返回None表示"我搞不定"，而不是硬猜一个类别。这个设计很重要，它明确了L0的能力边界，把不确定的情况干净地交给下一级。我见过很多人写规则层时喜欢"尽量给个结果"，结果规则层自己就开始产生错误分类，反而污染了后续流程。

2.2 时间、数量、路径这类结构化信息的抽取

需求里经常夹带结构化信息，比如"把上周的日志整理一下""处理前100条记录""输出到D盘的report文件夹"。这些信息用正则抽取比模型靠谱得多。

import re def extract_structured_info(text): info = {} # 数量：前100条、最近50个 m = re.search(r"(前|最近|最新)?\s*(\d+)\s*(条|个|份|篇)", text) if m: info["limit"] = int(m.group(2)) # 路径：盘符或常见路径写法 m = re.search(r"([A-Za-z]:\\[^\s，。]+|/[\w/]+)", text) if m: info["path"] = m.group(1) # 时间范围 if "上周" in text: info["time_range"] = "last_week" elif "本月" in text: info["time_range"] = "this_month" return info

这里有个经验：正则不要写得太贪心。我一开始写的路径正则把整句话都吞进去了，因为[^\s]+会一直匹配到空格为止。后来改成明确排除中文标点，才稳定下来。规则层的正则一定要拿真实数据反复测，别想当然。

2.3 输出格式的强制约束与校验

L0还负责一件事：定义并校验输出结构。不管后面L1怎么发挥，最终输出必须符合一个固定的schema。我一般用Pydantic或者简单的dict校验来做。

from pydantic import BaseModel, ValidationError from typing import List, Optional class SubTask(BaseModel): id: int action: str target: Optional[str] = None params: dict = {} class TaskPlan(BaseModel): category: str subtasks: List[SubTask] source: str # "L0" 或 "L1" def validate_plan(raw): try: return TaskPlan(**raw) except ValidationError as e: return None # 校验失败，回退或重试

校验失败时不要慌，这正是流水线的价值所在——失败可以被捕获、被记录、被重试，而不是像纯模型方案那样悄悄输出一个错误结果。我通常会在校验失败时把原始文本和失败原因一起记进日志，方便后续优化规则。

2.4 什么情况下必须把球踢给L1

L0不是万能的，以下几种情况我会明确标记为"需要L1介入"：

• 关键词表里一个类别都没命中，或者多个类别得分接近（比如都是1分），无法确定主类别。
• 需求里包含明显的模糊表述，比如"帮我看看这个怎么弄""优化一下体验"这类没有明确动作词的句子。
• 结构化信息抽取后，关键字段缺失，比如分类是"数据处理"但没抽到数据源。
• 需求涉及多步骤逻辑，规则无法判断步骤之间的依赖关系。

把这些情况统一打上一个need_l1=True的标记，交给下一级。这个标记本身就是L0的产出之一，它让整个流程的走向变得可观测。

3. L1模型兜底层怎么接才不翻车

3.1 本地模型的选型与量化取舍

L1要跑本地模型，选型是第一道坎。我的经验是：任务拆分这种结构化输出场景，不需要追求模型参数规模，7B到14B的指令微调模型足够，关键是量化要选对。

我实测过几档配置，用同一批模糊需求做对比：

可以看到，从13B Q4到34B Q4，可用率只提升了4个百分点，但耗时翻了一倍多。性价比拐点在13B Q4附近。如果你用的是Titan RTX这类24GB显存的卡，跑13B Q4绰绰有余，甚至能留出显存给并发。追求极致稳定可以上Q8，但大多数场景Q4够用。

注意：量化等级越低，模型对复杂指令的遵循能力下降越明显。Q4是可用下限，再往下（Q3、Q2）格式漂移会明显增多，不建议在任务拆分场景使用。

3.2 给模型的提示词要"窄"不要"宽"

L1的提示词设计和通用对话完全不同。因为L0已经把大部分确定性工作做完了，L1只需要处理残余的模糊判断，所以提示词要尽可能窄。

我的做法是把L0的中间结果作为上下文喂给模型，明确告诉它"你只需要做这一件事"：

L1_PROMPT = """你是一个任务拆分助手。用户需求已经过初步分析，结果如下： - 已识别类别：{category} - 已抽取参数：{params} - 待解决问题：{pending} 请只针对"待解决问题"输出补充的拆分结果，严格返回JSON数组，每个元素包含 action 和 target 两个字段。 不要输出任何解释文字，不要添加未提及的任务。""" def call_l1(text, l0_result): prompt = L1_PROMPT.format( category=l0_result.get("category", "未知"), params=l0_result.get("params", {}), pending=l0_result.get("pending", text) ) return model.generate(prompt)

这个提示词的关键点：明确边界（只处理待解决问题）、明确格式（JSON数组）、明确禁止（不要解释、不要脑补）。我试过把整段原始需求直接给模型，输出质量明显不如这种"带着L0结论去问"的方式。

3.3 用few-shot把输出格式钉死

光靠指令约束还不够，本地小模型对格式的敏感度需要few-shot来强化。我会在提示词里塞2到3个输入输出示例，覆盖最常见的拆分模式。

FEW_SHOT = """ 示例1： 待解决问题：把文档按类型分到不同文件夹 输出：[{"action": "classify", "target": "documents"}, {"action": "move", "target": "by_type"}] 示例2： 待解决问题：优化这段代码的可读性 输出：[{"action": "rename_variables", "target": "code"}, {"action": "extract_function", "target": "code"}] """

few-shot的示例要短、准、格式统一。我踩过的坑是示例写得太长太详细，结果模型开始模仿示例里的措辞而不是结构，反而限制了它的泛化。后来把示例压缩到一两行，效果反而更好。

3.4 模型输出的二次清洗与回退策略

模型输出永远不能直接信，必须过一遍清洗。我的清洗流程是：

1. 用正则把输出里的markdown代码块标记、前后解释文字剥掉，只留JSON部分。
2. 尝试解析JSON，失败则进入回退。
3. 解析成功后用L0的schema校验，字段缺失或类型错误则进入回退。
4. 回退策略：重试一次（换更严格的提示词），再失败就返回一个"降级结果"——把整个需求当成单个任务，标记为needs_review。

import json def clean_and_parse(raw_output): # 剥离代码块 raw_output = re.sub(r"```(json)?", "", raw_output).strip() # 截取第一个 [ 到最后一个 ] start, end = raw_output.find("["), raw_output.rfind("]") if start == -1 or end == -1: return None try: return json.loads(raw_output[start:end+1]) except json.JSONDecodeError: return None

这个"降级结果"的设计很重要。流水线不能因为某一级失败就整个崩掉，它必须能优雅退化。降级结果虽然不完美，但至少保证了流程能继续，而且被标记出来供人工复核。

4. 两级之间怎么衔接才顺畅

4.1 中间数据结构的统一约定

L0和L1之间传递的数据必须有一个统一的结构，否则衔接处会变成一团乱麻。我定义了一个中间态PipelineContext，贯穿整个流程：

from dataclasses import dataclass, field @dataclass class PipelineContext: raw_text: str category: str = None params: dict = field(default_factory=dict) subtasks: list = field(default_factory=list) need_l1: bool = False pending: str = "" source: str = "L0" errors: list = field(default_factory=list)

这个结构的好处是每一级只修改自己负责的字段，L0填category、params、need_l1，L1填subtasks。出问题时看哪个字段是空的，就知道卡在哪一级。

4.2 触发L1的条件判断与优先级

不是所有need_l1=True的情况都同等紧急。我给它分了个优先级：

• 高优先级：L0完全无法分类，且需求文本较长（超过50字），这类必须走L1。
• 中优先级：分类成功但关键参数缺失，走L1补全。
• 低优先级：分类成功、参数齐全，只是步骤依赖关系不明确，可以先用L0的默认拆分，L1作为可选优化。

这个优先级设计让流水线在资源紧张时（比如并发请求多、显存吃紧）可以只处理高优先级，把中低优先级的先缓存起来。实测下来，高优先级通常只占全部请求的15%左右，压力小很多。

4.3 超时、异常与降级的处理

本地模型推理偶尔会卡住或者超时，尤其是显存不足触发换页的时候。我给L1调用加了硬超时：

import signal class TimeoutError(Exception): pass def handler(signum, frame): raise TimeoutError("L1 inference timeout") def call_l1_with_timeout(text, l0_result, timeout=10): signal.signal(signal.SIGALRM, handler) signal.alarm(timeout) try: result = call_l1(text, l0_result) signal.alarm(0) return result except TimeoutError: return None # 触发降级

超时后直接走降级逻辑，把需求标记为needs_review，绝不阻塞整个流程。这个设计在批量处理场景里救过我好几次——有一次模型因为显存碎片卡了整整30秒，如果没有超时，整批任务都得等着。

4.4 日志埋点让流水线可观测

流水线跑起来之后，最怕的是"黑盒"——出了问题不知道卡在哪。我在每个关键节点都埋了日志：

• L0分类结果和命中关键词
• 是否触发L1、触发原因
• L1原始输出、清洗后输出、校验结果
• 最终source标记（L0还是L1）
• 耗时统计

import logging logger = logging.getLogger("pipeline") def log_stage(stage, ctx, extra=None): logger.info({ "stage": stage, "category": ctx.category, "need_l1": ctx.need_l1, "source": ctx.source, "errors": ctx.errors, "extra": extra or {} })

有了这些日志，我后来优化规则时能精确知道"哪些需求被L0误判了""L1在哪些类别上表现差"，优化方向一下就清晰了。

5. 实测中那些文档不会告诉你的坑

5.1 中文标点和空格导致的规则失效

这个坑我踩得最冤。规则里写的是"整理" in text，结果用户输入的是"整理一下"，没问题；但用户输入"整 理"（中间带空格）或者用了全角标点，规则就失效了。中文文本的预处理比英文麻烦得多。

我的解决方案是在L0入口做一次归一化：

def normalize(text): # 全角转半角 text = text.replace(" ", " ") # 去除多余空格 text = re.sub(r"\s+", "", text) return text

注意这里我直接把所有空格去掉了，因为中文需求里空格基本没有语义价值，去掉反而让关键词匹配更稳。但如果是中英混合的需求，这个策略要慎用，得保留英文单词间的空格。

5.2 模型对"否定句"的理解偏差

本地小模型对否定句的处理很弱。比如"不要删除原文件"，它可能拆出一个"删除文件"的任务。这个坑在任务拆分场景里特别危险，因为否定往往意味着约束条件。

我的应对是在L0阶段就把否定词识别出来，作为约束参数传给L1：

NEGATION_WORDS = ["不要", "别", "禁止", "避免", "不能"] def extract_negations(text): return [w for w in NEGATION_WORDS if w in text]

然后在L1提示词里明确写"以下操作被禁止：{negations}"。实测下来，这样处理后否定相关的错误率下降了一大半。

5.3 并发调用时的显存竞争

如果你要批量处理需求，并发调用L1会撞上显存竞争。我一开始开了4个并发，结果显存直接爆了，模型开始疯狂换页，速度反而比单线程还慢。

后来改成单模型实例 + 请求队列的模式，用一个信号量控制并发数：

import threading semaphore = threading.Semaphore(1) # 单实例串行 def safe_l1_call(text, l0_result): with semaphore: return call_l1_with_timeout(text, l0_result)

并发数设成1看起来保守，但因为L0已经过滤掉大部分请求，实际吞吐并不低。如果你的显存充裕（比如24GB以上），可以设成2，但一定要监控显存占用，别让它触发换页。

5.4 规则和模型的"责任边界"会随数据漂移

最后一个坑比较隐蔽：随着业务数据变化，L0和L1的责任边界会漂移。比如一开始"数据处理"类需求很少，L0的关键词表覆盖得住；后来这类需求暴增，出现了很多新表述，L0的命中率就下降了，更多请求涌向L1，整体耗时上升。

我的做法是定期复盘日志，统计L0的未命中率。如果某个类别的未命中率超过20%，就说明规则该更新了，把新出现的表述补进关键词表。这个维护动作不能省，否则流水线会慢慢退化成"纯模型方案"。

6. 把这套流水线用到你自己的场景

6.1 从最小可用版本开始搭

别一上来就追求完美。我的建议是先搭一个最小版本：L0只做最简单的关键词分类，L1只接一个本地模型做兜底，中间用一个dict传递数据。跑通之后再逐步加规则、加校验、加日志。

最小版本的代码量其实很小，核心就是三段：分类函数、模型调用、结果合并。我第一版总共不到100行，但已经能跑通整个流程，后面所有的优化都是在这个骨架上长出来的。

6.2 规则库的迭代节奏

规则库不是一次写完的，它是跟着真实数据长出来的。我的节奏是：每周看一次日志，把L0未命中的高频表述补进关键词表，把误判的规则修掉。每次改动都跑一遍回归测试，确保没把原来能命中的搞坏。

回归测试集我建议至少攒100条真实需求，覆盖各个类别，每次改规则都跑一遍。这个投入很值，能避免"修一个坏三个"的尴尬。

6.3 什么时候该考虑升级方案

这套两级流水线适合需求类型相对固定、对稳定性要求高、硬件资源有限的场景。如果你遇到下面这些情况，可能要考虑升级：

• 需求类型极其多样，规则库维护成本超过收益，这时候可以考虑用更强的模型或者引入检索增强。
• 对延迟要求极高（比如实时交互），本地模型推理再快也有几百毫秒，这时候可能要把L1也规则化，或者用更小的模型。
• 需要多轮对话式拆分，这套流水线是单轮的，多轮场景需要额外的状态管理。

但说实话，大多数本地AI任务拆分的场景，两级流水线已经够用了。它最大的价值不是技术多先进，而是把不确定性关进了笼子里——确定的部分用规则锁死，不确定的部分用模型兜底，中间用清晰的边界和降级策略保证流程不崩。这套思路我在文档整理、代码重构辅助、数据清洗好几个场景里都复用过了，每次都是先搭骨架、再补规则、最后调模型，稳得很。

如果你也在本地折腾AI任务拆分，不妨先从L0的关键词表开始写起，把最确定的那部分先固化下来，你会发现后面的事情顺很多。