你有沒有遇到過這種情況想讓一個模型同時搞定寫腳本改bug寫使用說明三件事結果它顧此失彼代碼里夾著解釋文字文案里又帶著代碼片段最后兩邊都不像樣。我一開始也是這么干的直到我把任務真正拆開才意識到多智能體的價值不是炫技而是把一個復雜需求拆給不同崗位去并行思考。這篇文章就帶你從零做一個簡單的AI任務調度助手基于LangChain生態(tài)里的 Agent 和圖形編排能力把調度者、執(zhí)行者、質檢員跑通一條完整鏈路。整篇文章適合兩類人一類是剛接觸 LangChain想知道 Agent 怎么串起來另一類是被單一大模型一個指令干到底折磨過想看看多智能體到底怎么落地的新手。我不會把代碼堆成一堵墻而是會把每個設計決策的原因講清楚順手把調試時踩過的坑也記下來。1. 先聊聊為什么單個大模型干不好任務調度1.1 一個模型身兼數職指令會互相干擾你可能會問我直接給一個大模型寫一段很長的提示詞讓它先寫代碼再寫說明最后自查不也是一種調度嗎表面上可以實際跑幾次就發(fā)現問題了。大模型沒有真正的工作記憶它只是根據上下文里的所有內容生成下一個 token。當代碼任務和文案任務擠在同一段上下文中模型很容易把兩種風格混在一起寫代碼時突然冒出一句這段邏輯很清晰寫文案時又插入代碼塊。還有更麻煩的如果某個子任務執(zhí)行失敗你要回退重跑單一大模型往往無法判斷到底該重寫哪部分只能整段重新生成。我自己遇到最典型的一個場景是讓模型寫一個Python腳本統(tǒng)計CSV缺失值并同時生成說明文檔。模型確實給出了完整回復但腳本里的注釋是給非技術人員看的營銷文案風格文檔里又貼了一大段源碼。分開看好像每個部分都對合在一起就是沒法用。這就是角色打架。1.2 多智能體不是玄學本質是把流程拆給不同大腦多智能體的思路其實很像現實中的團隊分工調度者負責接收需求、拆解任務、判斷下一步交給誰執(zhí)行者各有專長一個寫代碼一個寫文案互不干擾質檢員最后檢查產物是否滿足要求不合格就打回重來。每個智能體只關注自己的窄范圍提示詞短了指令沖突少了上下文也更干凈了。從工程角度看多智能體的價值在于可控性和可觀測性。單一大模型的調用像一個黑盒你很難知道它內部到底先做了什么而多智能體系統(tǒng)里每個節(jié)點是什么、輸入什么、輸出什么都是你自己定義的。哪怕中間某個 Agent 崩了你也知道是哪一步崩的改起來也快得多。在 LangChain 生態(tài)里實現多智能體一般有兩種路線一種是讓一個 Agent 持有其他 Agent 的工具自己做路由另一種是用 LangGraph 這類編排框架把每個智能體做成圖里的一個節(jié)點節(jié)點之間用條件邊連接。我這個項目選的是后者因為條件路由更直觀而且對初學者來說圖結構比隱藏的 Agent 工具鏈更容易調試。2. 環(huán)境準備與LangChain生態(tài)里的現實坑2.1 安裝與版本選擇LangChain不是裝一個就完事很多新手第一次裝 LangChain 時會被一堆包搞暈?，F在 LangChain 已經不是單個庫而是拆成了很多配套包。如果你只裝了langchain會發(fā)現很多 Agent 相關功能根本用不了。我這個項目里實際需要這么幾個pip install langgraph langchain langchain-openai langchain-community langchain-corelanggraph是負責編排多智能體流程的核心langchain-openai是讓 LangChain 能調用 OpenAI 兼容格式的接口langchain-community里裝著社區(qū)維護的一堆工具集成。不裝langchain-core可能會缺基礎類型雖然有些包會自動帶上但顯式裝一遍能省掉不少版本沖突的煩惱。版本方面一定要留個心眼LangGraph 從 0.2 開始 API 有過一些調整有些教程里寫的老寫法會直接報錯或者被廢棄。安裝的時候可以在包名后給定版本范圍例如pip install langgraph0.2 langchain0.3提示如果后面發(fā)現某個函數調用報AttributeError先別急著懷疑自己代碼很可能是某個依賴被升級到了不兼容的新版本。檢查一下pip list里的版本再對照 langgraph 的變更說明基本能定位。2.2 模型接入隨便一個大模型API都能當大腦這個項目里所有智能體本質上都是對同一個大模型 API 的多次調用只不過每次用的系統(tǒng)提示詞不一樣。所以我只需要一個能通過OpenAI 兼容接口訪問的模型服務就行不限定具體廠商。用環(huán)境變量管理密鑰和地址這樣代碼換模型時不用改動邏輯部分import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( base_urlos.getenv(LLM_BASE_URL), api_keyos.getenv(LLM_API_KEY), modelos.getenv(LLM_MODEL_NAME), temperature0.1, )把LLM_BASE_URL、LLM_API_KEY、LLM_MODEL_NAME寫到.env文件里就行。我用的是一個中規(guī)模的通用大模型上下文長度中等響應速度還不錯。如果模型支持結構化輸出調度節(jié)點的準確率會高不少如果模型不支持后面我會說怎么用提示詞約束加解析兜底來處理。這里有一個跟新手相關的關鍵點temperature我設置成 0.1因為調度節(jié)點的任務是對接路由決策不需要創(chuàng)意需要確定性。寫文案的 Agent 可以把溫度調高到 0.4 甚至 0.7。同一份代碼里不同節(jié)點使用不同溫度完全沒問題只要在定義節(jié)點時給不同的 LLM 實例就行。3. 這個任務調度助手的設計調度者、執(zhí)行者、質檢員3.1 狀態(tài)設計是第一步先想清楚每個智能體之間傳什么搭多智能體之前千萬別直接寫代碼第一個要考慮的是狀態(tài)State。多智能體運行相當于一個狀態(tài)機每個節(jié)點讀入狀態(tài)執(zhí)行完畢后又把新狀態(tài)傳給下一個節(jié)點。狀態(tài)里放什么字段直接決定了每個 Agent 到底能看到多少信息。我這套助手的核心字段如下字段類型作用querystr用戶的原始需求next_actionstr調度節(jié)點決定的下一步動作assignmentstr調度器給執(zhí)行者的子任務說明code_resultstr代碼 Agent 的工作產物doc_resultstr文案 Agent 的工作產物inspect_feedbackstr質檢員的審查意見step_countint當前已經跑了多少個節(jié)點防止死循環(huán)有些字段你可能看不出用途但它們在調試時特別重要。比如step_count如果調度器出了問題會在節(jié)點之間瘋狂往返沒有這個計數器整個程序就跑死了也不自知。我會在調度節(jié)點的提示詞里強制要求當 step_count 大于 3 時必須輸出final動作。還要考慮一個問題code_result和doc_result都是直接覆蓋還是保留歷史我這個 demo 里選擇直接覆蓋因為每次重新生成時質檢員的意見已經通過assignment傳給了執(zhí)行者不需要保存多輪歷史狀態(tài)。如果你要迭代很多次最好改成列表類型保留每一步的產物方便追溯和對比。3.2 三條角色鏈路與回退機制我設計了三類節(jié)點調度節(jié)點dispatcher讀用戶 query 和已有產出物輸出 JSON決定下一步是去寫代碼、寫文案還是直接收尾。執(zhí)行節(jié)點代碼 Agent 負責寫腳本并運行驗證文案 Agent 負責寫說明文檔、使用指南或解釋段落。質檢節(jié)點inspector匯總已完成的產出判斷質量是否達標不達標就帶著反饋意見回到調度節(jié)點重新指派。這個流程是一個典型的規(guī)劃-執(zhí)行-審查環(huán)。第一次跑通的時候最容易困惑的是為什么代碼 Agent 執(zhí)行完還要再回到調度節(jié)點而不是直接進質檢節(jié)點因為用戶的需求可能是先寫代碼再寫文案也可能是只需要文案。如果代碼 Agent 執(zhí)行完就直接進質檢那一份沒有文案的產出根本沒法審查。所以我的圖里執(zhí)行節(jié)點回環(huán)到調度節(jié)點由調度節(jié)點判斷下一個該做什么當所有子任務都完成調度節(jié)點才會給出final動作進入質檢環(huán)節(jié)。質檢環(huán)節(jié)也支持回退如果質檢不通過它會帶著inspect_feedback回到調度節(jié)點調度器把反饋寫進新的assignment再指派對應的執(zhí)行者改一輪。這里需要留意的是回退次數必須有限制否則一個永遠改不對的任務會把整個流程拖死。我的做法是把回退次數換算進step_count超過閾值就讓質檢員網開一面直接放行。4. 核心代碼逐段拆解狀態(tài)圖、路由與Agent實現4.1 基礎設施狀態(tài)類型、模型初始化與Python執(zhí)行工具代碼從定義狀態(tài)類型開始。我使用TypedDict來描述狀態(tài)結構這樣寫節(jié)點函數時能獲得 IDE 提示不容易因拼寫錯誤翻車from typing import TypedDict class AssistantState(TypedDict): query: str next_action: str assignment: str code_result: str doc_result: str inspect_feedback: str step_count: int final_answer: str然后是 Python 執(zhí)行工具。代碼 Agent 生成腳本后不能光停在我寫好了這一步最好真跑一遍看看有沒有語法錯誤。我給 Agent 配一個自定義工具用exec執(zhí)行傳入的代碼片段并捕獲輸出import io import contextlib from langchain_core.tools import tool tool def run_python_snippet(code: str) - str: 在本地進程內執(zhí)行一段Python代碼返回標準輸出與錯誤信息。 僅用于demo驗證短腳本生產環(huán)境務必放到隔離容器里。 stdout_buf io.StringIO() stderr_buf io.StringIO() try: with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf): exec(code, {__name__: __main__}) except Exception as exc: return f運行異常: {type(exc).__name__}: {exc}\n標準輸出:\n{stdout_buf.getvalue()} return f標準輸出:\n{stdout_buf.getvalue()}\n標準錯誤:\n{stderr_buf.getvalue()}警告直接在本地exec運行代碼是很危險的做法尤其當代碼 Agent 的輸出不可信時。這個 demo 只是教學用我建議你在沙箱容器里跑。即使如此代碼 Agent 的執(zhí)行結果也要當作外部輸出看待不要盲目信任。4.2 調度節(jié)點的實現負責拆解與路由調度節(jié)點本身不需要工具它就是一次帶結構化輸出要求的 LLM 調用。我會讓模型返回一個 JSON包含next_action和assignment兩個字段。調度器的系統(tǒng)提示詞需要把可選項講清楚code表示去寫代碼doc表示去寫文案final表示所有產出已完成準備交給質檢。同時還要提醒模型參考已有的code_result和doc_result避免重復勞動。from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import ChatPromptTemplate DISPATCH_PROMPT 你是任務調度器負責拆解用戶需求并安排執(zhí)行者。 可選動作 - code: 讓代碼Agent去編寫并驗證Python代碼 - doc: 讓文案Agent去撰寫說明文檔或文案 - final: 所有必要產物都已生成進入質檢環(huán)節(jié) 用戶需求{query} 已有代碼產物{code_result} 已有文案產物{doc_result} 當前執(zhí)行計數{step_count} 規(guī)則 1. 如果用戶需要代碼而代碼產物為空優(yōu)先選擇code。 2. 如果用戶需要說明文檔而文案產物為空選擇doc。 3. 如果步驟超過3輪不管怎樣都選擇final。 4. 輸出JSON格式{next_action: code, assignment: 給執(zhí)行者的子任務描述, reason: 簡單說明理由} 只輸出JSON不要輸出其他內容。 parser JsonOutputParser() def dispatch_node(state: AssistantState) - dict: if state.get(step_count, 0) 3: return { next_action: final, assignment: 所有任務已經執(zhí)行過多輪請質檢后直接匯總, step_count: state.get(step_count, 0) 1, } prompt ChatPromptTemplate.from_template(DISPATCH_PROMPT) chain prompt | llm | parser result chain.invoke({ query: state[query], code_result: state.get(code_result, 還沒有代碼產物), doc_result: state.get(doc_result, 還沒有文案產物), step_count: state.get(step_count, 0), }) return { next_action: result.get(next_action, final), assignment: result.get(assignment, ), step_count: state.get(step_count, 0) 1, }這個實現里有一個很容易忽略但很重要的點我利用step_count作為硬性逃生閥即使大模型輸出的next_action一直不老實超過 3 輪也會被強行改成final。4.3 執(zhí)行者節(jié)點代碼Agent和文案Agent代碼 Agent 用 LangChain 的create_react_agent創(chuàng)建讓它能調用run_python_snippet工具來實際執(zhí)行生成的腳本。系統(tǒng)提示詞限定它的職責范圍只寫代碼、只驗證代碼不要寫說明文檔代碼需要直接能跑并給出必要注釋。from langchain.agents import create_react_agent from langchain_core.prompts import PromptTemplate CODE_AGENT_TMPL 你是一名Python工程師。你只負責根據任務描述編寫可運行的Python代碼 并用run_python_snippet工具運行驗證。不要把任務描述之外的文案混進來。 任務描述 {assignment} 要求 - 代碼要簡潔包含必要的注釋。 - 先調用run_python_snippet運行一次即使報錯也要把錯誤信息原樣反饋。 - 最終回復時先給完整代碼再附一行執(zhí)行結果摘要。 code_agent_prompt PromptTemplate.from_template(CODE_AGENT_TMPL) code_agent_runnable create_react_agent(llm, [run_python_snippet], promptcode_agent_prompt)執(zhí)行節(jié)點里的工作就是把這個 Agent 跑起來并把最終輸出塞回狀態(tài)里def code_agent_node(state: AssistantState) - dict: result code_agent_runnable.invoke({assignment: state.get(assignment, )}) return {code_result: result[output]}文案 Agent 其實不需要工具我用一個普通的鏈來生成文案DOC_PROMPT 你是文檔撰寫員。你不修改代碼只根據已有材料撰寫用戶能看懂的說明文字。 要求語氣清晰、步驟明確、不要空話。 已有代碼產物 {code_result} 任務描述 {assignment} 請輸出說明文檔正文。 def doc_agent_node(state: AssistantState) - dict: prompt ChatPromptTemplate.from_template(DOC_PROMPT) chain prompt | llm result chain.invoke({ code_result: state.get(code_result, 暫無代碼產物), assignment: state.get(assignment, ), }) return {doc_result: result.content if hasattr(result, content) else str(result)}可以看到文案 Agent 的輸入里包含了code_result。這是多智能體協(xié)作的關鍵上一個角色的輸出成為下一個角色的輸入上下文。如果把這一步省掉文案 Agent 就成了盲寫寫出來的說明和實際代碼可能對不上。4.4 質檢節(jié)點與最終匯總質檢節(jié)點的職責是站在用戶視角看整體結果。它對代碼和文案做一次通讀輸出兩個判斷字段pass或fail以及反饋意見。實現上依然是給 LLM 一個結構化的輸出要求。INSPECT_PROMPT 你是質量審查員。請檢查以下代碼和文案是否匹配、是否完整、是否有明顯錯誤。 代碼產物 {code_result} 文案產物 {doc_result} 用戶需求 {query} 輸出JSON格式 {verdict: pass 或 fail, feedback: 你的具體審查意見} 只輸出JSON。 def inspect_node(state: AssistantState) - dict: prompt ChatPromptTemplate.from_template(INSPECT_PROMPT) chain prompt | llm | JsonOutputParser() result chain.invoke({ code_result: state.get(code_result, ), doc_result: state.get(doc_result, ), query: state[query], }) verdict result.get(verdict, fail) feedback result.get(feedback, ) return {next_action: verdict, inspect_feedback: feedback}匯總節(jié)點的任務就簡單多了把代碼和文案拼成最終答案def aggregate_node(state: AssistantState) - dict: final_answer ( f### 代碼部分\n\n{state.get(code_result, )}\n\n f### 文案部分\n\n{state.get(doc_result, )}\n\n f### 質檢意見\n\n{state.get(inspect_feedback, )} ) return {final_answer: final_answer}4.5 把圖組裝起來并跑一個完整例子現在到了最關鍵的一步用 LangGraph 把這些節(jié)點連成一張圖。有四個節(jié)點需要分別處理dispatch、code_agent、doc_agent、inspect、aggregate。連接規(guī)則是dispatch根據next_action無條件選擇去code_agent、doc_agent還是inspect。code_agent結束后回到dispatch讓它繼續(xù)判斷是否需要doc_agent。doc_agent結束后同樣回到dispatch。inspect根據next_action判斷是回dispatch繼續(xù)修改還是去aggregate收尾。from langgraph.graph import StateGraph, END graph StateGraph(AssistantState) graph.add_node(dispatch, dispatch_node) graph.add_node(code_agent, code_agent_node) graph.add_node(doc_agent, doc_agent_node) graph.add_node(inspect, inspect_node) graph.add_node(aggregate, aggregate_node) graph.set_entry_point(dispatch) graph.add_conditional_edges( dispatch, lambda state: state.get(next_action, final), { code: code_agent, doc: doc_agent, final: inspect, }, ) graph.add_edge(code_agent, dispatch) graph.add_edge(doc_agent, dispatch) graph.add_conditional_edges( inspect, lambda state: aggregate if state.get(next_action) pass else dispatch, { aggregate: aggregate, dispatch: dispatch, }, ) graph.add_edge(aggregate, END) app graph.compile()然后就可以用一句話來測試整個調度助手result app.invoke({ query: 幫我寫一個Python腳本讀取data.csv并統(tǒng)計每列缺失值數量然后用中文寫一段腳本使用說明。, next_action: , assignment: , code_result: , doc_result: , inspect_feedback: , step_count: 0, final_answer: , }) print(result[final_answer])第一次跑通的時候你會看到 LangGraph 打印出一串節(jié)點執(zhí)行日志比如dispatch - code_agent - dispatch - doc_agent - dispatch - inspect - aggregate這個順序正是調度器自己規(guī)劃出來的。這種感覺挺奇妙的——你并沒有在代碼里寫死流程而是讓模型在幾個選項之間做選擇最終走出一條合理的執(zhí)行路徑。5. 第一次跑往往不會順利我踩過的三個實際問題5.1 路由結果不夠結構化帶來的解析危機第一次搭建時我讓調度器直接輸出文本然后在代碼里用if code in output這種辦法判斷下一步。聽起來很省事實際上一團糟。模型可能輸出我們應該先 code然后 doc也可能輸出CODE大小寫和格式都不穩(wěn)定判斷邏輯越寫越臟。后來我改成用JsonOutputParser明確要求只輸出 JSON。結果在小模型上又出現一個問題模型不遵守指令輸出的 JSON 前面多了一句解釋性文字或者直接把引號寫彎了解析器照樣報錯。解決思路有兩個一是換用支持response_format{type: json_object}的模型在 API 層面強制結構輸出二是做好冪等的解析兜底用正則把第一個{到最后一個}之間的內容截出來再解析。我現在的做法是寫一個小的解析函數如果直接解析失敗就嘗試提取 JSON 片段import json import re def safe_parse_json(text: str) - dict: text text.strip() try: return json.loads(text) except json.JSONDecodeError: match re.search(r\{.*\}, text, re.S) if match: return json.loads(match.group(0)) raise把safe_parse_json接到dispatch_node里之后路由解析基本穩(wěn)定了。這里想提醒所有新手不要讓提示詞依賴模型自覺要在代碼層面做防御。5.2 會無限循環(huán)的調度器遞歸限制來得太遲LangGraph 默認對遞歸深度有限制超過限制會拋RecursionLimit。我一開始沒加step_count硬保護結果有一次模型反復給code_agent派任務明明代碼已經生成了它還是說還需要進一步驗證于是code_agent執(zhí)行完回到dispatch又來一輪直到遞歸上限被觸發(fā)整個調用鏈崩掉。這個問題在日志里看起來非常奇怪code_agent和dispatch兩個節(jié)點互相跳像死機了一樣。我當時第一反應是檢查圖結構懷疑邊連錯了排查了半天才發(fā)現純粹是調度器的決策沒被約束。解決方案就是在dispatch_node里加硬編碼的輪次判斷。我把它放在所有 LLM 調用之前如果輪次超限直接返回final。這個改動很小效果卻立竿見影。后來我又把閾值從 3 改成 4因為有些合規(guī)流程確實需要代碼→文案→質檢→修改三個來回給定更大余量后減少誤殺。5.3 ReAct Agent的中間步驟被截斷結果缺胳膊少腿代碼 Agent 用的是create_react_agent它內部會執(zhí)行多輪思考-行動-觀察循環(huán)。有一次它生成了很長的代碼我拿到code_result后發(fā)現結尾是斷的明顯是被截斷了。排查下來不是上下文窗口不夠而是max_iterations或者輸出長度限制卡住了。create_react_agent支持通過max_iterations控制思考輪數默認值在有些版本里偏小。當模型需要先思考、再跑代碼、再看結果、再修正的時候輪數根本不夠。我給這個 Agent 顯式設置了max_iterations5并且把handle_parsing_errorsTrue打開讓解析異常不至于中斷整個 Agentcode_agent_runnable create_react_agent( llm, [run_python_snippet], promptcode_agent_prompt, max_iterations5, handle_parsing_errorsTrue, )還有一個容易被忽略的點代碼生成 Agent 返回內容可能包含 Markdown 代碼塊標記比如python ...。傳給文案 Agent 沒問題但如果你直接拿它去喂給exec就會因為多余的標記字符而報錯。我的建議是在run_python_snippet工具內部先做一層處理把代碼塊標記剝掉清理后再執(zhí)行。6. 這個Demo還能怎么長進從玩具到能用的調度助手6.1 參數和提示詞層面的性價比優(yōu)化如果想讓這套調度助手的產出更穩(wěn)定第一步不是堆更多節(jié)點而是調模型參數。我把調度器和質檢員的溫度都保持 0.1文案 Agent 的溫度設在 0.4 左右。原因前面說過路由和審查需要確定性文案生成需要一點變化性。提示詞方面也有優(yōu)化空間。DISPATCH_PROMPT 里我是把code_result和doc_result的完整內容塞進去的任務復雜后這會變得很長。你可以先用一個 LLM 調用把已有產出做壓縮摘要再交給調度器判斷能顯著降低 token 消耗。這個摘要智能體本質上是多智能體架構里的信息預處理節(jié)點加進去不需要改動整體圖結構。另外質檢節(jié)點現在只輸出pass/fail和一句反饋略顯粗糙?？梢陨墳榘创a可運行性、文檔匹配度、完整性三個維度打分并給出具體修復建議。模型在結構化打分任務上的表現通常比籠統(tǒng)的判斷是否通過好不少因為每個維度都有明確檢查目標。6.2 擴展方向記憶、人工確認、多渠道輸入我目前這個 demo 是無狀態(tài)的每次調用都是獨立任務不會記住上一次運行的歷史。如果想讓助手具備對話記憶StateGraph 編譯時啟用持久化機制把狀態(tài)存到后端用戶再次訪問時能接著上一輪繼續(xù)。這一步對任務調度的實際體驗提升很明顯比如用戶說上次那個腳本再改成支持Excel調度器能結合歷史上下文理解上次那個是誰。更實際的擴展是引入人工確認節(jié)點。比如代碼 Agent 生成了刪除數據庫數據的腳本這種高風險任務不應該讓審查 Agent 獨自拍板而是應該中斷流程等人類確認后再繼續(xù)。LangGraph 提供了中斷和恢復的機制可以在關鍵節(jié)點掛起。這樣整個調度系統(tǒng)就從全自動變成了人在回路更適合真正的工作流。還可以繼續(xù)加新角色資料搜索 Agent、表格數據分析 Agent、翻譯 Agent等等。每加一個執(zhí)行者只需要在調度器的動作列表里加一個枚舉值再加一個對應的執(zhí)行節(jié)點和一條條件邊其余部分完全不用動。這種可擴展性正是把任務調度拆成多智能體架構帶來的最大紅利。我個人最后想多提醒一句多智能體系統(tǒng)確實能解決單模型角色混亂的問題但它不是銀彈。節(jié)點越多、回退次數越多調用成本就越高延遲也越長。做的時候不要為了多智而多智先把用戶需求拆出必要角色能用兩個 Agent 就別加第三個保持圖結構簡單才能一眼看出問題出在哪。