这段时间后台收到不少关于 Agent 开发的私信问得最多的就是LangChain 我能跑通但一涉及 Agent 循环、条件分支、状态持久化就不知道怎么组织代码了。市面上不少教程要么只讲概念要么直接甩一段看不出全貌的代码。这次我们直接看 LangGraph一个专门为有状态 Agent 工作流设计的图编排框架。它并不是取代 LangChain而是把 LangChain 里的模型调用、工具调用、记忆管理组合成一个可控制的图结构。你能在图上定义节点、连边、条件路由、循环检测甚至把多个子图嵌套在一起。这篇文章会沿着“核心概念 - 环境安装 - 手写 Agent 循环 - 条件路由 - RAG 实战 - API 服务与批量任务”的顺序走一遍并提供可直接复制的代码。读完你能回答三个问题LangGraph 到底解决什么问题、它能跑在什么硬件上、怎么把它接到自己的业务里。1. 核心能力速览能力项说明项目类型开源 Agent 工作流编排框架核心定位基于图结构构建有状态、可控制、可恢复的 AI Agent与 LangChain 关系构建在 LangChain 生态之上复用其模型、工具、检索组件主要功能节点编排、条件路由、循环控制、图状态管理、人机交互断点、子图嵌套支持的模型来源OpenAI 兼容接口、Ollama 本地模型、各类在线模型 API支持平台Windows / Linux / macOSPython 3.9启动方式Python 脚本 / FastAPI 服务 / LangGraph Studio是否支持 API支持可封装为 REST API 或接入已有 Web 服务是否支持批量任务支持可在代码中循环调用或设计并发队列显存要求取决于底层模型纯编排框架本身不占用 GPU使用本地模型时以模型大小为基准适合人群想从“单轮模型调用”进阶到“多步 Agent 工作流”的开发者从这张表能看出来LangGraph 的价值不在模型本身而在“控制逻辑”。如果你已经解决了模型调用问题接下来真正花时间的往往是如何把工具调用、多步推理、失败重试、并行分支做成稳定可维护的系统这正是 LangGraph 解决的核心问题。2. 适用场景与使用边界2.1 适合解决什么问题我从实际使用角度来说LangGraph 最常见的落地场景是这三类。第一类是多工具 Agent。比如一个智能客服它需要先判断用户意图再决定调订单查询还是物流查询最后把结果整理成自然语言回复。这类任务如果有多个工具调用链式写法会变得极其难维护而图结构天然适合表达这种“先判断、再分流、后汇总”的流程。第二类是RAG 增强问答。基础 RAG 是“检索 - 生成”但实际业务经常需要“判断是否需要检索 - 检索 - 判断答案是否充分 - 不充分就重写问题再检索 - 最终生成”。这个流程里每个判断节点都是一次模型调用用 LangGraph 可以让所有分支一目了然。第三类是长流程任务。比如批量处理文档、多步代码生成、自动报表生成。这些任务的特点是有中间状态、可能要执行几十步并且中途可能失败。LangGraph 的状态管理能让每一步都可视、可恢复。2.2 不适合什么场景如果你的需求只是“调一次模型拿到结果”用 LangChain 或直接请求 API 就足够了。强行引入图编排会把简单问题复杂化。另外如果应用是完全实时、毫秒级响应的流式交互图编排会带来额外的调度开销。虽然 LangGraph 有流式输出支持但对极端低延迟场景仍需做压测确认。2.3 使用边界与合规提醒任何 Agent 框架都只是工具落地时要注意几点涉及用户隐私数据时要确认模型服务部署在哪、数据是否会发送到外部接口。涉及版权内容时要有明确的授权链路。Agent 自动执行的操作要有权限边界尤其是接数据库、发邮件、操作文件这类高权限动作。对外提供服务时要考虑 Prompt 注入风险不能把系统提示词和工具描述完全暴露给不可信输入。3. 环境准备与前置条件3.1 运行时选择LangGraph 是 Python 框架对硬件没有强制要求。你可以在没有 GPU 的普通开发机上完成全部工作流编排和调试。真正消耗 GPU 的是底层大模型这一层可以选用在线 API也可以用 Ollama 在本地运行小参数模型。我的建议是学习阶段用在线模型 API 或 Ollama 跑 7B 以下模型普通 CPU 也能完成小规模测试。生产阶段把 LangGraph 服务和模型服务分开部署模型服务单独分配 GPU。3.2 Python 与依赖安装先确认 Python 版本推荐 3.9 到 3.12。python --version然后创建虚拟环境并安装依赖python -m venv langgraph_env source langgraph_env/bin/activate # Windows 下执行 langgraph_env\Scripts\activate pip install --upgrade pip pip install langgraph langchain-core langchain-openai langchain-community如果你要接入本地 Ollama 模型还需要安装pip install ollama安装完成后验证版本python -c import langgraph; print(langgraph.__version__)能正常输出版本号说明环境已经就绪。3.3 模型服务准备LangGraph 本身不提供模型需要先有一个可调用的模型服务。这里给两种常见方案。方案一使用 OpenAI 兼容接口。很多在线模型服务都提供 OpenAI 格式的接口只需要在环境变量里配置 API Key 和接口地址。export OPENAI_API_KEYyour-api-key export OPENAI_BASE_URLhttps://your-endpoint/v1方案二使用本地 Ollama。先安装 Ollama然后拉取一个小参数模型ollama pull qwen2.5:7b ollama serve验证模型可用curl http://localhost:11434/api/generate -d {model: qwen2.5:7b, prompt: 你好}本教程后面示例会同时兼容这两种方式你只需要改一行模型初始化代码。4. 安装部署与快速搭建第一个 LangGraph4.1 LangGraph 核心概念State、Node、Edge在写代码之前先花一分钟理解 LangGraph 的三个核心概念。State状态整个图的全局状态是一个数据结构。所有节点都能读写这个状态。Node节点一个处理函数输入是当前状态输出是更新后的状态片段。Edge边定义节点之间的转移方向。普通边是“执行完 A 必执行 B”条件边是“根据状态决定下一个节点”。用一句话概括LangGraph 就是把你脑子里的流程图变成代码让 Agent 的执行过程可控制、可观察、可重放。4.2 第一个示例两节点顺序执行先写一个最小示例感受一下基本写法。from typing import TypedDict from langgraph.graph import StateGraph, START, END # 1. 定义状态结构 class MyState(TypedDict): input_text: str output_text: str # 2. 定义节点函数 def node_a(state: MyState) - dict: print(执行 node_a) return {output_text: state[input_text] - 已处理} def node_b(state: MyState) - dict: print(执行 node_b) return {output_text: state[output_text] - 已完善} # 3. 构建图 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. 编译并执行 app graph.compile() result app.invoke({input_text: hello}) print(result)运行后可以看到输出执行 node_a 执行 node_b {input_text: hello, output_text: hello - 已处理 - 已完善}从这段代码能看出 LangGraph 的基本套路定义状态 - 定义节点函数 - 加节点、加边 - 编译 - 调用。后面所有复杂功能都是在这个基础上扩展。4.3 接入真实模型让 Agent 具备“思考”能力现在把上面的示例升级为真正的 Agent引入大模型调用。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 兼容端点 api_keyollama, # Ollama 不需要真实密钥随便填即可 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})这个示例说明了一个关键点LangGraph 的 State 可以是消息列表模型调用只是图里的一个节点。你可以在这个节点前后加工具调用、加判断逻辑形成一个完整的执行链路。5. 功能测试与效果验证5.1 测试目标对于 LangGraph功能测试不能只看“能不能返回文本”要看这些维度状态是否按预期流转。条件分支是否走对路径。循环是否有退出条件。工具调用是否能正确传入参数。异常时是否能恢复。下面的章节会按这个思路逐项验证。5.2 基础 Agent 循环测试让 Agent 可以自主调用工具只调一次模型不是 Agent。真正的 Agent 应该能“思考 - 决定调工具 - 看到结果 - 再思考”直到得出最终答案。下面实现一个最简单的 ReAct 循环。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 # 定义一个计算器工具 def calculator(expression: str) - str: 计算数学表达式例如 1 2 * 3。 try: # 注意生产环境不要直接 eval这里仅做演示 result eval(expression) return str(result) except Exception as e: return f计算失败: {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] # 如果模型返回了工具调用请求就进入工具节点 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) # 工具执行完回到 agent app graph.compile() result app.invoke({ messages: [ SystemMessage(content你是计算助手需要计算时调用 calculator 工具。), HumanMessage(content计算 (12 34) * 5 的结果) ] }) for msg in result[messages]: print(f--- {msg.type} ---) print(msg.content) if msg.tool_calls: print(Tool calls:, msg.tool_calls)运行后观察输出应该能看到完整的循环过程agent节点返回一个带 tool_calls 的消息。should_continue判断为continue进入tools。tools执行计算器返回 ToolMessage。回到agent模型基于工具结果生成最终答案。should_continue判断为end流程结束。这个就是 Agent 循环的骨架。后面加再多的工具、再复杂的逻辑核心结构都不会变。5.3 条件路由与分支控制测试实际业务里不是每次都需要调用工具。更合理的流程是模型先判断问题是否需要工具需要就走工具分支不需要直接回答。这就是条件路由。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: 模拟模型判断是否需要工具。 # 实际项目中这里可以调用模型做意图识别 if 计算 in state[question] or 多少 in state[question]: return {need_tool: True} return {need_tool: False} def use_tool(state: RouteState) - dict: return {answer: f【工具计算】{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: 计算 3 * 2})) print(app.invoke({question: 你好}))测试结果{question: 计算 3 * 2, need_tool: True, answer: 【工具计算】计算 3 * 2 的答案是 100} {question: 你好, need_tool: False, answer: 【直接回答】你好}注意add_conditional_edges就是 LangGraph 的“条件路由”核心 API。你只需要写一个返回字符串的函数根据返回值映射到不同节点。5.4 循环检测与最大步数限制Agent 循环最怕的是“死循环”。LangGraph 本身不会无限执行它有递归限制默认情况下超过限制会报错。更稳妥的做法是在状态中记录轮次主动退出。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] [已达到最大轮次强制停止] } 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])这种“显式记录轮次 条件退出”的模式在生产环境里非常实用建议代码里强制保留。5.5 子图与并行分支测试当流程复杂后可以把一个完整流程封装成子图再嵌入到父图节点中。下面演示子图的用法。# 先构建一个子图负责文本清清洗 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最终结果: {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)子图的价值在于复用。你可以把“工具调用循环”“RAG 检索”“报告生成”分别封装成独立子图组合成不同的业务应用。并行分支方面LangGraph 的节点只要不依赖彼此状态可以在不同边中并行执行然后通过一个汇聚节点合并结果。对于并行工具调用场景LangChain 的ToolNode本身就支持一次返回多个工具调用这也是常见做法。6. 接口 API 与批量任务6.1 用 FastAPI 封装 Agent 服务LangGraph 编译后的app可以直接在 Python 进程内调用也可以封装成 REST API。下面是一个最小可用的 FastAPI 服务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 构建的 AI 助手。}把 Agent 封装成 API 后前端、后端、自动化脚本都能直接对接。6.2 批量任务的实现思路LangGraph 没有内置任务队列但批量处理的核心逻辑很简单循环调用 错误处理 结果汇总。import time from concurrent.futures import ThreadPoolExecutor, as_completed questions [ 什么是 Agent, 计算 12 34, 如何学习 LangGraph, 计算 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} # 串行执行 start time.time() results [process_one(q) for q in questions] print(f串行耗时: {time.time() - start:.2f}s) # 并行执行 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)批量任务有几个工程化要点每个任务要捕获异常不能因为一条失败而中断整个批次。并发数要控制避免把模型服务的请求队列打满。结果要落盘或写库方便失败重试和效果复盘。对耗时较长的任务可以加超时控制。6.3 通过 Checkpoint 实现持久化与恢复LangGraph 一个重要特性是状态检查点。通过MemorySaver或数据库持久化可以在中断后恢复执行。from langgraph.checkpoint.memory import MemorySaver # 编译时传入 checkpointer memory MemorySaver() app_with_memory graph.compile(checkpointermemory) # 第一次执行带 thread_id相当于一个会话 ID config {configurable: {thread_id: session-001}} app_with_memory.invoke( {messages: [HumanMessage(content你好)]}, configconfig ) # 第二次执行同一个 thread_id会带上之前的状态 app_with_memory.invoke( {messages: [HumanMessage(content我刚才问了什么)]}, configconfig )这个特性非常实用相当于给 Agent 加上了“工作记忆”。生产环境建议把MemorySaver换成基于 Redis 或数据库的持久化方案这样服务重启后状态也不会丢失。7. 资源占用与性能观察7.1 不同运行方式下的资源分布LangGraph 本身是纯 Python 图编排逻辑CPU 占用很低内存占用以几十到几百 MB 计。真正消耗资源的是模型推理服务在线 API 或本地 Ollama。向量检索服务如果做 RAG。长时间运行时的状态累积。如果使用本地模型7B 量化模型通常需要 6G 到 10G 内存14B 以上模型需要更大显存或内存。具体数字取决于模型量化方式和上下文长度建议用nvidia-smi或任务管理器实时观察。7.2 如何降低延迟与显存占用有几个实际操作方向。第一减少不必要的多轮循环。Agent 每多一次工具调用就多一次模型推理。设计提示词时明确要求模型“能直接回答就不要调用工具”可以显著降低平均延迟。第二控制上下文长度。LangGraph 的 State 会累积所有消息对话轮次多了之后每次请求的 token 数会快速增长直接推高延迟和成本。策略是定期摘要历史消息或裁剪早期消息。第三模型侧优化。本地模型可以根据显存选择更小的量化版本降低 temperature 也能减少输出波动。7.3 观察指标建议建议在服务里加三类日志图流转日志每个节点进入和退出的时间。模型调用日志每次调用的输入输出 token 数和耗时。错误日志工具调用失败、超时、状态不一致等问题。通过观察节点耗时分布你很快能定位性能瓶颈是在模型推理还是工具执行。8. 常见问题与排查方法问题现象可能原因排查方式解决方案安装 langgraph 后 import 报错Python 版本过低或依赖冲突检查python --version和pip list使用 Python 3.9-3.12重新创建虚拟环境安装调用模型报连接错误模型服务未启动或地址配置错误先 curl 测试模型接口启动 Ollama 或检查 base_url、api_keyAgent 不调用工具模型没绑定工具或提示词不清楚打印模型返回的 tool_calls使用 bind_tools并在提示词中明说可用工具死循环或一直运行缺少循环退出条件查看日志中节点流转次数在 State 中加入 step_count设置最大轮次工具返回结果没有传给模型状态字段没有包含 ToolMessage打印 messages 列表确保 ToolNode 返回的消息被正确追加进 State条件路由走了错误的边路由函数返回值与映射表不一致打印路由函数返回值检查 add_conditional_edges 中的映射 keyAPI 服务并发高时报错模型服务并发受限查看模型服务日志控制线程池大小或引入任务队列服务重启后会话丢失未配置持久化 Checkpoint检查 checkpointer 参数使用 Redis 或数据库 CheckpointSaver上下文越来越长响应变慢State 中消息持续累积打印每次请求的 token 数做历史摘要或裁剪LangGraph 版本升级后 API 报错API 变更查看官方文档变更日志固定版本号不要随意升级除了表格里的方案再补充两个定位问题的实用技巧。第一个是“拆开调试”。把一个长流程拆成多个子图分别测试每个子图的输入输出。LangGraph 的图结构允许单独编译调用子图这比整个跑一大段流程更容易定位问题。第二个是“打印中间状态”。在节点函数里用print(state)或日志记录状态变化能非常直观地看到每一步数据流转是否符合预期。9. 最佳实践与使用建议9.1 工程落地建议经过实际项目验证我建议把下面几条作为默认规范。第一状态结构要精简。State 里只放节点之间需要传递的字段。一些临时变量不要塞进图状态里否则会干扰可视化调试也容易造成内存增长。第二工具函数要负责。工具节点不要只定义函数要给工具写清晰的 docstring 和参数说明。LangGraph 的工具绑定依赖模型理解工具描述描述写得越清楚模型用错的概率越低。第三条件路由要显式。所有add_conditional_edges的返回值和映射表要保证完全匹配并提供一个默认的 fallback 分支。第四尽早引入 Checkpoint。哪怕开发阶段不用设计时也要先留出thread_id的传递链路。后期加持久化会容易得多。第五日志和追踪不能省。LangGraph 提供了大量回调接口建议从第一天就接入 LangSmith 或自建日志系统。出了问题能快速回放执行过程。9.2 安全与合规建议Agent 自动调用工具有一个容易被忽视的风险如果模型被恶意 Prompt 诱导可能会执行非预期的工具操作。因此工具权限要最小化关键操作必须二次确认。涉及外部请求时对 URL、文件路径等参数做校验。涉及用户数据时确认数据不出域。涉及版权内容和肖像授权时必须核实授权链路。9.3 学习路径建议如果你刚接触 LangGraph建议按这个顺序学习先把本章的“两节点顺序执行”跑通。再实现一个带工具调用的 Agent 循环。然后加条件路由和循环限制。接着把流程拆成子图。最后封装成 API 并接入业务。不要一上来就复制别人的复杂项目。图编排的核心是“你会不会拆流程”而不是“你记没记住 API”。10. 总结与下一步LangGraph 最值得尝试的点在于它把 Agent 开发从“自由发挥的脚本”变成了“结构清晰的工程图”。比 LangChain 的链式调用更灵活比手写 Agent 循环更规范。如果你已经在做 AI Agent 开发这个框架值得投入时间。建议你在本地完成两件事先运行一遍本文的 ReAct Agent 循环确认模型调用、工具调用、条件退出整条链路能跑通然后把一个你手头已有的业务场景改造成 LangGraph 版本对比两者的维护成本差异。最容易踩的坑集中在条件路由返回值不匹配、Agent 循环缺少退出条件、工具调用后消息传递断裂。这三个问题占了 LangGraph 开发阶段的大部分报错遇到时不要慌打印中间状态就能定位。下一步可以扩展的方向很多接入多模态模型、增加人工审核节点、把 Checkpoint 换成 Redis 持久化、用子图重组回答生成流程甚至是结合向量数据库做一套完整的 RAG Agent 服务。图结构的好处是每次扩展都只需要新增节点和边不需要重写整个框架。这篇文章建议收藏备用遇到问题时回来翻一翻排查表能省不少时间。