LangGraph 智能体实战从环境搭建到条件路由与记忆机制这一篇讲透如果你正在学习 Agent 开发大概率已经听过 LangGraph甚至已经看过几篇教程。但多数教程只停留在“画个流程图”“跑个 demo”的阶段真正到了自己写业务逻辑时节点怎么设计、状态怎么传递、条件分支怎么跳、记忆怎么保存、工具调用怎么兜底这些问题全冒出来了。这篇文章不打算重复那些官方案例的翻译版。我会从工程落地角度把 LangGraph 从环境准备、核心概念、第一个可运行示例到条件路由、子图、记忆机制、Human-in-the-loop 真实场景完整拆一遍。读完之后你能独立搭建一个带工具调用和对话记忆的 LangGraph Agent并且知道每一步为什么这样设计。先说一个明确判断LangGraph 真正降低的不是“写一个 AI 应用”的门槛而是“把 AI 应用做成有状态、可控制、可维护的工程系统”的门槛。它和普通 LangChain chain 的本质区别在于它把 Agent 的整个执行过程显式建模成了图结构让开发者对流程有完全的掌控力。这一点理解到位了后面的代码才不算白写。1. 为什么需要 LangGraph从 Chain 到 Graph 的必然演进很多刚接触 LangChain 的开发者会有个困惑LangChain 已经能调模型、调工具、做 RAG 了为什么还需要 LangGraph答案是LangChain 的 Chain 是线性管道而真实 Agent 是带分支、循环、状态和人工干预的复杂流程。举个例子一个最简单的客服 Agent 流程可能是接收用户提问。判断是否需要查询订单系统。如果需要调用订单查询工具。根据工具返回结果生成回答。如果回答过程中用户又追问了可能需要回到第 2 步再次判断。这种流程用 Chain 表达会非常别扭。因为 Chain 的假设是“输入一次走完一条固定管道输出结果”。而 Agent 的实际执行是“走一步看一步根据当前状态决定下一步去哪”。LangGraph 把这个过程建模成一张有向图每个处理步骤是一个Node节点。节点之间通过Edge边连接。边的走向可以由条件函数conditional_edge动态决定。整张图共享一个State状态对象任何节点都可以读写这个状态。这意味着你可以精确控制 Agent 的每一步哪些步骤必须顺序执行哪些步骤需要根据模型输出决定走向哪些步骤允许并行哪些步骤需要停下来等用户确认。另外LangGraph 早期版本的核心定位是“低阶编排”它把 LangChain Agent 那些封装好的高层接口拆开暴露底层控制能力。后来 LangChain 官方明确推荐新项目优先使用 LangGraph 进行 Agent 开发甚至可以完全不依赖 LangChain 的 Agent 封装只用 LangChain 的模型封装和工具封装。一句话总结如果你只需要一个固定流程的提示词管道用 LangChain Chain 就行。如果你需要一个可以自主决策、多轮交互、可人工干预的 Agent用 LangGraph。2. LangGraph 的核心概念与工作原理在写代码之前先把 LangGraph 的五个核心概念彻底理清。不理解这五个概念代码就是死记硬背。2.1 State全局共享状态State 是 LangGraph 的灵魂。它定义了整个图在任意时刻的数据快照。所有节点都接收当前 State 作为输入返回值会合并进 State供下一个节点使用。State 通常是一个 TypedDict 或 Pydantic 模型。每个字段可以配置不同的归约操作比如覆盖、追加、合并。from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): messages: Annotated[List[dict], operator.add] # 消息列表新消息追加 next_step: str # 下一步走向默认覆盖 tool_result: str # 工具返回结果这里operator.add是一个归约器reducer。当节点返回新的messages时LangGraph 会执行两段列表相加而不是直接覆盖。这个设计非常关键它让多个节点可以往同一个状态字段追加内容而不会互相覆盖。2.2 Node处理节点Node 就是一个普通的 Python 函数也可以是异步函数接收 State 参数返回一个字典字典的键是要更新的状态字段。def call_model(state: AgentState): # 调用大模型的逻辑 response llm.invoke(state[messages]) return {messages: [{role: assistant, content: response.content}]}Node 里可以写任何逻辑调用大模型、调用工具、查数据库、写日志、做业务校验。LangGraph 不限制节点内部做什么它只负责调度和状态流转。2.3 Edge静态边和条件边Edge 分两种。静态边Edge节点 A 执行完无条件走向节点 B。graph.add_edge(node_a, node_b)条件边Conditional Edge节点 A 执行完后根据一个路由函数的返回值决定走向哪个节点。这是 LangGraph 最核心的亮点之一。def route_after_tool(state: AgentState): if state[next_step] call_model: return call_model else: return end graph.add_conditional_edges( tool_node, route_after_tool, {call_model: call_model, end: END} )条件边让图具备了“循环”能力当 Agent 判断还需要继续调用工具时可以沿着条件边回到前面的节点而不是线性结束。2.4 Checkpointer记忆与断点Checkpointer 是 LangGraph 的持久化层。它把每一步执行后的 State 快照保存到存储后端让 Agent 具备多轮对话记忆和断点恢复能力。如果你不配置 CheckpointerLangGraph 每次执行完图之后 State 就丢了下一次对话从零开始。这是很多初学 LangGraph 的开发者容易忽略的点。LangGraph 官方提供基于 SQLite 的SqliteSaver和基于内存的MemorySaver。生产环境可以对接 Redis 或 Postgres。2.5 编译与执行LangGraph 图需要先编译.compile()再执行。编译后得到的是一个CompiledGraph对象可以直接调用invoke()或stream()方法。app graph.compile(checkpointercheckpointer) result app.invoke( {messages: [{role: user, content: 帮我查一下订单状态}]}, config{configurable: {thread_id: user-001}} )thread_id是对话会话的标识同一个thread_id的多轮调用会共享 Checkpointer 里保存的状态。3. 环境准备与安装配置LangGraph 的环境准备并不复杂但有几个版本匹配的细节值得注意。如果你在安装阶段就把环境配错了后面跑示例时会遇到莫名其妙的报错。3.1 环境要求Python 3.9 及以上版本3.10、3.11 更稳定。操作系统Windows、macOS、Linux 均可。包管理工具推荐使用pip或poetry不要直接在系统全局环境里装建议使用虚拟环境。3.2 创建虚拟环境以当前教程使用的 Python 虚拟环境为例python -m venv langgraph-env source langgraph-env/bin/activate # Windows 下执行 langgraph-env\Scripts\activate3.3 安装核心依赖pip install --upgrade langgraph langchain langchain-openai如果没有 API Key想先本地体验可以安装 Ollama 和对应依赖pip install ollama注意langgraph、langchain、langchain-openai这三个包的版本兼容性直接影响运行结果。推荐的稳妥做法是安装最新稳定版如果出现 API 不兼容报错优先检查langgraph与langchain-core的版本匹配关系。3.4 关于 MCP 的补充说明MCPModel Context Protocol最近在 Agent 生态中非常火它的目标是统一大模型与外部工具、数据源的接入协议。LangGraph 的 Agent 在工程实践中通常会结合 MCP Server 来接入数据库、代码仓库、设计稿等外部资源。但是请注意MCP 是工具接入层协议LangGraph 是编排层框架两者解决的问题不冲突。更准确的理解是通过 MCP Server 暴露工具LangGraph 负责将这些工具编排进 Agent 的决策流程。后面会有具体示例。4. 核心组件拆解Node、State、Edge 的工程语义把概念映射到工程语义你才能真正理解为什么 LangGraph 这么设计。4.1 Node 的工程语义在真实的智能体项目里Node 通常对应一个“可复用的处理单元”。一个常见的划分方式call_model负责调用大模型生成回复或决策。call_tool负责执行具体工具比如查数据库、调 API。check_hallucination负责校验模型输出是否可靠。save_to_db负责把结果持久化。这种划分的好处是每个 Node 都可以单独测试、单独替换、单独复用。比如你想把模型从 GPT-4 换成本地模型只需要改call_model这一个 Node 的内部实现图结构完全不用动。4.2 State 的工程语义State 在设计时需要区分两类信息对话数据用户消息、AI 消息、工具消息。这类数据通常是追加式的用operator.add。控制数据当前步骤标识、重试次数、错误信息、分支标记。这类数据通常是覆盖式的用默认 reducer。如果在设计 State 时混淆了这两类字段后续调试会陷入“不知道状态为什么变成这样了”的困境。建议每个 Node 里只修改自己负责的字段并在节点开头打印关键状态方便追踪。4.3 Edge 的工程语义Edge 的最重要价值是显式化流程控制。相比让模型自由发挥纯 ReAct通过条件边把一些“固定规则”和“模型决策”分开可以让系统更稳定固定规则走静态边比如“工具执行完必须回到模型”。模型决策走条件边比如“模型判断是否继续调用工具”。异常分支必须走条件边比如“工具调用报错时重试还是结束”。这个“规则与模型分离”的设计思路是 LangGraph Agent 在生产环境中比纯 AutoGPT 方案更可靠的重要原因。5. 第一个 LangGraph 示例带工具调用的智能客服 Agent现在开始写真实的代码。这里用一个“智能客服 Agent”作为示例它能根据用户问题决定是否调用订单查询工具然后基于工具结果生成回答。5.1 定义 Statefrom typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): # 对话消息列表新消息自动追加 messages: Annotated[List[dict], operator.add] # 是否需要调用工具控制字段默认覆盖 should_call_tool: str # 工具返回结果默认覆盖 tool_result: str5.2 定义工具工具函数用tool装饰器包装便于 LangChain 自动生成工具描述和参数 schema。from langchain_core.tools import tool tool def query_orders(user_id: str) - str: 根据用户ID查询订单信息 # 这里是模拟实现真实项目中换成数据库查询即可 orders { 1001: 订单A已发货预计明天送达, 1002: 订单B待付款, } return orders.get(user_id, 未查询到订单信息)5.3 定义 Node三个核心 Nodecall_model调用大模型让它判断下一步动作直接回答还是调用工具。execute_tool执行工具调用。respond基于工具结果生成最终回答。from langchain_openai import ChatOpenAI # 这里使用 OpenAI 兼容接口你可以替换成任何 OpenAI 兼容服务 llm ChatOpenAI(modelgpt-4o-mini, temperature0) def call_model(state: AgentState): last_message state[messages][-1][content] # 简单判断如果消息中包含“订单”则标记需要调用工具 # 真实项目中应让模型判断这里为了演示条件边先简化 if 订单 in last_message: should_call_tool yes else: should_call_tool no # 仍然调用一次模型生成中间回复 response llm.invoke(state[messages]) return { messages: [{role: assistant, content: response.content}], should_call_tool: should_call_tool, } def execute_tool(state: AgentState): 执行订单查询工具 # 生产环境应从用户资料中取 user_id这里写死演示 result query_orders.invoke({user_id: 1001}) return { tool_result: result, should_call_tool: no, # 执行完工具后重置标记 } def respond(state: AgentState): 基于工具结果生成最终回复 tool_result state.get(tool_result, ) if tool_result: answer f根据系统查询结果{tool_result} else: answer 抱歉我暂时无法处理这个问题请稍后再试。 return {messages: [{role: assistant, content: answer}]}5.4 定义条件路由这是 LangGraph 条件边的核心逻辑from langgraph.graph import StateGraph, START, END def route_after_model(state: AgentState): 根据模型判断结果路由到工具节点或回答节点 if state[should_call_tool] yes: return execute_tool return respond5.5 构建并编译图# 初始化图 graph StateGraph(AgentState) # 添加节点 graph.add_node(call_model, call_model) graph.add_node(execute_tool, execute_tool) graph.add_node(respond, respond) # 添加入口边 graph.add_edge(START, call_model) # 添加条件边 graph.add_conditional_edges( call_model, route_after_model, { execute_tool: execute_tool, respond: respond, } ) # 工具执行完后回到模型让模型基于工具结果再组织回答 graph.add_edge(execute_tool, call_model) graph.add_edge(respond, END) # 编译图 app graph.compile()5.6 运行并验证result app.invoke({ messages: [{role: user, content: 你好帮我查一下订单状态}], }) for msg in result[messages]: print(f{msg[role]}: {msg[content]})预期执行流程是call_model判断用户输入包含“订单”将should_call_tool设为yes。条件边路由到execute_tool。execute_tool执行查询工具把结果写入tool_result。静态边回到call_model。此时should_call_tool已经是no条件边路由到respond。最终回答来自工具查询结果。这个示例虽然简单但它已经包含了 Agent 的基本骨架模型决策、工具调用、条件路由、循环回退。6. 进阶实战给 Agent 加上记忆和人类确认环节上面的示例有一个明显缺陷它没有记忆。用户第二次说“那发货地址呢”Agent 根本不知道用户是谁也不知道之前查过哪笔订单。解决这个问题需要引入 Checkpointer 和thread_id。6.1 配置 MemorySaver内存级持久化from langgraph.checkpoint.memory import MemorySaver # 使用内存级 Checkpointer memory MemorySaver() app graph.compile(checkpointermemory)6.2 带 thread_id 的多轮调用config {configurable: {thread_id: customer-001}} # 第一轮 result1 app.invoke( {messages: [{role: user, content: 查一下订单状态}]}, configconfig ) # 第二轮同一 thread_id result2 app.invoke( {messages: [{role: user, content: 再说一下发货时间}]}, configconfig )加了 Checkpointer 之后LangGraph 会在每次节点执行后保存 State 快照。第二轮调用时模型能读到第一轮的历史消息这是多轮对话记忆的基础。6.3 Human-in-the-loop打断和恢复Human-in-the-loop人类介入是生产级 Agent 的必备能力。典型场景是工具要执行高风险操作比如退款、删数据、发邮件前需要暂停让用户确认。LangGraph 里可以用interrupt_before实现app graph.compile( checkpointermemory, interrupt_before[execute_tool], # 执行工具前打断 )运行时会发现程序停在execute_tool之前不会继续往下走。此时你可以打印当前状态给用户确认然后调用invoke(None, configconfig)从断点处继续执行。# 第一次调用会在 execute_tool 前被打断 app.invoke( {messages: [{role: user, content: 查一下订单状态}]}, configconfig ) # 打印当前状态人工确认 current_state app.get_state(config) print(待执行的操作, current_state.next) # 确认通过后继续执行 app.invoke(None, configconfig)这个机制在真实项目中非常实用。比如销售智能体要执行“发送营销邮件”或“创建客户订单”之前都需要人工确认。6.4 子图拆分当一个 Agent 的节点数量超过 10 个图结构会变得难以维护。LangGraph 支持子图Subgraph可以把一组相关节点封装成一个子图作为主图的一个节点。一个典型场景把“订单查询”的完整流程查询、校验、格式化封装成子图然后主图只调用一个order_process节点。def build_order_subgraph() - StateGraph: # 这里定义子图内部节点和边 subgraph StateGraph(AgentState) subgraph.add_node(query, execute_tool) subgraph.add_node(format, respond) subgraph.add_edge(query, format) subgraph.add_edge(format, END) return subgraph.compile() # 在主图中添加子图作为节点 app graph.compile(checkpointermemory)子图的价值是模块化。多个 Agent 共享同一个订单处理逻辑时只需要复用这个子图即可。7. LangGraph 与 MCP 的工程配合7.1 MCP 是什么MCPModel Context Protocol是一套开放的协议用于让 AI 应用连接外部工具、数据源、文件系统等。你可以把它理解成“AI 应用时代的 USB-C 接口”。7.2 LangGraph Agent 接入 MCP 服务器在生产环境中LangGraph 的 Node 内部可以调用 MCP Client 去连接远程的 MCP Server。例如连接一个提供订单查询能力的 MCP Server。关键代码示意import json def call_mcp_tool(state: AgentState): Node 内部调用 MCP Server 暴露的工具 # 这部分是伪代码MCP SDK 的使用方式以官方文档为准 # client MCPClient(http://localhost:8000/mcp) # result client.call_tool(query_orders, {user_id: 1001}) result {status: success, data: 订单A已发货} return { tool_result: json.dumps(result, ensure_asciiFalse), should_call_tool: no }MCP 和 LangGraph 不冲突。LangGraph 更关注“流程怎么编”MCP 更关注“工具怎么接”。一个完整的 Agent 研发链路是MCP Server 负责统一暴露企业内部工具LangGraph 负责这些工具的调用时序、状态管理和异常补偿。8. 常见问题与排查思路问题现象可能原因排查方式解决方案langgraph安装失败Python 版本过低或依赖冲突python --version检查 pip 列表升级 Python 到 3.10创建新的虚拟环境重新安装执行invoke时提示KeyError: messagesState 中字段名与 Node 返回键不一致打印 State 类型定义检查所有 Node 的返回字典键统一字段命名确保 State 字段在启动前已存在条件边路由不到预期节点路由函数返回的字符串与映射字典键不匹配在路由函数中添加print输出返回值确保返回值与add_conditional_edges映射字典完全一致多轮对话没有记忆未配置 Checkpointer或未传递thread_id检查compile(checkpointer...)以及 config 里是否有thread_id加上 Checkpointer并在invoke的 config 中传thread_id工具调用后模型还在重复调用工具没有清空should_call_tool标记检查工具节点返回时是否重置控制字段工具执行完后设置should_call_tool no或通过独立字段区分不同阶段与 LangChain 版本不兼容langgraph和langchain-core版本不匹配查看完整错误日志检查依赖树全部升级到最新稳定版或锁定到兼容版本组合9. 最佳实践与工程建议9.1 状态设计原则对话数据用追加式 reducer如operator.add。控制数据用覆盖式 reducer。避免在 State 中存放大数据对象比如完整文件内容建议存引用或摘要。将“用户 ID”“会话 ID”等上下文信息放在 config 的configurable中而不是塞进 State。9.2 节点设计原则每个 Node 只做一件事。call_tool不要顺便写日志入库拆成独立节点。节点函数保持无副作用幂等便于重放和调试。所有 Node 内部错误都应该被捕获并通过控制字段路由到异常处理分支而不是直接抛异常导致整个图崩溃。9.3 工具调用安全边界工具的参数校验、权限校验必须在工具内部完成不能只靠模型自觉。高风险工具删除、退款、发送消息必须配合interrupt_before实现人工确认。工具调用要设计超时和重试机制防止外部服务不可用时拖死整个 Agent。日志中禁止打印用户的敏感字段如密码、Token、身份证号。9.4 日志与可观测性LangGraph 的stream()方法可以逐步输出每个节点的输入输出这在调试时非常有用。for chunk in app.stream( {messages: [{role: user, content: 查订单}]}, configconfig, stream_modeupdates ): print(chunk) # 每个节点的状态更新 print(- * 50)生产环境建议把stream_modeupdates的输出接入日志系统这样每次 Agent 执行的完整轨迹都能被还原。10. 总结与后续学习规划这篇文章从 LangGraph 的定位讲起带你理清了 LangChain 与 LangGraph 的本质区别接着拆解了 State、Node、Edge、Checkpointer 等核心概念然后通过“带工具调用的客服 Agent”示例跑通了第一版 LangGraph 应用最后补充了记忆、人工确认、子图、MCP 配合等进阶能力。现在你可以试着做一个完整的练习把这个客服 Agent 扩展成“支持多用户、支持多个工具查订单、查物流、查退款、带人工确认、带日志持久化”的生产级智能体。这个过程会逼你把本文中提到的知识点全部串起来。下一步值得深入的方向有三个Agents 抽象层LangGraph 在langchain.agents里提供了高层封装可以大幅简化代码。并行与 Fan-out 模式LangGraph 支持一个节点并行分发到多个分支最后汇总这是复杂 Agent 的常见模式。与 RAG 结合把 RAG 检索作为一个工具节点接入 LangGraph构建“能查文档的 Agent”这是目前企业落地最多的形态。建议先把文中的示例代码在自己的环境里完整跑通然后在 State 里增加一个retry_count字段自己实现一版带重试机制的工具调用流程。这一步做完你对 LangGraph 的理解就不是看过教程的程度了而是真正能上手写项目的程度。