1. 為什么“Agent 編排”值得單獨(dú)拎出來(lái)講1.1 從一個(gè)真實(shí)踩坑場(chǎng)景說(shuō)起去年下半年我接手了一個(gè)內(nèi)部工具鏈的改造項(xiàng)目目標(biāo)是把幾個(gè)獨(dú)立的小助手串成一條能自動(dòng)跑通的流水線一個(gè)負(fù)責(zé)讀需求文檔一個(gè)負(fù)責(zé)拆任務(wù)一個(gè)負(fù)責(zé)寫(xiě)代碼草稿最后一個(gè)負(fù)責(zé)跑測(cè)試并回傳結(jié)果。聽(tīng)起來(lái)不復(fù)雜對(duì)吧我一開(kāi)始也是這么想的直接寫(xiě)了個(gè)主函數(shù)按順序調(diào)用四個(gè)模塊中間用字符串拼接傳參。結(jié)果第一版跑通只花了半天后面修bug修了整整兩周。問(wèn)題出在哪出在“編排”這兩個(gè)字上。單個(gè)助手的能力其實(shí)都還行但一旦把它們連起來(lái)各種邊界情況就冒出來(lái)了上游輸出格式變了下游解析直接崩某個(gè)環(huán)節(jié)超時(shí)了整條鏈路卡死兩個(gè)環(huán)節(jié)都想調(diào)用同一個(gè)資源互相搶鎖更別提中間某一步失敗之后到底該重試、該跳過(guò)、還是該回滾根本沒(méi)有統(tǒng)一策略。我那時(shí)候才意識(shí)到Agent 編排不是簡(jiǎn)單的函數(shù)調(diào)用鏈它是一套獨(dú)立的工程問(wèn)題。這篇文章就想把這件事講透。不管你是剛接觸多智能體協(xié)作的新手還是已經(jīng)寫(xiě)過(guò)幾版編排邏輯但總覺(jué)得別扭的老手我都希望這里的內(nèi)容能讓你少走點(diǎn)彎路。我會(huì)從設(shè)計(jì)思路、核心機(jī)制、落地實(shí)現(xiàn)、問(wèn)題排查四個(gè)維度展開(kāi)盡量把每個(gè)決策背后的“為什么”說(shuō)清楚而不是只丟一堆代碼讓你抄。1.2 編排到底在編排什么先把概念對(duì)齊。所謂 Agent 編排指的是把多個(gè)具備獨(dú)立決策能力的智能體按照某種控制邏輯組織起來(lái)協(xié)同完成一個(gè)超出單個(gè)智能體能力的任務(wù)。注意這里有兩個(gè)關(guān)鍵詞獨(dú)立決策、控制邏輯。獨(dú)立決策意味著每個(gè) Agent 有自己的輸入輸出、有自己的工具集、甚至有自己的“脾氣”比如有的擅長(zhǎng)發(fā)散、有的擅長(zhǎng)收斂?？刂七壿媱t決定了它們之間怎么傳數(shù)據(jù)、怎么定順序、怎么處理異常、怎么共享狀態(tài)。很多人把編排等同于“寫(xiě)個(gè)循環(huán)依次調(diào)用”這就把問(wèn)題想簡(jiǎn)單了。真正的編排要解決的是在不確定的執(zhí)行環(huán)境下如何讓一組不確定的組件穩(wěn)定地產(chǎn)生確定的結(jié)果。這個(gè)定義聽(tīng)起來(lái)有點(diǎn)繞但你可以這樣理解單個(gè) Agent 像是一個(gè)員工編排則是管理這群?jiǎn)T工的制度。員工能力再?gòu)?qiáng)制度混亂項(xiàng)目照樣黃。反過(guò)來(lái)制度設(shè)計(jì)得好哪怕單個(gè)員工能力一般整體產(chǎn)出也能很穩(wěn)。這就是為什么我說(shuō)編排值得單獨(dú)拎出來(lái)講——它是多智能體系統(tǒng)從“能跑”到“能上線”之間那道最關(guān)鍵的坎。1.3 本文適合誰(shuí)看如果你屬于下面幾類人這篇內(nèi)容應(yīng)該對(duì)你有用正在做多智能體協(xié)作項(xiàng)目但被狀態(tài)傳遞、異常處理搞得頭大的開(kāi)發(fā)者想從單 Agent 應(yīng)用升級(jí)到多 Agent 系統(tǒng)不確定該怎么設(shè)計(jì)控制流的工程師對(duì)編排框架選型猶豫不決想搞清楚各種方案取舍邏輯的技術(shù)負(fù)責(zé)人單純想了解“Agent 編排”到底是怎么回事為后續(xù)學(xué)習(xí)打基礎(chǔ)的好奇者。我會(huì)盡量不假設(shè)你用過(guò)某個(gè)特定框架因?yàn)榫幣诺暮诵乃枷胧强缈蚣芡ㄓ玫摹Ｄ憷斫饬诉@套邏輯換任何工具都能快速上手。2. 編排方案的整體設(shè)計(jì)思路2.1 先想清楚你的任務(wù)是什么形狀在動(dòng)手寫(xiě)任何編排代碼之前我強(qiáng)烈建議你先做一件事把任務(wù)畫(huà)出來(lái)。不是畫(huà)流程圖那種正式的東西就是拿張紙把每個(gè)環(huán)節(jié)的輸入、輸出、依賴關(guān)系、可能的失敗點(diǎn)標(biāo)出來(lái)。這一步花十分鐘能省后面十小時(shí)的調(diào)試。任務(wù)大致可以分成幾種形狀。第一種是線性鏈A 做完給 BB 做完給 C像流水線一樣。這種最簡(jiǎn)單但也是最容易出問(wèn)題的因?yàn)槿魏我画h(huán)卡住整條鏈就停了。第二種是扇出扇入一個(gè)環(huán)節(jié)產(chǎn)出多個(gè)結(jié)果分發(fā)給多個(gè)下游并行處理最后再匯總。這種適合可以并行拆分的任務(wù)但匯總邏輯往往很麻煩。第三種是帶環(huán)的圖某個(gè)環(huán)節(jié)的結(jié)果可能觸發(fā)上游重新執(zhí)行形成迭代。這種最靈活也最難控制因?yàn)槟阋乐顾姥h(huán)。我踩過(guò)的坑是一開(kāi)始把任務(wù)想成線性鏈寫(xiě)了個(gè)順序執(zhí)行器結(jié)果后來(lái)發(fā)現(xiàn)某個(gè)環(huán)節(jié)需要根據(jù)結(jié)果決定是否回退硬生生把線性結(jié)構(gòu)改成了圖結(jié)構(gòu)代碼重構(gòu)了一大半。所以在動(dòng)手前把任務(wù)形狀想清楚比選什么框架重要得多。2.2 控制流與數(shù)據(jù)流要分開(kāi)設(shè)計(jì)這是我最想強(qiáng)調(diào)的一點(diǎn)。很多初學(xué)者會(huì)把控制流和數(shù)據(jù)流混在一起比如用“上一步的輸出”直接作為“下一步是否執(zhí)行的判斷依據(jù)”。這樣做短期看很省事長(zhǎng)期看是災(zāi)難。正確的做法是控制流決定“誰(shuí)在什么時(shí)候執(zhí)行”數(shù)據(jù)流決定“執(zhí)行時(shí)拿到什么數(shù)據(jù)”。兩者通過(guò)一個(gè)共享的狀態(tài)容器來(lái)交互但邏輯上要解耦。舉個(gè)例子A 環(huán)節(jié)產(chǎn)出了一個(gè)結(jié)果控制流關(guān)心的是“這個(gè)結(jié)果是否滿足繼續(xù)條件”數(shù)據(jù)流關(guān)心的是“這個(gè)結(jié)果怎么傳給 B”。如果混在一起當(dāng)你想改判斷條件時(shí)可能會(huì)不小心改動(dòng)了數(shù)據(jù)傳遞格式引發(fā)連鎖 bug。我現(xiàn)在的習(xí)慣是在編排層維護(hù)一個(gè)上下文對(duì)象所有環(huán)節(jié)的輸入輸出都往里面讀寫(xiě)但每個(gè)環(huán)節(jié)只聲明自己需要讀哪些字段、寫(xiě)哪些字段?？刂屏鲃t單獨(dú)維護(hù)一張狀態(tài)機(jī)記錄每個(gè)環(huán)節(jié)的執(zhí)行狀態(tài)待執(zhí)行、執(zhí)行中、成功、失敗、跳過(guò)。這樣改控制邏輯不會(huì)影響數(shù)據(jù)格式改數(shù)據(jù)格式也不會(huì)影響控制邏輯。2.3 同步、異步還是事件驅(qū)動(dòng)編排的執(zhí)行模型有三種主流選擇各有各的適用場(chǎng)景。同步阻塞式最簡(jiǎn)單一個(gè)環(huán)節(jié)執(zhí)行完才執(zhí)行下一個(gè)。優(yōu)點(diǎn)是邏輯清晰、調(diào)試方便缺點(diǎn)是效率低遇到耗時(shí)環(huán)節(jié)整個(gè)鏈路都在等。適合環(huán)節(jié)少、耗時(shí)短、對(duì)實(shí)時(shí)性要求不高的場(chǎng)景。異步非阻塞式允許環(huán)節(jié)并行執(zhí)行通過(guò)回調(diào)或 Future 來(lái)協(xié)調(diào)。效率高但復(fù)雜度也高尤其是錯(cuò)誤傳播和狀態(tài)同步會(huì)變得很棘手。適合環(huán)節(jié)多、可以并行、對(duì)吞吐量有要求的場(chǎng)景。事件驅(qū)動(dòng)式把每個(gè)環(huán)節(jié)看作事件的消費(fèi)者和生產(chǎn)者通過(guò)事件總線解耦。靈活性最高擴(kuò)展性最好但調(diào)試難度也最大因?yàn)閳?zhí)行路徑不是顯式寫(xiě)出來(lái)的而是運(yùn)行時(shí)動(dòng)態(tài)決定的。適合環(huán)節(jié)動(dòng)態(tài)變化、需要熱插拔的場(chǎng)景。我的經(jīng)驗(yàn)是除非有明確的性能瓶頸否則優(yōu)先選同步阻塞式。先把邏輯跑通等真的遇到性能問(wèn)題了再針對(duì)性地把耗時(shí)環(huán)節(jié)改成異步。過(guò)早引入異步和事件驅(qū)動(dòng)往往會(huì)讓簡(jiǎn)單問(wèn)題復(fù)雜化。我見(jiàn)過(guò)太多項(xiàng)目一開(kāi)始就上事件驅(qū)動(dòng)結(jié)果調(diào)試一個(gè)簡(jiǎn)單 bug 要翻十幾個(gè)日志文件。2.4 狀態(tài)管理編排的心臟狀態(tài)管理是編排里最容易做砸的部分。我見(jiàn)過(guò)幾種典型的錯(cuò)誤做法把狀態(tài)存在全局變量里、把狀態(tài)序列化到文件里每次讀寫(xiě)、把狀態(tài)分散在各個(gè)環(huán)節(jié)內(nèi)部靠參數(shù)傳遞。全局變量的問(wèn)題是并發(fā)不安全多個(gè)環(huán)節(jié)同時(shí)讀寫(xiě)會(huì)互相污染。文件序列化的問(wèn)題是性能差而且容易產(chǎn)生不一致的中間狀態(tài)。參數(shù)傳遞的問(wèn)題是鏈路一長(zhǎng)參數(shù)列表就爆炸了而且中間環(huán)節(jié)可能根本不需要某些參數(shù)卻被迫透?jìng)?。我推薦的方案是集中式狀態(tài)容器 不可變快照。所有狀態(tài)存在一個(gè)容器里每個(gè)環(huán)節(jié)執(zhí)行前拿一份當(dāng)前狀態(tài)的只讀快照?qǐng)?zhí)行后提交一份變更。這樣既保證了并發(fā)安全又方便做回滾和審計(jì)。具體實(shí)現(xiàn)可以用一個(gè)簡(jiǎn)單的鍵值存儲(chǔ)加上版本號(hào)機(jī)制。每次提交變更時(shí)版本號(hào)加一回滾時(shí)直接恢復(fù)到指定版本。注意狀態(tài)容器里不要存大對(duì)象比如完整的文件內(nèi)容或圖片二進(jìn)制。存引用或路徑就行大對(duì)象單獨(dú)管理。否則狀態(tài)快照會(huì)變得非常大影響性能。2.5 異常處理策略要提前定異常處理是編排設(shè)計(jì)里最容易被忽視、又最不能忽視的部分。我建議在動(dòng)手前就明確幾個(gè)策略重試策略哪些環(huán)節(jié)失敗可以重試重試幾次重試間隔怎么算我的經(jīng)驗(yàn)是對(duì)于調(diào)用外部服務(wù)的環(huán)節(jié)重試三次、間隔指數(shù)退避是比較穩(wěn)妥的默認(rèn)值。對(duì)于純計(jì)算環(huán)節(jié)重試通常沒(méi)意義直接失敗更快。降級(jí)策略某個(gè)環(huán)節(jié)失敗后是否有備選方案比如主模型調(diào)用失敗是否切換到備用模型降級(jí)邏輯要顯式寫(xiě)出來(lái)不要指望運(yùn)行時(shí)自動(dòng)發(fā)現(xiàn)。熔斷策略如果某個(gè)環(huán)節(jié)連續(xù)失敗多次是否要暫時(shí)跳過(guò)它避免拖垮整條鏈路這在依賴外部服務(wù)的場(chǎng)景下特別重要。補(bǔ)償策略對(duì)于已經(jīng)產(chǎn)生副作用的環(huán)節(jié)比如寫(xiě)入了數(shù)據(jù)庫(kù)失敗后是否需要回滾這需要每個(gè)環(huán)節(jié)自己聲明是否支持補(bǔ)償操作。這些策略不要等到出問(wèn)題了再想那時(shí)候往往已經(jīng)來(lái)不及了。在編排層統(tǒng)一實(shí)現(xiàn)這些策略比在每個(gè)環(huán)節(jié)里各寫(xiě)一套要可靠得多。3. 核心機(jī)制與實(shí)操要點(diǎn)拆解3.1 環(huán)節(jié)抽象每個(gè) Agent 怎么接入編排層要能統(tǒng)一管理各種 Agent前提是每個(gè) Agent 都符合一個(gè)標(biāo)準(zhǔn)接口。這個(gè)接口不需要很復(fù)雜但必須包含幾個(gè)關(guān)鍵方法execute(context)執(zhí)行主體邏輯接收上下文返回結(jié)果validate(context)執(zhí)行前的校驗(yàn)檢查輸入是否滿足要求compensate(context)補(bǔ)償操作用于回滾副作用metadata()返回元信息比如名稱、版本、依賴、超時(shí)時(shí)間。我見(jiàn)過(guò)有人為了省事直接讓每個(gè) Agent 暴露一個(gè)run()方法其他什么都不管。結(jié)果編排層沒(méi)法做統(tǒng)一校驗(yàn)也沒(méi)法做補(bǔ)償異常處理只能靠 try-catch 包一層非常脆弱。多花半小時(shí)定義接口能省后面幾天的返工。接口定義好之后還需要一個(gè)注冊(cè)機(jī)制讓編排層知道有哪些 Agent 可用。簡(jiǎn)單的做法是用一個(gè)字典鍵是 Agent 名稱值是實(shí)例。復(fù)雜一點(diǎn)可以用插件式加載支持運(yùn)行時(shí)動(dòng)態(tài)注冊(cè)。我一般先用字典等真的有動(dòng)態(tài)需求了再升級(jí)。3.2 上下文傳遞數(shù)據(jù)怎么在環(huán)節(jié)間流動(dòng)上下文傳遞的核心問(wèn)題是下游環(huán)節(jié)怎么拿到上游環(huán)節(jié)的產(chǎn)出同時(shí)又不被上游的無(wú)關(guān)數(shù)據(jù)干擾。我的做法是給上下文加一個(gè)命名空間機(jī)制。每個(gè)環(huán)節(jié)寫(xiě)入數(shù)據(jù)時(shí)必須指定一個(gè)命名空間比如step_a.result。下游環(huán)節(jié)讀取時(shí)也明確指定要讀哪個(gè)命名空間。這樣既避免了鍵名沖突又讓數(shù)據(jù)依賴關(guān)系變得顯式。具體實(shí)現(xiàn)上上下文可以是一個(gè)支持嵌套讀寫(xiě)的字典。寫(xiě)入時(shí)用context.set(step_a.result, value)讀取時(shí)用context.get(step_a.result)。如果需要傳遞復(fù)雜結(jié)構(gòu)可以約定用 JSON 序列化但要注意序列化開(kāi)銷。還有一個(gè)細(xì)節(jié)上下文要不要在環(huán)節(jié)間傳遞引用還是拷貝。傳遞引用效率高但上游環(huán)節(jié)如果修改了數(shù)據(jù)下游會(huì)受影響。傳遞拷貝安全但大對(duì)象拷貝開(kāi)銷大。我的折中方案是小對(duì)象傳拷貝大對(duì)象傳引用但標(biāo)記為只讀。這樣兼顧了安全和效率。3.3 執(zhí)行調(diào)度誰(shuí)先誰(shuí)后怎么定調(diào)度邏輯決定了環(huán)節(jié)的執(zhí)行順序。最簡(jiǎn)單的調(diào)度是拓?fù)渑判蚋鶕?jù)依賴關(guān)系排出執(zhí)行序列。但實(shí)際場(chǎng)景往往比這復(fù)雜因?yàn)榭赡艽嬖跅l件分支、循環(huán)、并行。我一般把調(diào)度邏輯分成兩層靜態(tài)依賴圖 動(dòng)態(tài)決策器。靜態(tài)依賴圖在初始化時(shí)構(gòu)建記錄哪些環(huán)節(jié)必須在哪些環(huán)節(jié)之前執(zhí)行。動(dòng)態(tài)決策器在運(yùn)行時(shí)根據(jù)上下文決定下一步執(zhí)行哪個(gè)環(huán)節(jié)支持條件分支和循環(huán)。動(dòng)態(tài)決策器的實(shí)現(xiàn)可以很簡(jiǎn)單就是一個(gè)函數(shù)接收當(dāng)前上下文和已執(zhí)行環(huán)節(jié)列表返回下一個(gè)要執(zhí)行的環(huán)節(jié)名稱。返回空表示執(zhí)行結(jié)束。這樣調(diào)度邏輯和環(huán)節(jié)邏輯完全解耦改調(diào)度不影響環(huán)節(jié)改環(huán)節(jié)也不影響調(diào)度。提示動(dòng)態(tài)決策器里一定要加最大步數(shù)限制防止死循環(huán)。我一般設(shè)成環(huán)節(jié)數(shù)量的三到五倍超過(guò)就強(qiáng)制終止并報(bào)錯(cuò)。3.4 并發(fā)控制并行環(huán)節(jié)怎么協(xié)調(diào)當(dāng)多個(gè)環(huán)節(jié)可以并行執(zhí)行時(shí)需要協(xié)調(diào)它們的啟動(dòng)、等待和結(jié)果收集。最直接的方式是用線程池或協(xié)程池提交任務(wù)后等待所有任務(wù)完成。但并行會(huì)帶來(lái)幾個(gè)問(wèn)題。第一是資源競(jìng)爭(zhēng)多個(gè)環(huán)節(jié)同時(shí)讀寫(xiě)共享資源可能沖突。解決辦法是給資源加鎖或者讓環(huán)節(jié)聲明自己需要哪些資源調(diào)度層統(tǒng)一分配。第二是部分失敗如果并行任務(wù)中有一個(gè)失敗了其他任務(wù)要不要取消我的策略是默認(rèn)等待所有任務(wù)結(jié)束然后統(tǒng)一處理失敗除非失敗是致命的才主動(dòng)取消其他任務(wù)。第三是結(jié)果順序并行任務(wù)完成順序不確定但下游可能依賴特定順序。解決辦法是在收集結(jié)果時(shí)按環(huán)節(jié)名稱排序而不是按完成順序。我實(shí)測(cè)下來(lái)并行度不是越高越好。對(duì)于 IO 密集型環(huán)節(jié)并行度可以設(shè)高一些比如 CPU 核數(shù)的兩到四倍。對(duì)于計(jì)算密集型環(huán)節(jié)并行度設(shè)成 CPU 核數(shù)就行再高反而因?yàn)樯舷挛那袚Q變慢。3.5 日志與可觀測(cè)性出問(wèn)題怎么查編排系統(tǒng)一旦跑起來(lái)出問(wèn)題的時(shí)候如果沒(méi)有足夠的日志排查會(huì)非常痛苦。我的經(jīng)驗(yàn)是日志要記錄三類信息環(huán)節(jié)的開(kāi)始和結(jié)束、上下文的變更、異常和重試。環(huán)節(jié)開(kāi)始和結(jié)束的日志要包含環(huán)節(jié)名稱、時(shí)間戳、執(zhí)行時(shí)長(zhǎng)。上下文變更的日志要包含變更前后的值但要注意脫敏不要把敏感數(shù)據(jù)寫(xiě)進(jìn)日志。異常和重試的日志要包含異常類型、堆棧、重試次數(shù)。除了日志還建議加一個(gè)執(zhí)行軌跡功能把每個(gè)環(huán)節(jié)的輸入輸出、執(zhí)行狀態(tài)、耗時(shí)都記錄下來(lái)形成一個(gè)可查詢的軌跡。這樣出問(wèn)題時(shí)直接看軌跡就能定位到哪個(gè)環(huán)節(jié)出了問(wèn)題不用翻日志。軌跡可以存在內(nèi)存里也可以持久化到數(shù)據(jù)庫(kù)看需求。4. 完整實(shí)操流程與關(guān)鍵環(huán)節(jié)實(shí)現(xiàn)4.1 環(huán)境準(zhǔn)備與依賴安裝假設(shè)我們用 Python 來(lái)實(shí)現(xiàn)這套編排邏輯基礎(chǔ)環(huán)境需要 Python 3.9 以上因?yàn)橐玫揭恍┹^新的類型注解特性。依賴方面核心只需要標(biāo)準(zhǔn)庫(kù)但為了更好的開(kāi)發(fā)體驗(yàn)建議裝幾個(gè)輔助庫(kù)pip install pydantic typing-extensionspydantic用來(lái)做數(shù)據(jù)校驗(yàn)和序列化typing-extensions用來(lái)支持更豐富的類型注解。如果你需要異步支持可以再加asyncio不過(guò)它是標(biāo)準(zhǔn)庫(kù)自帶的不用額外裝。項(xiàng)目結(jié)構(gòu)建議這樣組織orchestrator/ __init__.py context.py # 上下文實(shí)現(xiàn) agent.py # Agent 接口定義 scheduler.py # 調(diào)度器實(shí)現(xiàn) executor.py # 執(zhí)行器實(shí)現(xiàn) exceptions.py # 自定義異常 agents/ # 具體 Agent 實(shí)現(xiàn) __init__.py reader.py planner.py coder.py tester.py這樣分層清晰后續(xù)擴(kuò)展也方便。4.2 上下文容器的實(shí)現(xiàn)先來(lái)實(shí)現(xiàn)上下文容器。核心需求是支持命名空間讀寫(xiě)、版本管理、快照。import copy from typing import Any, Dict, Optional class Context: def __init__(self): self._data: Dict[str, Any] {} self._version: int 0 self._snapshots: Dict[int, Dict[str, Any]] {} def set(self, key: str, value: Any) - None: self._data[key] value self._version 1 def get(self, key: str, default: Any None) - Any: return self._data.get(key, default) def snapshot(self) - int: self._snapshots[self._version] copy.deepcopy(self._data) return self._version def rollback(self, version: int) - None: if version not in self._snapshots: raise ValueError(fSnapshot version {version} not found) self._data copy.deepcopy(self._snapshots[version]) self._version version property def version(self) - int: return self._version這里有幾個(gè)設(shè)計(jì)決策值得說(shuō)明。第一set方法每次都會(huì)增加版本號(hào)這樣任何變更都可追蹤。第二snapshot方法會(huì)深拷貝當(dāng)前數(shù)據(jù)保證快照不受后續(xù)修改影響。第三rollback方法恢復(fù)到指定版本用于補(bǔ)償操作。注意深拷貝大對(duì)象開(kāi)銷很大如果上下文里存了大對(duì)象建議只存引用快照時(shí)只拷貝引用本身。或者用寫(xiě)時(shí)復(fù)制Copy-on-Write策略只在真正修改時(shí)才拷貝。4.3 Agent 接口與基類實(shí)現(xiàn)接下來(lái)定義 Agent 接口。我用抽象基類來(lái)實(shí)現(xiàn)強(qiáng)制子類實(shí)現(xiàn)必要方法。from abc import ABC, abstractmethod from typing import Any, Dict class BaseAgent(ABC): property abstractmethod def name(self) - str: pass abstractmethod def execute(self, context: Context) - Any: pass def validate(self, context: Context) - bool: return True def compensate(self, context: Context) - None: pass def metadata(self) - Dict[str, Any]: return { name: self.name, timeout: 30, retry: 3, compensable: False, }execute是必須實(shí)現(xiàn)的其他方法有默認(rèn)實(shí)現(xiàn)。metadata返回元信息調(diào)度器會(huì)根據(jù)這些信息決定超時(shí)、重試等策略。實(shí)現(xiàn)一個(gè)具體的 Agent 示例class ReaderAgent(BaseAgent): property def name(self) - str: return reader def execute(self, context: Context) - Any: doc_path context.get(input.doc_path) if not doc_path: raise ValueError(Missing doc_path in context) with open(doc_path, r, encodingutf-8) as f: content f.read() context.set(reader.content, content) return content def metadata(self) - Dict[str, Any]: return { name: self.name, timeout: 10, retry: 1, compensable: False, }這個(gè) Agent 讀取文檔內(nèi)容并寫(xiě)入上下文。注意它從上下文讀輸入也往上下文寫(xiě)輸出不通過(guò)參數(shù)傳遞這樣解耦得更徹底。4.4 調(diào)度器的實(shí)現(xiàn)調(diào)度器負(fù)責(zé)決定執(zhí)行順序。我實(shí)現(xiàn)一個(gè)基于依賴圖的調(diào)度器。from typing import List, Dict, Set class Scheduler: def __init__(self): self._dependencies: Dict[str, Set[str]] {} self._agents: Dict[str, BaseAgent] {} def register(self, agent: BaseAgent, depends_on: List[str] None): self._agents[agent.name] agent self._dependencies[agent.name] set(depends_on or []) def topological_sort(self) - List[str]: visited set() result [] def visit(node: str): if node in visited: return visited.add(node) for dep in self._dependencies.get(node, set()): visit(dep) result.append(node) for name in self._agents: visit(name) return result def next_agent(self, executed: Set[str]) - Optional[str]: for name in self.topological_sort(): if name in executed: continue deps self._dependencies.get(name, set()) if deps.issubset(executed): return name return Nonetopological_sort給出一個(gè)合法的執(zhí)行順序next_agent根據(jù)已執(zhí)行集合返回下一個(gè)可執(zhí)行的 Agent。這樣調(diào)度邏輯和 Agent 實(shí)現(xiàn)完全解耦。4.5 執(zhí)行器與異常處理執(zhí)行器負(fù)責(zé)實(shí)際調(diào)用 Agent并處理超時(shí)、重試、補(bǔ)償。import time from typing import Optional class Executor: def __init__(self, scheduler: Scheduler, context: Context): self.scheduler scheduler self.context context self.executed: Set[str] set() self.max_steps len(scheduler._agents) * 5 def run(self) - None: steps 0 while steps self.max_steps: steps 1 name self.scheduler.next_agent(self.executed) if name is None: break agent self.scheduler._agents[name] self._execute_with_retry(agent) self.executed.add(name) def _execute_with_retry(self, agent: BaseAgent) - None: meta agent.metadata() retry meta.get(retry, 1) timeout meta.get(timeout, 30) last_error None for attempt in range(retry): try: start time.time() if not agent.validate(self.context): raise ValueError(fValidation failed for {agent.name}) agent.execute(self.context) elapsed time.time() - start if elapsed timeout: raise TimeoutError(f{agent.name} exceeded timeout) return except Exception as e: last_error e if attempt retry - 1: time.sleep(2 ** attempt) raise last_error這個(gè)執(zhí)行器實(shí)現(xiàn)了重試和超時(shí)檢查。注意重試間隔用指數(shù)退避第一次等 1 秒第二次等 2 秒第三次等 4 秒避免密集重試壓垮下游。4.6 組裝與運(yùn)行最后把所有部分組裝起來(lái)跑一遍。def main(): context Context() context.set(input.doc_path, ./requirements.md) scheduler Scheduler() scheduler.register(ReaderAgent()) scheduler.register(PlannerAgent(), depends_on[reader]) scheduler.register(CoderAgent(), depends_on[planner]) scheduler.register(TesterAgent(), depends_on[coder]) executor Executor(scheduler, context) executor.run() print(Final result:, context.get(tester.result))這就是一個(gè)最小可用的編排系統(tǒng)。麻雀雖小五臟俱全包含了上下文管理、Agent 抽象、調(diào)度、執(zhí)行、異常處理這些核心機(jī)制。你可以在這個(gè)基礎(chǔ)上按需擴(kuò)展比如加并行執(zhí)行、加事件驅(qū)動(dòng)、加持久化。5. 常見(jiàn)問(wèn)題與排查技巧實(shí)錄5.1 環(huán)節(jié)卡死不動(dòng)怎么辦這是最常見(jiàn)的問(wèn)題。表現(xiàn)是程序跑著跑著就沒(méi)反應(yīng)了日志也不輸出?？赡艿脑蛴袔讉€(gè)第一某個(gè)環(huán)節(jié)在等待一個(gè)永遠(yuǎn)不會(huì)到來(lái)的信號(hào)比如等待一個(gè)已經(jīng)被消費(fèi)掉的事件。排查方法是看日志最后停在哪然后檢查那個(gè)環(huán)節(jié)的等待條件。第二死鎖兩個(gè)環(huán)節(jié)互相等待對(duì)方釋放資源。排查方法是看線程堆棧找出互相等待的線程。第三死循環(huán)調(diào)度器一直在兩個(gè)環(huán)節(jié)之間來(lái)回跳。排查方法是看執(zhí)行軌跡如果發(fā)現(xiàn)環(huán)節(jié)重復(fù)執(zhí)行就是死循環(huán)。我的經(jīng)驗(yàn)是給每個(gè)環(huán)節(jié)加超時(shí)是最有效的預(yù)防手段。不管什么原因卡住超時(shí)后強(qiáng)制終止并報(bào)錯(cuò)至少不會(huì)讓整個(gè)系統(tǒng)掛死。5.2 數(shù)據(jù)傳遞丟失或錯(cuò)亂表現(xiàn)是下游環(huán)節(jié)拿到的數(shù)據(jù)不是上游產(chǎn)出的或者干脆是空的。可能的原因鍵名拼寫(xiě)錯(cuò)誤、命名空間沖突、上下文被意外清空。排查方法是加一個(gè)上下文變更日志每次讀寫(xiě)都記錄鍵名和值。這樣出問(wèn)題時(shí)直接看日志就能定位到是哪一步出的錯(cuò)。我一般會(huì)在開(kāi)發(fā)環(huán)境開(kāi)啟這個(gè)日志生產(chǎn)環(huán)境關(guān)掉避免性能開(kāi)銷。預(yù)防措施是給上下文加類型校驗(yàn)。寫(xiě)入時(shí)檢查值的類型是否符合預(yù)期讀取時(shí)如果類型不對(duì)就報(bào)錯(cuò)。這樣能在早期發(fā)現(xiàn)問(wèn)題而不是等到下游環(huán)節(jié)才暴露。5.3 重試導(dǎo)致副作用重復(fù)執(zhí)行這是個(gè)隱蔽但危害很大的問(wèn)題。如果某個(gè)環(huán)節(jié)有副作用比如寫(xiě)數(shù)據(jù)庫(kù)、發(fā)消息重試時(shí)副作用會(huì)重復(fù)執(zhí)行導(dǎo)致數(shù)據(jù)不一致。解決辦法是讓有副作用的環(huán)節(jié)支持冪等。具體做法有幾種用唯一 ID 去重、用樂(lè)觀鎖檢查版本、用事務(wù)保證原子性。如果實(shí)在做不到冪等就要在編排層標(biāo)記這個(gè)環(huán)節(jié)不可重試失敗后直接走補(bǔ)償邏輯。我在實(shí)際項(xiàng)目里踩過(guò)這個(gè)坑一個(gè)環(huán)節(jié)負(fù)責(zé)給用戶發(fā)通知重試了三次用戶收到了三條一樣的通知。后來(lái)改成先寫(xiě)一個(gè)待發(fā)送記錄發(fā)送成功后標(biāo)記為已發(fā)送重試時(shí)先檢查記錄狀態(tài)就解決了。5.4 并行環(huán)節(jié)結(jié)果順序不確定并行執(zhí)行時(shí)結(jié)果收集順序和完成順序一致但下游可能依賴特定順序。解決辦法是在收集結(jié)果時(shí)按環(huán)節(jié)名稱排序而不是按完成順序。results {} for future in as_completed(futures): name futures[future] results[name] future.result() ordered [results[name] for name in sorted(results.keys())]這樣無(wú)論哪個(gè)環(huán)節(jié)先完成最終結(jié)果順序都是確定的。5.5 常見(jiàn)問(wèn)題速查表問(wèn)題現(xiàn)象可能原因排查方法解決措施環(huán)節(jié)卡死等待信號(hào)、死鎖、死循環(huán)看日志最后位置、線程堆棧、執(zhí)行軌跡加超時(shí)、檢查等待條件、限制最大步數(shù)數(shù)據(jù)丟失鍵名錯(cuò)誤、命名空間沖突上下文變更日志加類型校驗(yàn)、統(tǒng)一命名規(guī)范副作用重復(fù)重試未冪等檢查副作用記錄實(shí)現(xiàn)冪等、標(biāo)記不可重試結(jié)果順序亂并行完成順序不定檢查結(jié)果收集邏輯按名稱排序收集性能差串行執(zhí)行、大對(duì)象拷貝性能分析改并行、傳引用、寫(xiě)時(shí)復(fù)制5.6 幾個(gè)獨(dú)家避坑技巧第一個(gè)技巧在開(kāi)發(fā)環(huán)境加一個(gè)“慢動(dòng)作模式”每個(gè)環(huán)節(jié)執(zhí)行前暫停一秒方便你觀察執(zhí)行順序和上下文變化。這個(gè)模式在調(diào)試復(fù)雜編排邏輯時(shí)特別有用能讓你直觀看到數(shù)據(jù)怎么流動(dòng)。第二個(gè)技巧給每個(gè)環(huán)節(jié)加一個(gè)唯一執(zhí)行 ID格式是環(huán)節(jié)名-時(shí)間戳-隨機(jī)數(shù)。這個(gè) ID 會(huì)出現(xiàn)在所有相關(guān)日志里方便你追蹤一次執(zhí)行的完整鏈路。我試過(guò)在日志量很大的時(shí)候靠這個(gè) ID 快速定位問(wèn)題。第三個(gè)技巧上下文里加一個(gè)“調(diào)試信息”命名空間專門存中間結(jié)果和調(diào)試數(shù)據(jù)。這個(gè)命名空間的數(shù)據(jù)不參與業(yè)務(wù)邏輯只用于排查。生產(chǎn)環(huán)境可以關(guān)掉開(kāi)發(fā)環(huán)境全開(kāi)。第四個(gè)技巧定期做“混沌測(cè)試”故意讓某些環(huán)節(jié)失敗或超時(shí)看編排系統(tǒng)能不能正確處理。我一般會(huì)在測(cè)試環(huán)境隨機(jī)注入故障驗(yàn)證重試、降級(jí)、補(bǔ)償邏輯是否按預(yù)期工作。這個(gè)習(xí)慣幫我提前發(fā)現(xiàn)了好幾個(gè)隱藏 bug。6. 編排方案的擴(kuò)展與演進(jìn)方向6.1 從硬編碼到配置驅(qū)動(dòng)最初的編排邏輯往往是硬編碼的環(huán)節(jié)順序、依賴關(guān)系都寫(xiě)在代碼里。這種方式簡(jiǎn)單直接但改起來(lái)麻煩每次調(diào)整都要改代碼、重新部署。演進(jìn)方向是配置驅(qū)動(dòng)把編排邏輯抽到一個(gè)配置文件里比如 YAML 或 JSON。代碼只負(fù)責(zé)解析配置并執(zhí)行不關(guān)心具體有哪些環(huán)節(jié)。這樣調(diào)整編排邏輯只需要改配置不用動(dòng)代碼。配置大概長(zhǎng)這樣agents: - name: reader class: agents.reader.ReaderAgent depends_on: [] - name: planner class: agents.planner.PlannerAgent depends_on: [reader] - name: coder class: agents.coder.CoderAgent depends_on: [planner]解析配置后動(dòng)態(tài)加載類并注冊(cè)這樣擴(kuò)展新環(huán)節(jié)只需要加配置和實(shí)現(xiàn)類不用改調(diào)度代碼。6.2 從單機(jī)到分布式當(dāng)環(huán)節(jié)數(shù)量增多、耗時(shí)變長(zhǎng)單機(jī)執(zhí)行可能扛不住。這時(shí)候需要考慮分布式編排把環(huán)節(jié)分散到多臺(tái)機(jī)器上執(zhí)行。分布式編排的核心挑戰(zhàn)是狀態(tài)同步和任務(wù)調(diào)度。狀態(tài)同步可以用中心化的狀態(tài)存儲(chǔ)比如 Redis 或數(shù)據(jù)庫(kù)。任務(wù)調(diào)度可以用消息隊(duì)列把環(huán)節(jié)任務(wù)投遞到隊(duì)列工作節(jié)點(diǎn)消費(fèi)并執(zhí)行。分布式帶來(lái)的好處是擴(kuò)展性和容錯(cuò)性但復(fù)雜度也大幅上升。我的建議是除非單機(jī)真的扛不住否則不要輕易上分布式。很多項(xiàng)目其實(shí)單機(jī)加并行就夠了上分布式反而引入了一堆新問(wèn)題。6.3 從靜態(tài)編排到動(dòng)態(tài)編排靜態(tài)編排的依賴關(guān)系是預(yù)先定義好的運(yùn)行時(shí)不會(huì)變。動(dòng)態(tài)編排則允許在運(yùn)行時(shí)根據(jù)上下文決定下一步執(zhí)行什么甚至動(dòng)態(tài)生成新的環(huán)節(jié)。動(dòng)態(tài)編排的典型場(chǎng)景是任務(wù)分解一個(gè)規(guī)劃環(huán)節(jié)根據(jù)輸入生成一組子任務(wù)每個(gè)子任務(wù)對(duì)應(yīng)一個(gè)環(huán)節(jié)這些環(huán)節(jié)在運(yùn)行時(shí)才確定。這需要編排系統(tǒng)支持動(dòng)態(tài)注冊(cè)和動(dòng)態(tài)調(diào)度。實(shí)現(xiàn)動(dòng)態(tài)編排的關(guān)鍵是把環(huán)節(jié)定義和環(huán)節(jié)實(shí)例分開(kāi)。環(huán)節(jié)定義是靜態(tài)的描述這類環(huán)節(jié)能做什么環(huán)節(jié)實(shí)例是動(dòng)態(tài)的描述這次執(zhí)行具體要做什么。調(diào)度器根據(jù)實(shí)例來(lái)調(diào)度而不是根據(jù)定義。6.4 編排的可觀測(cè)性建設(shè)隨著編排系統(tǒng)變復(fù)雜可觀測(cè)性變得越來(lái)越重要。除了前面說(shuō)的日志和軌跡還可以加指標(biāo)監(jiān)控和告警。指標(biāo)包括每個(gè)環(huán)節(jié)的執(zhí)行次數(shù)、成功率、平均耗時(shí)、P95 耗時(shí)。這些指標(biāo)能幫你發(fā)現(xiàn)性能瓶頸和異常環(huán)節(jié)。告警則是在指標(biāo)超過(guò)閾值時(shí)主動(dòng)通知比如某個(gè)環(huán)節(jié)失敗率超過(guò) 10% 就告警。我一般用 Prometheus 收集指標(biāo)Grafana 做可視化Alertmanager 做告警。這套組合比較成熟接入也簡(jiǎn)單。如果不想引入這么多組件至少要把關(guān)鍵指標(biāo)打到日志里方便后續(xù)分析。6.5 我個(gè)人在實(shí)際操作中的體會(huì)做了幾個(gè)編排項(xiàng)目之后我最大的體會(huì)是編排的復(fù)雜度不在于技術(shù)而在于對(duì)業(yè)務(wù)的理解。技術(shù)方案再先進(jìn)如果對(duì)任務(wù)形狀、失敗模式、數(shù)據(jù)依賴?yán)斫獠坏轿徽諛幼霾缓?。所以我現(xiàn)在做編排第一步永遠(yuǎn)是跟業(yè)務(wù)方聊清楚這個(gè)任務(wù)有哪些環(huán)節(jié)每個(gè)環(huán)節(jié)的輸入輸出是什么哪些環(huán)節(jié)可能失敗失敗了怎么辦把這些聊透了技術(shù)實(shí)現(xiàn)反而是水到渠成的事。另外一點(diǎn)體會(huì)是不要追求一步到位。我見(jiàn)過(guò)太多項(xiàng)目一開(kāi)始就想設(shè)計(jì)一個(gè)完美的編排框架結(jié)果框架還沒(méi)寫(xiě)完業(yè)務(wù)需求已經(jīng)變了。更好的做法是先實(shí)現(xiàn)一個(gè)最小可用版本跑通核心流程然后根據(jù)實(shí)際遇到的問(wèn)題逐步演進(jìn)。這樣每一步都有明確的改進(jìn)目標(biāo)不會(huì)過(guò)度設(shè)計(jì)。最后分享一個(gè)小技巧給編排系統(tǒng)加一個(gè)“回放”功能。把每次執(zhí)行的上下文變更記錄下來(lái)出問(wèn)題時(shí)可以回放整個(gè)執(zhí)行過(guò)程一步步看數(shù)據(jù)怎么變化的。這個(gè)功能在排查復(fù)雜問(wèn)題時(shí)特別有用比看日志直觀得多。實(shí)現(xiàn)也不復(fù)雜就是在上下文變更時(shí)追加寫(xiě)日志回放時(shí)按順序重放。