如果你還在猶豫“多智能體到底是不是偽需求”或者收藏了一堆 LangGraph 資料卻始終沒有跑通一個(gè)完整的示例那么這篇文章就是為你準(zhǔn)備的。先說一個(gè)判斷**LangGraph 不是 LangChain 的簡(jiǎn)單升級(jí)而是把大模型應(yīng)用從“單次調(diào)用”推進(jìn)到“可編排狀態(tài)機(jī)”的關(guān)鍵基礎(chǔ)設(shè)施。**在這個(gè)體系里多智能體不是噱頭而是解決復(fù)雜任務(wù)拆解、工具調(diào)用編排、人工審核介入、長(zhǎng)流程狀態(tài)恢復(fù)等真實(shí)問題的一套工程化方案。這篇文章會(huì)從多智能體架構(gòu)講起逐步拆解 LangGraph 的核心組件最后用可復(fù)制的代碼帶你從零構(gòu)建一個(gè)帶條件路由、子圖和并行分支的多智能體應(yīng)用。本文不會(huì)教你“背概念”而是希望你讀完能回答三個(gè)問題LangGraph 為什么值得學(xué)它的核心組件是怎么配合工作的如果我只想做一個(gè)最小可用項(xiàng)目代碼該怎么寫如果這三個(gè)問題正是你關(guān)心的建議先收藏再跟著實(shí)操。1. 為什么你需要關(guān)注 LangGraph 多智能體1.1 單個(gè)大模型調(diào)用解決不了的問題很多開發(fā)者第一次接觸大模型應(yīng)用開發(fā)時(shí)最先寫的是這樣的代碼from openai import OpenAI client OpenAI() response client.chat.completions.create( modelgpt-4o, messages[{role: user, content: 寫一份周報(bào)}], ) print(response.choices[0].message.content)這段代碼本身沒有問題但它只能處理“單次輸入、單次輸出”的簡(jiǎn)單請(qǐng)求。一旦業(yè)務(wù)場(chǎng)景變成下面這樣單次調(diào)用就撐不住了用戶輸入一個(gè)問題需要先判斷該調(diào)用哪個(gè)領(lǐng)域?qū)＜夷Ｐ?。需要從多個(gè)數(shù)據(jù)源檢索資料再匯總成一份報(bào)告。生成結(jié)果之前需要人工審核確認(rèn)。長(zhǎng)任務(wù)執(zhí)行到一半失敗重新恢復(fù)后要繼續(xù)之前的進(jìn)度。這些問題本質(zhì)上不是“讓模型更聰明”能解決的而是需要一套工作流編排機(jī)制讓多個(gè)模型調(diào)用、工具調(diào)用和人工節(jié)點(diǎn)按順序或條件組合起來。LangGraph 就是為這類場(chǎng)景設(shè)計(jì)的。1.2 多智能體不是“多個(gè)模型聊天”一個(gè)常見的誤區(qū)是多智能體系統(tǒng)就是讓幾個(gè) AI 角色互相對(duì)話誰都能當(dāng)“智能體”。實(shí)際上工程意義上的多智能體系統(tǒng)包含四個(gè)核心要素要素說明多個(gè)能力邊界明確的智能體每個(gè)智能體負(fù)責(zé)一類任務(wù)而不是全部任務(wù)明確的調(diào)度/路由機(jī)制決定某個(gè)請(qǐng)求應(yīng)該進(jìn)入哪個(gè)智能體共享或可傳遞的狀態(tài)智能體之間需要傳遞中間結(jié)果可恢復(fù)的執(zhí)行流程支持暫停、恢復(fù)、回滾和檢查點(diǎn)如果只是讓兩個(gè)模型互相聊“你覺得呢”那只是聊天不是多智能體。真正的多智能體應(yīng)用必須是一個(gè)有狀態(tài)的、可控制的、可觀測(cè)的工作流。1.3 LangGraph 在這個(gè)生態(tài)中的定位LangGraph 是由 LangChain 團(tuán)隊(duì)推出的框架專門用于構(gòu)建有狀態(tài)、可編排的大模型應(yīng)用。它與 LangChain 的核心區(qū)別在于【LangChain】更偏向提供模型調(diào)用、Prompt 模板、工具封裝等基礎(chǔ)能力?！綥angGraph】更側(cè)重定義圖結(jié)構(gòu)節(jié)點(diǎn)Node做什么、邊Edge怎么走、狀態(tài)State怎么流轉(zhuǎn)。說得直白一點(diǎn)LangChain 是積木LangGraph 是圖紙。你要建一座房子積木決定你能用什么材料圖紙決定房子長(zhǎng)什么樣。2. LangGraph 核心概念詳解在進(jìn)入代碼之前必須先把 LangGraph 的五個(gè)核心組件講清楚。這幾個(gè)概念貫穿后續(xù)所有實(shí)戰(zhàn)代碼。2.1 State全局狀態(tài)State 是 LangGraph 的靈魂。它表示工作流在任意時(shí)刻的“快照”所有節(jié)點(diǎn)都從 State 讀取輸入并把輸出寫回 State。from typing_extensions import TypedDict class AgentState(TypedDict): messages: list current_step: str result: str你可以把 State 理解為前端里的全局 Store只是這里的 Store 不止存界面狀態(tài)還存整個(gè)工作流的中間數(shù)據(jù)。2.2 Node節(jié)點(diǎn)Node 是工作流里的一個(gè)執(zhí)行單元可以是大模型調(diào)用、工具函數(shù)、API 請(qǐng)求也可以只是一個(gè)普通 Python 函數(shù)。def call_model(state: AgentState) - AgentState: # 這里是節(jié)點(diǎn)邏輯 return {result: some_result}每個(gè)節(jié)點(diǎn)本質(zhì)上就是“輸入 State輸出 State 增量”的函數(shù)。2.3 Edge邊Edge 定義節(jié)點(diǎn)之間的連接關(guān)系。分為普通邊和條件邊【普通邊】上一個(gè)節(jié)點(diǎn)執(zhí)行完無條件進(jìn)入下一個(gè)節(jié)點(diǎn)。【條件邊】根據(jù)當(dāng)前 State 或節(jié)點(diǎn)返回值決定進(jìn)入哪個(gè)分支。2.4 Conditional Edge條件路由條件路由是 LangGraph 最強(qiáng)大的能力之一。它允許你根據(jù)某個(gè)字段的值動(dòng)態(tài)選擇下一個(gè)節(jié)點(diǎn)。from langgraph.graph import END, START, StateGraph def route_by_intent(state: AgentState) - str: if 天氣 in state[messages][-1]: return weather_agent return general_agent2.5 Checkpoint檢查點(diǎn)Checkpoint 讓工作流具備“記憶”和“恢復(fù)能力”。你可以把每一步狀態(tài)保存到內(nèi)存或數(shù)據(jù)庫即使中間某個(gè)節(jié)點(diǎn)失敗也可以從上一個(gè)檢查點(diǎn)繼續(xù)執(zhí)行。from langgraph.checkpoint.memory import InMemorySaver checkpointer InMemorySaver()3. 環(huán)境準(zhǔn)備與 LangGraph 安裝3.1 版本與環(huán)境要求LangGraph 是一個(gè) Python 框架目前同時(shí)支持 JavaScript/TypeScript 版本。本文以 Python 為例。建議使用 Python 3.10 及以上版本具體小版本以你本機(jī)環(huán)境為準(zhǔn)。3.2 安裝 LangGraph推薦使用 pip 安裝pip install langgraph如果需要調(diào)用 OpenAI 或其他模型還要安裝對(duì)應(yīng) SDKpip install openai如果你使用的是國(guó)產(chǎn)大模型或本地部署模型只要接口兼容 OpenAI 格式也可以直接通過配置 base_url 接入。3.3 確認(rèn)安裝成功python -c import langgraph; print(langgraph.__version__)如果輸出版本號(hào)說明安裝成功。注意LangGraph 版本更新很快API 可能會(huì)有細(xì)微變化。本文的代碼以較新的穩(wěn)定版本為參考如果你使用的版本過舊或過新請(qǐng)以官方文檔為準(zhǔn)。4. 從零構(gòu)建第一個(gè) LangGraph 多智能體應(yīng)用4.1 業(yè)務(wù)場(chǎng)景我們構(gòu)造一個(gè)非常典型的多智能體場(chǎng)景一個(gè)“客服工單分流系統(tǒng)”。用戶提交一句問題系統(tǒng)先做意圖識(shí)別然后根據(jù)意圖分配到不同的智能體如果涉及訂單問題轉(zhuǎn)給訂單處理智能體。如果涉及技術(shù)問題轉(zhuǎn)給技術(shù)支持智能體。如果意圖不明確轉(zhuǎn)給通用助手。這個(gè)場(chǎng)景雖然簡(jiǎn)單但已經(jīng)包含了多智能體的核心要素多個(gè)專用節(jié)點(diǎn)、條件路由、共享狀態(tài)。4.2 定義狀態(tài)與節(jié)點(diǎn)創(chuàng)建一個(gè)文件agent_demo.pyfrom typing_extensions import TypedDict from langgraph.graph import END, START, StateGraph class AgentState(TypedDict): user_input: str intent: str final_answer: str def intent_node(state: AgentState) - AgentState: 模擬意圖識(shí)別實(shí)際項(xiàng)目中可以用大模型調(diào)用代替 text state[user_input] if 訂單 in text or 發(fā)貨 in text: intent order elif 報(bào)錯(cuò) in text or 無法運(yùn)行 in text or bug in text: intent tech else: intent general return {intent: intent} def order_agent(state: AgentState) - AgentState: 訂單處理智能體 return {final_answer: f【訂單客服】收到你的問題{state[user_input]}我們會(huì)盡快核查訂單狀態(tài)。} def tech_agent(state: AgentState) - AgentState: 技術(shù)支持智能體 return {final_answer: f【技術(shù)支持】收到你的問題{state[user_input]}請(qǐng)?zhí)峁┩暾麍?bào)錯(cuò)信息我們將協(xié)助排查。} def general_agent(state: AgentState) - AgentState: 通用助手智能體 return {final_answer: f【通用助手】收到你的問題{state[user_input]}建議聯(lián)系人工客服獲得更多幫助。}這里有一點(diǎn)需要特別注意每個(gè)節(jié)點(diǎn)函數(shù)必須返回一個(gè)字典字典中的鍵名要與 State 定義對(duì)應(yīng)。LangGraph 會(huì)把這個(gè)返回值合并到全局 State 中。4.3 定義條件路由函數(shù)條件路由函數(shù)是 LangGraph 中決定“下一步去哪”的關(guān)鍵。def route_by_intent(state: AgentState) - str: 根據(jù)意圖字段決定進(jìn)入哪個(gè)智能體節(jié)點(diǎn) intent state[intent] if intent order: return order_agent elif intent tech: return tech_agent else: return general_agent4.4 構(gòu)建圖并運(yùn)行# 構(gòu)建圖 builder StateGraph(AgentState) # 添加節(jié)點(diǎn) builder.add_node(intent_node, intent_node) builder.add_node(order_agent, order_agent) builder.add_node(tech_agent, tech_agent) builder.add_node(general_agent, general_agent) # 添加邊 builder.add_edge(START, intent_node) builder.add_conditional_edges( intent_node, route_by_intent, { order_agent: order_agent, tech_agent: tech_agent, general_agent: general_agent, }, ) builder.add_edge(order_agent, END) builder.add_edge(tech_agent, END) builder.add_edge(general_agent, END) # 編譯圖 graph builder.compile() # 運(yùn)行 result graph.invoke({user_input: 我的訂單什么時(shí)候發(fā)貨}) print(意圖, result[intent]) print(回答, result[final_answer])4.5 運(yùn)行與驗(yàn)證python agent_demo.py預(yù)期輸出意圖 order 回答 【訂單客服】收到你的問題我的訂單什么時(shí)候發(fā)貨我們會(huì)盡快核查訂單狀態(tài)。再測(cè)試一個(gè)技術(shù)問題result graph.invoke({user_input: 程序運(yùn)行時(shí)報(bào)錯(cuò)怎么辦}) print(意圖, result[intent]) print(回答, result[final_answer])預(yù)期輸出意圖 tech 回答 【技術(shù)支持】收到你的問題程序運(yùn)行時(shí)報(bào)錯(cuò)怎么辦請(qǐng)?zhí)峁┩暾麍?bào)錯(cuò)信息我們將協(xié)助排查。如果看到這樣的輸出說明第一個(gè)多智能體工作流已經(jīng)成功跑通了。5. 接入真實(shí)大模型讓意圖識(shí)別變成模型能力5.1 為什么需要用大模型做意圖識(shí)別上面示例中意圖識(shí)別使用的是字符串匹配效果非常有限。真實(shí)項(xiàng)目中你應(yīng)該讓大模型來承擔(dān)意圖識(shí)別任務(wù)這樣可以處理更復(fù)雜、更模糊的表達(dá)。5.2 改造意圖識(shí)別節(jié)點(diǎn)from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-4o-mini, temperature0, base_urlhttps://api.openai.com/v1, # 按你的服務(wù)商配置 api_keyyour-api-key, ) def intent_node_with_llm(state: AgentState) - AgentState: prompt f 你是意圖識(shí)別引擎。用戶輸入如下 {state[user_input]} 請(qǐng)判斷該輸入屬于哪種意圖只能輸出三個(gè)選項(xiàng)之一 - order訂單、物流、售后相關(guān) - tech技術(shù)報(bào)錯(cuò)、代碼問題、使用故障 - general其他情況 輸出格式只輸出意圖單詞不要有其他內(nèi)容。 response llm.invoke(prompt) intent response.content.strip().lower() if intent not in [order, tech, general]: intent general return {intent: intent}注意把temperature設(shè)為 0讓模型輸出更穩(wěn)定。在 Prompt 中限定輸出格式避免模型返回多行解釋。即使模型輸出了意外內(nèi)容也要做兜底處理。5.3 接口兼容的模型都可以接入如果你的項(xiàng)目使用的是國(guó)產(chǎn)大模型或本地部署模型只要服務(wù)商提供了 OpenAI 兼容接口就可以這樣配置llm ChatOpenAI( modelyour-model-name, api_keyyour-api-key, base_urlhttps://your-model-endpoint.com/v1, )這里的base_url指向你的模型服務(wù)地址即可。LangGraph 不關(guān)心底層模型是誰它只關(guān)心如何編排這些模型節(jié)點(diǎn)的調(diào)用關(guān)系。6. 深入 Condition Edge分支控制與循環(huán)檢測(cè)6.1 多分支條件路由實(shí)際業(yè)務(wù)中條件路由往往不是“一次判斷、一次分流”這么簡(jiǎn)單而是會(huì)出現(xiàn)多級(jí)分支、循環(huán)重試甚至動(dòng)態(tài)決定是否終止的情況。比如在意圖識(shí)別之后我們還想加一個(gè)“回答質(zhì)量評(píng)估”節(jié)點(diǎn)如果評(píng)估結(jié)果不合格則重新調(diào)用模型生成一次答案。def quality_check(state: AgentState) - AgentState: answer state[final_answer] # 簡(jiǎn)單的質(zhì)量規(guī)則答案長(zhǎng)度太短就判為不合格 if len(answer) 10: return {need_retry: True} return {need_retry: False} def route_by_quality(state: AgentState) - str: if state.get(need_retry): return retry_node return END這樣圖結(jié)構(gòu)就出現(xiàn)了“環(huán)”LangGraph 是支持這種循環(huán)結(jié)構(gòu)的。但與普通編程不同LangGraph 的循環(huán)需要小心處理否則可能出現(xiàn)無限循環(huán)。6.2 循環(huán)檢測(cè)與最大步數(shù)限制LangGraph 在編譯圖的時(shí)候會(huì)做基礎(chǔ)的循環(huán)檢測(cè)但不會(huì)阻止合法的循環(huán)。運(yùn)行時(shí)我們可以通過參數(shù)控制最大執(zhí)行步數(shù)config {recursion_limit: 10} result graph.invoke({user_input: 測(cè)試}, configconfig)如果工作流執(zhí)行超過 10 個(gè)節(jié)點(diǎn)步驟LangGraph 會(huì)拋出異常提醒你可能存在無限循環(huán)。這是一個(gè)非常實(shí)用的保護(hù)機(jī)制。在生產(chǎn)環(huán)境中強(qiáng)烈建議顯式設(shè)置recursion_limit。6.3 一個(gè)帶重試的完整圖示例from typing_extensions import TypedDict from langgraph.graph import END, START, StateGraph class RetryState(TypedDict): user_input: str final_answer: str need_retry: bool def generate_answer(state: RetryState) - RetryState: answer f針對(duì)問題「{state[user_input]}」的自動(dòng)回復(fù) return {final_answer: answer} def quality_check(state: RetryState) - RetryState: answer state[final_answer] if len(answer) 20: return {need_retry: True} return {need_retry: False} def retry_node(state: RetryState) - RetryState: return {final_answer: state[final_answer] 已補(bǔ)充詳細(xì)說明} def route_by_quality(state: RetryState) - str: if state.get(need_retry): return retry_node return END builder StateGraph(RetryState) builder.add_node(generate_answer, generate_answer) builder.add_node(quality_check, quality_check) builder.add_node(retry_node, retry_node) builder.add_edge(START, generate_answer) builder.add_edge(generate_answer, quality_check) builder.add_conditional_edges(quality_check, route_by_quality, {retry_node: retry_node, END: END}) builder.add_edge(retry_node, quality_check) graph builder.compile() config {recursion_limit: 5} result graph.invoke({user_input: 你好}, configconfig) print(result)這個(gè)示例展示了 LangGraph 中非常核心的能力循環(huán) 條件跳出 狀態(tài)更新。這類結(jié)構(gòu)在真實(shí)項(xiàng)目中非常常見比如 AI 生成內(nèi)容后的格式校驗(yàn)、合規(guī)檢查、安全審查都可以用類似方式實(shí)現(xiàn)。7. 子圖Subgraph與并行分支7.1 什么是子圖當(dāng)工作流變得復(fù)雜時(shí)不可能把所有節(jié)點(diǎn)都平鋪在同一個(gè)圖里。子圖允許你把一部分節(jié)點(diǎn)封裝成一個(gè)獨(dú)立的 Graph再作為上層圖的一個(gè)節(jié)點(diǎn)使用。打個(gè)比方主圖是公司的整體流程子圖是某個(gè)部門內(nèi)部的詳細(xì)流程。對(duì)外其他部門只看到結(jié)果不關(guān)心內(nèi)部細(xì)節(jié)。7.2 子圖實(shí)戰(zhàn)示例假設(shè)我們要做一個(gè)“技術(shù)問答智能體”其中“日志分析”需要作為獨(dú)立子圖實(shí)現(xiàn)因?yàn)槿罩痉治鰞?nèi)部包含讀取日志、提取錯(cuò)誤、生成診斷建議三個(gè)步驟。from typing_extensions import TypedDict from langgraph.graph import END, START, StateGraph class LogSubGraphState(TypedDict): log_text: str error_summary: str def read_log(state: LogSubGraphState) - LogSubGraphState: return {error_summary: f從日志中提取到關(guān)鍵錯(cuò)誤{state[log_text][:20]}} def extract_error(state: LogSubGraphState) - LogSubGraphState: return {error_summary: state[error_summary] 錯(cuò)誤類型運(yùn)行時(shí)異常} def generate_diagnosis(state: LogSubGraphState) - LogSubGraphState: return {error_summary: state[error_summary] 建議檢查內(nèi)存配置并重啟服務(wù)。} log_subgraph_builder StateGraph(LogSubGraphState) log_subgraph_builder.add_node(read_log, read_log) log_subgraph_builder.add_node(extract_error, extract_error) log_subgraph_builder.add_node(generate_diagnosis, generate_diagnosis) log_subgraph_builder.add_edge(START, read_log) log_subgraph_builder.add_edge(read_log, extract_error) log_subgraph_builder.add_edge(extract_error, generate_diagnosis) log_subgraph_builder.add_edge(generate_diagnosis, END) log_subgraph log_subgraph_builder.compile()7.3 把子圖接入主圖class MainState(TypedDict): question: str log_text: str diagnosis: str def diagnostic_node(state: MainState) - MainState: sub_result log_subgraph.invoke({log_text: state[log_text]}) return {diagnosis: sub_result[error_summary]} def answer_node(state: MainState) - MainState: return {diagnosis: f最終答復(fù){state[diagnosis]}} main_builder StateGraph(MainState) main_builder.add_node(diagnostic_node, diagnostic_node) main_builder.add_node(answer_node, answer_node) main_builder.add_edge(START, diagnostic_node) main_builder.add_edge(diagnostic_node, answer_node) main_builder.add_edge(answer_node, END) main_graph main_builder.compile() result main_graph.invoke({question: 服務(wù)掛了怎么辦, log_text: OutOfMemoryError at com.example.Main}) print(result[diagnosis])子圖的價(jià)值在于大型項(xiàng)目可以按業(yè)務(wù)模塊拆分圖定義。子圖可以被復(fù)用比如日志分析子圖可以被多個(gè)主圖調(diào)用。調(diào)試時(shí)只需要關(guān)注當(dāng)前子圖范圍問題定位更清晰。7.4 并行分支縮短任務(wù)耗時(shí)LangGraph 支持在一個(gè)節(jié)點(diǎn)后派出多個(gè)并行分支等待所有分支完成后合并結(jié)果。比如做市場(chǎng)分析時(shí)需要同時(shí)抓取競(jìng)品信息、用戶評(píng)價(jià)和銷售數(shù)據(jù)from langgraph.graph import END, START, StateGraph class ParallelState(TypedDict): question: str competitor_result: str user_review_result: str sale_result: str final_report: str def competitor_task(state: ParallelState) - ParallelState: return {competitor_result: 競(jìng)品信息A 產(chǎn)品主打性價(jià)比B 產(chǎn)品主打高端體驗(yàn)} def user_review_task(state: ParallelState) - ParallelState: return {user_review_result: 用戶評(píng)價(jià)對(duì) A 產(chǎn)品好評(píng)集中在價(jià)格B 產(chǎn)品好評(píng)集中在設(shè)計(jì)} def sale_task(state: ParallelState) - ParallelState: return {sale_result: 銷售數(shù)據(jù)A 產(chǎn)品近一月銷量 1000 件B 產(chǎn)品近一月銷量 800 件} def merge_report(state: ParallelState) - ParallelState: report ( f競(jìng)品{state[competitor_result]}\n f用戶{state[user_review_result]}\n f銷售{state[sale_result]} ) return {final_report: report} parallel_builder StateGraph(ParallelState) parallel_builder.add_node(competitor_task, competitor_task) parallel_builder.add_node(user_review_task, user_review_task) parallel_builder.add_node(sale_task, sale_task) parallel_builder.add_node(merge_report, merge_report) parallel_builder.add_edge(START, competitor_task) parallel_builder.add_edge(START, user_review_task) parallel_builder.add_edge(START, sale_task) parallel_builder.add_edge(competitor_task, merge_report) parallel_builder.add_edge(user_review_task, merge_report) parallel_builder.add_edge(sale_task, merge_report) parallel_builder.add_edge(merge_report, END) parallel_graph parallel_builder.compile() result parallel_graph.invoke({question: 近期市場(chǎng)分析}) print(result[final_report])并行分支的關(guān)鍵在于多個(gè)節(jié)點(diǎn)都連接到同一個(gè)聚合節(jié)點(diǎn)。LangGraph 會(huì)等待所有上游分支完成后才執(zhí)行聚合節(jié)點(diǎn)不需要你自己寫多線程代碼。8. Checkpoint讓工作流具備記憶與恢復(fù)能力8.1 Checkpoint 解決了什么問題多智能體工作流往往需要多輪交互。比如用戶第一句話是“幫我查一下訂單”第二句話是“順便改一下收貨地址”。如果沒有狀態(tài)保存機(jī)制第二句話就失去了上下文。Checkpoint 解決了這個(gè)問題。它把每一步 State 保存到檢查點(diǎn)存儲(chǔ)中你可以隨時(shí)恢復(fù)到任意歷史步驟。8.2 使用 InMemorySaver 實(shí)現(xiàn)記憶from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import END, START, StateGraph class ConversationState(TypedDict): user_input: str response: str history: list def respond_node(state: ConversationState) - ConversationState: history state.get(history, []) history.append({user: state[user_input]}) response f這是第 {len(history)} 輪對(duì)話你的問題是{state[user_input]} history.append({ai: response}) return {response: response, history: history} checkpointer InMemorySaver() builder StateGraph(ConversationState) builder.add_node(respond_node, respond_node) builder.add_edge(START, respond_node) builder.add_edge(respond_node, END) graph builder.compile(checkpointercheckpointer) # 第一輪對(duì)話 result1 graph.invoke( {user_input: 你好}, config{configurable: {thread_id: user-123}} ) print(result1[response]) # 第二輪對(duì)話 result2 graph.invoke( {user_input: 幫我查訂單}, config{configurable: {thread_id: user-123}} ) print(result2[response])注意thread_id的使用。它相當(dāng)于一個(gè)會(huì)話標(biāo)識(shí)同一個(gè)thread_id下的多輪調(diào)用會(huì)共享狀態(tài)。8.3 生產(chǎn)環(huán)境 Checkpoint 選型InMemorySaver只適合測(cè)試和單機(jī)場(chǎng)景。生產(chǎn)環(huán)境中建議使用持久化存儲(chǔ)LangGraph 官方提供了 Postgres 等存儲(chǔ)方案的適配你也可以把 Checkpoint 數(shù)據(jù)寫入 Redis 或自研存儲(chǔ)。核心原則是State 具備可恢復(fù)性是生產(chǎn)級(jí)工作流的基本要求。9. 將 Checkpoint 內(nèi)容傳入大模型上下文9.1 一個(gè)容易被忽略的問題很多人在使用 LangGraph 做對(duì)話應(yīng)用時(shí)會(huì)直接拿state[messages]作為模型上下文。但如果你使用了 Checkpointmessages 可能已經(jīng)被序列化或包含了一些內(nèi)部字段直接傳入模型并不合適。更穩(wěn)妥的做法是從 Checkpoint 中取回歷史對(duì)話摘要再手動(dòng)組裝成模型需要的 messages 格式。9.2 示例從 Checkpoint 構(gòu)建模型上下文def build_context_from_history(state: ConversationState) - list[dict]: context [] for item in state.get(history, []): if user in item: context.append({role: user, content: item[user]}) elif ai in item: context.append({role: assistant, content: item[ai]}) return context # 使用方法 def respond_with_context(state: ConversationState) - ConversationState: history state.get(history, []) context build_context_from_history(state) context.append({role: user, content: state[user_input]}) # 此時(shí)把 context 傳給大模型即可 response f已收到當(dāng)前上下文共 {len(context)} 條消息。 history.append({user: state[user_input]}) history.append({ai: response}) return {response: response, history: history}要點(diǎn)是不要直接拿內(nèi)部 State 結(jié)構(gòu)當(dāng)模型輸入要顯式組裝。這樣可以避免字段污染也方便你接入不同的模型接口。10. 常見問題與排查思路問題現(xiàn)象可能原因排查方式解決方案安裝失敗依賴版本沖突查看 pip 錯(cuò)誤信息使用pip check在干凈虛擬環(huán)境中安裝統(tǒng)一 langchain 相關(guān)包版本節(jié)點(diǎn)返回值沒有被更新返回的 dict 鍵名與 State 定義不一致打印 State 內(nèi)容確認(rèn)鍵名嚴(yán)格使用 State 定義的鍵名運(yùn)行時(shí)報(bào) KeyError某節(jié)點(diǎn)讀取了不存在的鍵在節(jié)點(diǎn)函數(shù)中打印state在節(jié)點(diǎn)函數(shù)開頭使用state.get(key)而不是state[key]工作流無限循環(huán)條件邊始終返回同一個(gè)節(jié)點(diǎn)檢查路由函數(shù)的返回值和節(jié)點(diǎn)映射設(shè)置recursion_limit并檢查條件邏輯多輪對(duì)話不記住上下文沒有配置 checkpointer 或 thread_id 不一致檢查編譯時(shí)是否傳入 checkpointer編譯時(shí)傳入checkpointer...invoke 時(shí)傳入一致 thread_id大模型響應(yīng)格式不符合預(yù)期Prompt 約束不足打印模型原始輸出在 Prompt 中限定輸出格式并增加兜底解析并行分支結(jié)果丟失聚合節(jié)點(diǎn)讀取鍵名錯(cuò)誤打印聚合節(jié)點(diǎn)前的 State 內(nèi)容核對(duì)每個(gè)分支節(jié)點(diǎn)返回的鍵名是否與 State 定義一致11. LangGraph 多智能體最佳實(shí)踐建議11.1 圖設(shè)計(jì)原則不要在單個(gè)圖里堆太多節(jié)點(diǎn)。經(jīng)驗(yàn)閾值是如果主圖節(jié)點(diǎn)超過 10 個(gè)建議拆分子圖。每個(gè)子圖只負(fù)責(zé)一個(gè)領(lǐng)域比如“日志診斷子圖”“訂單處理子圖”“合規(guī)審查子圖”。11.2 狀態(tài)設(shè)計(jì)原則State 字段越少越好。只放需要跨節(jié)點(diǎn)共享的數(shù)據(jù)不要把所有中間變量都塞進(jìn) State。比較常見的做法是將中間變量用局部變量保存只把關(guān)鍵結(jié)果放入 State。11.3 安全邊界大模型生成內(nèi)容可能包含違規(guī)信息生產(chǎn)環(huán)境應(yīng)加入內(nèi)容審核節(jié)點(diǎn)。涉及用戶隱私數(shù)據(jù)時(shí)State 中不應(yīng)保存明文敏感信息必要時(shí)應(yīng)脫敏后再寫入 Checkpoint。工具調(diào)用節(jié)點(diǎn)應(yīng)遵循最小權(quán)限原則不要給智能體過寬的權(quán)限邊界。11.4 可觀測(cè)性在關(guān)鍵節(jié)點(diǎn)增加日志輸出記錄節(jié)點(diǎn)名、輸入 State 摘要、輸出結(jié)果。LangGraph 生態(tài)中有 LangSmith 等觀測(cè)工具但在沒有接入的情況下最簡(jiǎn)單的做法是def node_with_log(state: State) - State: print(f進(jìn)入節(jié)點(diǎn)node_with_log) print(f輸入字段{state.keys()}) result do_something(state) print(f節(jié)點(diǎn)輸出{result}) return result11.5 從 Java 或其他語言接入 LangGraph很多同學(xué)問“Java 怎么使用 LangGraph”。目前 LangGraph 官方有 JavaScript 版本如果你的團(tuán)隊(duì)是 Java 技術(shù)棧更推薦的方式是把 LangGraph 工作流封裝成獨(dú)立服務(wù)通過 HTTP API 暴露給 Java 后端調(diào)用。這樣既能使用 LangGraph 的編排能力又不會(huì)和現(xiàn)有 Java 體系沖突。12. 總結(jié)與下一步學(xué)習(xí)建議LangGraph 多智能體開發(fā)的核心不是“讓模型變強(qiáng)”而是把大模型調(diào)用、工具調(diào)用、人工審核、狀態(tài)恢復(fù)等環(huán)節(jié)用工程化方式組織起來。State、Node、Edge、Conditional Edge、Checkpoint 這五個(gè)組件構(gòu)成了 LangGraph 的最小認(rèn)知框架本文的示例基本都圍繞這五個(gè)組件展開。下一步你可以做三件事第一把本文的客服工單分流示例改造成使用真實(shí)大模型并替換成你熟悉的數(shù)據(jù)源和工具調(diào)用。這一步能幫你真正理解 State 和 Condition Edge 的配合。第二嘗試在一個(gè)項(xiàng)目中加入子圖和并行分支體會(huì)模塊化拆分帶來的維護(hù)收益。第三把內(nèi)存 Checkpoint 換成持久化方案并接入你自己的日志體系讓工作流從“能跑”變成“可運(yùn)維”。LangGraph 的版本迭代很快新概念也在持續(xù)出現(xiàn)但核心的圖編排思想不會(huì)變。只要把這一套狀態(tài)機(jī)和條件路由的思維方式掌握了不管未來框架怎么升級(jí)你都能快速上手。