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