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