戰(zhàn)：構(gòu)建可控、有狀態(tài)的Agent工作流)
這段時(shí)間后臺(tái)收到不少關(guān)于 Agent 開(kāi)發(fā)的私信問(wèn)得最多的就是LangChain 我能跑通但一涉及 Agent 循環(huán)、條件分支、狀態(tài)持久化就不知道怎么組織代碼了。市面上不少教程要么只講概念要么直接甩一段看不出全貌的代碼。這次我們直接看 LangGraph一個(gè)專(zhuān)門(mén)為有狀態(tài) Agent 工作流設(shè)計(jì)的圖編排框架。它并不是取代 LangChain而是把 LangChain 里的模型調(diào)用、工具調(diào)用、記憶管理組合成一個(gè)可控制的圖結(jié)構(gòu)。你能在圖上定義節(jié)點(diǎn)、連邊、條件路由、循環(huán)檢測(cè)甚至把多個(gè)子圖嵌套在一起。這篇文章會(huì)沿著“核心概念 - 環(huán)境安裝 - 手寫(xiě) Agent 循環(huán) - 條件路由 - RAG 實(shí)戰(zhàn) - API 服務(wù)與批量任務(wù)”的順序走一遍并提供可直接復(fù)制的代碼。讀完你能回答三個(gè)問(wèn)題LangGraph 到底解決什么問(wèn)題、它能跑在什么硬件上、怎么把它接到自己的業(yè)務(wù)里。1. 核心能力速覽能力項(xiàng)說(shuō)明項(xiàng)目類(lèi)型開(kāi)源 Agent 工作流編排框架核心定位基于圖結(jié)構(gòu)構(gòu)建有狀態(tài)、可控制、可恢復(fù)的 AI Agent與 LangChain 關(guān)系構(gòu)建在 LangChain 生態(tài)之上復(fù)用其模型、工具、檢索組件主要功能節(jié)點(diǎn)編排、條件路由、循環(huán)控制、圖狀態(tài)管理、人機(jī)交互斷點(diǎn)、子圖嵌套支持的模型來(lái)源OpenAI 兼容接口、Ollama 本地模型、各類(lèi)在線模型 API支持平臺(tái)Windows / Linux / macOSPython 3.9啟動(dòng)方式Python 腳本 / FastAPI 服務(wù) / LangGraph Studio是否支持 API支持可封裝為 REST API 或接入已有 Web 服務(wù)是否支持批量任務(wù)支持可在代碼中循環(huán)調(diào)用或設(shè)計(jì)并發(fā)隊(duì)列顯存要求取決于底層模型純編排框架本身不占用 GPU使用本地模型時(shí)以模型大小為基準(zhǔn)適合人群想從“單輪模型調(diào)用”進(jìn)階到“多步 Agent 工作流”的開(kāi)發(fā)者從這張表能看出來(lái)LangGraph 的價(jià)值不在模型本身而在“控制邏輯”。如果你已經(jīng)解決了模型調(diào)用問(wèn)題接下來(lái)真正花時(shí)間的往往是如何把工具調(diào)用、多步推理、失敗重試、并行分支做成穩(wěn)定可維護(hù)的系統(tǒng)這正是 LangGraph 解決的核心問(wèn)題。2. 適用場(chǎng)景與使用邊界2.1 適合解決什么問(wèn)題我從實(shí)際使用角度來(lái)說(shuō)LangGraph 最常見(jiàn)的落地場(chǎng)景是這三類(lèi)。第一類(lèi)是多工具 Agent。比如一個(gè)智能客服它需要先判斷用戶(hù)意圖再?zèng)Q定調(diào)訂單查詢(xún)還是物流查詢(xún)最后把結(jié)果整理成自然語(yǔ)言回復(fù)。這類(lèi)任務(wù)如果有多個(gè)工具調(diào)用鏈?zhǔn)綄?xiě)法會(huì)變得極其難維護(hù)而圖結(jié)構(gòu)天然適合表達(dá)這種“先判斷、再分流、后匯總”的流程。第二類(lèi)是RAG 增強(qiáng)問(wèn)答?；A(chǔ) RAG 是“檢索 - 生成”但實(shí)際業(yè)務(wù)經(jīng)常需要“判斷是否需要檢索 - 檢索 - 判斷答案是否充分 - 不充分就重寫(xiě)問(wèn)題再檢索 - 最終生成”。這個(gè)流程里每個(gè)判斷節(jié)點(diǎn)都是一次模型調(diào)用用 LangGraph 可以讓所有分支一目了然。第三類(lèi)是長(zhǎng)流程任務(wù)。比如批量處理文檔、多步代碼生成、自動(dòng)報(bào)表生成。這些任務(wù)的特點(diǎn)是有中間狀態(tài)、可能要執(zhí)行幾十步并且中途可能失敗。LangGraph 的狀態(tài)管理能讓每一步都可視、可恢復(fù)。2.2 不適合什么場(chǎng)景如果你的需求只是“調(diào)一次模型拿到結(jié)果”用 LangChain 或直接請(qǐng)求 API 就足夠了。強(qiáng)行引入圖編排會(huì)把簡(jiǎn)單問(wèn)題復(fù)雜化。另外如果應(yīng)用是完全實(shí)時(shí)、毫秒級(jí)響應(yīng)的流式交互圖編排會(huì)帶來(lái)額外的調(diào)度開(kāi)銷(xiāo)。雖然 LangGraph 有流式輸出支持但對(duì)極端低延遲場(chǎng)景仍需做壓測(cè)確認(rèn)。2.3 使用邊界與合規(guī)提醒任何 Agent 框架都只是工具落地時(shí)要注意幾點(diǎn)涉及用戶(hù)隱私數(shù)據(jù)時(shí)要確認(rèn)模型服務(wù)部署在哪、數(shù)據(jù)是否會(huì)發(fā)送到外部接口。涉及版權(quán)內(nèi)容時(shí)要有明確的授權(quán)鏈路。Agent 自動(dòng)執(zhí)行的操作要有權(quán)限邊界尤其是接數(shù)據(jù)庫(kù)、發(fā)郵件、操作文件這類(lèi)高權(quán)限動(dòng)作。對(duì)外提供服務(wù)時(shí)要考慮 Prompt 注入風(fēng)險(xiǎn)不能把系統(tǒng)提示詞和工具描述完全暴露給不可信輸入。3. 環(huán)境準(zhǔn)備與前置條件3.1 運(yùn)行時(shí)選擇LangGraph 是 Python 框架對(duì)硬件沒(méi)有強(qiáng)制要求。你可以在沒(méi)有 GPU 的普通開(kāi)發(fā)機(jī)上完成全部工作流編排和調(diào)試。真正消耗 GPU 的是底層大模型這一層可以選用在線 API也可以用 Ollama 在本地運(yùn)行小參數(shù)模型。我的建議是學(xué)習(xí)階段用在線模型 API 或 Ollama 跑 7B 以下模型普通 CPU 也能完成小規(guī)模測(cè)試。生產(chǎn)階段把 LangGraph 服務(wù)和模型服務(wù)分開(kāi)部署模型服務(wù)單獨(dú)分配 GPU。3.2 Python 與依賴(lài)安裝先確認(rèn) Python 版本推薦 3.9 到 3.12。python --version然后創(chuàng)建虛擬環(huán)境并安裝依賴(lài)python -m venv langgraph_env source langgraph_env/bin/activate # Windows 下執(zhí)行 langgraph_env\Scripts\activate pip install --upgrade pip pip install langgraph langchain-core langchain-openai langchain-community如果你要接入本地 Ollama 模型還需要安裝pip install ollama安裝完成后驗(yàn)證版本python -c import langgraph; print(langgraph.__version__)能正常輸出版本號(hào)說(shuō)明環(huán)境已經(jīng)就緒。3.3 模型服務(wù)準(zhǔn)備LangGraph 本身不提供模型需要先有一個(gè)可調(diào)用的模型服務(wù)。這里給兩種常見(jiàn)方案。方案一使用 OpenAI 兼容接口。很多在線模型服務(wù)都提供 OpenAI 格式的接口只需要在環(huán)境變量里配置 API Key 和接口地址。export OPENAI_API_KEYyour-api-key export OPENAI_BASE_URLhttps://your-endpoint/v1方案二使用本地 Ollama。先安裝 Ollama然后拉取一個(gè)小參數(shù)模型ollama pull qwen2.5:7b ollama serve驗(yàn)證模型可用curl http://localhost:11434/api/generate -d {model: qwen2.5:7b, prompt: 你好}本教程后面示例會(huì)同時(shí)兼容這兩種方式你只需要改一行模型初始化代碼。4. 安裝部署與快速搭建第一個(gè) LangGraph4.1 LangGraph 核心概念State、Node、Edge在寫(xiě)代碼之前先花一分鐘理解 LangGraph 的三個(gè)核心概念。State狀態(tài)整個(gè)圖的全局狀態(tài)是一個(gè)數(shù)據(jù)結(jié)構(gòu)。所有節(jié)點(diǎn)都能讀寫(xiě)這個(gè)狀態(tài)。Node節(jié)點(diǎn)一個(gè)處理函數(shù)輸入是當(dāng)前狀態(tài)輸出是更新后的狀態(tài)片段。Edge邊定義節(jié)點(diǎn)之間的轉(zhuǎn)移方向。普通邊是“執(zhí)行完 A 必執(zhí)行 B”條件邊是“根據(jù)狀態(tài)決定下一個(gè)節(jié)點(diǎn)”。用一句話概括LangGraph 就是把你腦子里的流程圖變成代碼讓 Agent 的執(zhí)行過(guò)程可控制、可觀察、可重放。4.2 第一個(gè)示例兩節(jié)點(diǎn)順序執(zhí)行先寫(xiě)一個(gè)最小示例感受一下基本寫(xiě)法。from typing import TypedDict from langgraph.graph import StateGraph, START, END # 1. 定義狀態(tài)結(jié)構(gòu) class MyState(TypedDict): input_text: str output_text: str # 2. 定義節(jié)點(diǎn)函數(shù) def node_a(state: MyState) - dict: print(執(zhí)行 node_a) return {output_text: state[input_text] - 已處理} def node_b(state: MyState) - dict: print(執(zhí)行 node_b) return {output_text: state[output_text] - 已完善} # 3. 構(gòu)建圖 graph StateGraph(MyState) graph.add_node(node_a, node_a) graph.add_node(node_b, node_b) graph.add_edge(START, node_a) graph.add_edge(node_a, node_b) graph.add_edge(node_b, END) # 4. 編譯并執(zhí)行 app graph.compile() result app.invoke({input_text: hello}) print(result)運(yùn)行后可以看到輸出執(zhí)行 node_a 執(zhí)行 node_b {input_text: hello, output_text: hello - 已處理 - 已完善}從這段代碼能看出 LangGraph 的基本套路定義狀態(tài) - 定義節(jié)點(diǎn)函數(shù) - 加節(jié)點(diǎn)、加邊 - 編譯 - 調(diào)用。后面所有復(fù)雜功能都是在這個(gè)基礎(chǔ)上擴(kuò)展。4.3 接入真實(shí)模型讓 Agent 具備“思考”能力現(xiàn)在把上面的示例升級(jí)為真正的 Agent引入大模型調(diào)用。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage, AIMessage # 使用 OpenAI 兼容接口 llm ChatOpenAI( modelqwen2.5:7b, base_urlhttp://localhost:11434/v1, # Ollama 的 OpenAI 兼容端點(diǎn) api_keyollama, # Ollama 不需要真實(shí)密鑰隨便填即可 temperature0.7 ) class AgentState(TypedDict): messages: Annotated[list, lambda x, y: x y] def call_model(state: AgentState) - dict: response llm.invoke(state[messages]) return {messages: [response]} graph StateGraph(AgentState) graph.add_node(agent, call_model) graph.add_edge(START, agent) graph.add_edge(agent, END) app graph.compile() result app.invoke({ messages: [ SystemMessage(content你是一個(gè)樂(lè)于助人的助手。), HumanMessage(content用一句話介紹大模型 Agent 是什么。) ] }) for msg in result[messages]: print(f{msg.type}: {msg.content})這個(gè)示例說(shuō)明了一個(gè)關(guān)鍵點(diǎn)LangGraph 的 State 可以是消息列表模型調(diào)用只是圖里的一個(gè)節(jié)點(diǎn)。你可以在這個(gè)節(jié)點(diǎn)前后加工具調(diào)用、加判斷邏輯形成一個(gè)完整的執(zhí)行鏈路。5. 功能測(cè)試與效果驗(yàn)證5.1 測(cè)試目標(biāo)對(duì)于 LangGraph功能測(cè)試不能只看“能不能返回文本”要看這些維度狀態(tài)是否按預(yù)期流轉(zhuǎn)。條件分支是否走對(duì)路徑。循環(huán)是否有退出條件。工具調(diào)用是否能正確傳入?yún)?shù)。異常時(shí)是否能恢復(fù)。下面的章節(jié)會(huì)按這個(gè)思路逐項(xiàng)驗(yàn)證。5.2 基礎(chǔ) Agent 循環(huán)測(cè)試讓 Agent 可以自主調(diào)用工具只調(diào)一次模型不是 Agent。真正的 Agent 應(yīng)該能“思考 - 決定調(diào)工具 - 看到結(jié)果 - 再思考”直到得出最終答案。下面實(shí)現(xiàn)一個(gè)最簡(jiǎn)單的 ReAct 循環(huán)。import json from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, ToolMessage # 定義一個(gè)計(jì)算器工具 def calculator(expression: str) - str: 計(jì)算數(shù)學(xué)表達(dá)式例如 1 2 * 3。 try: # 注意生產(chǎn)環(huán)境不要直接 eval這里僅做演示 result eval(expression) return str(result) except Exception as e: return f計(jì)算失敗: {str(e)} tools [calculator] llm ChatOpenAI( modelqwen2.5:7b, base_urlhttp://localhost:11434/v1, api_keyollama, temperature0 ).bind_tools(tools) class AgentState(TypedDict): messages: Annotated[list, lambda x, y: x y] def call_agent(state: AgentState) - dict: response llm.invoke(state[messages]) return {messages: [response]} def should_continue(state: AgentState) - str: last_message state[messages][-1] # 如果模型返回了工具調(diào)用請(qǐng)求就進(jìn)入工具節(jié)點(diǎn) if last_message.tool_calls: return continue return end graph StateGraph(AgentState) graph.add_node(agent, call_agent) graph.add_node(tools, ToolNode(tools)) graph.add_edge(START, agent) graph.add_conditional_edges( agent, should_continue, { continue: tools, end: END } ) graph.add_edge(tools, agent) # 工具執(zhí)行完回到 agent app graph.compile() result app.invoke({ messages: [ SystemMessage(content你是計(jì)算助手需要計(jì)算時(shí)調(diào)用 calculator 工具。), HumanMessage(content計(jì)算 (12 34) * 5 的結(jié)果) ] }) for msg in result[messages]: print(f--- {msg.type} ---) print(msg.content) if msg.tool_calls: print(Tool calls:, msg.tool_calls)運(yùn)行后觀察輸出應(yīng)該能看到完整的循環(huán)過(guò)程agent節(jié)點(diǎn)返回一個(gè)帶 tool_calls 的消息。should_continue判斷為continue進(jìn)入tools。tools執(zhí)行計(jì)算器返回 ToolMessage?；氐絘gent模型基于工具結(jié)果生成最終答案。should_continue判斷為end流程結(jié)束。這個(gè)就是 Agent 循環(huán)的骨架。后面加再多的工具、再?gòu)?fù)雜的邏輯核心結(jié)構(gòu)都不會(huì)變。5.3 條件路由與分支控制測(cè)試實(shí)際業(yè)務(wù)里不是每次都需要調(diào)用工具。更合理的流程是模型先判斷問(wèn)題是否需要工具需要就走工具分支不需要直接回答。這就是條件路由。from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class RouteState(TypedDict): question: str need_tool: bool answer: str def judge(state: RouteState) - dict: 模擬模型判斷是否需要工具。 # 實(shí)際項(xiàng)目中這里可以調(diào)用模型做意圖識(shí)別 if 計(jì)算 in state[question] or 多少 in state[question]: return {need_tool: True} return {need_tool: False} def use_tool(state: RouteState) - dict: return {answer: f【工具計(jì)算】{state[question]} 的答案是 100} def direct_answer(state: RouteState) - dict: return {answer: f【直接回答】{state[question]}} def route_by_need(state: RouteState) - Literal[tool, direct]: if state[need_tool]: return tool return direct graph StateGraph(RouteState) graph.add_node(judge, judge) graph.add_node(tool_node, use_tool) graph.add_node(direct_node, direct_answer) graph.add_edge(START, judge) graph.add_conditional_edges( judge, route_by_need, { tool: tool_node, direct: direct_node } ) graph.add_edge(tool_node, END) graph.add_edge(direct_node, END) app graph.compile() print(app.invoke({question: 計(jì)算 3 * 2})) print(app.invoke({question: 你好}))測(cè)試結(jié)果{question: 計(jì)算 3 * 2, need_tool: True, answer: 【工具計(jì)算】計(jì)算 3 * 2 的答案是 100} {question: 你好, need_tool: False, answer: 【直接回答】你好}注意add_conditional_edges就是 LangGraph 的“條件路由”核心 API。你只需要寫(xiě)一個(gè)返回字符串的函數(shù)根據(jù)返回值映射到不同節(jié)點(diǎn)。5.4 循環(huán)檢測(cè)與最大步數(shù)限制Agent 循環(huán)最怕的是“死循環(huán)”。LangGraph 本身不會(huì)無(wú)限執(zhí)行它有遞歸限制默認(rèn)情況下超過(guò)限制會(huì)報(bào)錯(cuò)。更穩(wěn)妥的做法是在狀態(tài)中記錄輪次主動(dòng)退出。class LoopState(TypedDict): messages: list step_count: int def agent_step(state: LoopState) - dict: # 模擬一次 Agent 處理 new_step state[step_count] 1 if new_step 5: return { step_count: new_step, messages: state[messages] [已達(dá)到最大輪次強(qiáng)制停止] } return { step_count: new_step, messages: state[messages] [f第 {new_step} 輪] } def should_stop(state: LoopState) - str: if state[step_count] 5: return end return continue graph StateGraph(LoopState) graph.add_node(agent, agent_step) graph.add_edge(START, agent) graph.add_conditional_edges( agent, should_stop, { continue: agent, end: END } ) app graph.compile() result app.invoke({messages: [], step_count: 0}) print(result[messages])這種“顯式記錄輪次 條件退出”的模式在生產(chǎn)環(huán)境里非常實(shí)用建議代碼里強(qiáng)制保留。5.5 子圖與并行分支測(cè)試當(dāng)流程復(fù)雜后可以把一個(gè)完整流程封裝成子圖再嵌入到父圖節(jié)點(diǎn)中。下面演示子圖的用法。# 先構(gòu)建一個(gè)子圖負(fù)責(zé)文本清清洗 from langgraph.graph import StateGraph, START, END class CleanState(TypedDict): raw_text: str clean_text: str def clean_step(state: CleanState) - dict: return {clean_text: state[raw_text].strip()} subgraph StateGraph(CleanState) subgraph.add_node(clean, clean_step) subgraph.add_edge(START, clean) subgraph.add_edge(clean, END) clean_app subgraph.compile() # 父圖引用子圖 class ParentState(TypedDict): raw_text: str clean_text: str final_text: str def use_subgraph(state: ParentState) - dict: sub_result clean_app.invoke({raw_text: state[raw_text]}) return {clean_text: sub_result[clean_text]} def final_step(state: ParentState) - dict: return {final_text: f最終結(jié)果: {state[clean_text]}} parent_graph StateGraph(ParentState) parent_graph.add_node(clean_node, use_subgraph) parent_graph.add_node(final, final_step) parent_graph.add_edge(START, clean_node) parent_graph.add_edge(clean_node, final) parent_graph.add_edge(final, END) parent_app parent_graph.compile() result parent_app.invoke({raw_text: 需要清洗的文本 }) print(result)子圖的價(jià)值在于復(fù)用。你可以把“工具調(diào)用循環(huán)”“RAG 檢索”“報(bào)告生成”分別封裝成獨(dú)立子圖組合成不同的業(yè)務(wù)應(yīng)用。并行分支方面LangGraph 的節(jié)點(diǎn)只要不依賴(lài)彼此狀態(tài)可以在不同邊中并行執(zhí)行然后通過(guò)一個(gè)匯聚節(jié)點(diǎn)合并結(jié)果。對(duì)于并行工具調(diào)用場(chǎng)景LangChain 的ToolNode本身就支持一次返回多個(gè)工具調(diào)用這也是常見(jiàn)做法。6. 接口 API 與批量任務(wù)6.1 用 FastAPI 封裝 Agent 服務(wù)LangGraph 編譯后的app可以直接在 Python 進(jìn)程內(nèi)調(diào)用也可以封裝成 REST API。下面是一個(gè)最小可用的 FastAPI 服務(wù)from fastapi import FastAPI from pydantic import BaseModel from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage app FastAPI(titleLangGraph Demo Service) llm ChatOpenAI( modelqwen2.5:7b, base_urlhttp://localhost:11434/v1, api_keyollama, temperature0.7 ) class ChatState(TypedDict): messages: list def call_model(state: ChatState) - dict: response llm.invoke(state[messages]) return {messages: [response]} graph StateGraph(ChatState) graph.add_node(agent, call_model) graph.add_edge(START, agent) graph.add_edge(agent, END) agent_app graph.compile() class ChatRequest(BaseModel): message: str class ChatResponse(BaseModel): reply: str app.post(/chat, response_modelChatResponse) def chat(req: ChatRequest): result agent_app.invoke({ messages: [ SystemMessage(content你是一個(gè)簡(jiǎn)潔的助手。), HumanMessage(contentreq.message) ] }) return ChatResponse(replyresult[messages][-1].content) # 啟動(dòng)方式uvicorn main:app --host 0.0.0.0 --port 8000啟動(dòng)后用 curl 測(cè)試curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {message: 你好請(qǐng)簡(jiǎn)單介紹一下你自己}返回示例{reply: 你好我是一個(gè)基于 LangGraph 構(gòu)建的 AI 助手。}把 Agent 封裝成 API 后前端、后端、自動(dòng)化腳本都能直接對(duì)接。6.2 批量任務(wù)的實(shí)現(xiàn)思路LangGraph 沒(méi)有內(nèi)置任務(wù)隊(duì)列但批量處理的核心邏輯很簡(jiǎn)單循環(huán)調(diào)用 錯(cuò)誤處理 結(jié)果匯總。import time from concurrent.futures import ThreadPoolExecutor, as_completed questions [ 什么是 Agent, 計(jì)算 12 34, 如何學(xué)習(xí) LangGraph, 計(jì)算 100 / 4, ] def process_one(q: str) - dict: try: result agent_app.invoke({ messages: [ SystemMessage(content你是一個(gè)簡(jiǎn)潔的助手。), HumanMessage(contentq) ] }) return {question: q, answer: result[messages][-1].content, status: success} except Exception as e: return {question: q, answer: str(e), status: failed} # 串行執(zhí)行 start time.time() results [process_one(q) for q in questions] print(f串行耗時(shí): {time.time() - start:.2f}s) # 并行執(zhí)行 start time.time() with ThreadPoolExecutor(max_workers4) as executor: futures [executor.submit(process_one, q) for q in questions] parallel_results [f.result() for f in as_completed(futures)] print(f并行耗時(shí): {time.time() - start:.2f}s) for r in parallel_results: print(r)批量任務(wù)有幾個(gè)工程化要點(diǎn)每個(gè)任務(wù)要捕獲異常不能因?yàn)橐粭l失敗而中斷整個(gè)批次。并發(fā)數(shù)要控制避免把模型服務(wù)的請(qǐng)求隊(duì)列打滿(mǎn)。結(jié)果要落盤(pán)或?qū)憥?kù)方便失敗重試和效果復(fù)盤(pán)。對(duì)耗時(shí)較長(zhǎng)的任務(wù)可以加超時(shí)控制。6.3 通過(guò) Checkpoint 實(shí)現(xiàn)持久化與恢復(fù)LangGraph 一個(gè)重要特性是狀態(tài)檢查點(diǎn)。通過(guò)MemorySaver或數(shù)據(jù)庫(kù)持久化可以在中斷后恢復(fù)執(zhí)行。from langgraph.checkpoint.memory import MemorySaver # 編譯時(shí)傳入 checkpointer memory MemorySaver() app_with_memory graph.compile(checkpointermemory) # 第一次執(zhí)行帶 thread_id相當(dāng)于一個(gè)會(huì)話 ID config {configurable: {thread_id: session-001}} app_with_memory.invoke( {messages: [HumanMessage(content你好)]}, configconfig ) # 第二次執(zhí)行同一個(gè) thread_id會(huì)帶上之前的狀態(tài) app_with_memory.invoke( {messages: [HumanMessage(content我剛才問(wèn)了什么)]}, configconfig )這個(gè)特性非常實(shí)用相當(dāng)于給 Agent 加上了“工作記憶”。生產(chǎn)環(huán)境建議把MemorySaver換成基于 Redis 或數(shù)據(jù)庫(kù)的持久化方案這樣服務(wù)重啟后狀態(tài)也不會(huì)丟失。7. 資源占用與性能觀察7.1 不同運(yùn)行方式下的資源分布LangGraph 本身是純 Python 圖編排邏輯CPU 占用很低內(nèi)存占用以幾十到幾百 MB 計(jì)。真正消耗資源的是模型推理服務(wù)在線 API 或本地 Ollama。向量檢索服務(wù)如果做 RAG。長(zhǎng)時(shí)間運(yùn)行時(shí)的狀態(tài)累積。如果使用本地模型7B 量化模型通常需要 6G 到 10G 內(nèi)存14B 以上模型需要更大顯存或內(nèi)存。具體數(shù)字取決于模型量化方式和上下文長(zhǎng)度建議用nvidia-smi或任務(wù)管理器實(shí)時(shí)觀察。7.2 如何降低延遲與顯存占用有幾個(gè)實(shí)際操作方向。第一減少不必要的多輪循環(huán)。Agent 每多一次工具調(diào)用就多一次模型推理。設(shè)計(jì)提示詞時(shí)明確要求模型“能直接回答就不要調(diào)用工具”可以顯著降低平均延遲。第二控制上下文長(zhǎng)度。LangGraph 的 State 會(huì)累積所有消息對(duì)話輪次多了之后每次請(qǐng)求的 token 數(shù)會(huì)快速增長(zhǎng)直接推高延遲和成本。策略是定期摘要?dú)v史消息或裁剪早期消息。第三模型側(cè)優(yōu)化。本地模型可以根據(jù)顯存選擇更小的量化版本降低 temperature 也能減少輸出波動(dòng)。7.3 觀察指標(biāo)建議建議在服務(wù)里加三類(lèi)日志圖流轉(zhuǎn)日志每個(gè)節(jié)點(diǎn)進(jìn)入和退出的時(shí)間。模型調(diào)用日志每次調(diào)用的輸入輸出 token 數(shù)和耗時(shí)。錯(cuò)誤日志工具調(diào)用失敗、超時(shí)、狀態(tài)不一致等問(wèn)題。通過(guò)觀察節(jié)點(diǎn)耗時(shí)分布你很快能定位性能瓶頸是在模型推理還是工具執(zhí)行。8. 常見(jiàn)問(wèn)題與排查方法問(wèn)題現(xiàn)象可能原因排查方式解決方案安裝 langgraph 后 import 報(bào)錯(cuò)Python 版本過(guò)低或依賴(lài)沖突檢查python --version和pip list使用 Python 3.9-3.12重新創(chuàng)建虛擬環(huán)境安裝調(diào)用模型報(bào)連接錯(cuò)誤模型服務(wù)未啟動(dòng)或地址配置錯(cuò)誤先 curl 測(cè)試模型接口啟動(dòng) Ollama 或檢查 base_url、api_keyAgent 不調(diào)用工具模型沒(méi)綁定工具或提示詞不清楚打印模型返回的 tool_calls使用 bind_tools并在提示詞中明說(shuō)可用工具死循環(huán)或一直運(yùn)行缺少循環(huán)退出條件查看日志中節(jié)點(diǎn)流轉(zhuǎn)次數(shù)在 State 中加入 step_count設(shè)置最大輪次工具返回結(jié)果沒(méi)有傳給模型狀態(tài)字段沒(méi)有包含 ToolMessage打印 messages 列表確保 ToolNode 返回的消息被正確追加進(jìn) State條件路由走了錯(cuò)誤的邊路由函數(shù)返回值與映射表不一致打印路由函數(shù)返回值檢查 add_conditional_edges 中的映射 keyAPI 服務(wù)并發(fā)高時(shí)報(bào)錯(cuò)模型服務(wù)并發(fā)受限查看模型服務(wù)日志控制線程池大小或引入任務(wù)隊(duì)列服務(wù)重啟后會(huì)話丟失未配置持久化 Checkpoint檢查 checkpointer 參數(shù)使用 Redis 或數(shù)據(jù)庫(kù) CheckpointSaver上下文越來(lái)越長(zhǎng)響應(yīng)變慢State 中消息持續(xù)累積打印每次請(qǐng)求的 token 數(shù)做歷史摘要或裁剪LangGraph 版本升級(jí)后 API 報(bào)錯(cuò)API 變更查看官方文檔變更日志固定版本號(hào)不要隨意升級(jí)除了表格里的方案再補(bǔ)充兩個(gè)定位問(wèn)題的實(shí)用技巧。第一個(gè)是“拆開(kāi)調(diào)試”。把一個(gè)長(zhǎng)流程拆成多個(gè)子圖分別測(cè)試每個(gè)子圖的輸入輸出。LangGraph 的圖結(jié)構(gòu)允許單獨(dú)編譯調(diào)用子圖這比整個(gè)跑一大段流程更容易定位問(wèn)題。第二個(gè)是“打印中間狀態(tài)”。在節(jié)點(diǎn)函數(shù)里用print(state)或日志記錄狀態(tài)變化能非常直觀地看到每一步數(shù)據(jù)流轉(zhuǎn)是否符合預(yù)期。9. 最佳實(shí)踐與使用建議9.1 工程落地建議經(jīng)過(guò)實(shí)際項(xiàng)目驗(yàn)證我建議把下面幾條作為默認(rèn)規(guī)范。第一狀態(tài)結(jié)構(gòu)要精簡(jiǎn)。State 里只放節(jié)點(diǎn)之間需要傳遞的字段。一些臨時(shí)變量不要塞進(jìn)圖狀態(tài)里否則會(huì)干擾可視化調(diào)試也容易造成內(nèi)存增長(zhǎng)。第二工具函數(shù)要負(fù)責(zé)。工具節(jié)點(diǎn)不要只定義函數(shù)要給工具寫(xiě)清晰的 docstring 和參數(shù)說(shuō)明。LangGraph 的工具綁定依賴(lài)模型理解工具描述描述寫(xiě)得越清楚模型用錯(cuò)的概率越低。第三條件路由要顯式。所有add_conditional_edges的返回值和映射表要保證完全匹配并提供一個(gè)默認(rèn)的 fallback 分支。第四盡早引入 Checkpoint。哪怕開(kāi)發(fā)階段不用設(shè)計(jì)時(shí)也要先留出thread_id的傳遞鏈路。后期加持久化會(huì)容易得多。第五日志和追蹤不能省。LangGraph 提供了大量回調(diào)接口建議從第一天就接入 LangSmith 或自建日志系統(tǒng)。出了問(wèn)題能快速回放執(zhí)行過(guò)程。9.2 安全與合規(guī)建議Agent 自動(dòng)調(diào)用工具有一個(gè)容易被忽視的風(fēng)險(xiǎn)如果模型被惡意 Prompt 誘導(dǎo)可能會(huì)執(zhí)行非預(yù)期的工具操作。因此工具權(quán)限要最小化關(guān)鍵操作必須二次確認(rèn)。涉及外部請(qǐng)求時(shí)對(duì) URL、文件路徑等參數(shù)做校驗(yàn)。涉及用戶(hù)數(shù)據(jù)時(shí)確認(rèn)數(shù)據(jù)不出域。涉及版權(quán)內(nèi)容和肖像授權(quán)時(shí)必須核實(shí)授權(quán)鏈路。9.3 學(xué)習(xí)路徑建議如果你剛接觸 LangGraph建議按這個(gè)順序?qū)W習(xí)先把本章的“兩節(jié)點(diǎn)順序執(zhí)行”跑通。再實(shí)現(xiàn)一個(gè)帶工具調(diào)用的 Agent 循環(huán)。然后加條件路由和循環(huán)限制。接著把流程拆成子圖。最后封裝成 API 并接入業(yè)務(wù)。不要一上來(lái)就復(fù)制別人的復(fù)雜項(xiàng)目。圖編排的核心是“你會(huì)不會(huì)拆流程”而不是“你記沒(méi)記住 API”。10. 總結(jié)與下一步LangGraph 最值得嘗試的點(diǎn)在于它把 Agent 開(kāi)發(fā)從“自由發(fā)揮的腳本”變成了“結(jié)構(gòu)清晰的工程圖”。比 LangChain 的鏈?zhǔn)秸{(diào)用更靈活比手寫(xiě) Agent 循環(huán)更規(guī)范。如果你已經(jīng)在做 AI Agent 開(kāi)發(fā)這個(gè)框架值得投入時(shí)間。建議你在本地完成兩件事先運(yùn)行一遍本文的 ReAct Agent 循環(huán)確認(rèn)模型調(diào)用、工具調(diào)用、條件退出整條鏈路能跑通然后把一個(gè)你手頭已有的業(yè)務(wù)場(chǎng)景改造成 LangGraph 版本對(duì)比兩者的維護(hù)成本差異。最容易踩的坑集中在條件路由返回值不匹配、Agent 循環(huán)缺少退出條件、工具調(diào)用后消息傳遞斷裂。這三個(gè)問(wèn)題占了 LangGraph 開(kāi)發(fā)階段的大部分報(bào)錯(cuò)遇到時(shí)不要慌打印中間狀態(tài)就能定位。下一步可以擴(kuò)展的方向很多接入多模態(tài)模型、增加人工審核節(jié)點(diǎn)、把 Checkpoint 換成 Redis 持久化、用子圖重組回答生成流程甚至是結(jié)合向量數(shù)據(jù)庫(kù)做一套完整的 RAG Agent 服務(wù)。圖結(jié)構(gòu)的好處是每次擴(kuò)展都只需要新增節(jié)點(diǎn)和邊不需要重寫(xiě)整個(gè)框架。這篇文章建議收藏備用遇到問(wèn)題時(shí)回來(lái)翻一翻排查表能省不少時(shí)間。 LangGraph實(shí)戰(zhàn)：構(gòu)建可控、有狀態(tài)的Agent工作流
這段時(shí)間后臺(tái)收到不少關(guān)于 Agent 開(kāi)發(fā)的私信問(wèn)得最多的就是LangChain 我能跑通但一涉及 Agent 循環(huán)、條件分支、狀態(tài)持久化就不知道怎么組織代碼了。市面上不少教程要么只講概念要么直接甩一段看不出全貌的代碼。這次我們直接看 LangGraph一個(gè)專(zhuān)門(mén)為有狀態(tài) Agent 工作流設(shè)計(jì)的圖編排框架。它并不是取代 LangChain而是把 LangChain 里的模型調(diào)用、工具調(diào)用、記憶管理組合成一個(gè)可控制的圖結(jié)構(gòu)。你能在圖上定義節(jié)點(diǎn)、連邊、條件路由、循環(huán)檢測(cè)甚至把多個(gè)子圖嵌套在一起。這篇文章會(huì)沿著“核心概念 - 環(huán)境安裝 - 手寫(xiě) Agent 循環(huán) - 條件路由 - RAG 實(shí)戰(zhàn) - API 服務(wù)與批量任務(wù)”的順序走一遍并提供可直接復(fù)制的代碼。讀完你能回答三個(gè)問(wèn)題LangGraph 到底解決什么問(wèn)題、它能跑在什么硬件上、怎么把它接到自己的業(yè)務(wù)里。1. 核心能力速覽能力項(xiàng)說(shuō)明項(xiàng)目類(lèi)型開(kāi)源 Agent 工作流編排框架核心定位基于圖結(jié)構(gòu)構(gòu)建有狀態(tài)、可控制、可恢復(fù)的 AI Agent與 LangChain 關(guān)系構(gòu)建在 LangChain 生態(tài)之上復(fù)用其模型、工具、檢索組件主要功能節(jié)點(diǎn)編排、條件路由、循環(huán)控制、圖狀態(tài)管理、人機(jī)交互斷點(diǎn)、子圖嵌套支持的模型來(lái)源OpenAI 兼容接口、Ollama 本地模型、各類(lèi)在線模型 API支持平臺(tái)Windows / Linux / macOSPython 3.9啟動(dòng)方式Python 腳本 / FastAPI 服務(wù) / LangGraph Studio是否支持 API支持可封裝為 REST API 或接入已有 Web 服務(wù)是否支持批量任務(wù)支持可在代碼中循環(huán)調(diào)用或設(shè)計(jì)并發(fā)隊(duì)列顯存要求取決于底層模型純編排框架本身不占用 GPU使用本地模型時(shí)以模型大小為基準(zhǔn)適合人群想從“單輪模型調(diào)用”進(jìn)階到“多步 Agent 工作流”的開(kāi)發(fā)者從這張表能看出來(lái)LangGraph 的價(jià)值不在模型本身而在“控制邏輯”。如果你已經(jīng)解決了模型調(diào)用問(wèn)題接下來(lái)真正花時(shí)間的往往是如何把工具調(diào)用、多步推理、失敗重試、并行分支做成穩(wěn)定可維護(hù)的系統(tǒng)這正是 LangGraph 解決的核心問(wèn)題。2. 適用場(chǎng)景與使用邊界2.1 適合解決什么問(wèn)題我從實(shí)際使用角度來(lái)說(shuō)LangGraph 最常見(jiàn)的落地場(chǎng)景是這三類(lèi)。第一類(lèi)是多工具 Agent。比如一個(gè)智能客服它需要先判斷用戶(hù)意圖再?zèng)Q定調(diào)訂單查詢(xún)還是物流查詢(xún)最后把結(jié)果整理成自然語(yǔ)言回復(fù)。這類(lèi)任務(wù)如果有多個(gè)工具調(diào)用鏈?zhǔn)綄?xiě)法會(huì)變得極其難維護(hù)而圖結(jié)構(gòu)天然適合表達(dá)這種“先判斷、再分流、后匯總”的流程。第二類(lèi)是RAG 增強(qiáng)問(wèn)答?；A(chǔ) RAG 是“檢索 - 生成”但實(shí)際業(yè)務(wù)經(jīng)常需要“判斷是否需要檢索 - 檢索 - 判斷答案是否充分 - 不充分就重寫(xiě)問(wèn)題再檢索 - 最終生成”。這個(gè)流程里每個(gè)判斷節(jié)點(diǎn)都是一次模型調(diào)用用 LangGraph 可以讓所有分支一目了然。第三類(lèi)是長(zhǎng)流程任務(wù)。比如批量處理文檔、多步代碼生成、自動(dòng)報(bào)表生成。這些任務(wù)的特點(diǎn)是有中間狀態(tài)、可能要執(zhí)行幾十步并且中途可能失敗。LangGraph 的狀態(tài)管理能讓每一步都可視、可恢復(fù)。2.2 不適合什么場(chǎng)景如果你的需求只是“調(diào)一次模型拿到結(jié)果”用 LangChain 或直接請(qǐng)求 API 就足夠了。強(qiáng)行引入圖編排會(huì)把簡(jiǎn)單問(wèn)題復(fù)雜化。另外如果應(yīng)用是完全實(shí)時(shí)、毫秒級(jí)響應(yīng)的流式交互圖編排會(huì)帶來(lái)額外的調(diào)度開(kāi)銷(xiāo)。雖然 LangGraph 有流式輸出支持但對(duì)極端低延遲場(chǎng)景仍需做壓測(cè)確認(rèn)。2.3 使用邊界與合規(guī)提醒任何 Agent 框架都只是工具落地時(shí)要注意幾點(diǎn)涉及用戶(hù)隱私數(shù)據(jù)時(shí)要確認(rèn)模型服務(wù)部署在哪、數(shù)據(jù)是否會(huì)發(fā)送到外部接口。涉及版權(quán)內(nèi)容時(shí)要有明確的授權(quán)鏈路。Agent 自動(dòng)執(zhí)行的操作要有權(quán)限邊界尤其是接數(shù)據(jù)庫(kù)、發(fā)郵件、操作文件這類(lèi)高權(quán)限動(dòng)作。對(duì)外提供服務(wù)時(shí)要考慮 Prompt 注入風(fēng)險(xiǎn)不能把系統(tǒng)提示詞和工具描述完全暴露給不可信輸入。3. 環(huán)境準(zhǔn)備與前置條件3.1 運(yùn)行時(shí)選擇LangGraph 是 Python 框架對(duì)硬件沒(méi)有強(qiáng)制要求。你可以在沒(méi)有 GPU 的普通開(kāi)發(fā)機(jī)上完成全部工作流編排和調(diào)試。真正消耗 GPU 的是底層大模型這一層可以選用在線 API也可以用 Ollama 在本地運(yùn)行小參數(shù)模型。我的建議是學(xué)習(xí)階段用在線模型 API 或 Ollama 跑 7B 以下模型普通 CPU 也能完成小規(guī)模測(cè)試。生產(chǎn)階段把 LangGraph 服務(wù)和模型服務(wù)分開(kāi)部署模型服務(wù)單獨(dú)分配 GPU。3.2 Python 與依賴(lài)安裝先確認(rèn) Python 版本推薦 3.9 到 3.12。python --version然后創(chuàng)建虛擬環(huán)境并安裝依賴(lài)python -m venv langgraph_env source langgraph_env/bin/activate # Windows 下執(zhí)行 langgraph_env\Scripts\activate pip install --upgrade pip pip install langgraph langchain-core langchain-openai langchain-community如果你要接入本地 Ollama 模型還需要安裝pip install ollama安裝完成后驗(yàn)證版本python -c import langgraph; print(langgraph.__version__)能正常輸出版本號(hào)說(shuō)明環(huán)境已經(jīng)就緒。3.3 模型服務(wù)準(zhǔn)備LangGraph 本身不提供模型需要先有一個(gè)可調(diào)用的模型服務(wù)。這里給兩種常見(jiàn)方案。方案一使用 OpenAI 兼容接口。很多在線模型服務(wù)都提供 OpenAI 格式的接口只需要在環(huán)境變量里配置 API Key 和接口地址。export OPENAI_API_KEYyour-api-key export OPENAI_BASE_URLhttps://your-endpoint/v1方案二使用本地 Ollama。先安裝 Ollama然后拉取一個(gè)小參數(shù)模型ollama pull qwen2.5:7b ollama serve驗(yàn)證模型可用curl http://localhost:11434/api/generate -d {model: qwen2.5:7b, prompt: 你好}本教程后面示例會(huì)同時(shí)兼容這兩種方式你只需要改一行模型初始化代碼。4. 安裝部署與快速搭建第一個(gè) LangGraph4.1 LangGraph 核心概念State、Node、Edge在寫(xiě)代碼之前先花一分鐘理解 LangGraph 的三個(gè)核心概念。State狀態(tài)整個(gè)圖的全局狀態(tài)是一個(gè)數(shù)據(jù)結(jié)構(gòu)。所有節(jié)點(diǎn)都能讀寫(xiě)這個(gè)狀態(tài)。Node節(jié)點(diǎn)一個(gè)處理函數(shù)輸入是當(dāng)前狀態(tài)輸出是更新后的狀態(tài)片段。Edge邊定義節(jié)點(diǎn)之間的轉(zhuǎn)移方向。普通邊是“執(zhí)行完 A 必執(zhí)行 B”條件邊是“根據(jù)狀態(tài)決定下一個(gè)節(jié)點(diǎn)”。用一句話概括LangGraph 就是把你腦子里的流程圖變成代碼讓 Agent 的執(zhí)行過(guò)程可控制、可觀察、可重放。4.2 第一個(gè)示例兩節(jié)點(diǎn)順序執(zhí)行先寫(xiě)一個(gè)最小示例感受一下基本寫(xiě)法。from typing import TypedDict from langgraph.graph import StateGraph, START, END # 1. 定義狀態(tài)結(jié)構(gòu) class MyState(TypedDict): input_text: str output_text: str # 2. 定義節(jié)點(diǎn)函數(shù) def node_a(state: MyState) - dict: print(執(zhí)行 node_a) return {output_text: state[input_text] - 已處理} def node_b(state: MyState) - dict: print(執(zhí)行 node_b) return {output_text: state[output_text] - 已完善} # 3. 構(gòu)建圖 graph StateGraph(MyState) graph.add_node(node_a, node_a) graph.add_node(node_b, node_b) graph.add_edge(START, node_a) graph.add_edge(node_a, node_b) graph.add_edge(node_b, END) # 4. 編譯并執(zhí)行 app graph.compile() result app.invoke({input_text: hello}) print(result)運(yùn)行后可以看到輸出執(zhí)行 node_a 執(zhí)行 node_b {input_text: hello, output_text: hello - 已處理 - 已完善}從這段代碼能看出 LangGraph 的基本套路定義狀態(tài) - 定義節(jié)點(diǎn)函數(shù) - 加節(jié)點(diǎn)、加邊 - 編譯 - 調(diào)用。后面所有復(fù)雜功能都是在這個(gè)基礎(chǔ)上擴(kuò)展。4.3 接入真實(shí)模型讓 Agent 具備“思考”能力現(xiàn)在把上面的示例升級(jí)為真正的 Agent引入大模型調(diào)用。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage, AIMessage # 使用 OpenAI 兼容接口 llm ChatOpenAI( modelqwen2.5:7b, base_urlhttp://localhost:11434/v1, # Ollama 的 OpenAI 兼容端點(diǎn) api_keyollama, # Ollama 不需要真實(shí)密鑰隨便填即可 temperature0.7 ) class AgentState(TypedDict): messages: Annotated[list, lambda x, y: x y] def call_model(state: AgentState) - dict: response llm.invoke(state[messages]) return {messages: [response]} graph StateGraph(AgentState) graph.add_node(agent, call_model) graph.add_edge(START, agent) graph.add_edge(agent, END) app graph.compile() result app.invoke({ messages: [ SystemMessage(content你是一個(gè)樂(lè)于助人的助手。), HumanMessage(content用一句話介紹大模型 Agent 是什么。) ] }) for msg in result[messages]: print(f{msg.type}: {msg.content})這個(gè)示例說(shuō)明了一個(gè)關(guān)鍵點(diǎn)LangGraph 的 State 可以是消息列表模型調(diào)用只是圖里的一個(gè)節(jié)點(diǎn)。你可以在這個(gè)節(jié)點(diǎn)前后加工具調(diào)用、加判斷邏輯形成一個(gè)完整的執(zhí)行鏈路。5. 功能測(cè)試與效果驗(yàn)證5.1 測(cè)試目標(biāo)對(duì)于 LangGraph功能測(cè)試不能只看“能不能返回文本”要看這些維度狀態(tài)是否按預(yù)期流轉(zhuǎn)。條件分支是否走對(duì)路徑。循環(huán)是否有退出條件。工具調(diào)用是否能正確傳入?yún)?shù)。異常時(shí)是否能恢復(fù)。下面的章節(jié)會(huì)按這個(gè)思路逐項(xiàng)驗(yàn)證。5.2 基礎(chǔ) Agent 循環(huán)測(cè)試讓 Agent 可以自主調(diào)用工具只調(diào)一次模型不是 Agent。真正的 Agent 應(yīng)該能“思考 - 決定調(diào)工具 - 看到結(jié)果 - 再思考”直到得出最終答案。下面實(shí)現(xiàn)一個(gè)最簡(jiǎn)單的 ReAct 循環(huán)。import json from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, ToolMessage # 定義一個(gè)計(jì)算器工具 def calculator(expression: str) - str: 計(jì)算數(shù)學(xué)表達(dá)式例如 1 2 * 3。 try: # 注意生產(chǎn)環(huán)境不要直接 eval這里僅做演示 result eval(expression) return str(result) except Exception as e: return f計(jì)算失敗: {str(e)} tools [calculator] llm ChatOpenAI( modelqwen2.5:7b, base_urlhttp://localhost:11434/v1, api_keyollama, temperature0 ).bind_tools(tools) class AgentState(TypedDict): messages: Annotated[list, lambda x, y: x y] def call_agent(state: AgentState) - dict: response llm.invoke(state[messages]) return {messages: [response]} def should_continue(state: AgentState) - str: last_message state[messages][-1] # 如果模型返回了工具調(diào)用請(qǐng)求就進(jìn)入工具節(jié)點(diǎn) if last_message.tool_calls: return continue return end graph StateGraph(AgentState) graph.add_node(agent, call_agent) graph.add_node(tools, ToolNode(tools)) graph.add_edge(START, agent) graph.add_conditional_edges( agent, should_continue, { continue: tools, end: END } ) graph.add_edge(tools, agent) # 工具執(zhí)行完回到 agent app graph.compile() result app.invoke({ messages: [ SystemMessage(content你是計(jì)算助手需要計(jì)算時(shí)調(diào)用 calculator 工具。), HumanMessage(content計(jì)算 (12 34) * 5 的結(jié)果) ] }) for msg in result[messages]: print(f--- {msg.type} ---) print(msg.content) if msg.tool_calls: print(Tool calls:, msg.tool_calls)運(yùn)行后觀察輸出應(yīng)該能看到完整的循環(huán)過(guò)程agent節(jié)點(diǎn)返回一個(gè)帶 tool_calls 的消息。should_continue判斷為continue進(jìn)入tools。tools執(zhí)行計(jì)算器返回 ToolMessage?；氐絘gent模型基于工具結(jié)果生成最終答案。should_continue判斷為end流程結(jié)束。這個(gè)就是 Agent 循環(huán)的骨架。后面加再多的工具、再?gòu)?fù)雜的邏輯核心結(jié)構(gòu)都不會(huì)變。5.3 條件路由與分支控制測(cè)試實(shí)際業(yè)務(wù)里不是每次都需要調(diào)用工具。更合理的流程是模型先判斷問(wèn)題是否需要工具需要就走工具分支不需要直接回答。這就是條件路由。from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class RouteState(TypedDict): question: str need_tool: bool answer: str def judge(state: RouteState) - dict: 模擬模型判斷是否需要工具。 # 實(shí)際項(xiàng)目中這里可以調(diào)用模型做意圖識(shí)別 if 計(jì)算 in state[question] or 多少 in state[question]: return {need_tool: True} return {need_tool: False} def use_tool(state: RouteState) - dict: return {answer: f【工具計(jì)算】{state[question]} 的答案是 100} def direct_answer(state: RouteState) - dict: return {answer: f【直接回答】{state[question]}} def route_by_need(state: RouteState) - Literal[tool, direct]: if state[need_tool]: return tool return direct graph StateGraph(RouteState) graph.add_node(judge, judge) graph.add_node(tool_node, use_tool) graph.add_node(direct_node, direct_answer) graph.add_edge(START, judge) graph.add_conditional_edges( judge, route_by_need, { tool: tool_node, direct: direct_node } ) graph.add_edge(tool_node, END) graph.add_edge(direct_node, END) app graph.compile() print(app.invoke({question: 計(jì)算 3 * 2})) print(app.invoke({question: 你好}))測(cè)試結(jié)果{question: 計(jì)算 3 * 2, need_tool: True, answer: 【工具計(jì)算】計(jì)算 3 * 2 的答案是 100} {question: 你好, need_tool: False, answer: 【直接回答】你好}注意add_conditional_edges就是 LangGraph 的“條件路由”核心 API。你只需要寫(xiě)一個(gè)返回字符串的函數(shù)根據(jù)返回值映射到不同節(jié)點(diǎn)。5.4 循環(huán)檢測(cè)與最大步數(shù)限制Agent 循環(huán)最怕的是“死循環(huán)”。LangGraph 本身不會(huì)無(wú)限執(zhí)行它有遞歸限制默認(rèn)情況下超過(guò)限制會(huì)報(bào)錯(cuò)。更穩(wěn)妥的做法是在狀態(tài)中記錄輪次主動(dòng)退出。class LoopState(TypedDict): messages: list step_count: int def agent_step(state: LoopState) - dict: # 模擬一次 Agent 處理 new_step state[step_count] 1 if new_step 5: return { step_count: new_step, messages: state[messages] [已達(dá)到最大輪次強(qiáng)制停止] } return { step_count: new_step, messages: state[messages] [f第 {new_step} 輪] } def should_stop(state: LoopState) - str: if state[step_count] 5: return end return continue graph StateGraph(LoopState) graph.add_node(agent, agent_step) graph.add_edge(START, agent) graph.add_conditional_edges( agent, should_stop, { continue: agent, end: END } ) app graph.compile() result app.invoke({messages: [], step_count: 0}) print(result[messages])這種“顯式記錄輪次 條件退出”的模式在生產(chǎn)環(huán)境里非常實(shí)用建議代碼里強(qiáng)制保留。5.5 子圖與并行分支測(cè)試當(dāng)流程復(fù)雜后可以把一個(gè)完整流程封裝成子圖再嵌入到父圖節(jié)點(diǎn)中。下面演示子圖的用法。# 先構(gòu)建一個(gè)子圖負(fù)責(zé)文本清清洗 from langgraph.graph import StateGraph, START, END class CleanState(TypedDict): raw_text: str clean_text: str def clean_step(state: CleanState) - dict: return {clean_text: state[raw_text].strip()} subgraph StateGraph(CleanState) subgraph.add_node(clean, clean_step) subgraph.add_edge(START, clean) subgraph.add_edge(clean, END) clean_app subgraph.compile() # 父圖引用子圖 class ParentState(TypedDict): raw_text: str clean_text: str final_text: str def use_subgraph(state: ParentState) - dict: sub_result clean_app.invoke({raw_text: state[raw_text]}) return {clean_text: sub_result[clean_text]} def final_step(state: ParentState) - dict: return {final_text: f最終結(jié)果: {state[clean_text]}} parent_graph StateGraph(ParentState) parent_graph.add_node(clean_node, use_subgraph) parent_graph.add_node(final, final_step) parent_graph.add_edge(START, clean_node) parent_graph.add_edge(clean_node, final) parent_graph.add_edge(final, END) parent_app parent_graph.compile() result parent_app.invoke({raw_text: 需要清洗的文本 }) print(result)子圖的價(jià)值在于復(fù)用。你可以把“工具調(diào)用循環(huán)”“RAG 檢索”“報(bào)告生成”分別封裝成獨(dú)立子圖組合成不同的業(yè)務(wù)應(yīng)用。并行分支方面LangGraph 的節(jié)點(diǎn)只要不依賴(lài)彼此狀態(tài)可以在不同邊中并行執(zhí)行然后通過(guò)一個(gè)匯聚節(jié)點(diǎn)合并結(jié)果。對(duì)于并行工具調(diào)用場(chǎng)景LangChain 的ToolNode本身就支持一次返回多個(gè)工具調(diào)用這也是常見(jiàn)做法。6. 接口 API 與批量任務(wù)6.1 用 FastAPI 封裝 Agent 服務(wù)LangGraph 編譯后的app可以直接在 Python 進(jìn)程內(nèi)調(diào)用也可以封裝成 REST API。下面是一個(gè)最小可用的 FastAPI 服務(wù)from fastapi import FastAPI from pydantic import BaseModel from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage app FastAPI(titleLangGraph Demo Service) llm ChatOpenAI( modelqwen2.5:7b, base_urlhttp://localhost:11434/v1, api_keyollama, temperature0.7 ) class ChatState(TypedDict): messages: list def call_model(state: ChatState) - dict: response llm.invoke(state[messages]) return {messages: [response]} graph StateGraph(ChatState) graph.add_node(agent, call_model) graph.add_edge(START, agent) graph.add_edge(agent, END) agent_app graph.compile() class ChatRequest(BaseModel): message: str class ChatResponse(BaseModel): reply: str app.post(/chat, response_modelChatResponse) def chat(req: ChatRequest): result agent_app.invoke({ messages: [ SystemMessage(content你是一個(gè)簡(jiǎn)潔的助手。), HumanMessage(contentreq.message) ] }) return ChatResponse(replyresult[messages][-1].content) # 啟動(dòng)方式uvicorn main:app --host 0.0.0.0 --port 8000啟動(dòng)后用 curl 測(cè)試curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {message: 你好請(qǐng)簡(jiǎn)單介紹一下你自己}返回示例{reply: 你好我是一個(gè)基于 LangGraph 構(gòu)建的 AI 助手。}把 Agent 封裝成 API 后前端、后端、自動(dòng)化腳本都能直接對(duì)接。6.2 批量任務(wù)的實(shí)現(xiàn)思路LangGraph 沒(méi)有內(nèi)置任務(wù)隊(duì)列但批量處理的核心邏輯很簡(jiǎn)單循環(huán)調(diào)用 錯(cuò)誤處理 結(jié)果匯總。import time from concurrent.futures import ThreadPoolExecutor, as_completed questions [ 什么是 Agent, 計(jì)算 12 34, 如何學(xué)習(xí) LangGraph, 計(jì)算 100 / 4, ] def process_one(q: str) - dict: try: result agent_app.invoke({ messages: [ SystemMessage(content你是一個(gè)簡(jiǎn)潔的助手。), HumanMessage(contentq) ] }) return {question: q, answer: result[messages][-1].content, status: success} except Exception as e: return {question: q, answer: str(e), status: failed} # 串行執(zhí)行 start time.time() results [process_one(q) for q in questions] print(f串行耗時(shí): {time.time() - start:.2f}s) # 并行執(zhí)行 start time.time() with ThreadPoolExecutor(max_workers4) as executor: futures [executor.submit(process_one, q) for q in questions] parallel_results [f.result() for f in as_completed(futures)] print(f并行耗時(shí): {time.time() - start:.2f}s) for r in parallel_results: print(r)批量任務(wù)有幾個(gè)工程化要點(diǎn)每個(gè)任務(wù)要捕獲異常不能因?yàn)橐粭l失敗而中斷整個(gè)批次。并發(fā)數(shù)要控制避免把模型服務(wù)的請(qǐng)求隊(duì)列打滿(mǎn)。結(jié)果要落盤(pán)或?qū)憥?kù)方便失敗重試和效果復(fù)盤(pán)。對(duì)耗時(shí)較長(zhǎng)的任務(wù)可以加超時(shí)控制。6.3 通過(guò) Checkpoint 實(shí)現(xiàn)持久化與恢復(fù)LangGraph 一個(gè)重要特性是狀態(tài)檢查點(diǎn)。通過(guò)MemorySaver或數(shù)據(jù)庫(kù)持久化可以在中斷后恢復(fù)執(zhí)行。from langgraph.checkpoint.memory import MemorySaver # 編譯時(shí)傳入 checkpointer memory MemorySaver() app_with_memory graph.compile(checkpointermemory) # 第一次執(zhí)行帶 thread_id相當(dāng)于一個(gè)會(huì)話 ID config {configurable: {thread_id: session-001}} app_with_memory.invoke( {messages: [HumanMessage(content你好)]}, configconfig ) # 第二次執(zhí)行同一個(gè) thread_id會(huì)帶上之前的狀態(tài) app_with_memory.invoke( {messages: [HumanMessage(content我剛才問(wèn)了什么)]}, configconfig )這個(gè)特性非常實(shí)用相當(dāng)于給 Agent 加上了“工作記憶”。生產(chǎn)環(huán)境建議把MemorySaver換成基于 Redis 或數(shù)據(jù)庫(kù)的持久化方案這樣服務(wù)重啟后狀態(tài)也不會(huì)丟失。7. 資源占用與性能觀察7.1 不同運(yùn)行方式下的資源分布LangGraph 本身是純 Python 圖編排邏輯CPU 占用很低內(nèi)存占用以幾十到幾百 MB 計(jì)。真正消耗資源的是模型推理服務(wù)在線 API 或本地 Ollama。向量檢索服務(wù)如果做 RAG。長(zhǎng)時(shí)間運(yùn)行時(shí)的狀態(tài)累積。如果使用本地模型7B 量化模型通常需要 6G 到 10G 內(nèi)存14B 以上模型需要更大顯存或內(nèi)存。具體數(shù)字取決于模型量化方式和上下文長(zhǎng)度建議用nvidia-smi或任務(wù)管理器實(shí)時(shí)觀察。7.2 如何降低延遲與顯存占用有幾個(gè)實(shí)際操作方向。第一減少不必要的多輪循環(huán)。Agent 每多一次工具調(diào)用就多一次模型推理。設(shè)計(jì)提示詞時(shí)明確要求模型“能直接回答就不要調(diào)用工具”可以顯著降低平均延遲。第二控制上下文長(zhǎng)度。LangGraph 的 State 會(huì)累積所有消息對(duì)話輪次多了之后每次請(qǐng)求的 token 數(shù)會(huì)快速增長(zhǎng)直接推高延遲和成本。策略是定期摘要?dú)v史消息或裁剪早期消息。第三模型側(cè)優(yōu)化。本地模型可以根據(jù)顯存選擇更小的量化版本降低 temperature 也能減少輸出波動(dòng)。7.3 觀察指標(biāo)建議建議在服務(wù)里加三類(lèi)日志圖流轉(zhuǎn)日志每個(gè)節(jié)點(diǎn)進(jìn)入和退出的時(shí)間。模型調(diào)用日志每次調(diào)用的輸入輸出 token 數(shù)和耗時(shí)。錯(cuò)誤日志工具調(diào)用失敗、超時(shí)、狀態(tài)不一致等問(wèn)題。通過(guò)觀察節(jié)點(diǎn)耗時(shí)分布你很快能定位性能瓶頸是在模型推理還是工具執(zhí)行。8. 常見(jiàn)問(wèn)題與排查方法問(wèn)題現(xiàn)象可能原因排查方式解決方案安裝 langgraph 后 import 報(bào)錯(cuò)Python 版本過(guò)低或依賴(lài)沖突檢查python --version和pip list使用 Python 3.9-3.12重新創(chuàng)建虛擬環(huán)境安裝調(diào)用模型報(bào)連接錯(cuò)誤模型服務(wù)未啟動(dòng)或地址配置錯(cuò)誤先 curl 測(cè)試模型接口啟動(dòng) Ollama 或檢查 base_url、api_keyAgent 不調(diào)用工具模型沒(méi)綁定工具或提示詞不清楚打印模型返回的 tool_calls使用 bind_tools并在提示詞中明說(shuō)可用工具死循環(huán)或一直運(yùn)行缺少循環(huán)退出條件查看日志中節(jié)點(diǎn)流轉(zhuǎn)次數(shù)在 State 中加入 step_count設(shè)置最大輪次工具返回結(jié)果沒(méi)有傳給模型狀態(tài)字段沒(méi)有包含 ToolMessage打印 messages 列表確保 ToolNode 返回的消息被正確追加進(jìn) State條件路由走了錯(cuò)誤的邊路由函數(shù)返回值與映射表不一致打印路由函數(shù)返回值檢查 add_conditional_edges 中的映射 keyAPI 服務(wù)并發(fā)高時(shí)報(bào)錯(cuò)模型服務(wù)并發(fā)受限查看模型服務(wù)日志控制線程池大小或引入任務(wù)隊(duì)列服務(wù)重啟后會(huì)話丟失未配置持久化 Checkpoint檢查 checkpointer 參數(shù)使用 Redis 或數(shù)據(jù)庫(kù) CheckpointSaver上下文越來(lái)越長(zhǎng)響應(yīng)變慢State 中消息持續(xù)累積打印每次請(qǐng)求的 token 數(shù)做歷史摘要或裁剪LangGraph 版本升級(jí)后 API 報(bào)錯(cuò)API 變更查看官方文檔變更日志固定版本號(hào)不要隨意升級(jí)除了表格里的方案再補(bǔ)充兩個(gè)定位問(wèn)題的實(shí)用技巧。第一個(gè)是“拆開(kāi)調(diào)試”。把一個(gè)長(zhǎng)流程拆成多個(gè)子圖分別測(cè)試每個(gè)子圖的輸入輸出。LangGraph 的圖結(jié)構(gòu)允許單獨(dú)編譯調(diào)用子圖這比整個(gè)跑一大段流程更容易定位問(wèn)題。第二個(gè)是“打印中間狀態(tài)”。在節(jié)點(diǎn)函數(shù)里用print(state)或日志記錄狀態(tài)變化能非常直觀地看到每一步數(shù)據(jù)流轉(zhuǎn)是否符合預(yù)期。9. 最佳實(shí)踐與使用建議9.1 工程落地建議經(jīng)過(guò)實(shí)際項(xiàng)目驗(yàn)證我建議把下面幾條作為默認(rèn)規(guī)范。第一狀態(tài)結(jié)構(gòu)要精簡(jiǎn)。State 里只放節(jié)點(diǎn)之間需要傳遞的字段。一些臨時(shí)變量不要塞進(jìn)圖狀態(tài)里否則會(huì)干擾可視化調(diào)試也容易造成內(nèi)存增長(zhǎng)。第二工具函數(shù)要負(fù)責(zé)。工具節(jié)點(diǎn)不要只定義函數(shù)要給工具寫(xiě)清晰的 docstring 和參數(shù)說(shuō)明。LangGraph 的工具綁定依賴(lài)模型理解工具描述描述寫(xiě)得越清楚模型用錯(cuò)的概率越低。第三條件路由要顯式。所有add_conditional_edges的返回值和映射表要保證完全匹配并提供一個(gè)默認(rèn)的 fallback 分支。第四盡早引入 Checkpoint。哪怕開(kāi)發(fā)階段不用設(shè)計(jì)時(shí)也要先留出thread_id的傳遞鏈路。后期加持久化會(huì)容易得多。第五日志和追蹤不能省。LangGraph 提供了大量回調(diào)接口建議從第一天就接入 LangSmith 或自建日志系統(tǒng)。出了問(wèn)題能快速回放執(zhí)行過(guò)程。9.2 安全與合規(guī)建議Agent 自動(dòng)調(diào)用工具有一個(gè)容易被忽視的風(fēng)險(xiǎn)如果模型被惡意 Prompt 誘導(dǎo)可能會(huì)執(zhí)行非預(yù)期的工具操作。因此工具權(quán)限要最小化關(guān)鍵操作必須二次確認(rèn)。涉及外部請(qǐng)求時(shí)對(duì) URL、文件路徑等參數(shù)做校驗(yàn)。涉及用戶(hù)數(shù)據(jù)時(shí)確認(rèn)數(shù)據(jù)不出域。涉及版權(quán)內(nèi)容和肖像授權(quán)時(shí)必須核實(shí)授權(quán)鏈路。9.3 學(xué)習(xí)路徑建議如果你剛接觸 LangGraph建議按這個(gè)順序?qū)W習(xí)先把本章的“兩節(jié)點(diǎn)順序執(zhí)行”跑通。再實(shí)現(xiàn)一個(gè)帶工具調(diào)用的 Agent 循環(huán)。然后加條件路由和循環(huán)限制。接著把流程拆成子圖。最后封裝成 API 并接入業(yè)務(wù)。不要一上來(lái)就復(fù)制別人的復(fù)雜項(xiàng)目。圖編排的核心是“你會(huì)不會(huì)拆流程”而不是“你記沒(méi)記住 API”。10. 總結(jié)與下一步LangGraph 最值得嘗試的點(diǎn)在于它把 Agent 開(kāi)發(fā)從“自由發(fā)揮的腳本”變成了“結(jié)構(gòu)清晰的工程圖”。比 LangChain 的鏈?zhǔn)秸{(diào)用更靈活比手寫(xiě) Agent 循環(huán)更規(guī)范。如果你已經(jīng)在做 AI Agent 開(kāi)發(fā)這個(gè)框架值得投入時(shí)間。建議你在本地完成兩件事先運(yùn)行一遍本文的 ReAct Agent 循環(huán)確認(rèn)模型調(diào)用、工具調(diào)用、條件退出整條鏈路能跑通然后把一個(gè)你手頭已有的業(yè)務(wù)場(chǎng)景改造成 LangGraph 版本對(duì)比兩者的維護(hù)成本差異。最容易踩的坑集中在條件路由返回值不匹配、Agent 循環(huán)缺少退出條件、工具調(diào)用后消息傳遞斷裂。這三個(gè)問(wèn)題占了 LangGraph 開(kāi)發(fā)階段的大部分報(bào)錯(cuò)遇到時(shí)不要慌打印中間狀態(tài)就能定位。下一步可以擴(kuò)展的方向很多接入多模態(tài)模型、增加人工審核節(jié)點(diǎn)、把 Checkpoint 換成 Redis 持久化、用子圖重組回答生成流程甚至是結(jié)合向量數(shù)據(jù)庫(kù)做一套完整的 RAG Agent 服務(wù)。圖結(jié)構(gòu)的好處是每次擴(kuò)展都只需要新增節(jié)點(diǎn)和邊不需要重寫(xiě)整個(gè)框架。這篇文章建議收藏備用遇到問(wèn)題時(shí)回來(lái)翻一翻排查表能省不少時(shí)間。