先說一個很多人都會踩的坑你以為把任務(wù)丟給一個大模型加上上下文提示詞它就能幫你把一個十步的活兒一口氣干完。實際上大模型單次輸出經(jīng)常在中間某個環(huán)節(jié)悄悄跑偏或者直接因為超過了輸出上限給你半截結(jié)果。一旦出問題就得從頭再來前面的所有工作全部作廢。這也是我后來決定重做一套“多 Agent 串行流水線”的原因——把任務(wù)拆成一串可重試、可續(xù)跑的 Pipeline 節(jié)點鏈每個節(jié)點只負責一個被明確定義的子任務(wù)Agent 在其中扮演該環(huán)節(jié)的執(zhí)行者整條鏈路支持失敗重試和斷點續(xù)跑。這套方案的適用面很廣。無論是讓大模型批量處理幾十份文檔還是把一個“翻譯 → 潤色 → 校對”的流程穩(wěn)定跑完都能用得上。對個人開發(fā)者和后端工程來說它不需要重型調(diào)度系統(tǒng)一個 SQLite 文件外加一個循環(huán)就能實現(xiàn)很可靠的恢復機制。我在自己的某個內(nèi)部項目里用這套骨架跑了上百條多步驟任務(wù)最大的感受并不是“花了多少錢”而是“再也不用半夜爬起來盯著日志看它有沒有從頭再來一遍”。這篇文章就把這套架構(gòu)的拆解思路、重試機制、斷點續(xù)跑實現(xiàn)和實戰(zhàn)問題完整寫出來。1. 為什么要搞一套可重試、可續(xù)跑的多 Agent 流水線1.1 單次大模型調(diào)用解決復雜任務(wù)時的尷尬如果只是“請從這段文章里提取三個關(guān)鍵詞”這種一步任務(wù)直接調(diào)用大模型當然沒有任何問題。麻煩的是那種帶有明顯先后順序的復合任務(wù)比如“讀取原始報告 → 提取關(guān)鍵指標 → 生成摘要 → 整理成表格 → 寫一段結(jié)論”。把整個流程塞進一次調(diào)用不是不能跑而是風險極高。模型在長上下文里容易“選擇性失憶”前兩步還在遵守格式要求第三步就開始自由發(fā)揮輸出一個完全無法解析的結(jié)構(gòu)。上下文一旦逼近窗口上限調(diào)用可能直接失敗而且這個失敗往往發(fā)生在最后階段前面花費的 token 全部浪費。我做過一次很粗糙的實驗讓同一個模型完成一個五步驟的文本處理任務(wù)單次調(diào)用成功率勉強超過三成。失敗的原因五花八門有的是 JSON 解析失敗有的是輸出被截斷有的是中途開始胡編內(nèi)容。更麻煩的是當你只拿到一個失敗堆棧時幾乎不可能知道問題到底出在“提示詞說不清楚”“模型能力不夠”還是“輸入數(shù)據(jù)格式有問題”。這種黑盒式調(diào)試非常痛苦逼著我換一條路。1.2 串行流水線在 AI 任務(wù)里的價值串行流水線的思路其實是從 CI/CD 那里抄來的。軟件發(fā)布時編譯、測試、打包、部署是分階段的每個階段有獨立的日志、重試策略和產(chǎn)物。AI 任務(wù)完全可以照搬把一個大任務(wù)拆成若干節(jié)點按順序執(zhí)行前一個節(jié)點的輸出成為后一個節(jié)點的輸入。每個節(jié)點可以是一個大模型調(diào)用也可以是一個普通函數(shù)、一次外部 API 請求。這樣天然獲得三個好處隔離性——某一步崩潰不會把已經(jīng)算好的結(jié)果拖下水可觀測性——任何一步的輸入輸出都能被審計可恢復性——失敗后不必從頭開始。我常用一個工廠流水線來類比。零件在傳送帶上經(jīng)過 A、B、C 三個工位C 工位發(fā)現(xiàn)零件有問題只需要把零件退回 A 工位返工B、C 已完成的步驟根本不需要重做。AI Pipeline 也一樣如果第二步的模型超時第三步已經(jīng)生成的成果要繼續(xù)保留等第二步補跑成功后再接著往下走。長耗時、高成本的任務(wù)尤其受益畢竟每次全量重跑都意味著真金白銀的 token 消耗。1.3 這套方案適合誰、能解決什么問題如果你符合下面幾種情況之一我很推薦嘗試這套設(shè)計。第一種你經(jīng)常用大模型處理多步驟文檔任務(wù)比如“翻譯 → 潤色 → 風格統(tǒng)一”希望每一步失敗時能單獨重跑。第二種你的任務(wù)里有部分步驟依賴外部服務(wù)比如數(shù)據(jù)庫查詢、搜索 API、內(nèi)部 HTTP 接口這些服務(wù)偶爾超時或限流需要重試。第三種你需要向別人說明“這個最終結(jié)果是怎么一步步得到的”這時每個節(jié)點獨立的輸入輸出就是很好的審計記錄。這套方案對個人項目和小團隊很友好。它不需要引入分布式工作流引擎本地文件系統(tǒng)加少量代碼就能實現(xiàn)斷點續(xù)跑。等節(jié)點數(shù)量增加到幾十甚至上百個再考慮把 checkpoint 遷移到 Redis 或數(shù)據(jù)庫也不遲。核心能力“可重試、可續(xù)跑”與底層存儲無關(guān)重點在于狀態(tài)管理和節(jié)點契約的設(shè)計。2. 核心設(shè)計任務(wù)拆解與節(jié)點鏈模型2.1 任務(wù)拆解的原則粒度、邊界、依賴關(guān)系任務(wù)拆解最常見的兩個錯誤是拆得太粗和拆得太細。拆得太粗時一個節(jié)點里仍然塞了一大堆邏輯失敗后重試的范圍還是太大拆得太細時節(jié)點數(shù)量爆炸每個節(jié)點都要保存上下文的輸入輸出重復開銷反而壓過收益。我的判斷標準是如果某個節(jié)點只依賴上一個節(jié)點的輸出并且它的輸出能被下一個節(jié)點直接使用那么粒度基本合適。如果節(jié)點內(nèi)部需要維護額外狀態(tài)比如多次累積計數(shù)、跨步驟的臨時變量說明這里應(yīng)該繼續(xù)拆分或者重新設(shè)計。具體拆解時我會從最終目標反推先列出為了得到最終結(jié)果必須生成哪些中間物。舉一個“生成產(chǎn)品周報”的例子第一步從數(shù)據(jù)庫導出本周數(shù)據(jù)第二步把數(shù)據(jù)整理成指標表格第三步根據(jù)表格寫總結(jié)第四步排版成 Markdown。為什么要單獨拆“生成表格”和“寫總結(jié)”因為總結(jié)環(huán)節(jié)可能要用更強的模型或者需要多次重試而表格生成屬于機械轉(zhuǎn)換失敗概率小重試策略可以更激進。不同的節(jié)點各配各的參數(shù)這就是拆分帶來的靈活性。節(jié)點之間只通過輸入輸出契約互動這個約束必須寫進代碼里。我在代碼中禁止節(jié)點之間訪問彼此的運行上下文所有數(shù)據(jù)都通過統(tǒng)一接口傳遞。這條規(guī)則能讓流水線保持簡單也保證了斷點續(xù)跑時上下文重建完全可預測。2.2 節(jié)點與鏈路的抽象輸入輸出契約、Agent 封裝為了統(tǒng)一管理我把每個節(jié)點定義成下面的數(shù)據(jù)結(jié)構(gòu)from dataclasses import dataclass, field dataclass class PipelineNode: id: str agent_type: str # llm, function, api input_schema: dict output_schema: dict params: dict field(default_factorydict) max_retries: int 3這里的id是節(jié)點在流水線中的唯一標識agent_type決定執(zhí)行方式。input_schema和output_schema用來描述這個節(jié)點期望收到什么、會輸出什么。例如一個“生成摘要”節(jié)點輸入可能是{instructions: str, text: str}輸出可能是{summary: str, keywords: list}。params保存這個節(jié)點特有的參數(shù)比如模型名、溫度、超時。max_retries控制重試次數(shù)。我還會把一條流水線定義成一個按順序排列的PipelineNode列表。串行意味著節(jié)點的next_node_id隱含在列表順序里不需要額外聲明依賴關(guān)系。這樣做的缺點是靈活性差但換來了極低的認知負擔。在真實場景中絕大多數(shù)任務(wù)都是天然的線性流程先把它跑通比盲目追求 DAG 編排更重要。2.3 為什么選擇串行而不是編排、并行有人會問市面上有那么多工作流編排工具支持并行分支、條件判斷為什么偏要用這么原始的串行方式我的回答是對于“生成 → 校驗 → 再生成”這類任務(wù)串行已經(jīng)足夠而且更不容易出錯。翻譯之后才潤色潤色之后才校對這是強順序依賴。如果強行并行還要額外處理分支合并和結(jié)果一致性檢查復雜度一下子拉高好幾倍。但這不代表排斥并行。如果任務(wù)里有完全獨立的子問題比如同時請求多個數(shù)據(jù)源我會先并行拉取再把所有采集結(jié)果合并成一個 JSON 灌回主鏈路。整個任務(wù)依然是一條主串行鏈并行只發(fā)生在“采集節(jié)點”內(nèi)部。這樣斷點續(xù)跑時調(diào)度器只需要關(guān)注主鏈的節(jié)點狀態(tài)不用處理復雜的并行分支恢復邏輯。3. 可重試機制的落地細節(jié)3.1 失敗的分類與錯誤分級重試之前的頭等大事是區(qū)分“值得重試的失敗”和“不值得重試的失敗”。如果不做區(qū)分重試機制反而會把問題放大比如網(wǎng)絡(luò)超時重試兩三次可能就成功了但輸入數(shù)據(jù)為空這種確定性錯誤重試一萬次也救不回來。我通常把錯誤分成三類基礎(chǔ)設(shè)施類錯誤網(wǎng)絡(luò)超時、服務(wù)限流、臨時 5xx這類錯誤重試很有效。模型業(yè)務(wù)類錯誤JSON 解析失敗、輸出不符合 schema、內(nèi)容質(zhì)量不合格這類要具體分析有時重試能緩解有時要改提示詞。確定性錯誤必填字段缺失、參數(shù)非法、數(shù)據(jù)不存在這類重試沒有任何意義應(yīng)該立刻終止。在代碼里我會讓所有 Agent 包裝器把底層異常統(tǒng)一拋成自定義異常并帶上retryable標志。執(zhí)行器拿到異常后根據(jù)標志決定是否進入重試循環(huán)。這樣分層之后重試策略看起來清晰很多。3.2 重試策略次數(shù)、退避、超時附參數(shù)示例我常用的重試策略是每個節(jié)點默認最多重試 3 次初始等待 1 秒之后按指數(shù)增長每次乘以 2同時加入 0 到 200 毫秒的隨機抖動。超時根據(jù)節(jié)點耗時靈活設(shè)置簡單 LLM 調(diào)用給 30 秒長文本分析給 120 秒外部 API 額外設(shè)置連接超時和讀超時。這是我在 Python 里常用的裝飾器寫法import time import random from functools import wraps class RetryableError(Exception): pass class FatalError(Exception): pass def retry_policy(max_retries3, base_interval1.0, max_interval60.0, jitter0.2): def decorator(func): wraps(func) def wrapper(*args, **kwargs): attempt 0 while True: try: return func(*args, **kwargs) except RetryableError as e: attempt 1 if attempt max_retries: raise interval min(max_interval, base_interval * (2 ** attempt)) interval random.uniform(0, jitter) time.sleep(interval) return wrapper return decorator加抖動的目的很實際多個節(jié)點幾乎同時失敗時如果不加抖動下一輪重試還是會擠在同一個時間點再次觸發(fā)服務(wù)端限流。指數(shù)退避的底數(shù)我選 2間隔增長不會太慢整個任務(wù)不會等太久。如果你對成本極度敏感可以把底數(shù)改為 3代價是用戶等待時間變長。參數(shù)沒有絕對標準建議先從小值開始觀察重試成功率再調(diào)整。3.3 冪等性與狀態(tài)處理讓節(jié)點可以安全重跑重試機制有一個容易忽略的副作用節(jié)點重跑時可能重復執(zhí)行有外部影響的操作。比如某個節(jié)點不僅調(diào)用大模型還會把結(jié)果寫入數(shù)據(jù)庫第一次重試前寫入了一部分第二次重試又寫一遍數(shù)據(jù)就重復了。為了解決這個問題我為每個節(jié)點執(zhí)行單元引入了一個全局唯一的execution_id在節(jié)點開始前生成寫入 checkpoint 狀態(tài)然后才開始執(zhí)行。具體流程是節(jié)點要執(zhí)行任何外部寫操作之前先檢查該execution_id的狀態(tài)。如果狀態(tài)是 done說明之前已經(jīng)成功過直接讀取之前的輸出返回如果狀態(tài)是 running 但之前殘留了部分寫入則需要先清理再重新執(zhí)行。大模型調(diào)用本身是相對純凈的“生成型”操作重復調(diào)用不會產(chǎn)生外部副作用但會多花 token因此我在策略上會把 LLM 節(jié)點的重試次數(shù)控制得很低避免無效燒錢。冪等性的核心不復雜但很容易被忽略。尤其是當你第一次寫這套系統(tǒng)時會想當然地認為“失敗就是干凈地失敗”現(xiàn)實往往是“失敗發(fā)生在寫入一半之后”。所以我會建議所有涉及外部寫操作的節(jié)點都要把“寫入前登記”和“寫入后標記”當成強制要求。4. 斷點續(xù)跑的 Pipeline 調(diào)度實現(xiàn)4.1 節(jié)點狀態(tài)機的設(shè)計pending/running/done/failed/skipped斷點續(xù)跑的前提是每個節(jié)點都有明確、可持久化的狀態(tài)。狀態(tài)機的設(shè)計盡量簡單我只使用五種狀態(tài)pending還沒開始執(zhí)行running正在執(zhí)行中done已經(jīng)成功完成輸出已保存failed重試次數(shù)用盡最終失敗skipped條件不滿足或前置節(jié)點失敗選擇跳過狀態(tài)遷移是嚴格單向的pending可以到running、failed或skippedrunning可以到done或failed?；謴蜁r調(diào)度器只關(guān)心第一個不是done和skipped的節(jié)點。這里有一條鐵律done狀態(tài)必須和實際輸出保持一致不允許出現(xiàn)狀態(tài)是 done 但輸出為空的情況。我在保存狀態(tài)和輸出時把它寫在同一個事務(wù)里要么都成功要么都回滾。4.2 持久化 checkpoint什么信息必須記錄checkpoint 我默認用 SQLite 存儲因為它輕量、支持事務(wù)而且恢復時不需要全部加載到內(nèi)存。表結(jié)構(gòu)大致長這樣CREATE TABLE pipeline_state ( pipeline_id TEXT NOT NULL, node_id TEXT NOT NULL, status TEXT NOT NULL, attempt_count INTEGER DEFAULT 0, input_ref TEXT, output_ref TEXT, error_message TEXT, updated_at TEXT, PRIMARY KEY (pipeline_id, node_id) );這里的input_ref和output_ref可以是指向具體內(nèi)容的 ID也可以指向外部文件路徑。我強烈建議不要直接把幾萬字的大模型輸出全文塞進 checkpoint。之前我犯過這個錯誤一個二十節(jié)點的流水線跑完checkpoint 文件超過 200MB恢復一次要好幾秒。后來改成輸出超過 10KB 就存成外部文件checkpoint 里只記錄路徑恢復速度立刻正常。另一個經(jīng)驗是每個節(jié)點完成之后必須立刻把它的輸出快照持久化后續(xù)節(jié)點讀取輸入時統(tǒng)一從快照加載而不是從內(nèi)存變量拿數(shù)據(jù)。這個規(guī)則聽起來像廢話但在進程被殺、容器重啟等場景下少了這一步就是斷點續(xù)跑失效的根源。我現(xiàn)在所有節(jié)點讀取輸入都通過checkpoint.load_input(node_id)從不直接引用內(nèi)存對象。4.3 調(diào)度器如何從斷點恢復并跳過已完成節(jié)點調(diào)度器的核心邏輯比想象中簡單。我把它寫成一個循環(huán)遍歷節(jié)點列表讀取節(jié)點狀態(tài)如果狀態(tài)是done或skipped就跳過否則執(zhí)行。真正復雜的是處理running狀態(tài)如果上一次進程在執(zhí)行中途被殺節(jié)點狀態(tài)停留在running恢復時要先把它改回pending再交給重試機制處理。偽代碼大致如下def run_pipeline(pipeline, checkpoint, resume_fromNone): start_index 0 if resume_from: start_index pipeline.index_of(resume_from) for node in pipeline.nodes[start_index:]: state checkpoint.get_node_state(node.id) if state and state.status in (done, skipped): continue if state and state.status running: checkpoint.mark_pending(node.id) input_data checkpoint.build_node_input(node.id, pipeline) try: result execute_node_with_retry(node, input_data, checkpoint) checkpoint.save_output(node.id, result) checkpoint.mark_done(node.id) except FatalError: checkpoint.mark_failed(node.id) raisebuild_node_input會從 checkpoint 中讀取所有前置節(jié)點的輸出然后按照當前節(jié)點的input_schema重組為字典。如果前置節(jié)點沒有輸出它直接拋錯避免下游拿到殘缺數(shù)據(jù)。整個恢復流程就是不斷循環(huán)這個函數(shù)直到遇見一個failed節(jié)點或跑完全部節(jié)點。5. 實操一個可參考的多 Agent 串行 Pipeline 骨架5.1 整體架構(gòu)與目錄結(jié)構(gòu)我實際使用的項目骨架長這樣pipeline_runner/ ├── models.py # Pipeline、PipelineNode、ExecutionState 等數(shù)據(jù)結(jié)構(gòu) ├── checkpoint.py # SQLite 持久化 ├── retry.py # 重試策略 ├── executor.py # 節(jié)點執(zhí)行與調(diào)度 ├── agent_wrappers.py # LLM、函數(shù)、API 的統(tǒng)一包裝器 └── cli.py # 命令行入口這個結(jié)構(gòu)不追求分布式只保證在單進程內(nèi)實現(xiàn)斷點續(xù)跑。checkpoint.py負責所有持久化executor.py負責調(diào)度循環(huán)agent_wrappers.py負責把不同類型的 Agent 封裝成統(tǒng)一接口。如果以后需要多機并行只需要把checkpoint換成 Redis 或數(shù)據(jù)庫然后讓多個 worker 搶占pending節(jié)點即可核心狀態(tài)機不用變。5.2 節(jié)點定義與 Agent 接入示例我寫一個最簡單的 LLM Agent 包裝器核心思路就是“接收一個 dict返回一個 dict”。模型調(diào)用、超時、解析 JSON 都藏在包裝器內(nèi)部節(jié)點定義不關(guān)心這些細節(jié)。# agent_wrappers.py class LLMWrapper: def __init__(self, modeldefault-model, temperature0.2, timeout120): self.model model self.temperature temperature self.timeout timeout def run(self, input_data: dict) - dict: messages [ {role: system, content: input_data[instructions]}, {role: user, content: input_data[text]}, ] raw call_model( modelself.model, messagesmessages, temperatureself.temperature, timeoutself.timeout, ) return parse_json_response(raw)節(jié)點定義則專注于輸入輸出和參數(shù)node_summary PipelineNode( idsummary, agent_typellm, input_schema{instructions: str, text: str}, output_schema{summary: str, keywords: list}, params{model: fast-model, temperature: 0.3}, max_retries3, )這里最關(guān)鍵的是input_schema和output_schema。它們既是文檔也是約束執(zhí)行器在調(diào)用 Agent 前會校驗傳入字段是否齊全在 Agent 返回后校驗字段類型是否正確。schema 一出問題馬上就能定位到具體節(jié)點而不是含糊地說“模型不行”。我用 pydantic 做實際校驗但簡單的isinstance判斷也夠用。5.3 重試與斷點續(xù)跑的完整流程執(zhí)行器executor.py里的核心邏輯我已經(jīng)在 4.3 展示過這里補上更完整的重試交互。每個節(jié)點執(zhí)行前會先檢查 checkpoint 里有沒有殘留的retry_wait狀態(tài)如果上一次在等待重試時進程被殺恢復后會直接重新計算等待時間而不是再次執(zhí)行模型調(diào)用。這個細節(jié)保證了重試語義不會因為中斷而混亂。下面是我實際使用的execute_node_with_retry簡化版def execute_node_with_retry(node, input_data, checkpoint): attempt 0 while True: try: agent create_agent(node.agent_type, node.params) result agent.run(input_data) checkpoint.record_attempt(node.id, attempt, statussuccess) return result except FatalError as e: checkpoint.record_attempt(node.id, attempt, statusfatal_error, errorstr(e)) raise except RetryableError as e: attempt 1 if attempt node.max_retries: checkpoint.record_attempt(node.id, attempt, statusfailed, errorstr(e)) raise wait_time compute_backoff(attempt, basenode.params.get(retry_base, 1)) checkpoint.record_attempt(node.id, attempt, statusretry_wait, errorstr(e)) time.sleep(wait_time)在每次time.sleep之前我都先把retry_wait狀態(tài)記錄下來。因為 sleep 期間進程可能被殺恢復時能明確知道當前處于“等待重試”階段而不是“執(zhí)行中”。我在實踐中發(fā)現(xiàn)這種顯式記錄對排查問題非常有幫助。看 checkpoint 表格就能知道某節(jié)點是模型調(diào)用慢還是在等待退避還是真的執(zhí)行失敗了。5.4 極簡命令行工具實現(xiàn)命令行入口讓整個流水線變成一個可以獨立運行的進程。我常用的命令是python cli.py run --pipeline configs/report_pipeline.json --checkpoint ./checks.db python cli.py resume --pipeline configs/report_pipeline.json --checkpoint ./checks.dbrun從頭開始執(zhí)行resume加載已有 checkpoint從第一個非 done 狀態(tài)開始。--pipeline參數(shù)指向一個 JSON 文件描述節(jié)點列表--checkpoint指向 SQLite 文件名。CLI 啟動時先加載配置初始化 checkpoint再調(diào)用executor.run_pipeline。這套骨架看起來簡單但其實已經(jīng)能支撐真實任務(wù)。我不少項目就是靠它批量跑了幾百份材料中途斷網(wǎng)、API 限流、進程被殺只要補一條resume命令流水線還能從斷點繼續(xù)。6. 常見問題與排查技巧實錄6.1 重試導致重復副作用、費用飆升怎么避免一位朋友用了我的思路后反饋他的第四步是發(fā)送通知郵件模型輸出內(nèi)容后調(diào)用郵件 API 失敗重試時重復發(fā)送了兩封郵件。原因就是把發(fā)送郵件當成了普通節(jié)點直接重試沒有做冪等處理。我在 3.3 里提到過execution_id這里再重申一遍所有有外部副作用的節(jié)點執(zhí)行前必須先登記執(zhí)行成功后必須標記。重試時如果發(fā)現(xiàn)該execution_id已經(jīng)成功直接返回成功結(jié)果不再執(zhí)行副作用。更好的做法是把“生成內(nèi)容”和“執(zhí)行副作用”拆成兩個節(jié)點。生成郵件內(nèi)容屬于純模型操作失敗可以隨便重試發(fā)送郵件屬于外部副作用一旦成功就永久記錄。這樣既可以利用重試修復內(nèi)容質(zhì)量問題又不會重復發(fā)郵件。費用方面如果某個節(jié)點反復因模型輸出格式問題失敗建議先停機檢查提示詞而不是一味加大重試次數(shù)。重試不是萬靈藥它只對瞬時故障有效。6.2 checkpoint 文件太大或太碎恢復慢怎么辦checkpoint 變成性能瓶頸通常是存了太多大文本。我最早遇到的問題是每個節(jié)點的完整輸出都存成一個文件節(jié)點多之后文件碎片化嚴重恢復時需要遍歷大量文件。后來我做了兩件事一是把輸出按大小分流小對象直接存在 SQLite BLOB 字段大對象存到outputs/目錄checkpoint 只記錄引用二是定期合并 checkpoint比如用一次新快照替換歷史狀態(tài)。SQLite 還有一個額外好處支持按行讀取不需要反序列化整個文件。如果節(jié)點數(shù)量達到幾百直接在pipeline_state表上加索引查詢速度依然不錯。如果未來checkpoint 落到 Redis可以考慮用哈希表按節(jié)點存儲恢復時只拉取第一個未完成節(jié)點附近的少量狀態(tài)效率和 SQLite 類似。6.3 節(jié)點之間上下文太長導致成本膨脹的優(yōu)化這是非常隱蔽的問題。每個節(jié)點都拿到上游節(jié)點的完整輸出然后拼進提示詞越到后面上下文越長token 費用自然飆升。比如第一步生成了一份兩萬字的文檔第二步只是提取標題卻把全文重新傳給模型一次這就很浪費。更麻煩的是上下文太長還可能引發(fā)模型“走神”降低輸出質(zhì)量。我建議的優(yōu)化是兩個方向并行。第一精確裁剪利用input_schema定義下游真正需要的字段執(zhí)行器在build_node_input時只提取這些字段而不是把整個輸出對象塞過去。第二增加摘要節(jié)點如果下游確實需要全文概覽先對長文本做分段摘要再把摘要作為下游輸入。這兩個方向配合起來通常能削減 60% 以上的上下文體積我實測五節(jié)點流水線的成本能下降將近一半。6.4 邊界情況任務(wù)中途被用戶終止、版本變更兼容運行流水線時用戶可能會按 CtrlC或者部署新版本后想繼續(xù)舊 checkpoint。如果不在代碼層面對這種情況做處理恢復時很容易出現(xiàn)狀態(tài)錯亂。我的做法有兩點一是 checkpoint 表里加一個schema_version字段每次升級代碼時手動遞增?；謴蜁r發(fā)現(xiàn)版本不匹配跑一個遷移函數(shù)把舊的節(jié)點輸出結(jié)構(gòu)轉(zhuǎn)成新版本兼容的結(jié)構(gòu)。無法遷移時提示用戶選擇強制重跑受影響節(jié)點。二是在執(zhí)行循環(huán)外層捕獲KeyboardInterrupt先把當前節(jié)點標記為running表示執(zhí)行到一半。下次 resume 時會把running改回pending再繼續(xù)。這樣雖然不能保證完全無差錯但至少能把斷點定位到具體節(jié)點不會讓整個任務(wù)從零再來。另外要提醒的是如果你改了某個已done節(jié)點的提示詞或者 Agent 參數(shù)舊的 checkpoint 里已經(jīng)保存了它的輸出下次 resume 時會直接跳過你的修改不會生效。這是“斷點續(xù)跑”的一種副作用。我解決的辦法是每次調(diào)整節(jié)點配置后把對應(yīng)節(jié)點的狀態(tài)重置為pending或者用一個force參數(shù)強制重跑指定節(jié)點。我在實際使用這套架構(gòu)之后最明顯的感覺是心不慌了。以前遇到長任務(wù)失敗只能心疼地看著它從頭再燒一遍 token現(xiàn)在無論哪個節(jié)點中斷我都能準確說出失敗原因然后一條resume命令讓它接著斷點走。設(shè)計 Pipeline 的過程也是在訓練自己把復雜的 AI 任務(wù)拆解成更小、更可控的單位。如果你也被“一步到位”式的大模型調(diào)用坑過不妨用這套思路試一次。