“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 Chain | LangGraph |
|---|---|---|
| 核心模型 | 线性链 / 固定流程 | 图 / 状态机 |
| 分支和循环 | 弱 | 强 |
| 状态管理 | 靠外部传入,节点间不共享 | 内置 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_KEY="your-api-key"如果使用国内模型服务或本地部署的模型,只要兼容 OpenAI 接口协议,都可以通过ChatOpenAI(base_url=...)接入。
from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="your-model-name", base_url="http://localhost:8000/v1", # 根据实际情况填写 api_key="your-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.txt:
langgraph>=0.2.0 langchain>=0.2.0 langchain-openai>=0.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(model="your-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 "end"6.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 要接入数据库、文件系统、浏览器、设计稿、企业内网服务等大量外部系统,工具编写和权限管理会迅速失控。这就是 MCP(Model Context Protocol)要解决的核心问题。
7.1 MCP 是什么
MCP 是一个开放协议,它把“AI 应用需要的数据源和工具”标准化成统一的客户端-服务端模型。AI 应用是 MCP Client,外部系统通过 MCP Server 暴露工具和资源,两边通过 JSON-RPC 通信。这样,同一套 Agent 代码,只需要切换 MCP Server 配置,就能连接完全不同的外部系统。
对比一下:传统做法是你要为每个外部系统写自定义 SDK 集成代码;MCP 做法是外部系统提供一个标准 MCP Server,LangChain 客户端只需要加载这个 Server 暴露的工具列表,即可像调用本地工具一样调用远程能力。
7.2 Skill 与 MCP 有什么区别
随着 Agent 生态发展,很多开发者开始区分“Skill”和“MCP”:
| 对比项 | MCP | Skill |
|---|---|---|
| 本质 | 标准化连接协议 | 可复用的能力包 / 操作指令集合 |
| 侧重点 | 工具、资源、上下文的传输 | 行为流程和知识封装 |
| 例子 | 数据库 MCP Server、浏览器 MCP Server | “处理订单退款的技能”“写日报的技能” |
| 与 Agent 关系 | Agent 通过 MCP 调用外部工具 | Agent 通过加载 Skill 获得特定任务方法论 |
更通俗地说,MCP 更接近“插头标准”,解决的是“如何连接外部工具”;Skill 更接近“岗位手册”,解决的是“如何处理一类任务”。实际项目中两者往往配合使用:Agent 通过 Skill 知道任务该怎么拆解,通过 MCP 调取数据和能力。
7.3 在 LangGraph 中接入 MCP Server
LangChain 生态提供了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( command="python", 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 线程记忆:Checkpointer
LangGraph 提供 Checkpointer,把每一步的 State 都持久化到存储介质,并以thread_id区分不同会话。下次用同一个thread_id调用时,Agent 会自动恢复上次的 State。
# 文件路径:langgraph-demo/agent_memory.py from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() app = graph.compile(checkpointer=memory) config = {"configurable": {"thread_id": "user-1001"}} # 第一轮 result1 = app.invoke( {"messages": [HumanMessage(content="我叫小明,喜欢看科幻小说")]}, config=config, ) # 第二轮,用同一个 thread_id result2 = app.invoke( {"messages": [HumanMessage(content="你还记得我喜欢什么类型的小说吗?")]}, config=config, ) print(result2["messages"][-1].content)这里MemorySaver只是内存实现,适合开发和测试。生产环境建议使用 PostgreSQL、Redis 等外部存储作为 Checkpointer,LangGraph 官方提供了对应适配扩展。使用检查点后,还可以实现人工中断、回溯和容错恢复,这是 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 / END | langgraph 版本差异 | 查看安装版本文档 | 确认从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 敏感信息不要放进 State
State 会被 Checkpointer 持久化。密码、API Key、身份证号等敏感信息如果出现在 State 中,就可能被写入存储系统。生产环境要做好字段脱敏,必要时单独存储并引用 ID。
10.6 用 LangSmith 或自建日志追踪 Agent 运行链路
Agent 调试比普通接口难得多。建议从项目第一天就接入追踪:记录每个节点的输入输出、工具调用参数、token 消耗、耗时。没有追踪数据,Agent 出问题只能靠猜。
11. 总结与后续学习方向
这篇文章从 LangChain 与 LangGraph 的概念边界讲起,逐一跑通了图状态、节点、边、条件路由、工具调用循环、MCP 协议和 Memory 三层记忆体系。如果你把这些代码亲手跑一遍,已经具备了开发一个中等复杂度 Agent 的核心能力。下一步,可以往这几个方向深入:多 Agent 协作与子图拆分、人工审批中断与恢复、基于向量库的长期记忆检索、以及把 Agent 封装成可对外服务的 API。真实项目里,Agent 的稳定性往往取决于工具设计和状态管理,而不是模型本身;把这句话放在心里,能少走很多弯路。