打造确定性 LLM 管道:从结构化约束到规则决策的工程实践 📅 发布时间:2026/9/4 3:29:35 👁 浏览次数: 做基于大模型的应用时不知道你有没有遇到过这种场景同一个问题提交两次一次得到“用户想取消订单”一次得到“客户要求退款”第二次甚至把输出里的intent字段名改成了intention。如果只是做一个聊天 Demo这种差异还能忍受可一旦这套逻辑被接进客服工单分类、合同信息抽取、日志结构化分析这类生产管道后面所有任务都是根据这次输出结果去触发流程一丁点字段漂移都可能导致解析报错、规则失效、甚至把错误数据写进数据库。这几年有一个话题在 LLM 工程社区里被反复讨论——怎么开发“确定性更高”的 LLM pipelines。很多人以为把temperature调到 0 就能得到固定输出结果发现误差还是存在也有人开始把 JSON Schema、校验器、缓存、规则引擎一股脑塞进管道但并不知道每一层到底在解决什么。这篇文章会从问题根源讲起整理一套更适合工程落地的设计思路并给出一个可以照着改写的 Python 示例。无论你是在做 RAG 问答、信息抽取还是自动化 Agent这套思路都能直接套用。1. 先厘清LLM Pipeline 的“确定性”到底指什么1.1 从业务视角理解确定性在自然语言处理领域确定性通常指“相同输入永远得到相同输出”。但对 LLM 应用来说这种定义不一定切合实际。同一个输入模型第一次返回“取消订单”第二次返回“取消订单。”对下游结构化解析来说本质上是同一个结果。所以在工程文档里我更喜欢把确定性拆成三个层次来沟通确定性层次具体表现适用场景字符级确定性相同输入多次调用文本完全一致需要签名、比对、内容指纹的场景结构化确定性文本措辞可能略有差异但字段名、字段类型、枚举值符合约定信息抽取、工单分类、API 参数填充业务决策确定性模型输出可能不同但经过代码决策后进入的业务分支完全一致风控判断、审批流程、自动化工单处理对大多数生产管道来说不需要追求第一层因为成本太高也不现实。重点应放在第二层和第三层。也就是说我们要保证“接口契约稳定”和“下游业务动作稳定”至于中间自然语言怎么表述只要不破坏契约就允许在有限范围内漂移。1.2 哪些 LLM 应用最容易受不确定性影响需要接入下游系统或数据库的应用往往最容易受影响。比如自动化工单系统模型负责从用户描述中抽取“问题类型”“紧急程度”“涉及系统”如果字段值没有落在预设枚举里后续的责任人分配逻辑就无法工作。另一个典型场景是 RAG 查询改写。用户提问后LLM 先改写检索词再把改写结果交给召回模块。改一次和改两次召回结果不同最终答案自然不同。类似需求还很常见关键词抽取、格式转换、代码审查结论分析、产品评论归因等。凡是“模型输出需要被进一步加工”的场景都应该专门设计确定性层。1.3 一个常见误区确定性不等于控制力很多初学同学会陷入一个误区认为只要把 prompt 写得更严格、更详细模型就会乖乖听话。实际上 prompt 只能“增加倾向性”不能从机制上保证结果稳定。模型每次生成 token 都是在下一次概率分布上抽样哪怕某个词概率已经达到 0.95多次调用仍可能出现另外 5% 的路径。真正可靠的确定性来自三层配合参数控制、输出约束、后置校验与决策。只靠其中任意一层都很难构建出能长期维护的稳定管道。2. LLM 管道不稳定根源通常不在模型本身2.1 生成机制天然存在概率行为这里需要理解一下大模型生成文本的基本原理。模型会根据当前上下文为词表里的所有 token 计算一个概率分布然后从中选择下一个 token。选择时通常会受temperature、top_p等采样参数影响。当temperature趋近 0 时模型会倾向于选概率最高的 token但只要某个位置存在两个概率相近的候选最优选择也有可能在不同推理环境下发生变化。更麻烦的是不同模型服务商在后端可能使用不同的批处理方式、量化精度、并行策略即使同一个模型名输出也可能存在细微差异。有些接口支持seed参数用于尽量复现随机数序列但服务端是否严格遵循以及模型版本是否更新都会影响最终效果。这部分不确定性来自基础设施层单靠应用层无法彻底消除。2.2 提示词的微小变化会在管道里被放大不少研发团队习惯在线上直接改提示词改完发现一组测试用例通过另一组却失败了。原因是提示词本来就是一个高维输入空间任何新增描述、示例顺序调整都会改变模型内部的注意力分布。尤其是分类任务中如果存在多个语义相近的类别模型很容易在边界上来回摇摆。管道越长这种“微扰”就越明显。第一层模型把用户的话改写成检索词第二层模型基于检索结果生成回答第三层再做信息抽取。只要第一层的检索词出现轻微偏差第二层拿到的文档集合就不一样第三层能抽到的信息自然不同。这就是工程上常说的级联误差放大。2.3 自由文本输出本身就是不确定性的主要来源如果你让模型直接用自然语言回答“这个工单属于什么类型”它很大概率会输出“我觉得这是一个支付问题”“这个看起来是账号问题”这类带有主观表达的内容。这类输出哪怕语义正确也不方便被程序解析。实际工程中更需要的是让模型输出严格 JSON字段名、枚举值全部由调用方预先定义。自由文本越少校验成本越低管道也就越稳定。2.4 别忘了评估“环境变更”这种隐式不确定性除了模型自身行为线上管道所处的环境也在随时间变化模型提供方发布了新版本Embedding 模型被替换向量数据库里的数据被更新甚至某个参数路由开关发生了变化。无论代码逻辑是否改动只要这些外部条件变了管道输出就可能变化。因此“确定性”不是一次性工作而是一种需要持续关注的工程质量属性。稳定输出不是简单记为“代码写好了”而是要把模型版本、提示词版本、特征规则版本都纳入到可追踪的范围里。3. 开发确定性 LLM 管道要记住三个原则3.1 原则一先固定采样参数但不迷信采样参数在调用模型时应尽量固定可以固定的参数。比如信息抽取和文本分类场景通常会把temperature设置为 0 或较低值把top_p设置为 1。部分模型接口允许设置seed如果业务场景要求高可复现性可以在请求中带上固定的seed值。下面是一段以 OpenAI Chat Completions 风格为例的调用代码很多兼容该协议的国内模型厂商也支持类似字段。具体参数名需要按你实际使用的 SDK 调整。from openai import OpenAI client OpenAI( api_keyyour-api-key, base_urlhttps://your-model-endpoint, ) response client.chat.completions.create( modelyour-model, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_text}, ], temperature0, seed42, response_format{type: json_object}, )注意seed并不能提供绝对保证。它只作用在支持该能力的推理服务上而且当服务端对模型进行更新或推理引擎调整后相同 seed 仍然可能产生不同结果。因此参数控制是基础条件不是充分条件。3.2 原则二用结构化约束代替自由发挥如果业务允许一定要让模型输出结构化数据而不是自然语言段落。最常见的做法包括使用模型的 JSON Mode 或response_format参数要求输出合法 JSON。使用 Function Calling / Tool Calling让模型把结果填充到人预设的函数参数里。在系统提示词里给出明确的 JSON Schema 或示例。在后端使用 Pydantic、jsonschema 等工具做严格校验。这套组合的实际意义是提前把“输出契约”定义好。模型只能在契约允许的范围内尝试表达即使措辞有变化解析器仍然能稳定地把结果读出来。编写 Schema 时建议给每个字段都指定enum或明确的类型避免模型自由发挥。3.3 原则三让代码做决策让模型做理解这是所有确定性改造里最关键的一条。我见过很多管道把模型当作“万能开关”让模型直接给出是否通过、属于哪个部门、紧急程度是高是低。虽然看上去很方便但本质上是用自然语言完成了一次难以审计、难以稳定的判断。更稳妥的思路是把任务拆分模型只负责“抽取”“改写”“生成”这些需要语义理解的动作代码负责“匹配”“映射”“判断”这些可以枚举、可测试的动作。举个例子与其让模型判断“这个工单是否紧急”不如让模型先抽取“用户提到的系统、状态码、关键词、操作动作”再由一段规则代码判断是否命中紧急条件。规则可以被单测覆盖也能被业务人员直接审查远比把判断逻辑埋在模型里可控。4. 一套可复用的确定性 Pipeline 设计套路4.1 管道整体分层一套适合生产环境的 LLM 管道建议至少分成以下 7 层输入规范化 - Prompt 版本管理 - LLM 调用 - 结构化解析 - Schema 校验 - 代码规则决策 - 日志与缓存每一层都有明确职责输入规范化清理空白字符、限制长度、统一语言标识减少无意义扰动。Prompt 版本管理同一个业务场景只使用带版本号的提示词禁止线上随手改。LLM 调用固定模型名、温度、seed并通过超时和重试策略兜底。结构化解析将模型输出从文本解析成可处理的数据结构。Schema 校验用 JSON Schema 校验字段类型、枚举值、必填项。代码规则决策将业务规则用代码实现模型只提供特征输入。日志与缓存记录模型版本、提示词版本、输入输出摘要并利用缓存避免重复请求。4.2 为什么要把 Prompt 也纳入版本管理很多团队把提示词当成“会变化的配置”今天加一句话明天删一个例子。这在普通聊天场景还好但在管道场景里会让问题排查变得非常困难。因为当你发现线上输出不稳定时往往很难判断是哪一次提示词改动导致的。推荐做法是每个提示词模板都有唯一 ID 或版本号并在日志里记录“当前请求使用哪个版本的提示词”。这样即使输出发生变化也能快速定位到改动来源。4.3 缓存不是可选项而是稳定性基础设施对于同一用户输入、同一提示词版本、同一模型参数的请求完全可以复用上一次结果。缓存不仅能降低成本和延迟还能让重复验证得到一致的历史结果。设计缓存键时需要把模型名、seed、temperature、提示词版本、用户输入都计算进去避免因为参数不同而错误命中。需要注意的是如果下游依赖外部数据库或实时检索结果则不能只缓存模型输出还要把外部上下文纳入缓存键。5. 实战案例构建一个“确定性工单分析”管道5.1 场景与需求假设业务方需要处理大量用户工单希望程序自动判断工单的紧急程度并转给正确的负责部门。直接让大模型输出“紧急/不紧急”并不可控因为不同人甚至不同业务时期对“紧急”的定义都可能不同。因此这里采用“模型抽取特征 代码规则决策”的方案。管道处理目标如下输入一段用户工单描述文本。输出JSON 对象包含urgency、owner_team、confidence等字段。稳定性要求在相同输入、相同规则版本下输出结果保持一致。5.2 项目结构先规划一个最小项目结构。workshop/ ├── config.py ├── llm_client.py ├── pipeline.py ├── decision.py └── requirements.txt文件说明config.py放置模型名、温度、seed、Prompt 版本常量。llm_client.py封装模型调用、解析、缓存。decision.py用代码实现紧急程度与责任部门判断。pipeline.py把上面模块串成完整管道。5.3 编写系统提示词要让模型只做特征抽取别做分类决策。系统提示词这里只允许模型输出限定字段的 JSONSYSTEM_PROMPT 你是一个工单特征抽取引擎。你的任务是从用户描述中抽取结构化特征。 规则 1. 只输出一个 JSON 对象不要输出任何解释。 2. JSON 对象必须包含以下字段 - keywords: string[], 用户描述中的关键短语 - systems: string[], 可能涉及的系统或模块名称 - error_codes: string[], 出现的错误码、状态码或错误信息 - stale: boolean, 用户是否提到“从昨天开始”“持续了X天”等时间线索 3. 如果某个字段没有对应内容返回空数组或 false。 4. 不要给出维修建议不要判断紧急程度。 示例输出 {keywords: [无法登录, 503], systems: [登录系统], error_codes: [503], stale: false} 模型的任务被限制得很窄只做抽取不做判断。这样即使模型在不同调用里生成的关键词措辞稍有不同只要关键词能落到规则匹配层最终决策就可能是稳定的。5.4 实现 LLM 客户端与调用函数接下来封装一个call_extract_features函数负责调用模型、解析 JSON 并返回结果。为了代码简洁这里用字典做缓存示意生产环境可以替换成 Redis。# llm_client.py import hashlib import json from openai import OpenAI client OpenAI( api_keyyour-api-key, base_urlhttps://your-model-endpoint, ) MODEL_NAME your-model TEMPERATURE 0 SEED 42 SYSTEM_PROMPT_VERSION v1 SYSTEM_PROMPT 这里填写 5.3 中的完整系统提示词 _cache {} def _make_cache_key(user_text: str) - str: raw json.dumps( { model: MODEL_NAME, prompt_version: SYSTEM_PROMPT_VERSION, temperature: TEMPERATURE, seed: SEED, user_text: user_text, }, sort_keysTrue, ensure_asciiFalse, ) return hashlib.sha256(raw.encode(utf-8)).hexdigest() def extract_features(user_text: str) - dict: cache_key _make_cache_key(user_text) if cache_key in _cache: return _cache[cache_key] response client.chat.completions.create( modelMODEL_NAME, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_text}, ], temperatureTEMPERATURE, seedSEED, response_format{type: json_object}, ) content response.choices[0].message.content data json.loads(content) _cache[cache_key] data return data这段代码通过缓存减少重复请求也保证同一个用户输入在缓存有效期范围内能拿到完全一致的结果。需要特别注意的是如果模型返回的 JSON 缺少必填字段后面的json.loads虽然能成功但字段缺失问题会在决策层暴露出来所以要配合后续校验。5.5 用代码实现规则决策decision.py的作用是根据抽取出的特征给出稳定且可测试的结论。规则尽量写简单命中关键词则进入对应分支没有命中则使用“待人工确认”。# decision.py HIGH_RULES { urgent: [宕机, 无法登录, 崩溃, 全部不可用], teams: [登录系统, 交易系统, 订单系统], } MEDIUM_RULES { urgent: [报错, 超时, 503, 500], teams: [订单系统, 支付系统], } def decide_by_rules(features: dict) - dict: keywords [k.lower() for k in features.get(keywords, [])] systems [s.lower() for s in features.get(systems, [])] error_codes [e.lower() for e in features.get(error_codes, [])] stale features.get(stale, False) urgency low if any(k in HIGH_RULES[urgent] for k in keywords) or any( s in HIGH_RULES[teams] for s in systems ): urgency high elif any(k in MEDIUM_RULES[urgent] for k in keywords error_codes): urgency medium owner_team unknown for team in [登录系统, 订单系统, 支付系统, 交易系统]: if team.lower() in .join(systems keywords): owner_team team break if stale: owner_team customer-service if urgency low: urgency medium return { urgency: urgency, owner_team: owner_team, needs_human_review: owner_team unknown, matched_features: { keywords: features.get(keywords, []), systems: features.get(systems, []), error_codes: features.get(error_codes, []), }, }注意这里把“很久以前就出现”的时间线索单独映射到了客服团队并提升了紧急程度。这样的规则对业务来说清晰可见业务方可以直接告诉你“我们不希望 503 被分给登录系统”而不需要去修改晦涩的提示词。这就是把决策逻辑放在代码层带来的可维护性优势。5.6 串成完整管道pipeline.py负责把抽取和决策拼接起来并对最终结果补上版本信息方便后续追踪。# pipeline.py import json from decision import decide_by_rules from llm_client import SYSTEM_PROMPT_VERSION, extract_features RULE_VERSION 2025.01 def run_pipeline(user_text: str) - dict: features extract_features(user_text) decision decide_by_rules(features) return { input: user_text, features: features, decision: decision, meta: { prompt_version: SYSTEM_PROMPT_VERSION, rule_version: RULE_VERSION, }, } if __name__ __main__: demo_input 从昨天开始一直提示503用户无法正常下单系统看起来已经不可用了 result run_pipeline(demo_input) print(json.dumps(result, ensure_asciiFalse, indent2))运行这个脚本后理想输出会像下面这样{ input: 从昨天开始一直提示503用户无法正常下单系统看起来已经不可用了, features: { keywords: [提示503, 无法正常下单, 系统不可用], systems: [下单系统], error_codes: [503], stale: true }, decision: { urgency: high, owner_team: 登录系统, needs_human_review: true, matched_features: { keywords: [提示503, 无法正常下单, 系统不可用], systems: [下单系统], error_codes: [503] } }, meta: { prompt_version: v1, rule_version: 2025.01 } }示例中系统可能报错因为人工设计的抽出字段和规则阈值比较粗糙这正说明规则层需要不断补充关键词和维护。实际生产环境中建议把这类规则放在本地规则中心或配置中心方便灰度修改和回滚。5.7 编写回归测试把行为固化下来想要让管道长期稳定最关键的动作是使用自动化测试把关键输出“锁住”。无论提示词怎么修改只要回归测试不通过就不能发布。测试中为了不依赖外部模型服务可以对extract_features做 mock只测试规则决策部分。# test_pipeline.py from decision import decide_by_rules def test_urgent_when_system_down(): features { keywords: [无法登录], systems: [登录系统], error_codes: [], stale: False, } result decide_by_rules(features) assert result[urgency] high assert result[owner_team] 登录系统 def test_low_when_only_question(): features { keywords: [怎么修改密码], systems: [], error_codes: [], stale: False, } result decide_by_rules(features) assert result[urgency] low assert result[owner_team] unknown assert result[needs_human_review] is True这类测试并不复杂但能显著提高重构提示词时的安全感。只要规则层不变即使 LLM 抽出的关键词有细微差异原则上也不会影响最终决策。这也是“确定性管道”在真实项目中最重要的收益。6. 常见问题与排查思路无论采用哪种设计实际开发中总会遇到新的问题。下面按经验整理了高频问题与排查方向。问题现象常见原因解决思路把 temperature 设置为 0输出仍然不同采样参数不能消除服务端全部不确定性检查是否支持 seed并用结构化约束 后置校验对强一致要求做缓存模型偶尔输出非法 JSON提示词约束不够强或输出被截断使用 JSON Mode/response_format调大 max_tokens增加解析失败重试重试后输出比第一次差重试逻辑重复调用但未固定模型参数每次重试使用相同参数最好再加一层“重试日志”记录上下文修改提示词后一批测试用例失败prompt 变更影响范围超出预期建立回归测试集所有 prompt 变更走 git 评审流程模型输出合法但业务结果反复变化决策逻辑仍被模型承担而不是代码承担重构为“模型抽取特征、代码执行决策”相同代码在不同时间结果不同上游模型版本更新或外部数据变化在日志中记录模型版本、embedding 版本、规则版本并加入版本告警排查这类问题有一个通用顺序先看日志中记录的 model、prompt version、temperature、seed 是否一致再检查模型返回内容是否有明显差异接着分析解析后的结构化特征是否稳定最后定位到是模型层、规则层还是外部依赖层发生了变化。不要一上来就调整提示词否则问题可能越调越乱。7. 工程最佳实践与可维护性建议7.1 对管道进行分层稳定性管理把系统拆分为“模型边界”和“代码边界”。模型边界内允许语义波动代码边界内必须严格稳定。比如前面案例中的extract_features可以接受关键词的小差异但decide_by_rules输出的urgency和owner_team必须能被测试锁定。这样思考的好处是团队能明确知道哪里需要投入精力做校验哪里可以容忍误差。7.2 不追求绝对零随机但要确保“失败模式可见”完全消除随机性在很多平台上是不现实的。更实际的做法是当模型输出进入无法解析或无法映射到规则的状态时系统不要悄无声息地返回默认值而是显式标记为“needs_human_review”并写入待处理队列。很多时候管道出错并不可怕可怕的是错误数据混进正常业务流程最后连排查线索都找不到。7.3 将 prompt、规则、模型版本都纳入配置管理建议建立一份版本变更记录类似下面这样rule_version: 2025.01 prompt_version: v1 model_name: your-model updated_by: backend-user change_note: 新增 503 关键命中规则这样当业务方提出“为什么上周分类结果和这周不一样”时可以直接从版本对比中寻找原因而不是去猜哪次部署改变了行为。7.4 新模型上线前先跑回归集再切流工程中经常会遇到“新模型效果看起来更好”的情况。但即使基准指标提升也不能直接替换旧模型因为新模型的输出风格、失败模式、对特定 prompt 的响应方式可能完全不同。正确做法是先准备一个覆盖典型场景和边界条件的回归集让新旧模型在相同输入下跑出结果对比结构化字段的分布、非法输出率以及规则命中率然后通过灰度切流让一部分请求走新模型观察一段时间再全量放开。7.5 把“请求级日志”当作管道的一部分每次请求至少记录以下信息用户输入摘要、模型名称、prompt 版本、temperature、seed、原始返回内容、解析后数据结构、规则版本、最终决策、耗时、是否命中缓存。长期积累后这套日志不仅能用于问题追溯还能用来分析模型输出漂移的规律帮助你判断到底是模型更新了还是用户输入分布变了。8. 结语从“模型魔法”走向“可测试的工程系统”开发确定性 LLM pipeline核心不是把模型变成一台机器而是用工程手段压缩模型的不确定空间。记住这条主线模型只负责需要语言理解的任务凡是能被规则、枚举、代码测试覆盖的决策都不要让模型去做。当你感觉管道输出不再“神出鬼没”时通常不是因为你找到了一个更听话的模型而是因为你已经在模型外围搭好了契约、校验、缓存和回归测试这些确定性基础设施。如果你正在开发一个分类或抽取类管道可以从今天开始做一个最小改造先把输出改成固定 JSON Schema再把所有业务判断移到代码层最后补一个只针对规则层的小测试集。这个改造不需要动模型也不需要等算法团队支持只要后端开发愿意花半天时间就能明显降低线上行为漂移的概率。