今天看一个经常出现在 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-K，K 相同但召回内容因为 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( model="your-model", messages=[ {"role": "system", "content": "你只做文本分类，禁止额外解释。"}, {"role": "user", "content": text}, ], temperature=0, response_format=Judgment, )不同 SDK 的写法有差异，务必以你的 LLM 官方文档为准。关键不是函数叫什么名字，而是做到两点：模型被强制输出 schema，而不是“尽量输出 schema”。
5.2 function calling 比口头约束更可靠
当模型支持 function calling 或 tool calling 时，可以把任务定义成一个“函数”。强制tool_choice调用该函数后，服务端通常会在更严格的结构约束下生成参数。
response = client.chat.completions.create( model="your-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"}}, temperature=0, )这个方案比“请输出 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, flags=re.MULTILINE).strip() try: data = json.loads(text) return Judgment(**data).model_dump() except (json.JSONDecodeError, ValidationError) as exc: raise ValueError(f"unexpected model output: {raw}") from exc这里的要点是：模型输出不可信，解析之后必须再做一次 schema 校验。校验失败不能静默吞掉，也不能直接重试三次造成重复动作，要按业务失败处理。
6. 采样参数、seed 与模型版本固定
输出结构稳定之后，还需要让内容本身尽量可复现。一个低成本改动是收敛采样参数：temperature调低到 0，top_p固定为 1，部分服务商支持seed参数，可以一并设置。
params = { "temperature": 0, "top_p": 1.0, # 如果服务商支持 seed，设为固定值 "seed": 42, }需要说明的是，temperature=0不等于绝对确定。很多模型服务端仍可能使用非确定性采样、批处理填充方式、模型分片，不同推理实例之间的结果也会有轻微差异。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_ascii=False, sort_keys=True, 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_k=1) 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( model=MODEL_VERSION, messages=build_messages(normalized_text), schema_version=payload.schema_version, params=LLM_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, ttl=86400) return result接口层做的事是：去重、缓存、路由、校验、保存结果。只要每个环节都留日志，即使出了错，也可以回放同一个 request 看问题出在哪一步。
9.2 批量任务状态机
批量处理不能简单用 for 循环同步请求模型。真实场景中会有超时、限流、单条 prompt 格式错误、某条记录数据异常。建议为每一条任务维护状态，至少包括 pending、running、succeeded、failed。
| 状态 | 含义 | 动作 |
|---|---|---|
| pending | 等待处理 | 入队 |
| running | 正在调用模型 | 加分布式锁，防止重复处理 |
| succeeded | 已通过校验并保存 | 可被查询 |
| failed | 校验不通过或达到最大重试 | 进入人工复核或死信队列 |
| skipped | 规则判定无需处理 | 记录原因 |
批量任务运行时，每条记录要有独立job_id。模型调用成功不代表任务成功，必须等结构化校验通过后才更新为 succeeded。如果校验失败，要先区分是“瞬时错误”还是“永久错误”。
def process_item(item, max_retry=3): 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, reason="schema_validation_error") return except TimeoutError as exc: if attempt >= max_retry - 1: mark_failed(item.job_id, reason="timeout") 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_format |
| schema 校验失败率偏高 | 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 的确定性不是靠某个模型或某个参数瞬间解决的，而是靠你对系统每一层都建立了约束、缓存、校验和回退机制之后，才慢慢逼近的目标。
