1. 多 Agent 編排到底在解決什么問題1.1 從單 Agent 到多 Agent 的必然演進先說結(jié)論單 Agent 能做的事天花板比大多數(shù)人想象的低得多。我去年幫一個團隊做智能客服的 POC一開始就是一個 Agent 掛幾個工具——查訂單、查物流、發(fā)郵件。demo 階段跑得挺漂亮一上真實流量就崩了。問題不是模型不行而是一個 Agent 同時要理解意圖、規(guī)劃步驟、調(diào)用工具、校驗結(jié)果、生成回復(fù)上下文里塞了七八種職責(zé)稍微復(fù)雜一點的請求就開始胡言亂語工具調(diào)用參數(shù)錯得離譜。這就是單 Agent 的根本瓶頸職責(zé)耦合導(dǎo)致上下文污染。你可以把它想象成一個公司只雇了一個人既當(dāng)前臺又當(dāng)銷售又當(dāng)財務(wù)還當(dāng)技術(shù)客服短期能撐業(yè)務(wù)一復(fù)雜必然出錯。多 Agent 編排的核心思路就是分而治之把一個大任務(wù)拆成若干子任務(wù)每個子任務(wù)交給一個專職 Agent再由一個編排層Orchestrator負責(zé)調(diào)度、傳遞狀態(tài)、匯總結(jié)果。這跟微服務(wù)架構(gòu)的思路是一模一樣的——單體應(yīng)用拆成服務(wù)每個服務(wù)只干一件事通過明確的接口通信。1.2 編排層到底編排什么很多人對編排這個詞有誤解以為就是按順序調(diào)用幾個 Agent。實際上編排層要處理的事情遠比想象中復(fù)雜我把它拆成四個核心職責(zé)任務(wù)分解把用戶的自然語言請求拆成可執(zhí)行的子任務(wù)圖。比如幫我分析這份財報并生成 PPT要拆成讀取文件→提取關(guān)鍵指標→計算同比環(huán)比→生成圖表→組織文案→輸出 PPT。路由決策決定每個子任務(wù)交給哪個 Agent。是走檢索 Agent 還是走計算 Agent是走代碼執(zhí)行還是走知識庫查詢。狀態(tài)管理在 Agent 之間傳遞上下文。這里有個大坑——不能把上一個 Agent 的全部輸出無腦塞給下一個否則上下文會爆炸必須做摘要和結(jié)構(gòu)化提取。結(jié)果聚合與校驗多個 Agent 的輸出可能沖突編排層要負責(zé)仲裁、重試、兜底。提示編排層本身也是一個Agent只不過它的工具是其他 Agent。這個視角轉(zhuǎn)換很關(guān)鍵想通了之后整個架構(gòu)就順了。1.3 什么場景真的需要多 Agent不是所有項目都值得上多 Agent。我見過太多團隊為了技術(shù)先進硬上多 Agent結(jié)果復(fù)雜度翻了三倍效果還不如一個調(diào)好的單 Agent。判斷標準很簡單滿足下面任意兩條才考慮多 Agent判斷維度單 Agent 夠用建議上多 Agent任務(wù)步驟數(shù)3 步以內(nèi)5 步以上且步驟間有依賴工具數(shù)量5 個以內(nèi)10 個以上且分屬不同領(lǐng)域上下文長度穩(wěn)定在 8K 以內(nèi)經(jīng)常超過 32K職責(zé)類型單一如只做問答混合檢索計算生成校驗錯誤容忍度低錯了重來高需要交叉驗證舉個我實際做過的例子一個合同審查系統(tǒng)需要提取條款、比對標準模板、識別風(fēng)險點、生成修改建議、輸出審查報告。這五個環(huán)節(jié)每個都需要不同的 prompt 策略和工具集硬塞進一個 Agent 里prompt 寫到 3000 字還是顧此失彼。拆成五個專職 Agent 之后每個 prompt 控制在 500 字以內(nèi)準確率直接從 62% 提到 89%。2. 主流編排框架選型與 LangGraph 核心概念2.1 框架選型的幾個真實考量現(xiàn)在市面上做多 Agent 編排的框架不少我實際用過并且踩過坑的主要有這么幾類第一類是 LangGraph 這種圖結(jié)構(gòu)編排。核心抽象是節(jié)點邊狀態(tài)你把每個 Agent 當(dāng)成圖里的一個節(jié)點用邊定義流轉(zhuǎn)邏輯狀態(tài)在整個圖里共享。它的優(yōu)勢是控制流極其明確支持條件分支、循環(huán)、并行而且有 checkpoint 機制可以做斷點續(xù)跑。缺點是學(xué)習(xí)曲線陡得先理解狀態(tài)機和圖的概念。第二類是 CrewAI、AutoGen 這種角色扮演式編排。你定義幾個角色Researcher、Writer、Reviewer框架自動幫你協(xié)調(diào)它們對話。上手快但控制粒度粗一旦流程復(fù)雜起來Agent 之間的對話會失控token 消耗也嚇人。第三類是自己擼。用 Python 原生寫一個調(diào)度循環(huán)Agent 就是函數(shù)狀態(tài)就是一個 dict。靈活度最高但所有輪子都得自己造重試、超時、并發(fā)、日志全要手寫。我的建議是流程明確、需要精細控制的選 LangGraph快速驗證想法選 CrewAI生產(chǎn)環(huán)境且團隊有工程能力LangGraph 是當(dāng)前最穩(wěn)的選擇。下面重點講 LangGraph因為它的抽象最接近生產(chǎn)需求。2.2 LangGraph 的三個核心概念LangGraph 看著復(fù)雜其實就三個概念理解了就能上手State狀態(tài)一個貫穿整個圖的共享數(shù)據(jù)結(jié)構(gòu)通常是個 TypedDict。每個節(jié)點讀取它、修改它。關(guān)鍵是要定義好 reducer——比如多個節(jié)點都想往一個列表里 append 消息你得告訴 LangGraph 用operator.add來合并否則后寫的會覆蓋先寫的。Node節(jié)點就是一個函數(shù)輸入是 State輸出是要更新的 State 字段。一個節(jié)點可以是一個 Agent、一次工具調(diào)用、一段純 Python 邏輯。Edge邊定義節(jié)點之間的流轉(zhuǎn)。普通邊是固定的 A→B條件邊是add_conditional_edges根據(jù)當(dāng)前 State 決定下一步走哪。條件邊是多 Agent 編排的靈魂路由決策全靠它。from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: Annotated[list, operator.add] next_agent: str task_result: str def researcher_node(state: AgentState): # 檢索 Agent 的邏輯 return {messages: [(assistant, 檢索完成)], task_result: ...} def router_node(state: AgentState): # 根據(jù)任務(wù)結(jié)果決定下一步 if 需要計算 in state[task_result]: return {next_agent: calculator} return {next_agent: writer} graph StateGraph(AgentState) graph.add_node(researcher, researcher_node) graph.add_node(router, router_node) graph.add_edge(researcher, router) graph.add_conditional_edges( router, lambda s: s[next_agent], {calculator: calc_node, writer: writer_node} )這段代碼就是最小可用的多 Agent 編排骨架。注意Annotated[list, operator.add]這個寫法這是 LangGraph 里最容易踩的坑不加 reducer 的話多個節(jié)點寫 messages 會互相覆蓋。2.3 狀態(tài)設(shè)計決定架構(gòu)成敗我踩過最大的坑就是狀態(tài)設(shè)計得太隨意。一開始我把整個對話歷史都塞進 State結(jié)果跑幾輪之后上下文爆了token 費用飆升模型還開始失憶——因為關(guān)鍵信息被淹沒在噪音里。后來我總結(jié)出一套狀態(tài)分層設(shè)計全局狀態(tài)只放任務(wù) ID、用戶 ID、最終結(jié)果這類必須貫穿全程的字段。階段狀態(tài)每個 Agent 有自己的局部狀態(tài)用完就清理不往全局塞。消息通道Agent 之間的通信走結(jié)構(gòu)化的消息對象而不是裸字符串。比如{from: researcher, type: finding, content: ..., confidence: 0.9}。這樣設(shè)計之后上下文長度穩(wěn)定控制在 4K 以內(nèi)成本降了 60%準確率反而升了。3. 從零搭建一個多 Agent 編排系統(tǒng)3.1 環(huán)境準備與依賴安裝先把環(huán)境搭起來。Python 版本建議 3.10 以上LangGraph 對類型注解依賴比較重低版本會有各種奇怪的報錯。# 創(chuàng)建虛擬環(huán)境強烈建議別在全局環(huán)境里裝 python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate # 安裝核心依賴 pip install langgraph langchain langchain-openai pip install python-dotenv # 管理 API key如果你還沒裝 Python去官網(wǎng)下載安裝包安裝時務(wù)必勾選 Add Python to PATH否則后面命令行調(diào)不起來。裝完用python --version驗證一下。numpy 這類科學(xué)計算庫如果后面要用直接pip install numpy就行別去折騰源碼編譯。注意LangGraph 的版本迭代很快API 偶爾會有 breaking change。生產(chǎn)項目里一定要把版本號鎖死比如langgraph0.2.x別用latest。3.2 定義 Agent 的標準化接口多 Agent 系統(tǒng)最容易亂的地方就是每個 Agent 的輸入輸出格式不統(tǒng)一。我見過一個項目檢索 Agent 返回字符串計算 Agent 返回 dict生成 Agent 又返回 list編排層光做格式轉(zhuǎn)換就寫了一堆膠水代碼。我的做法是定義一個基類所有 Agent 都繼承它from abc import ABC, abstractmethod from pydantic import BaseModel class AgentInput(BaseModel): task: str context: dict {} history: list [] class AgentOutput(BaseModel): result: str confidence: float metadata: dict {} needs_next: str | None None class BaseAgent(ABC): name: str abstractmethod def run(self, inp: AgentInput) - AgentOutput: pass這樣編排層只需要處理AgentInput和AgentOutput兩種類型所有 Agent 對編排層來說都是黑盒替換、新增、刪除都不影響其他部分。這就是面向接口編程在多 Agent 場景下的價值。3.3 編排圖的完整實現(xiàn)下面是一個我實際項目里用過的編排圖任務(wù)是分析一份文檔并生成結(jié)構(gòu)化報告。涉及四個 AgentReader讀取解析、Analyzer分析提取、Calculator數(shù)值計算、Writer生成報告。from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver class ReportState(TypedDict): doc_path: str raw_content: str analysis: dict calc_result: dict final_report: str error: str | None retry_count: int def reader_node(state: ReportState): try: content read_document(state[doc_path]) return {raw_content: content, error: None} except Exception as e: return {error: str(e), retry_count: state.get(retry_count, 0) 1} def analyzer_node(state: ReportState): analysis analyze(state[raw_content]) return {analysis: analysis} def calculator_node(state: ReportState): result calculate(state[analysis]) return {calc_result: result} def writer_node(state: ReportState): report generate_report(state[analysis], state[calc_result]) return {final_report: report} def should_retry(state: ReportState): if state.get(error) and state.get(retry_count, 0) 3: return retry if state.get(error): return fail return continue builder StateGraph(ReportState) builder.add_node(reader, reader_node) builder.add_node(analyzer, analyzer_node) builder.add_node(calculator, calculator_node) builder.add_node(writer, writer_node) builder.set_entry_point(reader) builder.add_conditional_edges( reader, should_retry, {retry: reader, continue: analyzer, fail: END} ) builder.add_edge(analyzer, calculator) builder.add_edge(calculator, writer) builder.add_edge(writer, END) memory MemorySaver() app builder.compile(checkpointermemory)跑起來就是config {configurable: {thread_id: task-001}} result app.invoke({doc_path: ./report.pdf, retry_count: 0}, config) print(result[final_report])這里有幾個設(shè)計決策值得說清楚第一為什么 reader 要單獨成節(jié)點而不是塞進 analyzer因為文檔讀取是最容易失敗的一步文件不存在、格式不支持、編碼錯誤單獨成節(jié)點才能針對性地做重試。這就是失敗隔離原則。第二為什么用條件邊做重試而不是 try-except 循環(huán)因為 LangGraph 的 checkpoint 機制會在每個節(jié)點執(zhí)行后保存狀態(tài)用條件邊重試的話即使進程崩了重啟后能從上次的 checkpoint 繼續(xù)不用從頭跑。這在長流程任務(wù)里能省大量時間和 token。第三MemorySaver 只是開發(fā)用的。生產(chǎn)環(huán)境要換成SqliteSaver或PostgresSaver否則進程一重啟狀態(tài)就沒了。3.4 并行編排與結(jié)果聚合有些子任務(wù)之間沒有依賴可以并行跑。比如分析文檔時提取關(guān)鍵指標和識別風(fēng)險點互不干擾串行跑就是浪費時間。LangGraph 支持并行節(jié)點但并行分支的寫入必須用 reducer 合并否則會報錯。我一般這樣處理class ParallelState(TypedDict): content: str metrics: Annotated[list, operator.add] risks: Annotated[list, operator.add] summary: str def extract_metrics(state): return {metrics: [extract_metrics_logic(state[content])]} def detect_risks(state): return {risks: [detect_risks_logic(state[content])]} def merge_node(state): summary summarize(state[metrics], state[risks]) return {summary: summary} builder.add_node(extract_metrics, extract_metrics) builder.add_node(detect_risks, detect_risks) builder.add_node(merge, merge_node) # 從同一個節(jié)點分叉出去就是并行 builder.add_edge(start, extract_metrics) builder.add_edge(start, detect_risks) builder.add_edge(extract_metrics, merge) builder.add_edge(detect_risks, merge)并行跑下來整體耗時從 12 秒降到 7 秒效果立竿見影。但要注意并行節(jié)點之間不能有共享的可變狀態(tài)否則會有競態(tài)問題。4. 生產(chǎn)環(huán)境的關(guān)鍵問題與排查實錄4.1 上下文爆炸與 token 成本控制多 Agent 系統(tǒng)最大的成本黑洞就是 token。我統(tǒng)計過一個項目單次任務(wù)平均消耗 45K token其中 70% 是 Agent 之間傳遞的冗余上下文?？刂剖侄挝铱偨Y(jié)了三條按效果排序第一條Agent 間通信必須結(jié)構(gòu)化。別傳自然語言傳 JSON。自然語言里全是好的我已經(jīng)完成了檢索找到了以下內(nèi)容……這種廢話結(jié)構(gòu)化消息能砍掉一半 token。第二條做上下文摘要。上一個 Agent 的輸出如果超過 500 字先讓一個小模型比如 gpt-4o-mini壓縮成 200 字以內(nèi)的要點再傳給下一個。這一步能再省 30%。第三條設(shè)置硬性截斷。每個 Agent 的輸入上下文設(shè)一個上限超了就按重要性排序截斷。別指望模型自己忽略無關(guān)信息它做不到?？刂剖侄螌嵤╇y度token 節(jié)省副作用結(jié)構(gòu)化通信中40-50%需要定義 schema上下文摘要低25-35%可能丟細節(jié)硬性截斷低15-25%可能截掉關(guān)鍵信息緩存重復(fù)調(diào)用中10-20%需要設(shè)計緩存鍵4.2 Agent 死循環(huán)與超時處理多 Agent 系統(tǒng)最惡心的 bug 就是死循環(huán)。A 讓 B 做事B 覺得信息不夠讓 A 補充A 補充完 B 還是覺得不夠……兩個 Agent 能來回扯皮幾十輪token 燒光任務(wù)還沒完成。我的處理方案是三重保險全局步數(shù)上限整個圖最多執(zhí)行 N 步我一般設(shè) 20超了直接終止并返回當(dāng)前最優(yōu)結(jié)果。單 Agent 調(diào)用次數(shù)上限同一個 Agent 在一次任務(wù)里最多被調(diào)用 3 次超了就走兜底邏輯。超時熔斷單次任務(wù)總耗時超過 60 秒強制結(jié)束。def check_limits(state): if state.get(step_count, 0) 20: return force_end if state.get(elapsed, 0) 60: return force_end return continue提示兜底邏輯一定要設(shè)計好。寧可返回一個信息不完整但可用的結(jié)果也不要讓用戶干等或者拿到一個報錯。4.3 常見問題速查表下面這張表是我和團隊踩坑踩出來的基本覆蓋了 90% 的常見問題現(xiàn)象可能原因排查方向解決方案狀態(tài)字段被覆蓋沒定義 reducer檢查 TypedDict 注解加Annotated[type, reducer]圖跑不起來入口點沒設(shè)檢查 set_entry_point顯式設(shè)置入口條件邊不生效路由函數(shù)返回值不匹配打印路由函數(shù)輸出確保返回值在映射表里Agent 反復(fù)調(diào)用路由邏輯有環(huán)畫圖看流轉(zhuǎn)路徑加步數(shù)上限token 超限上下文沒清理統(tǒng)計每步 token加摘要和截斷結(jié)果不穩(wěn)定溫度參數(shù)太高檢查 LLM 配置降到 0.1 以下checkpoint 失效用了 MemorySaver檢查 saver 類型換持久化 saver4.4 一個真實的排查案例有次線上任務(wù)成功率突然從 92% 掉到 67%日志里全是analyzer 節(jié)點超時。我一開始以為是模型服務(wù)不穩(wěn)定查了半天 API 延遲正常。后來把每個節(jié)點的輸入輸出打出來對比發(fā)現(xiàn)reader 節(jié)點返回的內(nèi)容長度從平均 3K 漲到了 15K——因為上游數(shù)據(jù)源改了格式把整個 PDF 的原始文本都塞進來了。analyzer 處理 15K 的輸入自然超時。解決方案是在 reader 和 analyzer 之間加了一個預(yù)處理節(jié)點做文本清洗和分段把輸入壓回 3K 以內(nèi)。改完之后成功率恢復(fù)到 94%。這個案例的教訓(xùn)是多 Agent 系統(tǒng)里節(jié)點之間的數(shù)據(jù)契約比節(jié)點本身更重要。任何一個上游節(jié)點的輸出格式變化都可能引發(fā)下游雪崩。所以我在每個節(jié)點入口都加了輸入校驗長度、格式、必填字段全檢查一遍不合法直接走錯誤分支別讓它污染下游。5. 多 Agent 編排的進階實踐5.1 Agent 記憶的分層設(shè)計多 Agent 系統(tǒng)里記憶是個繞不開的話題。我的經(jīng)驗是分三層短期記憶當(dāng)前任務(wù)內(nèi)的上下文存在 State 里任務(wù)結(jié)束就清。中期記憶跨任務(wù)的會話歷史存數(shù)據(jù)庫按用戶 ID 索引定期摘要壓縮。長期記憶沉淀下來的知識和經(jīng)驗存向量庫用檢索的方式按需調(diào)用。關(guān)鍵點是別把三層混在一起。我見過把全部歷史都塞進 State 的做法跑十幾個任務(wù)之后上下文直接爆炸。正確的做法是State 里只放當(dāng)前任務(wù)相關(guān)的需要歷史的時候主動去查中期記憶需要知識的時候去檢索長期記憶。5.2 Agent 安全與權(quán)限隔離多 Agent 系統(tǒng)一旦接入真實工具文件系統(tǒng)、數(shù)據(jù)庫、外部 API安全問題就必須重視。我的做法是最小權(quán)限原則每個 Agent 只給它完成本職工作的最小工具集。檢索 Agent 只能讀不能寫計算 Agent 只能跑沙箱里的代碼不能碰文件系統(tǒng)寫報告 Agent 只能生成文本不能調(diào)用外部 API。另外所有 Agent 的工具調(diào)用都要過一層審計記錄誰在什么時候調(diào)了什么工具、傳了什么參數(shù)、返回了什么。出問題的時候能快速定位。注意如果 Agent 能執(zhí)行代碼一定要跑在沙箱里限制 CPU、內(nèi)存、執(zhí)行時間禁止網(wǎng)絡(luò)訪問。這是底線別省這一步。5.3 從 LangGraph 到自研編排的取舍LangGraph 很好用但不是銀彈。當(dāng)你的流程復(fù)雜到一定程度或者有特殊的性能要求時可能需要考慮自研編排層。我判斷的標準是如果 LangGraph 的抽象能覆蓋你 80% 的需求就用它如果超過 30% 的邏輯是在跟框架對抗就該考慮自研了。自研的核心其實就是一個調(diào)度循環(huán)加一個狀態(tài)機沒那么神秘class Orchestrator: def __init__(self, agents: dict, router): self.agents agents self.router router def run(self, task, max_steps20): state {task: task, history: [], step: 0} while state[step] max_steps: next_agent self.router(state) if next_agent END: break agent self.agents[next_agent] output agent.run(state) state[history].append(output) state[step] 1 return state就這么幾十行核心邏輯全在里面。自研的好處是完全可控壞處是所有基礎(chǔ)設(shè)施checkpoint、并發(fā)、可觀測性都得自己搭。所以我的建議是先用 LangGraph 快速驗證驗證通過后再評估要不要自研別一上來就造輪子。5.4 可觀測性建設(shè)多 Agent 系統(tǒng)不上可觀測性等于閉著眼睛開車。我一般會埋這幾類數(shù)據(jù)每個節(jié)點的執(zhí)行耗時找出性能瓶頸。每個節(jié)點的 token 消耗找出成本黑洞。Agent 之間的流轉(zhuǎn)路徑可視化整個任務(wù)的執(zhí)行軌跡。失敗率和失敗原因分布定位系統(tǒng)性問題。這些數(shù)據(jù)用 LangSmith 或者自己接 OpenTelemetry 都能采集。關(guān)鍵是別等到出問題才想起來加日志一開始就埋好后面排查問題能省一半時間。6. 我踩過的坑和幾條實在建議做多 Agent 編排這一年多踩的坑比寫的代碼還多。挑幾條最有價值的分享出來希望能幫你少走彎路。第一條別為了多 Agent 而多 Agent。我見過太多項目明明一個 Agent 加幾個工具就能搞定非要拆成五個 Agent結(jié)果復(fù)雜度翻倍效果還更差。先跑通單 Agent遇到明確的瓶頸再拆這是最務(wù)實的路徑。第二條狀態(tài)設(shè)計要克制。State 里每多一個字段就多一份維護成本和上下文開銷。我現(xiàn)在的原則是能推導(dǎo)出來的字段不放能放局部的不放全局能結(jié)構(gòu)化的不存原文。第三條錯誤處理比正常流程更重要。多 Agent 系統(tǒng)里任何一個節(jié)點都可能失敗而且失敗會沿著圖傳播。所以每個節(jié)點都要有明確的失敗語義編排層要有統(tǒng)一的兜底策略。寧可返回降級結(jié)果也不要讓整個任務(wù)崩掉。第四條測試要覆蓋流轉(zhuǎn)路徑。單元測試測單個 Agent 不夠一定要測整個圖的流轉(zhuǎn)。我一般會構(gòu)造幾類典型輸入正常流程、需要重試的、需要并行的、會觸發(fā)兜底的確保每條路徑都跑通。第五條成本要實時監(jiān)控。多 Agent 系統(tǒng)的 token 消耗是單 Agent 的好幾倍不監(jiān)控的話月底賬單能嚇死人。我一般會設(shè)一個日消耗上限超了就告警別等燒完了才發(fā)現(xiàn)。最后分享一個我最近在用的技巧給每個 Agent 加一個自信度字段。Agent 輸出結(jié)果的時候讓它自己評估一下對這個結(jié)果的把握有多大0-1。編排層根據(jù)自信度決定要不要交叉驗證——自信度低于 0.7 的就再派一個 Agent 獨立跑一遍兩個結(jié)果對比。這個機制讓我的系統(tǒng)準確率又提了 8 個百分點成本只增加了 15%性價比很高。