今天看一个经常出现在 HN 技术区的问题How do you develop more deterministic LLM pipelines? 中文直译是怎么把 LLM Pipeline 做得更可控、更不随机。团队一旦把 LLM 从聊天玩具接到线上业务几乎都会撞上同一个痛点同一份输入上一次结果还能用这一次就变成完全不同的结构重试三次得到三种答案下游流程根本不敢自动接。于是大家开始寻找确定性工程方案而不是继续堆 prompt 技巧。先把结论放在最前面确定性不是把 temperature 调到 0 就算赢而是一套「约束、路由、缓存、校验、回归」的工程闭环。所谓 deterministic LLM pipeline不要求模型在创意表达上一模一样而是做到输入相同时输出格式稳定、业务结论稳定、对外动作稳定并且整个过程可被审计和重放。要做到这些不能只靠模型参数还要改造你调用模型的那一层业务系统。本文不会介绍某个一键包也不绑定具体厂商而是以一个典型的 LLM 后端服务为例演示从 prompt、API 参数、结构化输出、缓存、批量任务到评估的完整改造思路。重点包括Pydantic/JSON Schema 约束、强制工具调用、seed 与采样参数、请求缓存与幂等键、批量任务状态机以及 golden set 回归。适合正在做 RAG、Agent、批量文本处理、文档结构化、客服工单路由的开发者和算法工程师阅读。1. LLM Pipeline 的确定性到底指什么在决定怎么改之前先定义清楚“确定性”。如果你追求的是“同一段 prompt 一字不差地复现输出”那在绝大多数商业模型上都不现实哪怕 temperature 设为 0 也可能出现少量波动因为模型服务端可能做了量化、批处理、内核版本更新甚至负载均衡到了不同推理实例。更务实的定义是确定性等于三个可验证目标。第一输入输出可重放。使用相同业务输入、相同模型版本、相同参数时最终返回的核心字段应该保持一致至少不能被下游感知出显著差异。第二格式可解析。无论模型说什么都应该落在预定义 JSON Schema、函数参数或枚举范围内程序不需要靠正则去猜。第三副作用可控。如果 LLM 驱动 Agent 调用工具、写数据库或发工单必须让这些动作发生在校验之后并且失败时不能重复执行。这三层目标对应不同手段第一层用采样参数、seed、固定模型版本来处理第二层用 function calling、JSON Schema 和响应校验器来处理第三层则必须靠规则路由、幂等键和人工审批兜底。把它们拆开之后排查速度会快很多。1.1 不确定性从哪来LLM Pipeline 的不确定性并不只来自模型本身的采样。常见来源至少有以下几类第一类是采样参数。temperature、top_p、top_k 会影响 token 概率分布不同取值下同一 prompt 可能得到不同表达。第二类是模型版本和推理环境。同样的模型名服务商背后可能悄悄更新了权重、量化方式或 prompt 模板返回内容就会发生批量变化。第三类是 prompt 构造不稳定。用户输入里的多余空格、换行、字段顺序或者从外部拿到的上下文偶尔带入了时间、ID、排序变化都会造成结果漂移。第四类容易被忽略的是非 LLM 依赖。向量检索召回内容变了、知识库文档更新了、上游 API 返回了重复字段都会传导到最终答案。第五类是并发和重试。同一个请求因为超时被重发模型实际执行了两次如果任务不是幂等的就会产生重复副作用。所以在写代码前先给系统画一条数据流输入从哪来、prompt 怎么拼、模型怎么调、结果怎么解析、后续动作怎么触发。哪些环节有随机因素标注出来这就是你的确定性改造清单。2. 确定性 LLM Pipeline 核心方法速览为了不过度设计可以先把方法分成五层。它们不是五选一而是按照优先级逐层叠加。治理层关键动作达到的效果Prompt 层固定系统提示词、固定字段顺序、给少样本示例降低因为 prompt 微小变化导致的结果漂移采样层temperature 调低、固定 seed、固定模型版本让模型生成路径尽量可复现输出层使用 JSON Schema、function calling、参数枚举保证输出结构稳定程序可直接解析调度层规则优先、缓存重复请求、幂等键去重减少无谓的模型调用避免随机性被放大评估层golden set、字段断言、回归测试持续发现模型升级和 prompt 调整带来的破坏从实现顺序看第一件要做的是把模型调用参数收敛下来第二件是强制结构化输出第三件是补缓存与规则路由第四件才是搭自动化回归。如果顺序颠倒你会在不稳定系统上面做复杂的评估越跑越迷茫。3. 适用场景与使用边界这种“确定性工程”特别适合以下几类任务。第一类是信息抽取和文档结构化。比如解析发票、合同、工单你需要的是稳定字段不是发挥文采输出必须落到 schema 里。第二类是分类与路由任务。情感判断、意图识别、安全审核都会进入下游动作错误分类造成的成本很高。第三类是 RAG 和 Agent 的工具调用。模型需要先决定调用哪个工具、传什么参数如果生成参数不稳定整个 Agent 流程会非常脆弱。第四类是批量文本处理。几千条数据后台跑如果失败率不可控很难靠人工补救。不是所有任务都适合套这套确定性方案。纯创意文案、标题生成、社交内容改写原本就需要多样性和“惊喜感”硬性要求稳定反而会牺牲质量。另外如果你的业务要求毫秒级响应最好不要让每一条请求都走完整 LLM而应尽量用规则或小模型过滤公共高频场景。使用边界同样要明确。任何确定性手段都不能消除幻觉temperature 为 0 也不代表内容绝对真实。涉及个人隐私、版权素材、人脸或声音数据时必须先确认数据来源合法并获得授权不要让模型输出直接变更业务状态更不要在未经测试的情况下让 Agent 自动执行敏感操作。外部 API 调用建议关闭日志中的原始文本或做脱敏处理后再上报。4. 先定位问题不稳定到底出在哪一层如果你的 LLM pipeline 已经不稳定不要急着改 prompt先做一次隔离实验。把同一个请求连续调用 10 次观察变化出现在哪个字段。如果只是措辞不同采样方差占主导如果字段突然多了少了结构约束不够如果一半好一半差往往是用在 prompt 里的外部上下文不稳定。还有一点容易被忽略非 LLM 依赖也会造成非确定性。比如你从向量数据库取 Top-KK 相同但召回内容因为 embedding 模型更新而改变上游文档多了一个空行甚至系统时间、用户 ID 被拼进 prompt都会传导到输出。诊断顺序是先固定输入再固定模型参数再固定输出格式最后检查周边依赖。现象大概率原因优先检查内容同一输入每次语义相同但用词不同采样随机性temperature、top_p、seed同一输入反复缺字段或多字段输出约束不足JSON Schema、function calling 是否强制白天正常、晚上批量失败率升高模型版本或服务端限流模型 fingerprint、限流日志修改 prompt 后一批结果变差prompt 回归golden set、历史采样并发重试后数据重复缺少幂等控制request_id、任务状态机建议把每次模型调用都记录一份审计日志包含模型名、参数、输入签名、输出摘要、耗时和异常类型。没有日志前面所有排查都只能靠猜。日志不一定要存完整 prompt但至少要存可复现请求的哈希以及服务端返回的 fingerprint 或版本标识。5. 输出结构化最直接有效的确定性手段所谓“让模型输出 JSON”其实是最不可靠的方案。如果你只写一句“请输出 JSON”模型可能给你 Markdown 代码块、注释、多余逗号甚至和 JSON 无关的自然语言。正确的做法是定义明确 schema并在 API 层强制模型走结构化输出。5.1 先定义明确 Schema假设你要做一个文本分类 Pipeline核心输出需要三个字段类别、置信度、理由。用 Pydantic 定义如下from pydantic import BaseModel class Judgment(BaseModel): label: str confidence: float reason: str如果你所在的服务商支持response_format或json_schema可以直接把模型绑定到该结构。以 OpenAI 风格 API 为例伪代码如下from openai import OpenAI client OpenAI() response client.beta.chat.completions.parse( modelyour-model, messages[ {role: system, content: 你只做文本分类禁止额外解释。}, {role: user, content: text}, ], temperature0, response_formatJudgment, )不同 SDK 的写法有差异务必以你的 LLM 官方文档为准。关键不是函数叫什么名字而是做到两点模型被强制输出 schema而不是“尽量输出 schema”。5.2 function calling 比口头约束更可靠当模型支持 function calling 或 tool calling 时可以把任务定义成一个“函数”。强制tool_choice调用该函数后服务端通常会在更严格的结构约束下生成参数。response client.chat.completions.create( modelyour-model, messages[ {role: system, content: 只做文本分类任务。}, {role: user, content: text}, ], tools[ { type: function, function: { name: submit_judgment, description: 提交分类结果, parameters: { type: object, properties: { label: {type: string, enum: [positive, negative, neutral]}, confidence: {type: number, minimum: 0, maximum: 1}, reason: {type: string} }, required: [label, confidence] } } } ], tool_choice{type: function, function: {name: submit_judgment}}, temperature0, )这个方案比“请输出 JSON”稳得多因为它把字段枚举、必填项、类型都直接塞给了推理过程。对你来说解析结果只需找到模型返回的tool_calls参数而不是在纯文本里做正则。5.3 下游解析也要防御式即便用了结构化输出实际返回仍可能因为服务商降级、超时或内部错误而出现异常。正确的下游处理不是拿到字符串就json.loads而是先清洗再校验最后兜底。import json import re from pydantic import ValidationError def parse_judgment(raw: str) - dict: text raw.strip() # 防御式清理如果模型仍包了 Markdown 代码块 text re.sub(r^(?:json)?|$, , text, flagsre.MULTILINE).strip() try: data json.loads(text) return Judgment(**data).model_dump() except (json.JSONDecodeError, ValidationError) as exc: raise ValueError(funexpected model output: {raw}) from exc这里的要点是模型输出不可信解析之后必须再做一次 schema 校验。校验失败不能静默吞掉也不能直接重试三次造成重复动作要按业务失败处理。6. 采样参数、seed 与模型版本固定输出结构稳定之后还需要让内容本身尽量可复现。一个低成本改动是收敛采样参数temperature调低到 0top_p固定为 1部分服务商支持seed参数可以一并设置。params { temperature: 0, top_p: 1.0, # 如果服务商支持 seed设为固定值 seed: 42, }需要说明的是temperature0不等于绝对确定。很多模型服务端仍可能使用非确定性采样、批处理填充方式、模型分片不同推理实例之间的结果也会有轻微差异。seed 的作用通常是“尽量复现同一次生成”但不能保证跨模型版本或跨服务商得到相同结果。因此比调参数更重要的是固定模型版本。生产环境里尽量不要使用“latest”这类自动升级别名否则一次无声升级可能让整批结果漂移。你应该在配置里显式指定模型版本并保存每次上线时的模型指纹。回归集跑完确认没有破坏性变化再允许接入新版本。模型参数也不是越低越好。如果你做的是创意生成或需要多样性强行把 temperature 降到 0 会让内容变得单调。这类场景不应该放进确定性 Pipeline而应单独设置一套高随机参数。7. 缓存与幂等相同请求不应该重复掷骰子在业务里大量模型请求其实是重复或近似重复的。同一个用户刷新页面、同一批存量数据重新处理、多个服务调用同一个公共分类器都会请求相同内容。若每次都调用模型不仅浪费 token还会让不稳定输出的影响面变大。7.1 文本缓存最直接的确定性手段是缓存相同输入、相同参数、相同 schema 下直接返回上一次结果。缓存 key 建议包含模型名、schema 版本、消息序列化后的哈希、采样参数。import hashlib import json def build_cache_key(model: str, messages: list, schema_version: str, params: dict) - str: canonical json.dumps( { model: model, messages: messages, schema_version: schema_version, temperature: params.get(temperature), top_p: params.get(top_p), seed: params.get(seed), }, ensure_asciiFalse, sort_keysTrue, separators(,, :), ) return hashlib.sha256(canonical.encode(utf-8)).hexdigest()注意不能把原始 prompt 直接字符串拼接后哈希因为用户输入里的空格、换行、emoji 都可能不一致。最好先做 canonicalization例如把 messages 中每个字段做 strip、去掉多余的空白符再 JSON 序列化。这样能让同义但微小差异的输入尽量命中同一个缓存。7.2 语义缓存文本缓存的局限是只处理完全相同的输入。如果两条请求语义相同但措辞不同文本缓存不命中。业界常用 embedding 相似度做 top-1 召回再设置高阈值判断是否命中。def semantic_cache_get(text: str, vector_store, threshold: float 0.97): query_vec embed(text) hit vector_store.search(query_vec, top_k1) if hit and hit.score threshold: return hit.payload.get(result) return None语义缓存能减少重复请求但阈值不能拍脑袋设太低否则会把并不等价的问题错误合并。建议只对“标准化程度高”的任务开启比如工单分类、FAQ 查询。涉及法律、医疗等高风险场景时不要因为语义相似就复用结果。幂等设计也要和缓存配合。如果外部请求自带request_id你应该在服务里记录这个 ID 的处理状态。重复请求到达时直接返回上一次处理结果而不是再调一次模型。8. 规则优先确定性路由与降级LLM 不应该处理所有请求。一个成熟的确定性 Pipeline会在入口处先做规则判断能确定的直接返回能匹配模板的直接走模板只有真正需要推理的内容才交给模型。这样既提升确定性也降低成本和延迟。import re POSTCODE_PATTERN re.compile(r^\d{6}$) def route_request(text: str): text text.strip() # 规则能覆盖的场景不调用模型 if POSTCODE_PATTERN.fullmatch(text): return {route: rule_region, region: lookup_region_by_postcode(text)} if text in FREQUENT_FAQ_MAP: return {route: faq, answer: FREQUENT_FAQ_MAP[text]} # 其余才走 LLM return {route: llm, result: call_llm_classifier(text)}路由逻辑本身是代码因此是确定的。对业务方来说凡是能写成正则、字典、名单、数据库映射的逻辑都应该在模型之前完成。模型只负责处理真正开放、不确定的输入场景。模型调用失败时也需要降级策略。例如分类置信度低于阈值时不返回猜测结果而是进入人工队列工具参数校验失败时不让 Agent 重试死循环而是返回“需要用户澄清”。把降级行为设计成明确定义的路径Pipeline 才能从“随机输出”变成“有边界的输出”。9. 接口 API 与批量任务设计当你要把确定性 Pipeline 暴露给其他服务或支撑批量任务时接口协议、幂等键和任务状态机是关键。返回结构也需要稳定否则下游每次都要适配新格式。9.1 统一处理入口建议设计一个/v1/process接口传入 request_id、业务类型、正文、schema_version。伪代码如下# Python / FastAPI 风格伪代码需要按实际项目调整 app.post(/v1/process) def process(payload: ProcessRequest): normalized_text normalize(payload.text) # 1. 幂等重复请求直接返回历史结果 existing store.get(payload.request_id) if existing: return existing # 2. 文本缓存命中 cache_key build_cache_key( modelMODEL_VERSION, messagesbuild_messages(normalized_text), schema_versionpayload.schema_version, paramsLLM_PARAMS, ) cached cache.get(cache_key) if cached: store.set(payload.request_id, cached) return cached # 3. 规则路由 routed route_request(normalized_text) if routed[route] ! llm: result routed else: result call_and_validate(routed[llm]) store.set(payload.request_id, result) cache.set(cache_key, result, ttl86400) return result接口层做的事是去重、缓存、路由、校验、保存结果。只要每个环节都留日志即使出了错也可以回放同一个 request 看问题出在哪一步。9.2 批量任务状态机批量处理不能简单用 for 循环同步请求模型。真实场景中会有超时、限流、单条 prompt 格式错误、某条记录数据异常。建议为每一条任务维护状态至少包括 pending、running、succeeded、failed。状态含义动作pending等待处理入队running正在调用模型加分布式锁防止重复处理succeeded已通过校验并保存可被查询failed校验不通过或达到最大重试进入人工复核或死信队列skipped规则判定无需处理记录原因批量任务运行时每条记录要有独立job_id。模型调用成功不代表任务成功必须等结构化校验通过后才更新为 succeeded。如果校验失败要先区分是“瞬时错误”还是“永久错误”。def process_item(item, max_retry3): for attempt in range(max_retry): try: raw call_model(item) parsed validate_and_parse(raw) save_result(item.job_id, parsed) mark_succeeded(item.job_id) return except ValidationError: # schema 校验失败重试大概率也一样 mark_failed(item.job_id, reasonschema_validation_error) return except TimeoutError as exc: if attempt max_retry - 1: mark_failed(item.job_id, reasontimeout) return sleep_with_backoff(attempt) def sleep_with_backoff(attempt: int): import time time.sleep(min(2 ** attempt, 30))这里的关键是不要对永久错误重试。JSON Schema 校验失败往往意味着 prompt、schema 或模型行为有问题重试只会烧钱。只有超时、限流、5xx 这类瞬时错误才适合退避重试。10. 性能、成本与可观测性做确定性改造时性能和成本不能只看单次调用延迟。一次请求如果因为格式错误重试了三次花费就变成三倍。所以更要关注缓存命中率、重试率、无效输出率和最终成功率。几个值得长期跟踪的指标总请求数、模型调用数、缓存命中数p95、p99 延迟无效 JSON / schema 校验失败占比重试次数与失败原因分布不同模型版本的输出变化单条业务请求的平均 token 成本置信度低于阈值的数量以及进入人工队列的数量。这些指标可以在日志系统里用结构化字段打点也可以直接写 JSONL 文件然后用脚本统计。示例metrics: total_requests: 10000 rule_hit: 3200 cache_hit: 4100 llm_calls: 2700 schema_failed: 31 timeout: 12 gold_set_accuracy: 0.982比指标更重要的是一套回归集。你应该准备 100 到 500 条典型输入每条的预期结果不一定是完整文本但可以是“分类正确”“包含必须字段”“工具调用正确”。每次升级模型、修改 prompt、调整 schema 前都在回归集上跑一遍对比失败样例数。要注意LLM Pipeline 的回归断言不能只做字符串完全相等。更合适的做法是字段级断言label是否在允许范围内、confidence是否大于阈值、reason是否非空、结构化校验是否通过。只有明确要求“逐字一致”的场景才做字符串比较。11. 确定性改造常见问题与排查方法问题现象可能原因排查方式解决方案同样输入结果仍不一致服务端 seed 不稳定或模型自动升级查看模型 fingerprint 与响应日志固定模型版本观察 fingerprint 变化输出偶尔带 Markdown 代码块结构化输出约束不够强检查 messages 里是否出现“输出JSON”和代码块改用 function calling 或 response_formatschema 校验失败率偏高schema 过于复杂或 prompt 信息不足看失败样本集中在哪个字段拆分子任务放宽必填项或增加 few-shot重试后出现重复工单/重复写入缺少幂等控制查 request_id 日志加幂等表任务只允许成功一次缓存命中率低key 包含时间、ID 等无关变量检查 cache key 序列化内容对输入做 canonicalization去掉无关字段批量任务到第几百条就卡住单条失败触发死循环或无限重试看 job 状态分布加最大重试次数、熔断和死信队列更新 prompt 后效果集体变差没有回归集用旧 prompt 重放历史请求建立 golden set发布前跑回归如果在结构化输出很稳的情况下模型偶尔仍返回不可解析内容不要强行反复重试。比较现实的处理是把它判定为“低置信度失败”进入人工处理队列。宁可让人工处理一小部分也不要让错误输出自动写入业务库。12. 落地建议与最佳实践确定性工程不是一次性能做完的改造而是一个持续收敛的过程。以下几条建议可以在项目中直接落地。第一先用最低成本组合temperature 设为 0强制 JSON Schema 或 function calling给接口加 request_id 和缓存。这四步能解决大部分“输出不可用”的问题。第二规则能覆盖的需求就不要交给模型。常见问题、格式校验、枚举映射都应前置到代码里模型只做开放语义理解。第三任何 Agent 工具调用都不能直接执行成功。模型返回工具参数后必须做参数校验和权限检查必要时由人确认后再触发后端动作。第四模型服务别名要谨慎。尽量固定版本升级前先跑满回归集比较历史输出差异。第五缓存与幂等键是确定性 Pipeline 的地基。没有这两者超时重试就会变成重复执行。第六日志要全链路保存。最好每个请求都记录 request_id、输入哈希、模型名、参数、耗时、输出摘要、schema 版本。第七避免把所有随机性寄托在一次调用上。复杂任务可以拆成多个原子步骤每个步骤单独校验而不是让一个大 prompt 一次吐出所有内容。最后给一个很现实的建议先把你现在最容易翻车的一个环节修到可控再扩展。LLM Pipeline 的确定性不是靠某个模型或某个参数瞬间解决的而是靠你对系统每一层都建立了约束、缓存、校验和回退机制之后才慢慢逼近的目标。