1. 為什么“Agent 編排”值得單獨(dú)拎出來講1.1 從一個(gè)真實(shí)踩坑場景說起去年下半年我接手了一個(gè)內(nèi)部工具鏈的改造項(xiàng)目目標(biāo)是把幾個(gè)獨(dú)立的小助手串成一條能自動(dòng)跑通的流水線一個(gè)負(fù)責(zé)讀需求文檔一個(gè)負(fù)責(zé)拆任務(wù)一個(gè)負(fù)責(zé)寫代碼草稿最后一個(gè)負(fù)責(zé)跑測試并回傳結(jié)果。聽起來不復(fù)雜對吧我一開始也是這么想的直接寫了個(gè)主函數(shù)按順序調(diào)用四個(gè)模塊中間用字符串拼接傳參。結(jié)果第一版跑通只花了半天后面修bug修了整整兩周。問題出在哪出在“編排”這兩個(gè)字上。單個(gè)助手的能力其實(shí)都還行但一旦把它們連起來各種邊界情況就冒出來了上游輸出格式變了下游解析直接崩某個(gè)環(huán)節(jié)超時(shí)了整條鏈路卡死兩個(gè)環(huán)節(jié)都想調(diào)用同一個(gè)資源互相搶鎖更別提中間某一步失敗之后到底該重試、該跳過、還是該回滾根本沒有統(tǒng)一策略。我那時(shí)候才意識到Agent 編排不是簡單的函數(shù)調(diào)用鏈它是一套獨(dú)立的工程問題。這篇文章就想把這件事講透。不管你是剛接觸多智能體協(xié)作的新手還是已經(jīng)寫過幾版編排邏輯但總覺得別扭的老手我都希望這里的內(nèi)容能讓你少走點(diǎn)彎路。我會(huì)從設(shè)計(jì)思路、核心機(jī)制、落地實(shí)現(xiàn)、問題排查四個(gè)維度展開盡量把每個(gè)決策背后的“為什么”說清楚而不是只丟一堆代碼讓你抄。1.2 編排到底在編排什么先把概念對齊。所謂 Agent 編排指的是把多個(gè)具備獨(dú)立決策能力的智能體按照某種控制邏輯組織起來協(xié)同完成一個(gè)超出單個(gè)智能體能力的任務(wù)。注意這里有兩個(gè)關(guān)鍵詞獨(dú)立決策、控制邏輯。獨(dú)立決策意味著每個(gè) Agent 有自己的輸入輸出、有自己的工具集、甚至有自己的“脾氣”比如有的擅長發(fā)散、有的擅長收斂?？刂七壿媱t決定了它們之間怎么傳數(shù)據(jù)、怎么定順序、怎么處理異常、怎么共享狀態(tài)。很多人把編排等同于“寫個(gè)循環(huán)依次調(diào)用”這就把問題想簡單了。真正的編排要解決的是在不確定的執(zhí)行環(huán)境下如何讓一組不確定的組件穩(wěn)定地產(chǎn)生確定的結(jié)果。這個(gè)定義聽起來有點(diǎn)繞但你可以這樣理解單個(gè) Agent 像是一個(gè)員工編排則是管理這群員工的制度。員工能力再強(qiáng)制度混亂項(xiàng)目照樣黃。反過來制度設(shè)計(jì)得好哪怕單個(gè)員工能力一般整體產(chǎn)出也能很穩(wěn)。這就是為什么我說編排值得單獨(dú)拎出來講——它是多智能體系統(tǒng)從“能跑”到“能上線”之間那道最關(guān)鍵的坎。1.3 本文適合誰看如果你屬于下面幾類人這篇內(nèi)容應(yīng)該對你有用正在做多智能體協(xié)作項(xiàng)目但被狀態(tài)傳遞、異常處理搞得頭大的開發(fā)者想從單 Agent 應(yīng)用升級到多 Agent 系統(tǒng)不確定該怎么設(shè)計(jì)控制流的工程師對編排框架選型猶豫不決想搞清楚各種方案取舍邏輯的技術(shù)負(fù)責(zé)人單純想了解“Agent 編排”到底是怎么回事為后續(xù)學(xué)習(xí)打基礎(chǔ)的好奇者。我會(huì)盡量不假設(shè)你用過某個(gè)特定框架因?yàn)榫幣诺暮诵乃枷胧强缈蚣芡ㄓ玫摹Ｄ憷斫饬诉@套邏輯換任何工具都能快速上手。2. 編排方案的整體設(shè)計(jì)思路2.1 先想清楚你的任務(wù)是什么形狀在動(dòng)手寫任何編排代碼之前我強(qiáng)烈建議你先做一件事把任務(wù)畫出來。不是畫流程圖那種正式的東西就是拿張紙把每個(gè)環(huán)節(jié)的輸入、輸出、依賴關(guān)系、可能的失敗點(diǎn)標(biāo)出來。這一步花十分鐘能省后面十小時(shí)的調(diào)試。任務(wù)大致可以分成幾種形狀。第一種是線性鏈A 做完給 BB 做完給 C像流水線一樣。這種最簡單但也是最容易出問題的因?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)。我踩過的坑是一開始把任務(wù)想成線性鏈寫了個(gè)順序執(zhí)行器結(jié)果后來發(fā)現(xiàn)某個(gè)環(huán)節(jié)需要根據(jù)結(jié)果決定是否回退硬生生把線性結(jié)構(gòu)改成了圖結(jié)構(gòu)代碼重構(gòu)了一大半。所以在動(dòng)手前把任務(wù)形狀想清楚比選什么框架重要得多。2.2 控制流與數(shù)據(jù)流要分開設(shè)計(jì)這是我最想強(qiáng)調(diào)的一點(diǎn)。很多初學(xué)者會(huì)把控制流和數(shù)據(jù)流混在一起比如用“上一步的輸出”直接作為“下一步是否執(zhí)行的判斷依據(jù)”。這樣做短期看很省事長期看是災(zāi)難。正確的做法是控制流決定“誰在什么時(shí)候執(zhí)行”數(shù)據(jù)流決定“執(zhí)行時(shí)拿到什么數(shù)據(jù)”。兩者通過一個(gè)共享的狀態(tà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è)上下文對象所有環(huán)節(jié)的輸入輸出都往里面讀寫但每個(gè)環(huán)節(jié)只聲明自己需要讀哪些字段、寫哪些字段。控制流則單獨(dú)維護(hù)一張狀態(tài)機(jī)記錄每個(gè)環(huán)節(jié)的執(zhí)行狀態(tài)待執(zhí)行、執(zhí)行中、成功、失敗、跳過。這樣改控制邏輯不會(huì)影響數(shù)據(jù)格式改數(shù)據(jù)格式也不會(huì)影響控制邏輯。2.3 同步、異步還是事件驅(qū)動(dòng)編排的執(zhí)行模型有三種主流選擇各有各的適用場景。同步阻塞式最簡單一個(gè)環(huán)節(jié)執(zhí)行完才執(zhí)行下一個(gè)。優(yōu)點(diǎn)是邏輯清晰、調(diào)試方便缺點(diǎn)是效率低遇到耗時(shí)環(huán)節(jié)整個(gè)鏈路都在等。適合環(huán)節(jié)少、耗時(shí)短、對實(shí)時(shí)性要求不高的場景。異步非阻塞式允許環(huán)節(jié)并行執(zhí)行通過回調(diào)或 Future 來協(xié)調(diào)。效率高但復(fù)雜度也高尤其是錯(cuò)誤傳播和狀態(tài)同步會(huì)變得很棘手。適合環(huán)節(jié)多、可以并行、對吞吐量有要求的場景。事件驅(qū)動(dòng)式把每個(gè)環(huán)節(jié)看作事件的消費(fèi)者和生產(chǎn)者通過事件總線解耦。靈活性最高擴(kuò)展性最好但調(diào)試難度也最大因?yàn)閳?zhí)行路徑不是顯式寫出來的而是運(yùn)行時(shí)動(dòng)態(tài)決定的。適合環(huán)節(jié)動(dòng)態(tài)變化、需要熱插拔的場景。我的經(jīng)驗(yàn)是除非有明確的性能瓶頸否則優(yōu)先選同步阻塞式。先把邏輯跑通等真的遇到性能問題了再針對性地把耗時(shí)環(huán)節(jié)改成異步。過早引入異步和事件驅(qū)動(dòng)往往會(huì)讓簡單問題復(fù)雜化。我見過太多項(xiàng)目一開始就上事件驅(qū)動(dòng)結(jié)果調(diào)試一個(gè)簡單 bug 要翻十幾個(gè)日志文件。2.4 狀態(tài)管理編排的心臟狀態(tài)管理是編排里最容易做砸的部分。我見過幾種典型的錯(cuò)誤做法把狀態(tài)存在全局變量里、把狀態(tài)序列化到文件里每次讀寫、把狀態(tài)分散在各個(gè)環(huán)節(jié)內(nèi)部靠參數(shù)傳遞。全局變量的問題是并發(fā)不安全多個(gè)環(huán)節(jié)同時(shí)讀寫會(huì)互相污染。文件序列化的問題是性能差而且容易產(chǎn)生不一致的中間狀態(tài)。參數(shù)傳遞的問題是鏈路一長參數(shù)列表就爆炸了而且中間環(huán)節(jié)可能根本不需要某些參數(shù)卻被迫透傳。我推薦的方案是集中式狀態(tài)容器 不可變快照。所有狀態(tài)存在一個(gè)容器里每個(gè)環(huán)節(jié)執(zhí)行前拿一份當(dāng)前狀態(tài)的只讀快照執(zhí)行后提交一份變更。這樣既保證了并發(fā)安全又方便做回滾和審計(jì)。具體實(shí)現(xiàn)可以用一個(gè)簡單的鍵值存儲加上版本號機(jī)制。每次提交變更時(shí)版本號加一回滾時(shí)直接恢復(fù)到指定版本。注意狀態(tài)容器里不要存大對象比如完整的文件內(nèi)容或圖片二進(jìn)制。存引用或路徑就行大對象單獨(dú)管理。否則狀態(tài)快照會(huì)變得非常大影響性能。2.5 異常處理策略要提前定異常處理是編排設(shè)計(jì)里最容易被忽視、又最不能忽視的部分。我建議在動(dòng)手前就明確幾個(gè)策略重試策略哪些環(huán)節(jié)失敗可以重試重試幾次重試間隔怎么算我的經(jīng)驗(yàn)是對于調(diào)用外部服務(wù)的環(huán)節(jié)重試三次、間隔指數(shù)退避是比較穩(wěn)妥的默認(rèn)值。對于純計(jì)算環(huán)節(jié)重試通常沒意義直接失敗更快。降級策略某個(gè)環(huán)節(jié)失敗后是否有備選方案比如主模型調(diào)用失敗是否切換到備用模型降級邏輯要顯式寫出來不要指望運(yùn)行時(shí)自動(dòng)發(fā)現(xiàn)。熔斷策略如果某個(gè)環(huán)節(jié)連續(xù)失敗多次是否要暫時(shí)跳過它避免拖垮整條鏈路這在依賴外部服務(wù)的場景下特別重要。補(bǔ)償策略對于已經(jīng)產(chǎn)生副作用的環(huán)節(jié)比如寫入了數(shù)據(jù)庫失敗后是否需要回滾這需要每個(gè)環(huán)節(jié)自己聲明是否支持補(bǔ)償操作。這些策略不要等到出問題了再想那時(shí)候往往已經(jīng)來不及了。在編排層統(tǒng)一實(shí)現(xiàn)這些策略比在每個(gè)環(huán)節(jié)里各寫一套要可靠得多。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í)間。我見過有人為了省事直接讓每個(gè) Agent 暴露一個(gè)run()方法其他什么都不管。結(jié)果編排層沒法做統(tǒng)一校驗(yàn)也沒法做補(bǔ)償異常處理只能靠 try-catch 包一層非常脆弱。多花半小時(shí)定義接口能省后面幾天的返工。接口定義好之后還需要一個(gè)注冊機(jī)制讓編排層知道有哪些 Agent 可用。簡單的做法是用一個(gè)字典鍵是 Agent 名稱值是實(shí)例。復(fù)雜一點(diǎn)可以用插件式加載支持運(yùn)行時(shí)動(dòng)態(tài)注冊。我一般先用字典等真的有動(dòng)態(tài)需求了再升級。3.2 上下文傳遞數(shù)據(jù)怎么在環(huán)節(jié)間流動(dòng)上下文傳遞的核心問題是下游環(huán)節(jié)怎么拿到上游環(huán)節(jié)的產(chǎn)出同時(shí)又不被上游的無關(guān)數(shù)據(jù)干擾。我的做法是給上下文加一個(gè)命名空間機(jī)制。每個(gè)環(huán)節(jié)寫入數(shù)據(jù)時(shí)必須指定一個(gè)命名空間比如step_a.result。下游環(huán)節(jié)讀取時(shí)也明確指定要讀哪個(gè)命名空間。這樣既避免了鍵名沖突又讓數(shù)據(jù)依賴關(guān)系變得顯式。具體實(shí)現(xiàn)上上下文可以是一個(gè)支持嵌套讀寫的字典。寫入時(shí)用context.set(step_a.result, value)讀取時(shí)用context.get(step_a.result)。如果需要傳遞復(fù)雜結(jié)構(gòu)可以約定用 JSON 序列化但要注意序列化開銷。還有一個(gè)細(xì)節(jié)上下文要不要在環(huán)節(jié)間傳遞引用還是拷貝。傳遞引用效率高但上游環(huán)節(jié)如果修改了數(shù)據(jù)下游會(huì)受影響。傳遞拷貝安全但大對象拷貝開銷大。我的折中方案是小對象傳拷貝大對象傳引用但標(biāo)記為只讀。這樣兼顧了安全和效率。3.3 執(zhí)行調(diào)度誰先誰后怎么定調(diào)度邏輯決定了環(huán)節(jié)的執(zhí)行順序。最簡單的調(diào)度是拓?fù)渑判蚋鶕?jù)依賴關(guān)系排出執(zhí)行序列。但實(shí)際場景往往比這復(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)可以很簡單就是一個(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ù)量的三到五倍超過就強(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ì)帶來幾個(gè)問題。第一是資源競爭多個(gè)環(huán)節(jié)同時(shí)讀寫共享資源可能沖突。解決辦法是給資源加鎖或者讓環(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í)測下來并行度不是越高越好。對于 IO 密集型環(huán)節(jié)并行度可以設(shè)高一些比如 CPU 核數(shù)的兩到四倍。對于計(jì)算密集型環(huán)節(jié)并行度設(shè)成 CPU 核數(shù)就行再高反而因?yàn)樯舷挛那袚Q變慢。3.5 日志與可觀測性出問題怎么查編排系統(tǒng)一旦跑起來出問題的時(shí)候如果沒有足夠的日志排查會(huì)非常痛苦。我的經(jīng)驗(yàn)是日志要記錄三類信息環(huán)節(jié)的開始和結(jié)束、上下文的變更、異常和重試。環(huán)節(jié)開始和結(jié)束的日志要包含環(huán)節(jié)名稱、時(shí)間戳、執(zhí)行時(shí)長。上下文變更的日志要包含變更前后的值但要注意脫敏不要把敏感數(shù)據(jù)寫進(jìn)日志。異常和重試的日志要包含異常類型、堆棧、重試次數(shù)。除了日志還建議加一個(gè)執(zhí)行軌跡功能把每個(gè)環(huán)節(jié)的輸入輸出、執(zhí)行狀態(tài)、耗時(shí)都記錄下來形成一個(gè)可查詢的軌跡。這樣出問題時(shí)直接看軌跡就能定位到哪個(gè)環(huán)節(jié)出了問題不用翻日志。軌跡可以存在內(nèi)存里也可以持久化到數(shù)據(jù)庫看需求。4. 完整實(shí)操流程與關(guān)鍵環(huán)節(jié)實(shí)現(xiàn)4.1 環(huán)境準(zhǔn)備與依賴安裝假設(shè)我們用 Python 來實(shí)現(xiàn)這套編排邏輯基礎(chǔ)環(huán)境需要 Python 3.9 以上因?yàn)橐玫揭恍┹^新的類型注解特性。依賴方面核心只需要標(biāo)準(zhǔn)庫但為了更好的開發(fā)體驗(yàn)建議裝幾個(gè)輔助庫pip install pydantic typing-extensionspydantic用來做數(shù)據(jù)校驗(yàn)和序列化typing-extensions用來支持更豐富的類型注解。如果你需要異步支持可以再加asyncio不過它是標(biāo)準(zhǔn)庫自帶的不用額外裝。項(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)先來實(shí)現(xiàn)上下文容器。核心需求是支持命名空間讀寫、版本管理、快照。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ì)決策值得說明。第一set方法每次都會(huì)增加版本號這樣任何變更都可追蹤。第二snapshot方法會(huì)深拷貝當(dāng)前數(shù)據(jù)保證快照不受后續(xù)修改影響。第三rollback方法恢復(fù)到指定版本用于補(bǔ)償操作。注意深拷貝大對象開銷很大如果上下文里存了大對象建議只存引用快照時(shí)只拷貝引用本身。或者用寫時(shí)復(fù)制Copy-on-Write策略只在真正修改時(shí)才拷貝。4.3 Agent 接口與基類實(shí)現(xiàn)接下來定義 Agent 接口。我用抽象基類來實(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)容并寫入上下文。注意它從上下文讀輸入也往上下文寫輸出不通過參數(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)行最后把所有部分組裝起來跑一遍。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. 常見問題與排查技巧實(shí)錄5.1 環(huán)節(jié)卡死不動(dòng)怎么辦這是最常見的問題。表現(xiàn)是程序跑著跑著就沒反應(yīng)了日志也不輸出?？赡艿脑蛴袔讉€(gè)第一某個(gè)環(huán)節(jié)在等待一個(gè)永遠(yuǎn)不會(huì)到來的信號比如等待一個(gè)已經(jīng)被消費(fèi)掉的事件。排查方法是看日志最后停在哪然后檢查那個(gè)環(huán)節(jié)的等待條件。第二死鎖兩個(gè)環(huán)節(jié)互相等待對方釋放資源。排查方法是看線程堆棧找出互相等待的線程。第三死循環(huán)調(diào)度器一直在兩個(gè)環(huán)節(jié)之間來回跳。排查方法是看執(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)出的或者干脆是空的?？赡艿脑蜴I名拼寫錯(cuò)誤、命名空間沖突、上下文被意外清空。排查方法是加一個(gè)上下文變更日志每次讀寫都記錄鍵名和值。這樣出問題時(shí)直接看日志就能定位到是哪一步出的錯(cuò)。我一般會(huì)在開發(fā)環(huán)境開啟這個(gè)日志生產(chǎn)環(huán)境關(guān)掉避免性能開銷。預(yù)防措施是給上下文加類型校驗(yàn)。寫入時(shí)檢查值的類型是否符合預(yù)期讀取時(shí)如果類型不對就報(bào)錯(cuò)。這樣能在早期發(fā)現(xiàn)問題而不是等到下游環(huán)節(jié)才暴露。5.3 重試導(dǎo)致副作用重復(fù)執(zhí)行這是個(gè)隱蔽但危害很大的問題。如果某個(gè)環(huán)節(jié)有副作用比如寫數(shù)據(jù)庫、發(fā)消息重試時(shí)副作用會(huì)重復(fù)執(zhí)行導(dǎo)致數(shù)據(jù)不一致。解決辦法是讓有副作用的環(huán)節(jié)支持冪等。具體做法有幾種用唯一 ID 去重、用樂觀鎖檢查版本、用事務(wù)保證原子性。如果實(shí)在做不到冪等就要在編排層標(biāo)記這個(gè)環(huán)節(jié)不可重試失敗后直接走補(bǔ)償邏輯。我在實(shí)際項(xiàng)目里踩過這個(gè)坑一個(gè)環(huán)節(jié)負(fù)責(zé)給用戶發(fā)通知重試了三次用戶收到了三條一樣的通知。后來改成先寫一個(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())]這樣無論哪個(gè)環(huán)節(jié)先完成最終結(jié)果順序都是確定的。5.5 常見問題速查表問題現(xiàn)象可能原因排查方法解決措施環(huán)節(jié)卡死等待信號、死鎖、死循環(huán)看日志最后位置、線程堆棧、執(zhí)行軌跡加超時(shí)、檢查等待條件、限制最大步數(shù)數(shù)據(jù)丟失鍵名錯(cuò)誤、命名空間沖突上下文變更日志加類型校驗(yàn)、統(tǒng)一命名規(guī)范副作用重復(fù)重試未冪等檢查副作用記錄實(shí)現(xiàn)冪等、標(biāo)記不可重試結(jié)果順序亂并行完成順序不定檢查結(jié)果收集邏輯按名稱排序收集性能差串行執(zhí)行、大對象拷貝性能分析改并行、傳引用、寫時(shí)復(fù)制5.6 幾個(gè)獨(dú)家避坑技巧第一個(gè)技巧在開發(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í)行的完整鏈路。我試過在日志量很大的時(shí)候靠這個(gè) ID 快速定位問題。第三個(gè)技巧上下文里加一個(gè)“調(diào)試信息”命名空間專門存中間結(jié)果和調(diào)試數(shù)據(jù)。這個(gè)命名空間的數(shù)據(jù)不參與業(yè)務(wù)邏輯只用于排查。生產(chǎn)環(huán)境可以關(guān)掉開發(fā)環(huán)境全開。第四個(gè)技巧定期做“混沌測試”故意讓某些環(huán)節(jié)失敗或超時(shí)看編排系統(tǒng)能不能正確處理。我一般會(huì)在測試環(huán)境隨機(jī)注入故障驗(yàn)證重試、降級、補(bǔ)償邏輯是否按預(yù)期工作。這個(gè)習(xí)慣幫我提前發(fā)現(xiàn)了好幾個(gè)隱藏 bug。6. 編排方案的擴(kuò)展與演進(jìn)方向6.1 從硬編碼到配置驅(qū)動(dòng)最初的編排邏輯往往是硬編碼的環(huán)節(jié)順序、依賴關(guān)系都寫在代碼里。這種方式簡單直接但改起來麻煩每次調(diào)整都要改代碼、重新部署。演進(jìn)方向是配置驅(qū)動(dòng)把編排邏輯抽到一個(gè)配置文件里比如 YAML 或 JSON。代碼只負(fù)責(zé)解析配置并執(zhí)行不關(guān)心具體有哪些環(huán)節(jié)。這樣調(diào)整編排邏輯只需要改配置不用動(dò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)加載類并注冊這樣擴(kuò)展新環(huán)節(jié)只需要加配置和實(shí)現(xiàn)類不用改調(diào)度代碼。6.2 從單機(jī)到分布式當(dāng)環(huán)節(jié)數(shù)量增多、耗時(shí)變長單機(jī)執(zhí)行可能扛不住。這時(shí)候需要考慮分布式編排把環(huán)節(jié)分散到多臺機(jī)器上執(zhí)行。分布式編排的核心挑戰(zhàn)是狀態(tài)同步和任務(wù)調(diào)度。狀態(tài)同步可以用中心化的狀態(tài)存儲比如 Redis 或數(shù)據(jù)庫。任務(wù)調(diào)度可以用消息隊(duì)列把環(huán)節(jié)任務(wù)投遞到隊(duì)列工作節(jié)點(diǎn)消費(fèi)并執(zhí)行。分布式帶來的好處是擴(kuò)展性和容錯(cuò)性但復(fù)雜度也大幅上升。我的建議是除非單機(jī)真的扛不住否則不要輕易上分布式。很多項(xiàng)目其實(shí)單機(jī)加并行就夠了上分布式反而引入了一堆新問題。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)編排的典型場景是任務(wù)分解一個(gè)規(guī)劃環(huán)節(jié)根據(jù)輸入生成一組子任務(wù)每個(gè)子任務(wù)對應(yīng)一個(gè)環(huán)節(jié)這些環(huán)節(jié)在運(yùn)行時(shí)才確定。這需要編排系統(tǒng)支持動(dòng)態(tài)注冊和動(dòng)態(tài)調(diào)度。實(shí)現(xiàn)動(dòng)態(tài)編排的關(guān)鍵是把環(huán)節(jié)定義和環(huán)節(jié)實(shí)例分開。環(huán)節(jié)定義是靜態(tài)的描述這類環(huán)節(jié)能做什么環(huán)節(jié)實(shí)例是動(dòng)態(tài)的描述這次執(zhí)行具體要做什么。調(diào)度器根據(jù)實(shí)例來調(diào)度而不是根據(jù)定義。6.4 編排的可觀測性建設(shè)隨著編排系統(tǒng)變復(fù)雜可觀測性變得越來越重要。除了前面說的日志和軌跡還可以加指標(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)超過閾值時(shí)主動(dòng)通知比如某個(gè)環(huán)節(jié)失敗率超過 10% 就告警。我一般用 Prometheus 收集指標(biāo)Grafana 做可視化Alertmanager 做告警。這套組合比較成熟接入也簡單。如果不想引入這么多組件至少要把關(guān)鍵指標(biāo)打到日志里方便后續(xù)分析。6.5 我個(gè)人在實(shí)際操作中的體會(huì)做了幾個(gè)編排項(xiàng)目之后我最大的體會(huì)是編排的復(fù)雜度不在于技術(shù)而在于對業(yè)務(wù)的理解。技術(shù)方案再先進(jìn)如果對任務(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ì)是不要追求一步到位。我見過太多項(xiàng)目一開始就想設(shè)計(jì)一個(gè)完美的編排框架結(jié)果框架還沒寫完業(yè)務(wù)需求已經(jīng)變了。更好的做法是先實(shí)現(xiàn)一個(gè)最小可用版本跑通核心流程然后根據(jù)實(shí)際遇到的問題逐步演進(jìn)。這樣每一步都有明確的改進(jìn)目標(biāo)不會(huì)過度設(shè)計(jì)。最后分享一個(gè)小技巧給編排系統(tǒng)加一個(gè)“回放”功能。把每次執(zhí)行的上下文變更記錄下來出問題時(shí)可以回放整個(gè)執(zhí)行過程一步步看數(shù)據(jù)怎么變化的。這個(gè)功能在排查復(fù)雜問題時(shí)特別有用比看日志直觀得多。實(shí)現(xiàn)也不復(fù)雜就是在上下文變更時(shí)追加寫日志回放時(shí)按順序重放。