這個(gè)標(biāo)題看起來像是一個(gè)純理論問題但背后是一個(gè)非?，F(xiàn)實(shí)的工程需求團(tuán)隊(duì)已經(jīng)有成熟的 Airflow、Spark、dbt 之類的數(shù)據(jù)管道現(xiàn)在想在 ETL 里加一個(gè)“用 LLM 做文本分類、實(shí)體抽取、摘要、打標(biāo)”的步驟。結(jié)果一接進(jìn)去就發(fā)現(xiàn)問題延遲高、成本不穩(wěn)、結(jié)果不可復(fù)現(xiàn)測試也不好寫。于是有人問能不能把 LLM 調(diào)用“編譯”成傳統(tǒng)數(shù)據(jù)管道那樣可重放、可緩存、可測試的確定性步驟。我的判斷是對(duì)于一批規(guī)則明確、輸入輸出穩(wěn)定的 LLM 調(diào)用這條路是可行的而且有很清楚的工程范式。關(guān)鍵不是讓 LLM 變得更像數(shù)據(jù)庫而是把它包裝成“帶外部依賴的算子”然后套上緩存、批量、重試、回放這些數(shù)據(jù)管道本來就有機(jī)制。本文會(huì)把問題拆開給出一個(gè)最小可運(yùn)行的管道示例再討論 API 接入、批量任務(wù)、資源占用、適用邊界和排查方法。1. 核心能力速覽在動(dòng)手之前先把這個(gè)問題映射成工程參數(shù)。能力項(xiàng)說明問題類型LLM 應(yīng)用架構(gòu)設(shè)計(jì)不是具體開源工具核心問題平凡的 LLM 調(diào)用能否被改造成傳統(tǒng)數(shù)據(jù)管道的確定性步驟目標(biāo)手段模板化、語義緩存、批量調(diào)用、失敗重試、確定性降級(jí)、蒸餾替換典型場景離線 ETL 中的文本分類、實(shí)體抽取、摘要、情感分析、標(biāo)簽生成適合讀者有數(shù)據(jù)管道建設(shè)經(jīng)驗(yàn)想把 LLM 接入批處理的工程師不適合場景實(shí)時(shí)對(duì)話、Agent 多輪規(guī)劃、MCP 工具調(diào)用等強(qiáng)交互任務(wù)資源要求取決于本地推理還是 API 調(diào)用API 調(diào)用不占用本地 GPU批量能力支持按批次處理比逐條請(qǐng)求更容易控成本可測試性引入緩存和回放機(jī)制后可以大幅提升結(jié)果可復(fù)現(xiàn)性合規(guī)模塊數(shù)據(jù)脫敏、版權(quán)授權(quán)、隱私保護(hù)屬于必須前置條件這里要特別說明一點(diǎn)標(biāo)題里的 “compiled” 不是傳統(tǒng)編譯器把高級(jí)語言變成機(jī)器碼的意思而是工程意義上的“把可變的外部調(diào)用改造成可驗(yàn)證、可重放、可預(yù)測的數(shù)據(jù)處理步驟”。2. 先拆問題trivial LLM call 到底指什么“trivial LLM call” 翻譯過來是“很普通的 LLM 調(diào)用”。這種調(diào)用通常有幾個(gè)特征。第一輸入輸出結(jié)構(gòu)非常固定。例如給定一條用戶評(píng)論輸出 positive/negative/neutral 三選一給定一段商品描述抽取品牌、型號(hào)、價(jià)格給一篇新聞輸出 100 字摘要。這類任務(wù)用提示詞模板就能做不需要復(fù)雜的多輪對(duì)話不需要工具調(diào)用也不依賴外部知識(shí)庫。第二業(yè)務(wù)邏輯相對(duì)穩(wěn)定。今天是“評(píng)論情感三分類”下個(gè)月大概率還是這個(gè)分類體系。即使樣本變化任務(wù)目標(biāo)不會(huì)頻繁變。這類任務(wù)很容易被固化成一個(gè)可重復(fù)執(zhí)行的步驟。第三單次調(diào)用包含的信息量小。輸入可能只有一段文本輸出是一個(gè)短字符串或 JSON。這就是為什么在傳統(tǒng)管道里它會(huì)顯得很“廉價(jià)”也因此被稱為 trivial。但真正的問題是LLM API 畢竟是一次遠(yuǎn)程調(diào)用它不是純函數(shù)。同樣一個(gè) prompt溫度設(shè)為 0 也可能出現(xiàn)輸出抖動(dòng)網(wǎng)絡(luò)超時(shí)會(huì)中斷整個(gè)管道費(fèi)用隨調(diào)用量線性增長下游任務(wù)不知道這條數(shù)據(jù)是“新算出來的”還是“命中緩存復(fù)用的”。這些才是阻礙 LLM 調(diào)用進(jìn)入傳統(tǒng)數(shù)據(jù)管道的真正原因。所以問題中 “compiled into conventional data pipelines” 的準(zhǔn)確含義是能不能把這種普通 LLM 調(diào)用改造成數(shù)據(jù)管道里一個(gè)標(biāo)準(zhǔn)算子讓它擁有確定性、可重放性、可觀測性。3. 傳統(tǒng)數(shù)據(jù)管道為什么不接受原生 LLM 調(diào)用傳統(tǒng)數(shù)據(jù)管道里的核心組件從 SQL 的 UDF 到 Spark 的 Transformation到 Airflow 的 PythonOperator都有幾個(gè)默認(rèn)前提結(jié)果可重放、運(yùn)行成本可控、失敗時(shí)可重試、輸入輸出可結(jié)構(gòu)化。LLM 原生調(diào)用在這些點(diǎn)上都不滿足。首先是不可復(fù)現(xiàn)。同一個(gè)輸入LLM 兩次調(diào)用的輸出可能不同。即使固定 temperature0不同模型版本、不同推理后端、甚至同一后端的量化參數(shù)都可能造成輸出漂移。數(shù)據(jù)管道下游往往需要穩(wěn)定結(jié)果做聚合和對(duì)比輸出漂移會(huì)直接污染報(bào)表。其次是失敗模型不同。傳統(tǒng)管道失敗一般是數(shù)據(jù)缺失、類型錯(cuò)誤、任務(wù)沖突LLM 調(diào)用失敗則是超時(shí)、限流、上下文超長、API key 失效、內(nèi)容安全攔截。這要求管道具備完全不同的重試策略和降級(jí)策略。第三是成本不可預(yù)估。管道里處理 1 萬條數(shù)據(jù)和 1000 萬條數(shù)據(jù)SQL 的邊際成本幾乎為 0LLM API 的成本卻隨文本長度和調(diào)用量線性增長。如果管道沒有緩存和去重機(jī)制一次重跑就可能是賬單翻倍。第四是延遲。生產(chǎn)數(shù)據(jù)管道通常對(duì)吞吐有硬要求。如果中間插入一個(gè)逐條調(diào)用 LLM API 的算子整個(gè)管道的吞吐會(huì)瞬間被外部服務(wù)的響應(yīng)延遲卡住。這四個(gè)問題合在一起結(jié)論已經(jīng)比較明顯不是 LLM 不能進(jìn)入數(shù)據(jù)管道而是需要給 LLM 套一層“編譯”機(jī)制先把上面四個(gè)問題解決掉。4. “編譯”在 LLM 管道里的三層含義把 LLM 調(diào)用編譯進(jìn)傳統(tǒng)數(shù)據(jù)管道工程上一般分三層來做。這三層可以獨(dú)立實(shí)施也可以疊加使用。4.1 第一層模板化與語義緩存這是最快見效的一層。做法是把 LLM 調(diào)用封裝成一個(gè)算子輸入是一行結(jié)構(gòu)化數(shù)據(jù)輸出是結(jié)構(gòu)化字段中間只做一件事——用模板拼出 prompt然后調(diào)用 LLM最后解析輸出。同時(shí)給算子加磁盤緩存或語義緩存。緩存鍵可以是一整個(gè) prompt 的哈希也可以是“任務(wù)名 輸入文本”的哈希。命中緩存就直接返回不發(fā)起 API 調(diào)用。這一步能把重復(fù)成本直接降到接近 0。對(duì)于數(shù)據(jù)管道來說同一條數(shù)據(jù)跑兩次、同一個(gè)批次被重放都是常見操作。如果沒有緩存每次重放都要為同樣的輸入付費(fèi)。語義緩存比哈希緩存更進(jìn)一步即使輸入文本有細(xì)微差異只要語義等價(jià)也能命中。但這個(gè)實(shí)現(xiàn)復(fù)雜度高需要向量化和相似度閾值適合在業(yè)務(wù)穩(wěn)定后再引入。第一步先用精確匹配緩存性價(jià)比最高。4.2 第二層蒸餾與確定性替換這一層的思路更徹底如果某個(gè) LLM 調(diào)用的任務(wù)是“穩(wěn)定分類”或“穩(wěn)定抽取”那就可以用第一批 LLM 輸出數(shù)據(jù)作為訓(xùn)練集把它蒸餾成一個(gè)小模型、規(guī)則集甚至一段純 Python 代碼。這才是標(biāo)題里 “compiled” 的最貼切含義——不是把 prompt 編譯成機(jī)器碼而是把一個(gè)“用自然語言描述的規(guī)則”編譯成確定性的代碼。舉個(gè)例子先用 LLM 批量標(biāo)注 5000 條評(píng)論情感然后訓(xùn)練一個(gè)很小的分類模型或者讓工程師從輸出中總結(jié)出一組關(guān)鍵詞規(guī)則。之后管道正式運(yùn)行時(shí)就不需要再調(diào) LLM直接跑小模型或規(guī)則即可。這一步適合高吞吐、低延遲、需要穩(wěn)定輸出的場景。它的代價(jià)是前期需要一批標(biāo)注數(shù)據(jù)和一個(gè)驗(yàn)證流程。如果任務(wù)本身經(jīng)常變蒸餾的成本可能比直接調(diào)用 LLM 還高。4.3 第三層編排系統(tǒng)里的 LLM 算子既不想完全蒸餾又想保留 LLM 的泛化能力那就把 LLM 調(diào)用封裝成管道里的標(biāo)準(zhǔn)算子并納入調(diào)度系統(tǒng)的重試、重放、監(jiān)控體系。在 Airflow 里可以寫一個(gè) PythonOperator內(nèi)部批處理一批文本而不是逐條調(diào)用在 Spark 里可以用 mapPartitions 對(duì)每個(gè)分片批量調(diào)用在 Ray 里可以建一個(gè)遠(yuǎn)程函數(shù)池并發(fā)調(diào)用。這一層解決的是“把 LLM 當(dāng)普通計(jì)算單元”的問題。管道調(diào)度器不知道里面跑的是 SQL 還是 LLM API它只知道這個(gè)算子有輸入、有輸出、可能失敗、可以重試。5. 一個(gè)最小可運(yùn)行的編譯型 LLM 管道示例紙上談兵沒有意義下面給一個(gè)可以直接跑通的最小示例。它演示了三個(gè)關(guān)鍵能力把 LLM 調(diào)用封裝成算子增加磁盤緩存避免重跑重復(fù)付費(fèi)增加 fallback即使 LLM 調(diào)用失敗管道也不會(huì)整體中斷。生產(chǎn)環(huán)境把BaseLLMClient替換成真實(shí) 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: 生產(chǎn)環(huán)境替換為真實(shí) 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) # 模擬網(wǎng)絡(luò)延遲 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. 構(gòu)造 prompt 并調(diào)用 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 ( 請(qǐng)從以下評(píng)論中提取情緒positive/negative/neutral 和主題關(guān)鍵詞輸出 JSON。\n f評(píng)論{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: 這個(gè)功能很好用我非常喜歡。}, {id: 2, text: 界面不太穩(wěn)定經(jīng)?？D。}, {id: 3, text: 功能正常速度可以接受。}, ] outputs run_pipeline(rows, [op]) for out in outputs: print(out)運(yùn)行第二次所有l(wèi)lm_extract_source字段都會(huì)變成cache說明沒有再次發(fā)起 LLM 調(diào)用。這個(gè)模式非常簡單但它已經(jīng)具備“編譯”的核心特征輸入確定、結(jié)果可緩存、失敗有降級(jí)、可以放進(jìn)任何調(diào)度器。把DummyLLMClient換成真實(shí)客戶端就是這個(gè)思路的真實(shí)落地版本。接著可以把它接到 Airflow 風(fēng)格的批處理任務(wù)里。以下是一個(gè)示意 DAG實(shí)際寫法需要按你的 Airflow 版本調(diào)整。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) # 從數(shù)倉或文件讀取 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 內(nèi)部逐條循環(huán)調(diào)用 LLM API。正確做法是每次執(zhí)行處理一批數(shù)據(jù)內(nèi)部再使用并發(fā)或分批請(qǐng)求。否則調(diào)度器一重試外部 API 會(huì)被同一批請(qǐng)求打爆。6. LLM API 接入、批量任務(wù)與成本控制真實(shí)項(xiàng)目不會(huì)用 DummyLLMClient下面講 API 接入時(shí)需要注意的工程細(xì)節(jié)。6.1 批量請(qǐng)求與隊(duì)列大多數(shù) LLM API 都有限流策略單位時(shí)間內(nèi)的請(qǐng)求次數(shù)和 token 數(shù)量都有限制。數(shù)據(jù)管道里最常見的錯(cuò)誤是一個(gè)批次 1 萬條數(shù)據(jù)直接用 for 循環(huán)調(diào)用結(jié)果觸發(fā)限流任務(wù)中途失敗。推薦的做法是引入并發(fā)池加令牌桶。把每個(gè)批次切成小塊用concurrent.futures.ThreadPoolExecutor控制并發(fā)度。寫一個(gè)具備自動(dòng)重試的通用調(diào)用函數(shù)比較穩(wěn)妥。import openai # 按實(shí)際 SDK 安裝版本不同參數(shù)可能有差異 from tenacity import retry, stop_after_attempt, wait_exponential client openai.OpenAI() # 生產(chǎn)環(huán)境從配置或密鑰服務(wù)讀取 retry(stopstop_after_attempt(3), waitwait_exponential(min1, max10)) def chat_once(prompt: str) - str: resp client.chat.completions.create( modelgpt-4o-mini, # 按實(shí)際可用模型名替換 messages[{role: user, content: prompt}], temperature0.0, ) return resp.choices[0].message.content重試策略不要對(duì)超時(shí)、限流、網(wǎng)絡(luò)抖動(dòng)和內(nèi)容安全攔截一視同仁。限流通常需要等更長的時(shí)間內(nèi)容安全攔截重試多少次都不會(huì)成功應(yīng)該把它標(biāo)記為錯(cuò)誤數(shù)據(jù)落入異常隊(duì)列而不是無腦重試。6.2 緩存鍵與去重在數(shù)據(jù)管道里緩存鍵設(shè)計(jì)很重要。建議把“任務(wù)名 模型名 prompt 版本 輸入哈希”組合成緩存鍵。只對(duì)輸入文本做哈希容易在 prompt 模板升級(jí)后拿到舊結(jié)果。如果一次要處理 1000 萬條評(píng)論很多文本可能是重復(fù)的。先做一次去重再對(duì)唯一文本調(diào)用 LLM能用較少的請(qǐng)求覆蓋較大比例的數(shù)據(jù)。這個(gè)去重本身就是成本和速度的雙重優(yōu)化。6.3 失敗重試與降級(jí)處理 LLM 調(diào)用失敗時(shí)至少要有三種策略。第一簡單重試。適合網(wǎng)絡(luò)抖動(dòng)、瞬時(shí)限流。第二延遲退避。適合 API 側(cè)壓力大讓出時(shí)間窗口。第三確定性降級(jí)。如果業(yè)務(wù)允許失敗時(shí)用關(guān)鍵詞規(guī)則或默認(rèn)值兜底。前面示例里的fallback_rule就是這個(gè)思路。降級(jí)結(jié)果必須標(biāo)記來源否則下游會(huì)把它當(dāng)成正常 LLM 輸出導(dǎo)致數(shù)據(jù)質(zhì)量失真。完整的數(shù)據(jù)管道里L(fēng)LM 調(diào)用的結(jié)論、來源、錯(cuò)誤信息都應(yīng)該作為列寫入結(jié)果表。這樣即使出現(xiàn)質(zhì)量波動(dòng)也能回溯到是哪一批數(shù)據(jù)、哪次 prompt 版本、哪個(gè)模型產(chǎn)生的問題。7. 資源占用與性能觀察資源占用要分兩種情況看。如果使用 LLM API本地管道資源主要是 CPU、內(nèi)存和帶寬不占用 GPU。這時(shí)要重點(diǎn)觀察的是API 延遲的 P50/P95、失敗率、限流次數(shù)、緩存命中率、token 消耗量。建議每批次結(jié)束后寫一條監(jiān)控日志至少包含這些指標(biāo)。如果使用本地模型推理比如在管道里部署一個(gè) 7B 或 14B 的模型重點(diǎn)看顯存占用。顯存占用跟模型精度、batch size、輸入長度都有關(guān)系。使用 fp16/bf16 這類低精度格式能明顯降低顯存但模型效果可能需要在小樣本上驗(yàn)證。這里不寫死某個(gè)模型具體占多少 GB因?yàn)椴煌炕燃?jí)、不同部署框架vLLM、llama.cpp、Ollama、Transformers差異很大需要以本機(jī)測試為準(zhǔn)。觀察方法很簡單本地推理用nvidia-smi -l 1看實(shí)時(shí)顯存和 GPU 利用率API 調(diào)用在客戶端記日志管道層面在批次開頭和結(jié)尾記錄處理?xiàng)l數(shù)、耗時(shí)、失敗數(shù)。影響性能的主要變量有三個(gè)輸入文本長度越長token 消耗越大延遲越高并發(fā)度API 模式下并發(fā)越高吞吐越高但有限流邊界緩存命中率命中率越高平均單條成本越低吞吐越穩(wěn)定。建議第一次跑通時(shí)先用 100 條數(shù)據(jù)做小批次性能測試觀察耗時(shí)和失敗率再逐步放大到全量。不要直接拿全量數(shù)據(jù)沖否則限流和成本都不可控。8. 適用場景與使用邊界這個(gè)思路適合什么場景適合輸入輸出結(jié)構(gòu)固定、單次調(diào)用信息量小、業(yè)務(wù)規(guī)則相對(duì)穩(wěn)定的批處理任務(wù)。比如用戶評(píng)論情感分類商品信息字段抽取新聞?wù)头未驑?biāo)郵件自動(dòng)分類文檔版式識(shí)別后的內(nèi)容清洗。不太適合什么場景不適合強(qiáng)交互、多輪決策、需要工具調(diào)用的復(fù)雜任務(wù)。Agent 規(guī)劃、MCP 工具調(diào)用、實(shí)時(shí)對(duì)話這類場景依賴狀態(tài)和多步反饋鏈路不可重放也缺少穩(wěn)定的輸出結(jié)構(gòu)硬塞進(jìn)傳統(tǒng)批處理管道反而會(huì)把問題搞復(fù)雜。這類任務(wù)更適合用專門的 Agent 編排框架而不是把每次調(diào)用都“編譯”成固定算子。另外從 RAG 和語義檢索角度說如果管道里的 LLM 調(diào)用依賴外部向量庫或動(dòng)態(tài)知識(shí)庫那輸入就不再是單行文本而是“文本 檢索上下文”。這種調(diào)用比 trivial call 復(fù)雜需要把檢索結(jié)果也納入緩存鍵和重放邏輯否則整個(gè)管道仍然不穩(wěn)定。使用邊界還包括數(shù)據(jù)和合規(guī)。文本數(shù)據(jù)進(jìn)入外部 LLM API 前必須確認(rèn)數(shù)據(jù)是否包含個(gè)人隱私、商業(yè)秘密、版權(quán)內(nèi)容。涉及人臉、聲音、肖像、受版權(quán)保護(hù)的素材時(shí)必須確認(rèn)授權(quán)。生產(chǎn)環(huán)境建議先做脫敏再評(píng)估能否使用外部 API。如果數(shù)據(jù)不能出域就要選擇本地部署推理服務(wù)成本和治理都不同。9. 常見問題與排查方法問題現(xiàn)象可能原因排查方式解決方案管道跑完發(fā)現(xiàn)很多結(jié)果來自 fallbackLLM 調(diào)用失敗但被降級(jí)處理檢查結(jié)果表中的_source和_error字段區(qū)分“LLM 正常結(jié)果”和“降級(jí)結(jié)果”統(tǒng)計(jì)失敗率同一份數(shù)據(jù)重跑成本翻倍緩存鍵設(shè)計(jì)不合理或緩存未生效檢查緩存命中率日志把任務(wù)名、模型名、prompt 版本納入緩存鍵API 請(qǐng)求大量超時(shí)并發(fā)度過高或限流查看客戶端日志和 API 錯(cuò)誤碼降低并發(fā)度增加指數(shù)退避重試結(jié)果不穩(wěn)定影響下游報(bào)表輸出解析失敗或模型輸出抖動(dòng)對(duì)比同一條輸入的多次輸出固定 temperature0增加輸出格式校驗(yàn)和重試顯存不足或推理很慢本地模型精度或 batch size 設(shè)置不當(dāng)用nvidia-smi觀察顯存記錄單批次耗時(shí)換低精度加載減小 batch size或改用 API模型升級(jí)后結(jié)果風(fēng)格變化prompt 模板或模型版本變更對(duì)比歷史輸出每次升級(jí)前用固定測試集做回歸對(duì)比大批量處理時(shí)中間斷掉沒有做批次檢查點(diǎn)和斷點(diǎn)續(xù)跑檢查調(diào)度器日志按批次寫結(jié)果重跑時(shí)跳過已完成批次輸出 JSON 解析失敗模型輸出包含多余文本記錄原始輸出增加輸出格式約束或解析后校驗(yàn)失敗重試10. 最佳實(shí)踐與使用建議先給一個(gè)保守的落地路徑。第一步把一個(gè) LLM 調(diào)用封裝成算子加入緩存和降級(jí)用小批量數(shù)據(jù)驗(yàn)證正確性。不要先上完整管道先驗(yàn)證單算子。第二步把算子接入現(xiàn)有調(diào)度器但保留“跳過 LLM 直接跑緩存”的模式。這樣在 prompt 調(diào)整和模型升級(jí)時(shí)能對(duì)比新舊結(jié)果。第三步建立固定的評(píng)測集。至少準(zhǔn)備 100 到 500 條帶標(biāo)準(zhǔn)答案的樣本每次模型或 prompt 變更后都跑一遍對(duì)比準(zhǔn)確率和格式合格率。沒有評(píng)測集的 LLM 管道后期維護(hù)會(huì)非常痛苦。第四步做蒸餾或規(guī)則替換。當(dāng)批量任務(wù)穩(wěn)定運(yùn)行一段時(shí)間后把高頻輸入和 LLM 輸出導(dǎo)出嘗試用規(guī)則、小模型替換。目標(biāo)是把 80% 的確定性請(qǐng)求從 LLM 調(diào)用中剝離出去只保留少數(shù)復(fù)雜樣本走 LLM。第五步監(jiān)控成本和質(zhì)量。每批次記錄 token 消耗、緩存命中率、fallback 次數(shù)、輸出解析失敗率。成本和質(zhì)量一旦異常能快速定位是數(shù)據(jù)變化、prompt 變化還是模型變化導(dǎo)致。關(guān)于 LLM 文本向量 API 未配置這類問題如果管道里還涉及向量化、語義檢索建議把“LLM 調(diào)用”和“向量化調(diào)用”分開配置、分開監(jiān)控?；煸谝黄饡?huì)導(dǎo)致故障定位困難尤其是其中一方限流或 key 失效時(shí)很難判斷是哪個(gè)服務(wù)導(dǎo)致整個(gè)管道卡住。11. 總結(jié)與下一步“Can trivial LLM calls be compiled into conventional data pipelines?” 這個(gè)問題的答案不是簡單的“能”或“不能”。對(duì)于輸入輸出固定、業(yè)務(wù)規(guī)則穩(wěn)定的普通 LLM 調(diào)用答案是“能但需要經(jīng)過工程改造”。改造的核心是把 LLM 從“隨時(shí)可能抖動(dòng)的外部服務(wù)”封裝成“帶緩存、帶重試、帶降級(jí)的管道算子”。最先應(yīng)該驗(yàn)證的功能是緩存能否正確命中、降級(jí)路徑能否在不中斷管道的情況下兜底。最容易踩的坑是直接逐條調(diào)用 API以及不記錄結(jié)果來源導(dǎo)致下游無法判斷數(shù)據(jù)質(zhì)量。后續(xù)可以從已有管道里挑一個(gè)最簡單的文本分類任務(wù)開始先跑通單算子再做批次調(diào)度和成本監(jiān)控。如果這個(gè)方向驗(yàn)證成功下一步可以把實(shí)驗(yàn)擴(kuò)展到模板化生成、摘要、字段抽取如果業(yè)務(wù)穩(wěn)定到一定程度再考慮把 LLM 輸出蒸餾成確定性組件徹底擺脫對(duì)在線模型的依賴。先跑通最小示例再逐步把評(píng)測集、影子比較、成本監(jiān)控加進(jìn)去是比較務(wù)實(shí)的路徑。