这个标题看起来像是一个纯理论问题但背后是一个非常现实的工程需求团队已经有成熟的 Airflow、Spark、dbt 之类的数据管道现在想在 ETL 里加一个“用 LLM 做文本分类、实体抽取、摘要、打标”的步骤。结果一接进去就发现问题延迟高、成本不稳、结果不可复现测试也不好写。于是有人问能不能把 LLM 调用“编译”成传统数据管道那样可重放、可缓存、可测试的确定性步骤。我的判断是对于一批规则明确、输入输出稳定的 LLM 调用这条路是可行的而且有很清楚的工程范式。关键不是让 LLM 变得更像数据库而是把它包装成“带外部依赖的算子”然后套上缓存、批量、重试、回放这些数据管道本来就有机制。本文会把问题拆开给出一个最小可运行的管道示例再讨论 API 接入、批量任务、资源占用、适用边界和排查方法。1. 核心能力速览在动手之前先把这个问题映射成工程参数。能力项说明问题类型LLM 应用架构设计不是具体开源工具核心问题平凡的 LLM 调用能否被改造成传统数据管道的确定性步骤目标手段模板化、语义缓存、批量调用、失败重试、确定性降级、蒸馏替换典型场景离线 ETL 中的文本分类、实体抽取、摘要、情感分析、标签生成适合读者有数据管道建设经验想把 LLM 接入批处理的工程师不适合场景实时对话、Agent 多轮规划、MCP 工具调用等强交互任务资源要求取决于本地推理还是 API 调用API 调用不占用本地 GPU批量能力支持按批次处理比逐条请求更容易控成本可测试性引入缓存和回放机制后可以大幅提升结果可复现性合规模块数据脱敏、版权授权、隐私保护属于必须前置条件这里要特别说明一点标题里的 “compiled” 不是传统编译器把高级语言变成机器码的意思而是工程意义上的“把可变的外部调用改造成可验证、可重放、可预测的数据处理步骤”。2. 先拆问题trivial LLM call 到底指什么“trivial LLM call” 翻译过来是“很普通的 LLM 调用”。这种调用通常有几个特征。第一输入输出结构非常固定。例如给定一条用户评论输出 positive/negative/neutral 三选一给定一段商品描述抽取品牌、型号、价格给一篇新闻输出 100 字摘要。这类任务用提示词模板就能做不需要复杂的多轮对话不需要工具调用也不依赖外部知识库。第二业务逻辑相对稳定。今天是“评论情感三分类”下个月大概率还是这个分类体系。即使样本变化任务目标不会频繁变。这类任务很容易被固化成一个可重复执行的步骤。第三单次调用包含的信息量小。输入可能只有一段文本输出是一个短字符串或 JSON。这就是为什么在传统管道里它会显得很“廉价”也因此被称为 trivial。但真正的问题是LLM API 毕竟是一次远程调用它不是纯函数。同样一个 prompt温度设为 0 也可能出现输出抖动网络超时会中断整个管道费用随调用量线性增长下游任务不知道这条数据是“新算出来的”还是“命中缓存复用的”。这些才是阻碍 LLM 调用进入传统数据管道的真正原因。所以问题中 “compiled into conventional data pipelines” 的准确含义是能不能把这种普通 LLM 调用改造成数据管道里一个标准算子让它拥有确定性、可重放性、可观测性。3. 传统数据管道为什么不接受原生 LLM 调用传统数据管道里的核心组件从 SQL 的 UDF 到 Spark 的 Transformation到 Airflow 的 PythonOperator都有几个默认前提结果可重放、运行成本可控、失败时可重试、输入输出可结构化。LLM 原生调用在这些点上都不满足。首先是不可复现。同一个输入LLM 两次调用的输出可能不同。即使固定 temperature0不同模型版本、不同推理后端、甚至同一后端的量化参数都可能造成输出漂移。数据管道下游往往需要稳定结果做聚合和对比输出漂移会直接污染报表。其次是失败模型不同。传统管道失败一般是数据缺失、类型错误、任务冲突LLM 调用失败则是超时、限流、上下文超长、API key 失效、内容安全拦截。这要求管道具备完全不同的重试策略和降级策略。第三是成本不可预估。管道里处理 1 万条数据和 1000 万条数据SQL 的边际成本几乎为 0LLM API 的成本却随文本长度和调用量线性增长。如果管道没有缓存和去重机制一次重跑就可能是账单翻倍。第四是延迟。生产数据管道通常对吞吐有硬要求。如果中间插入一个逐条调用 LLM API 的算子整个管道的吞吐会瞬间被外部服务的响应延迟卡住。这四个问题合在一起结论已经比较明显不是 LLM 不能进入数据管道而是需要给 LLM 套一层“编译”机制先把上面四个问题解决掉。4. “编译”在 LLM 管道里的三层含义把 LLM 调用编译进传统数据管道工程上一般分三层来做。这三层可以独立实施也可以叠加使用。4.1 第一层模板化与语义缓存这是最快见效的一层。做法是把 LLM 调用封装成一个算子输入是一行结构化数据输出是结构化字段中间只做一件事——用模板拼出 prompt然后调用 LLM最后解析输出。同时给算子加磁盘缓存或语义缓存。缓存键可以是一整个 prompt 的哈希也可以是“任务名 输入文本”的哈希。命中缓存就直接返回不发起 API 调用。这一步能把重复成本直接降到接近 0。对于数据管道来说同一条数据跑两次、同一个批次被重放都是常见操作。如果没有缓存每次重放都要为同样的输入付费。语义缓存比哈希缓存更进一步即使输入文本有细微差异只要语义等价也能命中。但这个实现复杂度高需要向量化和相似度阈值适合在业务稳定后再引入。第一步先用精确匹配缓存性价比最高。4.2 第二层蒸馏与确定性替换这一层的思路更彻底如果某个 LLM 调用的任务是“稳定分类”或“稳定抽取”那就可以用第一批 LLM 输出数据作为训练集把它蒸馏成一个小模型、规则集甚至一段纯 Python 代码。这才是标题里 “compiled” 的最贴切含义——不是把 prompt 编译成机器码而是把一个“用自然语言描述的规则”编译成确定性的代码。举个例子先用 LLM 批量标注 5000 条评论情感然后训练一个很小的分类模型或者让工程师从输出中总结出一组关键词规则。之后管道正式运行时就不需要再调 LLM直接跑小模型或规则即可。这一步适合高吞吐、低延迟、需要稳定输出的场景。它的代价是前期需要一批标注数据和一个验证流程。如果任务本身经常变蒸馏的成本可能比直接调用 LLM 还高。4.3 第三层编排系统里的 LLM 算子既不想完全蒸馏又想保留 LLM 的泛化能力那就把 LLM 调用封装成管道里的标准算子并纳入调度系统的重试、重放、监控体系。在 Airflow 里可以写一个 PythonOperator内部批处理一批文本而不是逐条调用在 Spark 里可以用 mapPartitions 对每个分片批量调用在 Ray 里可以建一个远程函数池并发调用。这一层解决的是“把 LLM 当普通计算单元”的问题。管道调度器不知道里面跑的是 SQL 还是 LLM API它只知道这个算子有输入、有输出、可能失败、可以重试。5. 一个最小可运行的编译型 LLM 管道示例纸上谈兵没有意义下面给一个可以直接跑通的最小示例。它演示了三个关键能力把 LLM 调用封装成算子增加磁盘缓存避免重跑重复付费增加 fallback即使 LLM 调用失败管道也不会整体中断。生产环境把BaseLLMClient替换成真实 LLM API 客户端即可。# llm_pipeline_demo.py import hashlib import json import time from dataclasses import dataclass from typing import Any, Callable, Dict, List, Optional class BaseLLMClient: 生产环境替换为真实 LLM API 客户端。 def complete(self, prompt: str, temperature: float 0.0) - str: raise NotImplementedError class DummyLLMClient(BaseLLMClient): def complete(self, prompt: str, temperature: float 0.0) - str: time.sleep(0.05) # 模拟网络延迟 return fresult_of: {prompt[:24]} class DiskCache: def __init__(self, cache_dir: str ./llm_cache): self.cache_dir cache_dir def _key(self, task: str, payload: Dict[str, Any]) - str: raw json.dumps( {task: task, payload: payload}, sort_keysTrue, ensure_asciiFalse, ) return hashlib.sha256(raw.encode(utf-8)).hexdigest() def get(self, task: str, payload: Dict[str, Any]) - Optional[str]: import os path os.path.join(self.cache_dir, self._key(task, payload) .json) if os.path.exists(path): with open(path, r, encodingutf-8) as f: return json.load(f)[output] return None def set(self, task: str, payload: Dict[str, Any], output: str) - None: import os os.makedirs(self.cache_dir, exist_okTrue) path os.path.join(self.cache_dir, self._key(task, payload) .json) with open(path, w, encodingutf-8) as f: json.dump({output: output}, f, ensure_asciiFalse) dataclass class LLMOperator: name: str llm: BaseLLMClient prompt_template: Callable[[Dict[str, Any]], str] cache: Optional[DiskCache] None fallback: Optional[Callable[[Dict[str, Any]], str]] None def execute(self, row: Dict[str, Any]) - Dict[str, Any]: # 1. 查缓存 if self.cache: cached self.cache.get(self.name, row) if cached is not None: row[self.name] cached row[f{self.name}_source] cache return row # 2. 构造 prompt 并调用 LLM prompt self.prompt_template(row) try: output self.llm.complete(prompt, temperature0.0) except Exception as exc: if self.fallback is None: raise output self.fallback(row) row[f{self.name}_error] str(exc) row[f{self.name}_source] fallback else: # 3. 写入缓存 if self.cache: self.cache.set(self.name, row, output) row[f{self.name}_source] llm row[self.name] output return row def run_pipeline( rows: List[Dict[str, Any]], operators: List[LLMOperator], ) - List[Dict[str, Any]]: results [] for row in rows: current dict(row) for op in operators: current op.execute(current) results.append(current) return results def build_prompt(row: Dict[str, Any]) - str: return ( 请从以下评论中提取情绪positive/negative/neutral 和主题关键词输出 JSON。\n f评论{row[text]} ) def fallback_rule(row: Dict[str, Any]) - str: text row[text] if any(w in text for w in [好, 喜欢, 赞]): return {sentiment: positive} return {sentiment: neutral} if __name__ __main__: llm DummyLLMClient() cache DiskCache() op LLMOperator( namellm_extract, llmllm, prompt_templatebuild_prompt, cachecache, fallbackfallback_rule, ) rows [ {id: 1, text: 这个功能很好用我非常喜欢。}, {id: 2, text: 界面不太稳定经常卡顿。}, {id: 3, text: 功能正常速度可以接受。}, ] outputs run_pipeline(rows, [op]) for out in outputs: print(out)运行第二次所有llm_extract_source字段都会变成cache说明没有再次发起 LLM 调用。这个模式非常简单但它已经具备“编译”的核心特征输入确定、结果可缓存、失败有降级、可以放进任何调度器。把DummyLLMClient换成真实客户端就是这个思路的真实落地版本。接着可以把它接到 Airflow 风格的批处理任务里。以下是一个示意 DAG实际写法需要按你的 Airflow 版本调整。from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime def extract_with_llm(**context): batch_id context[params][batch_id] batch load_batch(batch_id) # 从数仓或文件读取 results run_pipeline(batch, [llm_extract_operator]) write_to_parquet(results, foutput/batch_{batch_id}.parquet) with DAG( dag_idllm_compile_pipeline, start_datedatetime(2025, 1, 1), scheduledaily, catchupFalse, ) as dag: t1 PythonOperator( task_idrun_llm_batch, python_callableextract_with_llm, params{batch_id: {{ ds }}}, )注意这里最容易踩的坑是不要在 PythonOperator 内部逐条循环调用 LLM API。正确做法是每次执行处理一批数据内部再使用并发或分批请求。否则调度器一重试外部 API 会被同一批请求打爆。6. LLM API 接入、批量任务与成本控制真实项目不会用 DummyLLMClient下面讲 API 接入时需要注意的工程细节。6.1 批量请求与队列大多数 LLM API 都有限流策略单位时间内的请求次数和 token 数量都有限制。数据管道里最常见的错误是一个批次 1 万条数据直接用 for 循环调用结果触发限流任务中途失败。推荐的做法是引入并发池加令牌桶。把每个批次切成小块用concurrent.futures.ThreadPoolExecutor控制并发度。写一个具备自动重试的通用调用函数比较稳妥。import openai # 按实际 SDK 安装版本不同参数可能有差异 from tenacity import retry, stop_after_attempt, wait_exponential client openai.OpenAI() # 生产环境从配置或密钥服务读取 retry(stopstop_after_attempt(3), waitwait_exponential(min1, max10)) def chat_once(prompt: str) - str: resp client.chat.completions.create( modelgpt-4o-mini, # 按实际可用模型名替换 messages[{role: user, content: prompt}], temperature0.0, ) return resp.choices[0].message.content重试策略不要对超时、限流、网络抖动和内容安全拦截一视同仁。限流通常需要等更长的时间内容安全拦截重试多少次都不会成功应该把它标记为错误数据落入异常队列而不是无脑重试。6.2 缓存键与去重在数据管道里缓存键设计很重要。建议把“任务名 模型名 prompt 版本 输入哈希”组合成缓存键。只对输入文本做哈希容易在 prompt 模板升级后拿到旧结果。如果一次要处理 1000 万条评论很多文本可能是重复的。先做一次去重再对唯一文本调用 LLM能用较少的请求覆盖较大比例的数据。这个去重本身就是成本和速度的双重优化。6.3 失败重试与降级处理 LLM 调用失败时至少要有三种策略。第一简单重试。适合网络抖动、瞬时限流。第二延迟退避。适合 API 侧压力大让出时间窗口。第三确定性降级。如果业务允许失败时用关键词规则或默认值兜底。前面示例里的fallback_rule就是这个思路。降级结果必须标记来源否则下游会把它当成正常 LLM 输出导致数据质量失真。完整的数据管道里LLM 调用的结论、来源、错误信息都应该作为列写入结果表。这样即使出现质量波动也能回溯到是哪一批数据、哪次 prompt 版本、哪个模型产生的问题。7. 资源占用与性能观察资源占用要分两种情况看。如果使用 LLM API本地管道资源主要是 CPU、内存和带宽不占用 GPU。这时要重点观察的是API 延迟的 P50/P95、失败率、限流次数、缓存命中率、token 消耗量。建议每批次结束后写一条监控日志至少包含这些指标。如果使用本地模型推理比如在管道里部署一个 7B 或 14B 的模型重点看显存占用。显存占用跟模型精度、batch size、输入长度都有关系。使用 fp16/bf16 这类低精度格式能明显降低显存但模型效果可能需要在小样本上验证。这里不写死某个模型具体占多少 GB因为不同量化等级、不同部署框架vLLM、llama.cpp、Ollama、Transformers差异很大需要以本机测试为准。观察方法很简单本地推理用nvidia-smi -l 1看实时显存和 GPU 利用率API 调用在客户端记日志管道层面在批次开头和结尾记录处理条数、耗时、失败数。影响性能的主要变量有三个输入文本长度越长token 消耗越大延迟越高并发度API 模式下并发越高吞吐越高但有限流边界缓存命中率命中率越高平均单条成本越低吞吐越稳定。建议第一次跑通时先用 100 条数据做小批次性能测试观察耗时和失败率再逐步放大到全量。不要直接拿全量数据冲否则限流和成本都不可控。8. 适用场景与使用边界这个思路适合什么场景适合输入输出结构固定、单次调用信息量小、业务规则相对稳定的批处理任务。比如用户评论情感分类商品信息字段抽取新闻摘要客服工单打标邮件自动分类文档版式识别后的内容清洗。不太适合什么场景不适合强交互、多轮决策、需要工具调用的复杂任务。Agent 规划、MCP 工具调用、实时对话这类场景依赖状态和多步反馈链路不可重放也缺少稳定的输出结构硬塞进传统批处理管道反而会把问题搞复杂。这类任务更适合用专门的 Agent 编排框架而不是把每次调用都“编译”成固定算子。另外从 RAG 和语义检索角度说如果管道里的 LLM 调用依赖外部向量库或动态知识库那输入就不再是单行文本而是“文本 检索上下文”。这种调用比 trivial call 复杂需要把检索结果也纳入缓存键和重放逻辑否则整个管道仍然不稳定。使用边界还包括数据和合规。文本数据进入外部 LLM API 前必须确认数据是否包含个人隐私、商业秘密、版权内容。涉及人脸、声音、肖像、受版权保护的素材时必须确认授权。生产环境建议先做脱敏再评估能否使用外部 API。如果数据不能出域就要选择本地部署推理服务成本和治理都不同。9. 常见问题与排查方法问题现象可能原因排查方式解决方案管道跑完发现很多结果来自 fallbackLLM 调用失败但被降级处理检查结果表中的_source和_error字段区分“LLM 正常结果”和“降级结果”统计失败率同一份数据重跑成本翻倍缓存键设计不合理或缓存未生效检查缓存命中率日志把任务名、模型名、prompt 版本纳入缓存键API 请求大量超时并发度过高或限流查看客户端日志和 API 错误码降低并发度增加指数退避重试结果不稳定影响下游报表输出解析失败或模型输出抖动对比同一条输入的多次输出固定 temperature0增加输出格式校验和重试显存不足或推理很慢本地模型精度或 batch size 设置不当用nvidia-smi观察显存记录单批次耗时换低精度加载减小 batch size或改用 API模型升级后结果风格变化prompt 模板或模型版本变更对比历史输出每次升级前用固定测试集做回归对比大批量处理时中间断掉没有做批次检查点和断点续跑检查调度器日志按批次写结果重跑时跳过已完成批次输出 JSON 解析失败模型输出包含多余文本记录原始输出增加输出格式约束或解析后校验失败重试10. 最佳实践与使用建议先给一个保守的落地路径。第一步把一个 LLM 调用封装成算子加入缓存和降级用小批量数据验证正确性。不要先上完整管道先验证单算子。第二步把算子接入现有调度器但保留“跳过 LLM 直接跑缓存”的模式。这样在 prompt 调整和模型升级时能对比新旧结果。第三步建立固定的评测集。至少准备 100 到 500 条带标准答案的样本每次模型或 prompt 变更后都跑一遍对比准确率和格式合格率。没有评测集的 LLM 管道后期维护会非常痛苦。第四步做蒸馏或规则替换。当批量任务稳定运行一段时间后把高频输入和 LLM 输出导出尝试用规则、小模型替换。目标是把 80% 的确定性请求从 LLM 调用中剥离出去只保留少数复杂样本走 LLM。第五步监控成本和质量。每批次记录 token 消耗、缓存命中率、fallback 次数、输出解析失败率。成本和质量一旦异常能快速定位是数据变化、prompt 变化还是模型变化导致。关于 LLM 文本向量 API 未配置这类问题如果管道里还涉及向量化、语义检索建议把“LLM 调用”和“向量化调用”分开配置、分开监控。混在一起会导致故障定位困难尤其是其中一方限流或 key 失效时很难判断是哪个服务导致整个管道卡住。11. 总结与下一步“Can trivial LLM calls be compiled into conventional data pipelines?” 这个问题的答案不是简单的“能”或“不能”。对于输入输出固定、业务规则稳定的普通 LLM 调用答案是“能但需要经过工程改造”。改造的核心是把 LLM 从“随时可能抖动的外部服务”封装成“带缓存、带重试、带降级的管道算子”。最先应该验证的功能是缓存能否正确命中、降级路径能否在不中断管道的情况下兜底。最容易踩的坑是直接逐条调用 API以及不记录结果来源导致下游无法判断数据质量。后续可以从已有管道里挑一个最简单的文本分类任务开始先跑通单算子再做批次调度和成本监控。如果这个方向验证成功下一步可以把实验扩展到模板化生成、摘要、字段抽取如果业务稳定到一定程度再考虑把 LLM 输出蒸馏成确定性组件彻底摆脱对在线模型的依赖。先跑通最小示例再逐步把评测集、影子比较、成本监控加进去是比较务实的路径。