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