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