最近很多朋友問(wèn)我同一個(gè)問(wèn)題多智能體到底怎么上手看了不少概念文章一打開(kāi)代碼庫(kù)還是懵。我覺(jué)得與其啃復(fù)雜框架不如先手搓一個(gè)“AI任務(wù)調(diào)度助手”把多智能體的核心思路跑通。這篇文章就用LangChain生態(tài)里最常用的工具搭一個(gè)能拆解任務(wù)、分派Agent、匯總結(jié)果的簡(jiǎn)單調(diào)度器?？赐昴憔湍芾斫舛嘀悄荏w的協(xié)作邏輯也能直接跑出一版可用的Demo。先說(shuō)清楚這個(gè)Demo要解決什么問(wèn)題。平時(shí)我們調(diào)大模型基本是一個(gè)Agent從頭干到尾。比如讓它寫一份行業(yè)分析報(bào)告它又要聯(lián)網(wǎng)查資料又要整理數(shù)據(jù)還要潤(rùn)色文字最后出來(lái)的內(nèi)容常常顧此失彼。多智能體的思路是把一個(gè)大任務(wù)拆成多個(gè)子任務(wù)每個(gè)子任務(wù)交給一個(gè)專職Agent去處理——有負(fù)責(zé)調(diào)研的有負(fù)責(zé)寫初稿的有負(fù)責(zé)審核的。調(diào)度器Orchestrator負(fù)責(zé)拆任務(wù)、發(fā)任務(wù)、收集結(jié)果。這種結(jié)構(gòu)的好處是職責(zé)清晰每個(gè)Agent只干一件事模型就不容易“精神分裂”而且可以針對(duì)不同子任務(wù)選不同的模型或提示詞。我選的實(shí)現(xiàn)路線是LangGraph不是LangChain早期的AgentExecutor。LangGraph是LangChain團(tuán)隊(duì)專門做Agent編排的庫(kù)它把Agent流程定義成一張圖節(jié)點(diǎn)是Agent的動(dòng)作邊是節(jié)點(diǎn)之間的流轉(zhuǎn)狀態(tài)是節(jié)點(diǎn)間傳遞的數(shù)據(jù)。這種“狀態(tài)圖”的方式特別適合調(diào)度場(chǎng)景因?yàn)槟憧梢郧宄吹饺蝿?wù)流轉(zhuǎn)到哪一步、狀態(tài)里存了什么出問(wèn)題也容易排查。1. 整體設(shè)計(jì)與思路拆解1.1 為什么多智能體反而比單個(gè)大模型好用單個(gè)大模型Agent不是不能用而是在處理復(fù)雜任務(wù)時(shí)有幾個(gè)痛點(diǎn)。第一上下文容易被污染。一個(gè)Agent既要處理用戶需求又要調(diào)用工具還要根據(jù)中間結(jié)果調(diào)整策略所有信息都堆在一個(gè)上下文窗口里跑著跑著就忘了前面的關(guān)鍵約束。第二提示詞寫起來(lái)非常糾結(jié)。想讓一個(gè)Agent既懂“拆任務(wù)”又懂“調(diào)工具”又懂“匯總報(bào)告”這個(gè)系統(tǒng)提示詞可能要上千字而且模型很容易顧此失彼。第三調(diào)試?yán)щy。任務(wù)哪一步出錯(cuò)了只能看一長(zhǎng)串對(duì)話歷史定位問(wèn)題很痛苦。多智能體架構(gòu)把這些問(wèn)題拆開(kāi)了。每個(gè)Agent只維護(hù)一小段上下文提示詞短小專注一個(gè)環(huán)節(jié)出問(wèn)題只看那個(gè)Agent的狀態(tài)即可。像我這版調(diào)度助手就分了三個(gè)角色任務(wù)規(guī)劃Agent負(fù)責(zé)拆解需求、執(zhí)行Agent負(fù)責(zé)具體干活、質(zhì)檢Agent負(fù)責(zé)檢查結(jié)果。三個(gè)Agent各司其職調(diào)度器只在它們之間傳狀態(tài)不干涉具體執(zhí)行。這樣跑起來(lái)之后我可以單獨(dú)調(diào)試任何一個(gè)環(huán)節(jié)比如覺(jué)得寫稿質(zhì)量不好就只優(yōu)化寫稿Agent的提示詞其他都不用動(dòng)。1.2 選LangGraph做編排的理由如果你去搜“LangChain多智能體”會(huì)看到很多舊教程還在用initialize_agent或者AgentExecutor。坦白說(shuō)那套東西做簡(jiǎn)單問(wèn)答還行做調(diào)度就力不從心了。AgentExecutor是一個(gè)封裝好的黑盒你很難在里面插入“先檢查再?zèng)Q定下一步”的邏輯。后來(lái)LangChain團(tuán)隊(duì)推出LangGraph思路變成“把Agent流程畫成圖”這對(duì)調(diào)度場(chǎng)景是降維打擊。LangGraph里有兩個(gè)核心概念State狀態(tài)和Graph圖。State是一個(gè)全局?jǐn)?shù)據(jù)對(duì)象所有節(jié)點(diǎn)都能讀寫它Graph就是節(jié)點(diǎn)和邊的連接。調(diào)度器運(yùn)行時(shí)從入口節(jié)點(diǎn)開(kāi)始按邊走到下一個(gè)節(jié)點(diǎn)直到走到END節(jié)點(diǎn)。每個(gè)節(jié)點(diǎn)可以是一個(gè)Python函數(shù)也可以是一個(gè)內(nèi)置的Agent節(jié)點(diǎn)。節(jié)點(diǎn)執(zhí)行完之后返回一個(gè)字典更新State。邊可以是普通邊無(wú)條件跳轉(zhuǎn)也可以是條件邊根據(jù)當(dāng)前State內(nèi)容決定跳轉(zhuǎn)到哪個(gè)節(jié)點(diǎn)。這套表達(dá)能力和靈活性比AgentExecutor強(qiáng)太多了而且調(diào)試時(shí)能看到完整的運(yùn)行軌跡。1.3 調(diào)度助手的架構(gòu)長(zhǎng)什么樣我這版Demo的流程圖用文字描述大概是這樣的入口節(jié)點(diǎn)接收用戶輸入把任務(wù)文本存進(jìn)State然后進(jìn)入規(guī)劃節(jié)點(diǎn)讓大模型拆解任務(wù)輸出一個(gè)子任務(wù)列表每個(gè)子任務(wù)包含任務(wù)ID、任務(wù)描述、負(fù)責(zé)角色接下來(lái)進(jìn)入執(zhí)行階段每個(gè)子任務(wù)落到對(duì)應(yīng)的Agent節(jié)點(diǎn)上執(zhí)行所有子任務(wù)執(zhí)行完后進(jìn)入?yún)R總節(jié)點(diǎn)把各Agent的輸出整理成最終答復(fù)。為了讓新手好理解我故意沒(méi)做復(fù)雜的并行分發(fā)。實(shí)際項(xiàng)目中可以讓多個(gè)Agent并行執(zhí)行但在入門Demo里串行執(zhí)行更容易看清每個(gè)節(jié)點(diǎn)的輸入輸出。我在設(shè)計(jì)時(shí)刻意加了一個(gè)“人工審核”節(jié)點(diǎn)規(guī)劃節(jié)點(diǎn)拆解完任務(wù)后先不急著執(zhí)行而是把任務(wù)清單返回給用戶用戶確認(rèn)后再往下走。這樣既演示了LangGraph的中斷與恢復(fù)機(jī)制也防止模型拆出奇怪的任務(wù)列表浪費(fèi)token。這個(gè)設(shè)計(jì)思路來(lái)自實(shí)際踩坑——模型自己拆任務(wù)很容易拆出不可執(zhí)行的子任務(wù)加一道人工確認(rèn)效果立刻不一樣。2. 核心細(xì)節(jié)解析與實(shí)操要點(diǎn)2.1 這些概念先搞清楚再動(dòng)手在寫代碼之前有幾個(gè)LangGraph的高頻概念需要先有感覺(jué)。第一個(gè)是State它決定節(jié)點(diǎn)間怎么傳數(shù)據(jù)。你用TypedDict定義一個(gè)狀態(tài)結(jié)構(gòu)比如{input: str, tasks: list, results: dict}。所有節(jié)點(diǎn)函數(shù)接收一個(gè)state參數(shù)返回一個(gè)更新字典框架會(huì)自動(dòng)合并。第二個(gè)是節(jié)點(diǎn)節(jié)點(diǎn)就是普通Python函數(shù)簽名是(state) - dict。需要注意節(jié)點(diǎn)返回的字典不會(huì)覆蓋整個(gè)State只會(huì)更新你指定的key。第三個(gè)是邊。普通邊用add_edge表示執(zhí)行完節(jié)點(diǎn)A必然去節(jié)點(diǎn)B。條件邊用add_conditional_edges你需要傳一個(gè)路由函數(shù)函數(shù)根據(jù)當(dāng)前State計(jì)算一個(gè)分支名框架根據(jù)分支名找到對(duì)應(yīng)的目標(biāo)節(jié)點(diǎn)。第四個(gè)是狀態(tài)記憶LangGraph默認(rèn)會(huì)把State傳給每個(gè)節(jié)點(diǎn)但不會(huì)自動(dòng)保存歷史如果你需要多輪對(duì)話記憶要在State里加一個(gè)messages字段專門存歷史消息。2.2 模型與工具怎么選型這版Demo不綁定云廠商我推薦用兼容OpenAI接口的本地推理服務(wù)比如你在本地啟動(dòng)一個(gè)模型服務(wù)然后通過(guò)ChatOpenAI(base_urlhttp://localhost:11434/v1, api_keynot-needed, modellocal-chat-model)來(lái)調(diào)用。這樣做的原因是入門階段調(diào)試頻率很高如果每調(diào)一次都走API付費(fèi)心理上壓力大而且本地模型的響應(yīng)延遲低調(diào)試體驗(yàn)更流暢。不過(guò)要注意本地模型的能力參差不齊規(guī)劃Agent這種需要結(jié)構(gòu)化輸出的環(huán)節(jié)建議選指令跟隨能力強(qiáng)的模型。工具方面執(zhí)行Agent可以用內(nèi)置的TavilySearchResults做聯(lián)網(wǎng)檢索也可以寫一個(gè)簡(jiǎn)單的file_tool讀寫本地文件。我這版Demo里做了一個(gè)自定義工具叫now_time用來(lái)返回當(dāng)前時(shí)間。你別小看這個(gè)工具很多模型對(duì)“當(dāng)前時(shí)間”沒(méi)有概念寫日?qǐng)?bào)類任務(wù)時(shí)總會(huì)瞎編日期。給Agent掛一個(gè)取當(dāng)前時(shí)間的工具單位時(shí)間內(nèi)出錯(cuò)的概率立刻降下來(lái)。這其實(shí)說(shuō)明了一個(gè)道理多智能體系統(tǒng)里工具不在多而在準(zhǔn)。2.3 結(jié)構(gòu)化輸出是調(diào)度器的生命線在多智能體調(diào)度里最怕的就是模型返回一段亂七八糟的文本下游節(jié)點(diǎn)沒(méi)法解析。規(guī)劃節(jié)點(diǎn)的輸出必須是結(jié)構(gòu)化的子任務(wù)清單否則沒(méi)法分派。所以我會(huì)給規(guī)劃節(jié)點(diǎn)用強(qiáng)格式約束——要么用Pydantic的with_structured_output要么在提示詞里明確要求JSON格式再加一個(gè)JSON解析兜底函數(shù)。實(shí)測(cè)中提示詞寫得好不好直接決定解析成功率。我總結(jié)了一個(gè)比較穩(wěn)的JSON提示詞模板要求模型只輸出JSON不要輸出任何解釋提供JSON示例字段用蛇形命名明確說(shuō)“如果不符合條件返回空列表而不是報(bào)錯(cuò)”。除了模型輸出節(jié)點(diǎn)之間的消息格式也要統(tǒng)一。我這版里每個(gè)Agent返回的output字段固定是字符串字符串匯總節(jié)點(diǎn)直接拼接。如果你想傳更富的數(shù)據(jù)比如置信度、來(lái)源鏈接就用字典或Pydantic模型并寫在State定義里。切忌一會(huì)兒返回字符串一會(huì)兒返回字典排查問(wèn)題時(shí)會(huì)瘋掉。3. 實(shí)操過(guò)程與核心環(huán)節(jié)實(shí)現(xiàn)3.1 環(huán)境安裝與目錄結(jié)構(gòu)先創(chuàng)建一個(gè)干凈的虛擬環(huán)境然后安裝依賴。我用的是langgraph它會(huì)自動(dòng)帶起langchain-core等相關(guān)包。命令就一條pip install langgraph langchain-core langchain-openai如果你要用聯(lián)網(wǎng)搜索工具再加一個(gè)tavily-python并設(shè)置API Key。不用工具也能跑Demo可以先用一個(gè)模擬工具替代。我的項(xiàng)目結(jié)構(gòu)很簡(jiǎn)單只有兩個(gè)文件state.py放狀態(tài)定義agent.py放節(jié)點(diǎn)和構(gòu)圖邏輯。新手先不要學(xué)別人一上來(lái)搞十幾個(gè)文件單個(gè)文件跑通再拆分。3.2 定義狀態(tài)結(jié)構(gòu)我這樣定義Statefrom typing import TypedDict, List, Dict, Optional class SubTask(TypedDict): task_id: str desc: str owner: str status: str class AgentState(TypedDict): user_request: str tasks: List[SubTask] current_task: Optional[SubTask] results: Dict[str, str] error: Optional[str]這個(gè)結(jié)構(gòu)里的核心是results它是一個(gè)字典key是子任務(wù)IDvalue是Agent執(zhí)行后的文本結(jié)果。匯總節(jié)點(diǎn)只認(rèn)這個(gè)字典不用關(guān)心中間過(guò)程。current_task字段是給執(zhí)行節(jié)點(diǎn)用的每次從tasks里取一個(gè)未處理的任務(wù)放進(jìn)去執(zhí)行完再寫回results。error字段用來(lái)記錄錯(cuò)誤信息方便條件邊判斷是否需要重試。3.3 寫規(guī)劃節(jié)點(diǎn)規(guī)劃節(jié)點(diǎn)的核心工作是把用戶請(qǐng)求變成子任務(wù)清單。我用LangChain的with_structured_output來(lái)做感覺(jué)比手動(dòng)解析JSON穩(wěn)得多。定義好子任務(wù)的數(shù)據(jù)結(jié)構(gòu)from pydantic import BaseModel, Field class SubTaskOut(BaseModel): task_id: str Field(description任務(wù)編號(hào)例如 T01) desc: str Field(description任務(wù)描述) owner: str Field(description負(fù)責(zé)角色名稱) class TaskPlan(BaseModel): tasks: List[SubTaskOut] Field(description子任務(wù)清單)接著寫規(guī)劃節(jié)點(diǎn)函數(shù)from langchain_openai import ChatOpenAI system_planner 你是一個(gè)項(xiàng)目規(guī)劃助手。用戶會(huì)給你一個(gè)完整需求你需要把它拆解為多個(gè)獨(dú)立的子任務(wù)。 要求 1. 每個(gè)子任務(wù)只包含一個(gè)明確的動(dòng)作。 2. 子任務(wù)數(shù)量控制在3到5個(gè)。 3. 角色限定為research 或 write 或 review。 4. 最終匯總不需要單獨(dú)拆任務(wù)。 只輸出結(jié)構(gòu)化結(jié)果不允許額外解釋。用戶需求{request} def plan_node(state: AgentState) - dict: llm ChatOpenAI(modellocal-chat-model, temperature0) structured_llm llm.with_structured_output(TaskPlan) prompt system_planner.format(requeststate[user_request]) plan structured_llm.invoke(prompt) tasks [SubTask(task_idt.task_id, desct.desc, ownert.owner, statuspending) for t in plan.tasks] return {tasks: tasks}這里把temperature設(shè)為0是為了讓模型減少發(fā)散輸出更可預(yù)測(cè)。實(shí)際對(duì)比過(guò)temperature0.7時(shí)偶爾會(huì)拆出玄幻的任務(wù)描述0就很穩(wěn)。3.4 寫執(zhí)行節(jié)點(diǎn)和質(zhì)檢節(jié)點(diǎn)執(zhí)行節(jié)點(diǎn)不區(qū)分是research還是write統(tǒng)一走一個(gè)通用執(zhí)行函數(shù)。判斷標(biāo)準(zhǔn)就看owner字段如果是research就調(diào)用搜索工具如果是write就讓模型基于已有資料寫文本如果是review就檢查文本長(zhǎng)度和關(guān)鍵詞。def execute_node(state: AgentState): if not state.get(tasks): return {error: no tasks} for t in state[tasks]: if t[status] pending: state[current_task] t if t[owner] research: output do_research(t[desc]) elif t[owner] write: output do_write(t[desc]) else: output do_review(t[desc]) state[results][t[task_id]] output t[status] done break return {results: state[results], tasks: state[tasks]}這樣寫有一個(gè)好處一次執(zhí)行一個(gè)子任務(wù)下次再進(jìn)入節(jié)點(diǎn)時(shí)繼續(xù)處理下一個(gè)邏輯簡(jiǎn)單且不容易死循環(huán)。do_write其實(shí)就是一次普通的大模型生成調(diào)用do_review可以簡(jiǎn)單檢查字?jǐn)?shù)超過(guò)500字就算過(guò)。3.5 組裝LangGraph并加入人工確認(rèn)最關(guān)鍵的是構(gòu)圖。我用了StateGraph然后在規(guī)劃節(jié)點(diǎn)之后加了一個(gè)條件邊根據(jù)用戶輸入的“是否自動(dòng)執(zhí)行”標(biāo)志決定是去執(zhí)行還是返回給用戶確認(rèn)。如果你在終端環(huán)境跑用戶確認(rèn)要修改StateLangGraph提供了一個(gè)interrupt_before參數(shù)可以在進(jìn)入執(zhí)行節(jié)點(diǎn)前中斷然后你手動(dòng)修改State再恢復(fù)運(yùn)行。from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver g StateGraph(AgentState) g.add_node(planner, plan_node) g.add_node(executor, execute_node) g.add_node(summarizer, sum_node) g.set_entry_point(planner) g.add_edge(planner, executor) g.add_conditional_edges(executor, check_remaining_tasks, {continue: executor, done: summarizer}) g.add_edge(summarizer, END) app g.compile(checkpointerMemorySaver())關(guān)鍵在于check_remaining_tasksdef check_remaining_tasks(state): if any(t[status] pending for t in state[tasks]): return continue return done這樣執(zhí)行節(jié)點(diǎn)會(huì)反復(fù)進(jìn)入直到所有任務(wù)做完才進(jìn)入?yún)R總。如果你想要人工確認(rèn)可以在compile時(shí)傳interrupt_before[executor]然后在調(diào)用invoke后訪問(wèn)app.get_state(config)修改state后調(diào)用app.update_state(config, values)再繼續(xù)invoke。這個(gè)機(jī)制我第一次用的時(shí)候覺(jué)得很繞后來(lái)發(fā)現(xiàn)它其實(shí)是LangGraph給我們留的“安全剎車”。3.6 總結(jié)節(jié)點(diǎn)與整體運(yùn)行效果匯總節(jié)點(diǎn)就是把state[results]里的各段輸出整合一下有時(shí)也會(huì)讓大模型把多個(gè)結(jié)果轉(zhuǎn)成一篇連貫的最終報(bào)告。我這里是自己拼接加簡(jiǎn)單潤(rùn)色因?yàn)槌醢嬉M量少依賴大模型跑得快一些。def sum_node(state: AgentState): parts [state[results][t[task_id]] for t in state[tasks]] final_text \n\n.join(parts) return {final_answer: final_text}運(yùn)行一個(gè)例子看看效果。用戶輸入“幫我整理一篇關(guān)于本地模型推理加速的調(diào)研短文”規(guī)劃節(jié)點(diǎn)拆出來(lái)的任務(wù)大概是“搜集相關(guān)工具及性能數(shù)據(jù)”“撰寫調(diào)研短文”“檢查字?jǐn)?shù)是否滿足要求”。執(zhí)行節(jié)點(diǎn)分別處理后匯總節(jié)點(diǎn)把三塊內(nèi)容拼起來(lái)。整體跑下來(lái)耗時(shí)可能比單個(gè)Agent長(zhǎng)一點(diǎn)但每一段內(nèi)容的質(zhì)量是穩(wěn)定的不會(huì)出現(xiàn)“前面還在講推理加速后面突然扯到部署成本”這種跑題情況。4. 常見(jiàn)問(wèn)題與排查技巧實(shí)錄4.1 模型不遵守JSON輸出指令這個(gè)問(wèn)題我?guī)缀趺看巫龆嘀悄荏w項(xiàng)目都會(huì)遇到。要么是模型輸出里帶著解釋性文字要么是JSON字段名不對(duì)。解決辦法有兩個(gè)方向第一換更強(qiáng)指令跟隨的模型如果本地模型達(dá)不到要求就在結(jié)構(gòu)化輸出后面加一個(gè)兜底解析函數(shù)用正則把第一個(gè){到最后一個(gè)}切出來(lái)再json.loads第二在提示詞里把“不遵守格式的后果”寫清楚比如“任何非JSON輸出都會(huì)被系統(tǒng)丟棄你將拿不到任何報(bào)酬”實(shí)測(cè)下來(lái)威懾力比“請(qǐng)遵守JSON格式”管用。4.2 調(diào)度器一直在執(zhí)行節(jié)點(diǎn)里打轉(zhuǎn)這個(gè)問(wèn)題多半出在狀態(tài)更新邏輯上。比如忘了把已處理的任務(wù)改成done或者check_remaining_tasks判斷條件寫反了。我排查時(shí)會(huì)在每個(gè)節(jié)點(diǎn)入口打印一下State的關(guān)鍵字段print(, state.get(current_task), state.get(results))確認(rèn)任務(wù)狀態(tài)是否更新。還有個(gè)隱蔽坑LangGraph的State更新是按key覆蓋的如果你在節(jié)點(diǎn)里直接修改了傳入的列表對(duì)象而沒(méi)有返回新對(duì)象框架可能感知不到變化。所以返回時(shí)要顯式返回tasks和results。4.3 多個(gè)Agent上下文互相污染如果你把多個(gè)Agent塞進(jìn)同一個(gè)State里彼此歷史消息混在一起很容易造成模型行為錯(cuò)亂。我用的隔離技巧是每個(gè)Agent維護(hù)自己的消息列表key比如research_messages、write_messages節(jié)點(diǎn)之間互不相干。匯總節(jié)點(diǎn)也不去讀這些列表只讀最后的結(jié)果字符串。這樣設(shè)計(jì)雖然犧牲了一點(diǎn)信息傳遞效率但換來(lái)了調(diào)試時(shí)極大的清爽感。4.4 性能太慢怎么辦多智能體一定比單Agent慢因?yàn)槟Ｐ捅徽{(diào)用了多次。如果你覺(jué)得Demo響應(yīng)時(shí)間長(zhǎng)優(yōu)先看兩點(diǎn)第一是不是本地模型跑在CPU上如果是換GPU或換個(gè)小一點(diǎn)的量化模型第二是不是子任務(wù)拆得太多控制在3個(gè)以內(nèi)能顯著提速。還有一個(gè)技巧是并行化模型支持并發(fā)的話把執(zhí)行節(jié)點(diǎn)改造成線程池并發(fā)執(zhí)行整體耗時(shí)能壓縮到只比單任務(wù)慢一點(diǎn)。但入門階段不推薦并行因?yàn)椴l(fā)引入的競(jìng)態(tài)問(wèn)題特別不好排查。5. 從Demo到真正可用的任務(wù)調(diào)度助手5.1 加一個(gè)簡(jiǎn)單記憶功能目前的Demo是無(wú)狀態(tài)的用戶每次提問(wèn)都是全新上下文。實(shí)際使用時(shí)多輪對(duì)話很重要。LangGraph支持MemorySaver但那是給調(diào)度器存狀態(tài)用的不是模型記憶。模型記憶需要你自己把歷史對(duì)話傳給模型。我這版Demo里在寫Agent的系統(tǒng)提示詞里插入了歷史要點(diǎn)列表每次執(zhí)行任務(wù)前把上幾輪的關(guān)鍵結(jié)論傳進(jìn)去。注意不要傳原始對(duì)話模型會(huì)分不清新舊任務(wù)傳經(jīng)過(guò)壓縮的要點(diǎn)會(huì)更穩(wěn)。5.2 接入團(tuán)隊(duì)協(xié)作場(chǎng)景這個(gè)調(diào)度助手稍加改造就能模擬一個(gè)“內(nèi)容生產(chǎn)線”策劃Agent出選題調(diào)研Agent找資料寫作Agent出初稿校對(duì)Agent改病句。每個(gè)Agent可以屬于不同租戶或者獨(dú)立服務(wù)LangGraph支持節(jié)點(diǎn)函數(shù)內(nèi)部調(diào)用其他Agent服務(wù)。這樣其實(shí)就變成了一個(gè)簡(jiǎn)單的“Agent工作流引擎”。我自己跑下來(lái)的經(jīng)驗(yàn)是先把流程畫在紙上再翻譯成StateGraph順序不要反。直接寫代碼的話很容易漏掉某條邊。5.3 監(jiān)控、容錯(cuò)與重試生產(chǎn)環(huán)境里Agent一定會(huì)失敗。你可以給每個(gè)執(zhí)行節(jié)點(diǎn)包一層try-except當(dāng)某個(gè)子任務(wù)失敗時(shí)把錯(cuò)誤信息寫入error字段條件邊收到錯(cuò)誤后走一個(gè)“修復(fù)節(jié)點(diǎn)”。修復(fù)節(jié)點(diǎn)的邏輯很簡(jiǎn)單讓模型根據(jù)錯(cuò)誤信息重寫任務(wù)描述最多重試兩次還不行就把任務(wù)標(biāo)記為失敗直接返回部分結(jié)果。實(shí)測(cè)這個(gè)機(jī)制能挽回大概三成意外報(bào)錯(cuò)——比如模型臨時(shí)返回空內(nèi)容重寫任務(wù)描述后往往能正常執(zhí)行。重試要限制次數(shù)否則像網(wǎng)絡(luò)抖動(dòng)那種錯(cuò)誤會(huì)白白消耗大量時(shí)間。6. 寫在最后的幾點(diǎn)體會(huì)我自己做完這個(gè)Demo之后最大的感受是多智能體并沒(méi)有想象中那么玄乎本質(zhì)就是“把流程拆成節(jié)點(diǎn)讓模型在每個(gè)節(jié)點(diǎn)做一件事”。難的不是LangGraph的API而是任務(wù)拆分的邊界感。哪些事該讓模型決策哪些事該用代碼判定這個(gè)度需要反復(fù)調(diào)。建議你第一次跑的時(shí)候故意試幾個(gè)模糊需求比如“幫我處理一下這個(gè)數(shù)據(jù)”看看規(guī)劃節(jié)點(diǎn)怎么拆——它會(huì)提醒你模型不擅長(zhǎng)處理信息不完整的請(qǐng)求所以調(diào)度器最好在入口做一個(gè)需求澄清的節(jié)點(diǎn)那又是下一個(gè)迭代了。最后分享一個(gè)小技巧如果你在論文或業(yè)務(wù)項(xiàng)目里要用多智能體不要一開(kāi)始就上“多模型多工具多團(tuán)隊(duì)”的豪華配置先跑通一個(gè)三節(jié)點(diǎn)的最小閉環(huán)然后把你想加的功能逐個(gè)貼在邊上測(cè)試。多智能體系統(tǒng)的問(wèn)題絕大多數(shù)出在數(shù)據(jù)流上不在模型能力上。狀態(tài)設(shè)計(jì)得清楚這個(gè)系統(tǒng)就成功了一大半。