资讯动态

LangGraph保姆级教程:从LangChain到状态化Agent编排实践

发布时间:2026/8/30 22:12:31 来源:尧图企业网站定制
“LangChain 入门后下一步该学什么”是过去半年我在 CSDN 后台看到最多的问题。很多人照着官方文档跑通了 RAG、跑通了简单的 Agent 示例但一旦进入真实项目——多步任务编排、Agent 反复调用工具、跨会话保存状态、接入外部数据源——立刻卡住。这篇“新版 LangChain LangGraph 保姆级教程”要解决的正是这个问题。今天文章的核心判断是LangGraph 不是 LangChain 的替代品而是从“链式调用”到“状态化 Agent 编排”的范式跃迁。你真正要补的不只是新 API 的写法而是 Agent 系统的状态管理、流程控制和记忆机制。文章会从零开始用完整代码带你跑通最小 LangGraph 应用、条件路由、工具调用、MCP 接入和智能体记忆 Memory这也是码士集团 AI 大模型系列课程里最常被追问的一条主线。1. 这篇文章真正要解决的问题如果你只是用 LangChain 写一个调用 LLM 的脚本那 LangGraph 暂时帮不上什么忙。但如果你的目标是构建一个像样的 AI Agent问题就完全不同了。一个真实的 Agent 至少要回答这几个问题当前任务做到哪一步了哪些步骤已经完成根据当前状态下一步该调用模型、调用工具还是直接结束多轮对话之间Agent 是否还记得用户之前说过什么工具返回结果后Agent 应该继续思考还是终止循环整个对话跨 session 之后历史状态怎样恢复LangChain 的传统 Chain 模型很难优雅地解决这些问题因为 Chain 本质上是“提前定义好的固定流程”。而 Agent 要求的是“根据中间结果动态决定下一步”这是一种循环结构不是线性结构。LangGraph 把 Agent 建模成一张图节点是处理单元边是状态流转整个运行过程由共享的 State 驱动。这种设计解决的不只是写法问题更是架构思维问题——从“搭积木”变成“设计状态机”。所以这篇文章适合这些读者学过 LangChain 但没深入 Agent 的开发者、想在 Spring Boot / Python 服务里嵌入 Agent 能力的后端工程师、以及被“Agent 到底怎么落地”折磨过的 AI 应用开发者。你读完能完成的事情包括搭建 LangGraph 开发环境、理解 State / Node / Edge 核心概念、实现一个带工具调用循环的 Agent、接入 MCP 协议扩展工具、用 Memory 机制保存短期和长期记忆。2. LangChain 与 LangGraph先分清这对最容易混淆的概念2.1 LangChain 解决的是“模型接入与工具封装”LangChain 最初解决的问题很实在让开发者用一种统一的方式接入不同大模型把 Prompt、模型、输出解析、向量库、文档加载器这些组件封装成可组合的模块。你可以把 LangChain 理解成一个“大模型开发工具箱”。写 RAG 应用时Retriever、Embedding、ChatPromptTemplate这些组件确实非常方便。但 LangChain 的 Chain 在执行逻辑上有天然局限它适合顺序流程和固定分支很难表达“执行到一半根据结果回到前面重新跑”的循环。而 Agent 的核心恰恰是循环模型决定调用工具工具返回结果模型再决定下一步直到它认为任务结束。2.2 LangGraph 解决的是“Agent 的状态与流程编排”LangGraph 是 LangChain 团队推出的低层编排框架用于构建有状态、可循环、支持分支和并发的 Agent 应用。它的核心抽象是三样东西State状态贯穿整个图执行过程的数据结构每个节点都能读取和更新。Node节点实际执行逻辑的函数比如调用模型、调用工具、写数据库。Edge边节点之间的连接包括普通边和条件边根据状态决定走哪条边。这种设计思路和很多后端工程师熟悉的有限状态机FSM高度相似。你可以把 Agent 的一次运行看成一次状态机执行初始状态进入图各个节点不断读取状态、执行逻辑、更新状态条件边决定下一步走向最终到达 END 节点。2.3 什么时候该用谁对比维度LangChain ChainLangGraph核心模型线性链 / 固定流程图 / 状态机分支和循环弱强状态管理靠外部传入节点间不共享内置 State全图共享适合场景RAG、简单问答、固定流程复杂 Agent、多工具调用、人工审批流学习曲线平缓略陡一个常见误区是学 LangGraph 就必须抛弃 LangChain。实际上 LangGraph 节点内部往往要依赖 LangChain 的模型封装、Prompt 模板和 Tool 定义。这两者不是替代关系而是上层组件与编排框架的关系。后续代码你会看到节点内部用的就是ChatOpenAI、tool这些 LangChain 组件。3. 核心概念拆解State、Node、Edge、条件边与子图3.1 State整个 Agent 的“共享内存”在 LangGraph 中State 是一个跨节点共享的数据结构通常用 TypedDict 定义。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_input: str tool_result: str finished: bool这里最容易踩坑的是字段的合并策略。默认情况下节点返回的字段会直接覆盖原值。但messages这种需要追加而非覆盖的字段必须通过Annotated指定 reducer 函数。官方提供的add_messages就是专门处理消息列表的 reducer它会把新消息追加到旧消息后面而不是整体替换。如果只写messages: list而不是messages: Annotated[list, add_messages]第二次节点更新 state 时之前的历史消息就会被冲掉。这是 LangGraph 初学阶段最经典的 bug。3.2 Node一个普通的 Python 函数节点就是函数输入是当前整个 State输出是一个字典表示要更新的字段。def call_model(state: AgentState): messages state[messages] response llm.invoke(messages) return {messages: [response]}注意节点函数不要修改传入的 state 参数本身而是返回一个新的字典。LangGraph 会根据返回字典和字段的 reducer 策略生成新的 State。保持节点函数“纯净”会让调试变得简单很多。3.3 Edge普通边与条件边普通边表示“执行完 A 就执行 B”。条件边则根据状态内容动态决定下一步走哪个节点。graph.add_edge(START, call_model) graph.add_conditional_edges( call_model, should_continue, # 返回下一个节点名字符串或列表 {continue: call_tool, end: END} )条件边是 Agent 循环的核心。模型输出会先被解析如果模型想调用工具就走call_tool节点如果模型给出了最终答案就走向 END。3.4 子图与并行分支复杂项目里可以把一组节点封装成子图Subgraph作为主图中的一个节点使用。这在“审批流嵌套”“多 Agent 协作”场景下非常有用。并行分支则通过给add_conditional_edges返回多个节点名实现适合多个独立工具同时调用并汇总结果。LangGraph 的威力在于只要你把状态和转移路径定义清楚Agent 的复杂行为就可以被完全可视化、可测试、可回溯。这一点是传统 Agent 框架很难做到的。4. 环境准备与基础配置LangGraph 是纯 Python 库安装非常简单。建议使用 Python 3.9 以上版本并新建虚拟环境。# 创建虚拟环境 python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate # 安装核心依赖 pip install langgraph langchain langchain-openai如果你的 Agent 需要访问 OpenAI 兼容接口、本地大模型或其他模型服务请确保已经准备好对应的 API Key并写入环境变量。export OPENAI_API_KEYyour-api-key如果使用国内模型服务或本地部署的模型只要兼容 OpenAI 接口协议都可以通过ChatOpenAI(base_url...)接入。from langchain_openai import ChatOpenAI llm ChatOpenAI( modelyour-model-name, base_urlhttp://localhost:8000/v1, # 根据实际情况填写 api_keyyour-api-key )注意不同版本的 langgraph 在 API 细节上有差异。例如START/END从langgraph.graph导入在新旧版本中可能路径不同。建议安装后先打印版本号。python -c import langgraph; print(langgraph.__version__)后面示例代码以常见稳定 API 为准遇到 ImportError 时优先查看你安装版本的官方文档。5. 第一个 LangGraph 应用从 LangChain 链迁移过来为了理解 LangGraph 的执行模型我们先从最小可运行示例开始避免一上来就写复杂的 Agent。5.1 项目结构为方便演示我们建立一个简单项目目录langgraph-demo/ ├── requirements.txt └── basic_graph.pyrequirements.txtlanggraph0.2.0 langchain0.2.0 langchain-openai0.2.05.2 最小代码两个节点组成一张图# 文件路径langgraph-demo/basic_graph.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): message: str def node_1(state: State): print(执行 node_1) return {message: state[message] 来自 node_1} def node_2(state: State): print(执行 node_2) return {message: state[message] 来自 node_2} # 1. 创建图 graph StateGraph(State) # 2. 添加节点 graph.add_node(node_1, node_1) graph.add_node(node_2, node_2) # 3. 连接节点 graph.add_edge(START, node_1) graph.add_edge(node_1, node_2) graph.add_edge(node_2, END) # 4. 编译出可执行对象 app graph.compile() # 5. 调用 result app.invoke({message: 你好}) print(result)运行python basic_graph.py预期输出执行 node_1 执行 node_2 {message: 你好来自 node_1来自 node_2}5.3 关键点解释这个例子看起来很简单但它把 LangGraph 的执行模型完整展示出来了初始状态从 START 进入图按边依次执行节点每个节点的返回值更新 State最终从 END 退出。和普通for循环调用函数最大的区别是图可以被持久化、被中断、被恢复、被分步执行。后面讲的 Memory、人工审批、断点续跑都依赖这个执行模型。如果你已经熟悉 LangChain 的链式写法可以把这个过程理解为RunnableSequence只能直线执行而StateGraph允许你画一张任意拓扑结构的有向图。6. AI-Agent 实战工具调用与条件路由跑通最小图之后我们进入真正有 Agent 味道的例子。这个例子会包含用tool定义工具。让模型具备工具调用能力。通过条件边判断“是否还需要调用工具”。形成“模型 - 工具 - 模型”的循环。6.1 定义工具# 文件路径langgraph-demo/agent_tool.py from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气。 # 真实项目里这里会调用天气 API return f{city} 今天晴天气温 22 摄氏度。工具函数必须有清晰的 docstring因为大模型是根据函数签名和描述来决定是否调用它的。描述越清楚模型越不容易调错工具。6.2 编写 Agent 节点与条件路由# 文件路径langgraph-demo/agent_tool.py续 from typing import Annotated, TypedDict from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_core.messages import AIMessage, HumanMessage class AgentState(TypedDict): messages: Annotated[list, add_messages] llm ChatOpenAI(modelyour-model-name) llm_with_tools llm.bind_tools([get_weather]) # 节点1调用模型 def call_model(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]} # 节点2执行工具调用 def call_tool(state: AgentState): last_message state[messages][-1] tool_calls last_message.tool_calls results [] for tc in tool_calls: tool_name tc[name] tool_args tc[args] if tool_name get_weather: tool_output get_weather.invoke(tool_args) else: tool_output f未找到工具: {tool_name} results.append( { role: tool, name: tool_name, content: tool_output, tool_call_id: tc[id], } ) return {messages: results} # 条件路由如果模型想调用工具就继续否则结束 def should_continue(state: AgentState): last_message state[messages][-1] if last_message.tool_calls: return call_tool return end6.3 组装图并运行# 文件路径langgraph-demo/agent_tool.py续 graph StateGraph(AgentState) graph.add_node(call_model, call_model) graph.add_node(call_tool, call_tool) graph.add_edge(START, call_model) graph.add_conditional_edges( call_model, should_continue, { call_tool: call_tool, end: END, } ) graph.add_edge(call_tool, call_model) app graph.compile() result app.invoke({ messages: [HumanMessage(content北京天气怎么样)] }) for msg in result[messages]: print(msg.type, :, msg.content)6.4 运行流程说明一次典型执行过程是用户消息进入call_model模型返回一个AIMessage其中tool_calls包含了get_weather的调用请求。should_continue检测到tool_calls不为空走call_tool节点。call_tool真正执行工具并把工具结果封装成 tool 消息放回 State。图回到call_model模型拿到工具结果后生成最终回答此时tool_calls为空。should_continue返回end流程结束。这种“模型提议调用 - 框架执行工具 - 结果反馈给模型”的循环就是 Agent 区别于普通 RAG 应用的核心机制。很多项目里循环跑十几轮是正常的因此要非常小心工具的幂等性和超时控制避免 Agent 卡在死循环里空转。这里有个常见问题如果模型一直要求调用工具图就会一直循环。实际项目中建议在 State 里加一个step_count字段每次进入call_tool时加一超过阈值后强制走 END。LangGraph 的RecursionLimit也能兜底但业务层面的限制更友好。7. MCP 协议让 Agent 以标准方式连接外部工具手动用tool定义工具很直观但每个工具都要写一遍封装。如果 Agent 要接入数据库、文件系统、浏览器、设计稿、企业内网服务等大量外部系统工具编写和权限管理会迅速失控。这就是 MCPModel Context Protocol要解决的核心问题。7.1 MCP 是什么MCP 是一个开放协议它把“AI 应用需要的数据源和工具”标准化成统一的客户端-服务端模型。AI 应用是 MCP Client外部系统通过 MCP Server 暴露工具和资源两边通过 JSON-RPC 通信。这样同一套 Agent 代码只需要切换 MCP Server 配置就能连接完全不同的外部系统。对比一下传统做法是你要为每个外部系统写自定义 SDK 集成代码MCP 做法是外部系统提供一个标准 MCP ServerLangChain 客户端只需要加载这个 Server 暴露的工具列表即可像调用本地工具一样调用远程能力。7.2 Skill 与 MCP 有什么区别随着 Agent 生态发展很多开发者开始区分“Skill”和“MCP”对比项MCPSkill本质标准化连接协议可复用的能力包 / 操作指令集合侧重点工具、资源、上下文的传输行为流程和知识封装例子数据库 MCP Server、浏览器 MCP Server“处理订单退款的技能”“写日报的技能”与 Agent 关系Agent 通过 MCP 调用外部工具Agent 通过加载 Skill 获得特定任务方法论更通俗地说MCP 更接近“插头标准”解决的是“如何连接外部工具”Skill 更接近“岗位手册”解决的是“如何处理一类任务”。实际项目中两者往往配合使用Agent 通过 Skill 知道任务该怎么拆解通过 MCP 调取数据和能力。7.3 在 LangGraph 中接入 MCP ServerLangChain 生态提供了langchain-mcp-adapters可以把 MCP Server 暴露的工具做标准转换。由于该库仍处于快速迭代阶段下面代码给出的是常见写法具体 API 请以你安装的版本为准。# 文件路径langgraph-demo/mcp_demo.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools async def load_remote_tools(): # 以 stdio 方式启动一个本地 MCP Server server_params StdioServerParameters( commandpython, args[my_mcp_server.py], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await load_mcp_tools(session) return tools if __name__ __main__: tools asyncio.run(load_remote_tools()) for t in tools: print(t.name, t.description)调用时会启动一个本地 MCP Server这里是my_mcp_server.py通过标准的 stdio 通道进行通信。LangGraph 节点内部使用这些 tools 的方式和普通 LangChain 工具完全一致仍然是通过bind_tools绑定到模型。接入 MCP 后Agent 的外部工具扩展从“写代码”变成了“配置一个 Server 地址”。对于企业内网环境还可以通过 SSE 或 Streamable HTTP 方式连接远程 MCP Server团队可以共享一套工具服务而不是每个 Agent 重复实现一遍。需要特别提醒MCP Server 拥有工具执行能力和资源访问权限生产环境接入前必须做好身份认证、权限最小化和审计。不要直接把一个能执行任意 SQL 的 MCP Server 暴露给 Agent否则 Agent 可能拿着用户的自然语言请求做出不可控操作。8. 智能体记忆 Memory从单次会话到长期记忆没有记忆的 Agent每次对话都是一次“失忆”交互。用户需要在每轮对话里重复背景信息Agent 也无法感知任务上下文。LangGraph 解决记忆问题的方式分成三层。8.1 短期记忆State 本身刚才的 Agent 里messages字段就是短期记忆。每一轮“模型 - 工具 - 模型”产生的消息都会通过add_messages追加到 State。只要一次invoke没有结束所有节点都能看到完整上下文。但注意invoke结束后State 默认就丢失了。下一次invoke是全新状态。8.2 线程记忆CheckpointerLangGraph 提供 Checkpointer把每一步的 State 都持久化到存储介质并以thread_id区分不同会话。下次用同一个thread_id调用时Agent 会自动恢复上次的 State。# 文件路径langgraph-demo/agent_memory.py from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory) config {configurable: {thread_id: user-1001}} # 第一轮 result1 app.invoke( {messages: [HumanMessage(content我叫小明喜欢看科幻小说)]}, configconfig, ) # 第二轮用同一个 thread_id result2 app.invoke( {messages: [HumanMessage(content你还记得我喜欢什么类型的小说吗)]}, configconfig, ) print(result2[messages][-1].content)这里MemorySaver只是内存实现适合开发和测试。生产环境建议使用 PostgreSQL、Redis 等外部存储作为 CheckpointerLangGraph 官方提供了对应适配扩展。使用检查点后还可以实现人工中断、回溯和容错恢复这是 Agent 上生产环境非常关键的能力。8.3 长期记忆向量库与外部存储Checkpointer 保存的是“某个线程的完整状态”但无法回答这类问题所有用户都问过哪些高频问题某个用户的长期偏好是什么长期记忆通常把关键信息抽取出来写入外部存储。常见方案有三种向量数据库将对话历史切片做 Embedding后续检索与当前问题相关的历史片段。摘要记忆定期把旧对话摘要成一条精简记录随 State 一起传入。结构化存储把用户偏好、订单状态、权限角色等业务数据写入 MySQL / Redis随取随用。实际项目中长期记忆不是单一组件而是“State Checkpointer 业务表 向量库”的组合。你需要自己设计“什么时候把短期记忆写入长期记忆”。比如对话结束前可以让模型总结一条用户偏好并写入用户画像表。8.4 节点函数里如何更新多个 State 字段这是 LangGraph 新手最高频的问题。直接返回一个“普通字典”Key 就是 State 字段名Value 就是要更新的值即可。def process_node(state: AgentState): return { messages: append_new_message(state[messages]), # 注意 reducer user_input: state[user_input], tool_result: success, finished: True, }如果字段配置了Annotated[list, add_messages]LangGraph 会自动做追加合并如果字段是普通类型直接覆盖。不要在节点函数里手动修改state字典本身这可能导致状态不可追溯和调试困难。9. 常见问题与排查思路问题现象可能原因排查方式解决方案messages历史被覆盖字段未使用 reducer打印每次节点返回的 state 字段改为Annotated[list, add_messages]ImportError: 找不到 START / ENDlanggraph 版本差异查看安装版本文档确认从langgraph.graph导入升级到新版模型返回空tool_calls工具描述不清晰或模型不支持 function call打印AIMessage.tool_calls原始值优化 tool docstring更换支持工具调用的模型Agent 无限循环调用工具工具结果无法让模型下结论或工具报错被吞掉限制 step_count开启日志追踪加入最大轮数限制工具异常返回明确错误信息程序运行到某一步直接报 KeyError节点返回值包含 State 未定义的字段检查返回值 Key 与 TypedDict 定义只返回已定义字段或补充 State 定义MCP Server 连接超时stdio 启动命令错误、网络隔离直接运行 MCP Server 命令验证确认 Python 路径和依赖使用 HTTP 连接时检查端点上下文过长导致超限节点循环次数多消息全部追加查看 token 使用量日志加入摘要节点定期压缩messages内存持续增长大量中间状态被 Checkpointer 保存监控内存和存储占用使用外部持久化存储设置历史清理策略10. 最佳实践与工程建议10.1 设计 State 时留足扩展空间State 是 LangGraph 应用的数据契约。不要只定义当前节点需要的字段要为工具结果、step_count、错误信息等预留位置。字段命名保持一致避免user_input和query混用。10.2 节点函数保持单一职责一个节点只做一件事调模型、执行工具、查询数据库、写日志。不要把一个节点写成几百行的“上帝函数”。否则条件路由和错误重试都无从下手。10.3 有条件边必须可测试条件边映射函数是 Agent 行为的决策中枢。把should_continue这类函数单独抽出来用单元测试覆盖关键分支路径。LangGraph 的重点不是模型而是确定性逻辑。10.4 工具层做好超时、幂等与错误返回Agent 循环调用工具时工具超时必须小于整个图的超时时间。工具要保证幂等尤其是写入类操作。工具出错时不要直接抛异常而是把错误信息作为 tool message 返回给模型让模型决定怎么办。10.5 敏感信息不要放进 StateState 会被 Checkpointer 持久化。密码、API Key、身份证号等敏感信息如果出现在 State 中就可能被写入存储系统。生产环境要做好字段脱敏必要时单独存储并引用 ID。10.6 用 LangSmith 或自建日志追踪 Agent 运行链路Agent 调试比普通接口难得多。建议从项目第一天就接入追踪记录每个节点的输入输出、工具调用参数、token 消耗、耗时。没有追踪数据Agent 出问题只能靠猜。11. 总结与后续学习方向这篇文章从 LangChain 与 LangGraph 的概念边界讲起逐一跑通了图状态、节点、边、条件路由、工具调用循环、MCP 协议和 Memory 三层记忆体系。如果你把这些代码亲手跑一遍已经具备了开发一个中等复杂度 Agent 的核心能力。下一步可以往这几个方向深入多 Agent 协作与子图拆分、人工审批中断与恢复、基于向量库的长期记忆检索、以及把 Agent 封装成可对外服务的 API。真实项目里Agent 的稳定性往往取决于工具设计和状态管理而不是模型本身把这句话放在心里能少走很多弯路。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价