很多团队的 LLM 生产事故不是模型不够强而是把一类本来应该只做一次、然后固化成规则的问题做成了每次都要走模型推理的“薛定谔接口”。最近 Hacker News 上有一个讨论很值得注意Can trivial LLM calls be compiled into conventional data pipelines意思是那些“平凡”的 LLM 调用能不能被“编译”成传统的数据管道这个问题看起来有点学院派但它切中的是 LLM 应用落地时一个非常现实的痛点当你在生产环境里接入了十几个 LLM 调用点每次调用都付出延迟、成本和不确定性但其中相当一部分其实是重复、固定、可预测的。它们真的有每次都必须“请求一次模型”的理由吗本文会先拆解什么是“平凡 LLM 调用”再给出从即时调用演进到编译式管道的完整思路最后用工单自动分类这个最小场景写一套可运行、可验证的示例代码。读完你会理解LLM 应用不是只有“全交给大模型”和“全写死规则”两个极端中间还有一条工程化的中间路线。1. 先看清问题LLM 调用为什么不能直接当成管道用传统数据管道的核心特征是确定性。你给同一个输入管道一定会给出同一个输出每一步都可重放、可审计、可回溯。无论是 ETL 任务、消息处理、日志清洗还是实时风控数据管道强调的是“稳定”和“可预期”。LLM 即时调用则完全不同。同样的输入模型可能因为 temperature、采样、上下文漂移给出不同结果单次调用延迟在几百毫秒到几秒之间成本按 Token 计费批量处理时费用会线性增长更麻烦的是模型的输出不一定符合你预期的 JSON 结构你需要额外的解析、校验、重试逻辑。这带来几个直接的工程冲突批量任务很难做你需要把一万条数据做字段提取如果每条都走模型耗时和成本都不可控。审计困难数据管道需要解释“这条记录为什么被分类为高风险”但 LLM 给不出稳定的决策链。测试难写单元测试依赖确定性而 LLM 输出的随机性让 CI/CD 很难稳定断言。下游对接脆弱管道下游可能是数据库、消息队列或工单系统它们需要的是强类型、固定 Schema 的数据而不是一段“看起来像 JSON”的文本。所以业界对 LLM 调用的常见处理是加上函数调用、JSON Mode、结构化输出、重试与校验。这些手段缓解了问题但没有解决根源——你仍然在把每一次调用当作一次独立推理而不是在建立一套可积累、可编译的确定性资产。2. “平凡 LLM 调用”的识别标准要讨论编译先得定义什么值得编译。不是所有 LLM 调用都适合只有满足一定条件的“平凡调用”才值得。我建议用三个标准来判断第一输入输出的边界足够清晰。输入是一段固定范围的文本输出是有限集合里的一个分类、若干个字段或者一个可校验的结构化对象。比如“判断这条评论是正面还是负面”“提取工单里的客户姓名和产品型号”“把一句话改写为四个备选标题”。第二正确性可以被外部校验。模型返回结果后你能够用确定性的方式判断它是否合理。分类是否在合法枚举值里日期是否符合格式金额是否大于零。如果不能校验这个调用就不算平凡。第三任务不需要多步推理和开放知识。它不需要模型翻网页、查数据库、连续推导五步也不需要依赖实时世界知识。它更像“模式匹配 语义理解 字段映射”的组合而不是“深度推理”。符合这三个标准的场景非常普遍文本分类工单分类、邮件分类、评论打标、违规内容初筛。信息提取从简历提取姓名电话、从商品描述提取规格参数、从日志提取错误码。格式转换非结构化文本转 JSON、口语转书面语、一段话拆成多字段。路由决策判断用户问题属于哪个部门、该走哪条处理流程。这些任务确实需要语言理解能力但理解模式一旦固定就不需要每次都让模型自由发挥。更准确地说普通开发者写出一个能提取“产品型号”的规则可能很难但模型可以做到而模型能做到之后我们完全可以把它的经验“蒸馏”成一套更可控的执行路径。这里的核心判断是平凡 LLM 调用的价值不在于每次推理的随机性而在于它可以不断产生稳定的输入-输出样本并最终被编译为确定性的规则管道。3. 从即时调用到“编译式”管道的演进思路要理解“编译”这个词在 LLM 应用里的含义先看一个类比。传统编程中编译器把高级语言翻译成机器码目的是让程序运行得更快、更可预测。LLM 应用里的“编译”并不是把模型变成二进制而是指把“每次运行时请求模型推理”优化成“提前构建出确定性执行路径 缓存 规则回退”。这可以分三个层次来看第一层纯即时调用。所有请求都发给 LLM代码只负责拼接 Prompt 和解析结果。优点是灵活适合快速验证缺点是成本高、延迟高、不可控。第二层缓存 模板。对相同或相似的输入做缓存命中时直接返回历史结果同时把 Prompt 模板化减少自由发挥。优点是能解决重复请求的浪费缺点是没有改变“每次新输入都要推理”的根本模式。第三层编译式管道。分析历史调用数据把高频、稳定、可校验的输入模式固化为规则代码或配置表让大部分请求走确定性路径只有规则覆盖不到的情况才回退到 LLM。这已经不是缓存而是真正把一部分任务“编译”成了传统管道。有意思的是最近 Andrej Karpathy 提出的LLM wiki 范式也在讨论类似的分工问题不要把 LLM 当成一个什么都应该记住的计算核心而是把知识外置到可检索、可验证的系统里。编译式管道和这个思路一脉相承——LLM 是生成器和解释器但生产系统里应该沉淀出更多比“大模型推理”更廉价、更确定的资产。另外很多团队已经在用 Spring AI、LangChain、Agent 编排框架解决 LLM 应用的多工具协作问题AI Agent 也确实能处理动态决策。但注意编排框架解决的是“连接”问题编译式管道解决的是“不再需要每次推理”的问题。二者不冲突反而可以分层编排框架负责流程编译管道负责高频稳定路径。4. 编译式 LLM 管道的核心组成一个真正的编译式 LLM 管道通常由五部分组成。4.1 Schema 定义与约束第一步是给任务定义一个严格的数据契约。LLM 输出的字段名、类型、枚举值、嵌套结构都要提前定义。不要依赖“模型默认输出”要使用 JSON Schema 或结构化输出能力来约束它。例如工单分类的输出 Schema{ type: object, properties: { category: { type: string, enum: [咨询, 故障, 投诉, 建议] }, department: { type: string, enum: [技术部, 售后部, 产品部, 财务部] }, priority: { type: string, enum: [高, 中, 低] }, summary: { type: string, maxLength: 100 } }, required: [category, department, priority, summary] }这个 Schema 有两个作用一方面告诉 LLM 你要什么另一方面它也是未来编译产物的“接口标准”。规则管道、缓存层、下游系统全都面向它编码。4.2 样本采集与标注编译不是拍脑袋写规则而是从真实调用数据中沉淀。你需要先让 LLM 跑一段时间把输入文本和结构化输出成对保存下来。这些样本既是验证集也是规则提取的原料。每一条样本都应该包含原始输入文本LLM 结构化输出校验结果是否通过 Schema 检查业务侧最终使用的标签可能被人改过这一阶段不建议跳过。没有足够多样本后面编写的规则很容易过拟合覆盖不了真实输入的长尾。4.3 确定性执行层这是“编译产物”的核心通常表现为规则表、关键词映射、决策树或轻量级脚本。它必须完全确定性输入相同输出必然相同。设计上建议把规则配置化而不是散落在代码里。原因很简单规则会不断更新让业务人员参与维护比每次改代码更安全。# rules.yaml rules: - id: outage keywords: [故障, 崩溃, 无法启动, 服务器500, 宕机] result: category: 故障 department: 技术部 priority: 高 - id: refund keywords: [退款, 退钱, 扣款, 重复扣费] result: category: 投诉 department: 财务部 priority: 中 - id: advice keywords: [建议, 希望增加, 能不能优化] result: category: 建议 department: 产品部 priority: 低关键词规则不是唯一的编译形式。更复杂的场景可以使用决策树、正则表达式、词典 打分机制甚至一个小型的本地模型。关键是这些执行路径不依赖外部 API不消耗模型 Token运行时间在毫秒级。4.4 缓存与命中判定编译之后的管道仍然需要保留缓存层但缓存的作用变了。在纯调用阶段缓存是为了减少重复请求在编译管道里缓存是为了让“曾经 LLM 处理过、但规则尚未覆盖”的样本也能快速命中。缓存键不能直接用整段文本那样命中率太低。更合理的方式是取文本的归一化结果或语义向量。归一化包括大小写转换、去除多余空白、替换可变数字等。向量缓存则需要引入向量数据库或近似检索成本更高但命中率更好。实际项目中建议先做好文本归一化再考虑向量缓存。大多数场景下规则 归一化缓存已经能覆盖绝大部分高频路径。4.5 混合降级策略编译管道一定要设计兜底路径。当规则没有命中、缓存也没有命中时才调用 LLM。此时 LLM 的角色从“默认执行者”变成“异常处理者”。同时每次 LLM 兜底返回的结果都应该被异步记录下来进入下一轮规则迭代。这样管道会越用越“聪明”——不是模型变聪明了而是系统把 LLM 产出的经验不断固化成了确定性规则。4.6 监控与审计编译式管道比纯 LLM 调用更容易监控因为大部分请求走的是确定性路径它们的输入、输出、规则 ID、耗时都可以完整记录。只有回退到 LLM 的请求才带有模型调用成本和延迟。这反而让你第一次能精确回答“哪些请求是真的需要大模型”的问题。5. 完整示例把一条工单分类流程编译成管道下面用一个完整例子演示整个过程。场景是客服工单自动分类输入一段用户描述输出分类、责任部门、优先级和摘要。5.1 第一步先用 LLM 跑通定义调用代码首先写一个最朴素的 LLM 调用。这里以 Python 为例使用 OpenAI 风格的接口但重点是思路不绑定具体厂商。# pipeline_llm_v1.py 文件路径pipeline_llm_v1.py 说明纯 LLM 版本用于采集样本和验证 Schema。 import json from openai import OpenAI SYSTEM_PROMPT 你是客服工单分类器。请根据用户描述输出 JSON包含以下字段 - category: 咨询、故障、投诉、建议 - department: 技术部、售后部、产品部、财务部 - priority: 高、中、低 - summary: 不超过100字的问题摘要 只输出 JSON不要输出其他内容。 client OpenAI() def classify_ticket(text: str) - dict: response client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: text} ], response_format{type: json_object}, temperature0 ) raw response.choices[0].message.content data json.loads(raw) # 这里可以补充 Schema 校验逻辑 return data if __name__ __main__: sample 昨天下午开始系统一直提示500错误完全无法登录客户很着急 result classify_ticket(sample) print(json.dumps(result, ensure_asciiFalse, indent2))运行之后会输出类似{ category: 故障, department: 技术部, priority: 高, summary: 系统500错误无法登录 }这一步的目的是验证 Schema 合理、跑通链路并开始积累输入输出样本。注意不要急着优化先让数据说话。5.2 第二步编写 Schema 校验与样本入库LLM 输出并不总是合法 JSON所以需要加一层校验。同时把通过的样本保存到本地 JSONL 文件作为后续编译规则的数据集。# schema_validator.py 文件路径schema_validator.py 说明简单的 Schema 校验 样本入库。 import json VALID_CATEGORIES {咨询, 故障, 投诉, 建议} VALID_DEPARTMENTS {技术部, 售后部, 产品部, 财务部} VALID_PRIORITIES {高, 中, 低} def validate_result(data: dict) - tuple[bool, str]: if not isinstance(data, dict): return False, 结果不是对象 if data.get(category) not in VALID_CATEGORIES: return False, fcategory 非法: {data.get(category)} if data.get(department) not in VALID_DEPARTMENTS: return False, fdepartment 非法: {data.get(department)} if data.get(priority) not in VALID_PRIORITIES: return False, fpriority 非法: {data.get(priority)} if not data.get(summary) or len(data[summary]) 100: return False, summary 缺失或超过100字 return True, ok def append_sample(file_path: str, text: str, data: dict) - None: sample { text: text, result: data, valid: True } with open(file_path, a, encodingutf-8) as f: f.write(json.dumps(sample, ensure_asciiFalse) \n)校验层非常重要。它决定了哪些 LLM 输出可以进入样本库哪些需要重试或者人工处理。只有在校验层稳定的前提下后续编译出的规则才可信。5.3 第三步编译为规则管道当你积累了几百条样本之后就可以开始“编译”了。先分析样本找出高频关键词和对应结果的映射关系编写规则配置然后用确定性代码替代大部分 LLM 调用。# pipeline_compiled_v1.py 文件路径pipeline_compiled_v1.py 说明编译后的版本规则优先LLM 兜底。 import json from typing import Optional # 规则配置实际项目中建议从 rules.yaml 加载 RULES [ { id: outage, keywords: [故障, 崩溃, 无法启动, 服务器500, 宕机, 无法登录], result: {category: 故障, department: 技术部, priority: 高} }, { id: refund, keywords: [退款, 退钱, 扣款, 重复扣费], result: {category: 投诉, department: 财务部, priority: 中} }, { id: advice, keywords: [建议, 希望增加, 能不能优化], result: {category: 建议, department: 产品部, priority: 低} }, { id: inquiry, keywords: [怎么使用, 如何操作, 价格, 请问], result: {category: 咨询, department: 售后部, priority: 低} }, ] # 简单文本归一化去空白、转小写、保留中文英文数字 def normalize(text: str) - str: return .join(text.split()).lower() def match_rule(text: str) - Optional[dict]: normalized normalize(text) for rule in RULES: for keyword in rule[keywords]: if normalize(keyword) in normalized: return rule[result] return None def classify_ticket(text: str) - dict: rule_result match_rule(text) if rule_result: return {**rule_result, source: rule} return {**llm_fallback(text), source: llm} def llm_fallback(text: str) - dict: # 引用 pipeline_llm_v1.py 的调用逻辑 from pipeline_llm_v1 import classify_ticket as llm_classify return llm_classify(text) if __name__ __main__: samples [ 昨天下午开始系统一直提示500错误完全无法登录客户很着急, 想咨询一下你们的会员价格是多少, 你们怎么又重复扣款了请立刻退钱, 建议增加批量导出功能, ] for sample in samples: print(json.dumps(classify_ticket(sample), ensure_asciiFalse, indent2))这段代码最核心的变化是前四条样本都不再需要请求 LLM而是直接命中规则表返回结果。来源标记source可以用于监控和分析。5.4 第四步加入缓存和混合回退规则命中之后还需要处理“规则没有覆盖但曾经 LLM 处理过”的样本。这就要加一个简单缓存层。# pipeline_compiled_v2.py 文件路径pipeline_compiled_v2.py 说明在规则管道之上加入缓存层。 import json import os from typing import Optional from pipeline_compiled_v1 import match_rule, llm_fallback, normalize CACHE_FILE cache.jsonl _cache {} def _load_cache(): if not os.path.exists(CACHE_FILE): return {} with open(CACHE_FILE, r, encodingutf-8) as f: for line in f: line line.strip() if not line: continue item json.loads(line) _cache[item[key]] item[result] def _save_cache(key: str, result: dict): _cache[key] result with open(CACHE_FILE, a, encodingutf-8) as f: f.write(json.dumps({key: key, result: result}, ensure_asciiFalse) \n) def classify_ticket(text: str) - dict: # 规则优先 rule_result match_rule(text) if rule_result: return {**rule_result, source: rule} # 缓存其次 cache_key normalize(text) if cache_key in _cache: return {**_cache[cache_key], source: cache} # LLM 兜底并写入缓存 result llm_fallback(text) _save_cache(cache_key, result) return {**result, source: llm} _load_cache()注意这里的缓存键使用归一化文本简单但有效。只有归一化后完全相同的文本才会命中。如果你的输入变化很大可以考虑用语义向量做近似缓存但那是另一个复杂话题。5.5 第五步在 Java / Spring AI 项目中如何对接如果你所在的团队使用 Java 技术栈通常会用 Spring AI 或者直接封装 HTTP 调用。下面的示例展示同样的“规则优先、LLM 兜底”思路规则保存在 YAML 配置中业务代码通过一个 Router 组件统一处理。# application.yml 中增加自定义规则配置 llm-pipeline: rules: - id: outage keywords: [故障, 崩溃, 无法启动, 服务器500] category: 故障 department: 技术部 priority: 高 - id: refund keywords: [退款, 退钱, 扣款] category: 投诉 department: 财务部 priority: 中// 文件路径src/main/java/com/example/llmcompilerouter/LlmCompileRouter.java package com.example.llmcompilerouter; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; Component public class LlmCompileRouter { private final ListRuleConfig rules; private final LlmClient llmClient; private final MapString, MapString, Object cache new ConcurrentHashMap(); public LlmCompileRouter(ListRuleConfig rules, LlmClient llmClient) { this.rules rules; this.llmClient llmClient; } public MapString, Object classify(String text) { // 规则命中 MapString, Object ruleHit matchRule(text); if (ruleHit ! null) { ruleHit.put(source, rule); return ruleHit; } // 缓存命中 String key text.replaceAll(\\s, ).toLowerCase(); MapString, Object cached cache.get(key); if (cached ! null) { return cached; } // LLM 兜底 MapString, Object result llmClient.classify(text); cache.put(key, result); return result; } private MapString, Object matchRule(String text) { for (RuleConfig rule : rules) { for (String keyword : rule.getKeywords()) { if (text.contains(keyword)) { MapString, Object result new ConcurrentHashMap(); result.put(category, rule.getCategory()); result.put(department, rule.getDepartment()); result.put(priority, rule.getPriority()); return result; } } } return null; } }这段 Java 代码的核心思想和 Python 版本完全一致。规则优先、缓存其次、LLM 最后兜底。这种设计不需要引入额外的复杂框架普通 Spring 项目直接可以落地。5.6 如何测试与验证编译式管道的最大好处是可测试性极大提升。你可以对规则层写单元测试断言特定文本一定返回特定结果然后对 LLM 兜底层写集成测试只覆盖少量样本。这样 CI 就能稳定运行再也不会被 LLM 的随机输出干扰。6. 运行结果与效果验证运行上面的pipeline_compiled_v1.py预期输出{ category: 故障, department: 技术部, priority: 高, summary: 系统500错误无法登录, source: rule } { category: 咨询, department: 售后部, priority: 低, summary: 咨询会员价格, source: rule } { category: 投诉, department: 财务部, priority: 中, summary: 重复扣款要求退款, source: rule } { category: 建议, department: 产品部, priority: 低, summary: 建议增加批量导出功能, source: rule }如何判断编译成功建议用三个指标规则命中率。统计有多少请求走了规则层而不是 LLM。这是最核心的指标。假设你有 1000 条生产请求其中 600 条命中规则那么 LLM 调用量就下降到了原来的 40%。结果一致性。把编译后的规则输出和之前 LLM 处理的样本对比看有多少结果一致。如果一致性低于预期说明规则关键词覆盖不合理需要调整。LLM 兜底质量。关注兜底请求的输出是否仍然有效。如果某类输入长期反复走兜底说明应该为它新增规则。如果运行失败先按以下顺序排查配置文件是否加载成功规则列表是否为空。样本输入是否被正确归一化关键词大小写是否匹配。LLM 兜底调用是否超时或返回非 JSON。是否缺少rules.yaml或缓存文件目录没有写权限。7. 哪些场景不适合“编译”编译式管道不是银弹。下面这些场景不要强行套用编译思路。开放域问答。如果用户可能问任何问题答案需要实时知识、搜索引擎或外部 API你没法提前写规则覆盖长尾。这时候应该使用 RAG 或 Agent 框架而不是编译管道。多步推理任务。比如“分析这份合同里所有可能的风险并给出谈判建议”这类任务需要模型综合上下文、对比条款、推理潜在后果很难用规则强行模拟。强行编译会把质量压得很低。创意生成内容。标题改写、文案生成、诗歌创作这类任务的价值恰恰在多样性和非确定性。你希望每个用户得到不同的结果编译反而有害。高频变化的任务。如果业务需求每周都在变、分类体系频繁调整编译维护成本会很高。你可以等到 Schema 稳定后再编译不必一开始就追求规则化。判断标准很简单如果任务需要世界知识或开放性创造请继续使用 LLM如果任务只是在固定边界里做语义分类和字段提取请考虑编译。8. 常见问题与排查思路问题现象可能原因排查方式解决方案规则命中率很低关键词覆盖不足真实文本和样本差异大查看未命中样本分析常见表达扩充规则关键词或引入词典 打分机制规则结果和原来 LLM 结果不一致关键词冲突优先级设置不合理对比规则和 LLM 对同一输入的输出差异增加规则优先级建立规则冲突检测缓存命中率低缓存键粒度过粗或过细统计缓存键分布分析文本相似度调整归一化规则或使用语义向量缓存LLM 兜底返回非 JSON提示词约束不足模型版本变化打印原始响应检查是否走了旧版模型强化响应格式校验增加重试逻辑编译管道越来越复杂规则数量失控互相干扰梳理规则之间的重叠和冲突引入规则分组、周期性清理、决策树替代关键词表Schema 字段变动业务新增字段旧规则没有覆盖对比规则输出和 Schema 定义规范版本管理先升级 Schema 再更新规则在这些问题里最容易踩坑的是规则冲突。当规则逐渐增多一个输入可能同时命中“故障”和“投诉”关键词这时必须明确定义优先级。建议在规则表里增加priority字段高优先级规则先匹配并且为每条规则设置明确的作用域。9. 最佳实践与工程建议结合生产环境经验下面几条建议值得提前规划。第一先跑通 LLM再开始编译。不要跳过样本采集阶段直接从规则开始写。你以为的“常规场景”可能只是你以为的模型见到的真实输入会让你重新认识业务。第二为编译管道设计独立的配置中心。规则最好放在配置文件或配置中心里不要硬编码在代码里。业务人员不需要部署就能调整关键词和映射结果这能大大降低维护成本。第三监控规则覆盖率和 LLM 兜底率。建议为每个请求打上source标签并设置告警。如果某一天 LLM 兜底率突然飙升说明业务产生了新的高频模式需要及时更新规则。第四注意数据安全和权限边界。如果输入包含用户隐私数据要谨慎尽量不要把完整的明文文本发送给外部 LLM 服务。可以在进入编译管道前做字段脱敏LLM 兜底时只发送必要字段并在日志中隐藏敏感内容。所有生产环境变更都要经过测试环境验证保留回滚能力。第五与 Agent / 编排框架分层。编译管道适合“高频稳定”路径Agent 框架适合“低频动态”路径。一个复杂的客服系统可以先用编译管道处理 80% 的工单剩下的 20% 由 Agent 调用工具、查询订单系统、RAG 检索知识库。不要觉得用了 Agent 就一定比编译优雅大多数时候简单确定性逻辑比“让模型决定一切”更可靠、更省钱。第六给 Schema 做版本管理。当你新增一个枚举值或删除一个字段时旧的缓存和规则会失效。要像管理数据库 Schema 一样管理 LLM 输出 Schema先兼容再迁移最后清理。10. 总结与后续学习方向这篇文章想要说清楚的核心判断是平凡 LLM 调用确实可以编译成传统数据管道但前提是任务边界清晰、输出可校验、规则可迭代。编译不是否定 LLM 的价值而是把 LLM 从“每次都被迫上场”的位置上解放出来让它专注处理真正需要推理和创造的长尾问题。如果你正在做一个 LLM 应用建议用两周时间按这套方法做一次重构先采集样本再写 Schema然后设计规则层最后加上缓存和兜底。你会发现不仅成本降下来了系统的稳定性和可测试性也大幅提升。这也和 Karpathy 提出的 LLM wiki 思路一致LLM 应用最终要走向知识外置、逻辑确定性、工程可维护性。下一步可以继续深入的方向包括语义向量缓存的工程实现、规则自动挖掘用历史 LLM 输出反推模式、混合管道在 Agent 编排框架中的分层设计以及如何为编译式管道建立完整的监控和实验体系。建议先从本文的最小示例开始把样本采集和规则命中的闭环跑起来再逐步扩展。