一、LangGraph 是什么LangGraph 是 LangChain 团队官方推出的基于状态机的智能体 (Agent) 开发框架它基于 LangChain 生态构建专门解决了 LangChain 原生 LCEL 链式调用的核心短板用来构建有状态、可循环、可持久化、支持复杂分支逻辑的 Agent 工作流。它的核心设计思想是以「状态」为核心用「节点 边」的图结构编排工作流天然支持 Agent 所需的「思考 - 行动 - 反思」循环逻辑是目前 LangChain 生态中构建生产级 Agent 的官方标准方案。关键补充LangGraph 不是替代 LangChain而是 LangChain 生态的补充。LangChain 提供 LLM、工具、向量库、文档加载器等原子组件LangGraph 负责这些组件的复杂流程编排、状态管理和持久化两者通常配合使用。二、LangGraph 核心概念逐点详解 实战示例1. State状态LangGraph 的灵魂State 是 LangGraph 最核心的概念它是贯穿整个工作流的全局数据容器工作流中所有节点的输入都来自 State节点的输出都会更新 State整个工作流的执行过程就是 State 不断更新的过程。核心特性强类型定义通常用TypedDict或 PydanticBaseModel定义保证类型安全Reducer 机制通过Annotated绑定 reducer 函数定义 State 字段的更新规则避免手动处理数据合并全链路共享所有节点都可以读取 / 更新 State是节点间通信的唯一载体实战示例你正在使用的写法from typing import TypedDict, Sequence, Annotated from langchain_core.messages import BaseMessage from langgraph.graph import add_messages # 定义State结构 class ReActState(TypedDict): # 消息列表绑定add_messages reducer自动实现消息追加无需手动拼接列表 messages: Annotated[Sequence[BaseMessage], add_messages] loop_count: int # 防死循环计数器 user_id: str # 用户ID绑定长期记忆 error: str | None # 错误信息用于异常处理路由关键知识点add_messages是 LangGraph 内置的最常用 reducer专门处理消息列表的追加自动去重、合并消息是多轮对话的核心节点的返回值只需要包含 State 的子集不需要返回完整的 StateLangGraph 会自动根据 reducer 规则合并到全局 State 中你之前踩过的「HumanMessage 没有 tool_calls」的坑本质就是 State 中 messages 列表的最后一条消息类型判断错误属于 State 数据的处理问题2. Node节点工作流的执行单元Node 是工作流中的具体执行逻辑单元对应 Agent 中的一个动作比如 LLM 思考、工具调用、文档检索、反思校验本质是一个接收 State、返回 State 更新的纯函数。节点的 3 种核心类型节点类型作用示例普通自定义节点实现自定义业务逻辑LLM 思考节点、文档检索节点、反思校验节点内置工具节点ToolNodeLangGraph 内置封装专门处理 LLM 的工具调用你代码中的ToolNode(tools)自动解析AIMessage中的tool_calls并执行工具子图节点把一个完整的 LangGraph 图作为节点嵌入到另一个图中多智能体场景中把单个 Agent 的工作流封装为子图实战示例你正在使用的写法from langchain_core.messages import AIMessage # 自定义节点LLM思考节点 def llm_think(state: ReActState): 核心思考节点加载记忆、调用LLM生成回答/工具调用 messages state[messages] user_id state[user_id] error None try: # 加载长期记忆 user_preference store.get((user_memory, user_id), preference) if user_preference: messages [HumanMessage(contentf用户偏好{user_preference.value})] messages # 调用LLM response llm.invoke(messages) except Exception as e: response AIMessage(contentf请求失败{str(e)}) error str(e) # 返回State的更新LangGraph自动合并 return { messages: [response], loop_count: state[loop_count] 1, user_id: user_id, error: error } # 内置工具节点 tool_executor ToolNode(tools, handle_tool_errorsTrue)3. Edge边工作流的流程控制器Edge 是连接节点的「桥梁」定义了工作流的执行顺序和分支逻辑是 LangGraph 实现循环、条件判断、异常降级的核心。边的 3 种核心类型边类型作用语法普通边固定的执行顺序上一个节点执行完固定执行下一个节点builder.add_edge(START, llm_think)、builder.add_edge(tool_executor, llm_think)条件边核心分支逻辑根据当前 State 的数据动态决定下一个执行的节点builder.add_conditional_edges(llm_think, should_continue, 路由映射)起止边特殊的固定边START代表工作流入口END代表工作流结束builder.add_edge(llm_think, END)实战示例你正在使用的写法from typing import Literal from langgraph.constants import START, END # 条件路由函数接收State返回下一个节点的名称 def should_continue(state: ReActState) - Literal[llm_think, backup_llm_think, tool_executor, end]: # 优先处理异常有错误则走备用LLM节点 if state.get(error): if state[loop_count] 2: return backup_llm_think else: return end # 过滤出AI消息避免HumanMessage没有tool_calls的报错 ai_messages [msg for msg in state[messages] if isinstance(msg, AIMessage)] if not ai_messages: return llm_think last_message ai_messages[-1] # 正常流程判断有工具调用则执行工具否则结束 if not last_message.tool_calls or state[loop_count] 5: return end return tool_executor # 构建图时绑定条件边 builder StateGraph(ReActState) builder.add_node(llm_think, llm_think) builder.add_node(tool_executor, tool_executor) # 入口边 builder.add_edge(START, llm_think) # 条件边llm_think执行完后根据should_continue的返回值路由 builder.add_conditional_edges( llm_think, should_continue, # 路由映射函数返回值 → 目标节点 { backup_llm_think: backup_llm_think, tool_executor: tool_executor, end: END } ) # 普通边工具执行完后回到llm_think实现ReAct循环 builder.add_edge(tool_executor, llm_think)4. Checkpointer检查点持久化与会话恢复Checkpointer 是 LangGraph 内置的状态持久化组件它会在工作流的每一步执行完成后自动保存 State 的完整快照Checkpoint实现会话恢复、断点续跑、历史回溯。核心能力会话持久化服务重启后通过thread_id恢复之前的对话状态实现多轮对话不丢失断点续跑工作流执行中断后从断点继续执行无需从头开始历史回溯可以查看工作流每一步的 State 变化方便调试和问题排查时间旅行支持回滚到历史任意一个检查点重新执行工作流常用实现实现类适用场景特点MemorySaver开发学习、本地测试内存存储零配置重启数据丢失RedisSaver生产环境、分布式部署Redis 持久化重启不丢数据支持多实例共享PostgresSaver生产环境、企业级部署数据库持久化支持事务、数据备份实战示例你正在使用的写法from langgraph.checkpoint.redis import RedisSaver from langgraph.checkpoint.memory import MemorySaver # 1. 初始化持久化组件 checkpointer RedisSaver(redis://localhost:6379/0) # 开发环境用MemorySaver零配置 # checkpointer MemorySaver() # 2. 编译图时绑定checkpointer react_agent builder.compile(checkpointercheckpointer) # 3. 调用时通过thread_id绑定会话实现持久化 config {configurable: {thread_id: session_001, user_id: user_001}} # 同一个thread_id的调用会自动恢复之前的State result react_agent.invoke({messages: [HumanMessage(content你好)]}, configconfig)5. Store存储跨会话长期记忆Store 是 LangGraph 内置的跨会话全局存储组件和 Checkpointer 互补Checkpointer存储单会话内的工作流状态生命周期和会话绑定Store存储跨会话的全局数据比如用户长期偏好、知识库元数据、全局配置生命周期和用户 / 业务绑定核心能力键值对存储支持命名空间namespace实现数据隔离原生集成到 LangGraph 工作流中节点内可以直接读写支持 Redis、Postgres 等持久化实现实战示例你正在使用的写法from langgraph.store.redis import RedisStore # 初始化Store store RedisStore.from_url(redis://localhost:6379/1) # 编译图时绑定store react_agent builder.compile(checkpointercheckpointer, storestore) # 节点内读写Store def llm_think(state: ReActState): user_id state[user_id] # 读取用户长期记忆 user_preference store.get((user_memory, user_id), preference) # 写入用户长期记忆 store.put((user_memory, user_id), preference, 喜欢用摄氏度偏好简洁回答) ...6. 其他核心概念Thread会话通过thread_id唯一标识对应一个独立的工作流执行实例是 Checkpointer 隔离不同会话的核心标识。Compilation编译builder.compile()把定义好的图结构转换成可执行的运行时对象同时绑定 Checkpointer、Store、中断配置等。Interrupt中断LangGraph 原生支持工作流执行中暂停等待人工输入 / 审核后继续执行适合需要人工介入的复杂 Agent 场景。Stream流式输出支持多种流式模式values、updates、messages可以实时输出工作流的每一步执行结果提升用户体验。三、LangGraph vs LangChain 深度对比3.1 核心差异总表表格对比维度LangChainLangGraph核心设计范式链式调用LCEL线性 DAG有向无环图状态机图结构支持有环图流程控制能力仅支持线性执行、简单分支不支持循环原生支持任意循环、多条件分支、子图嵌套、异常降级状态管理无内置全局状态需要手动传递和管理上下文内置全局 State自动管理状态更新、合并、持久化持久化能力无原生持久化需要自己实现会话存储原生 Checkpointer 机制一行代码实现会话持久化、断点续跑Agent 能力仅支持简单的单轮工具调用复杂 Agent 需要大量二次开发原生适配 ReAct、Plan-Execute、多智能体等主流 Agent 范式是 LangChain 官方 Agent 标准方案可调试性链式执行黑盒难以回溯中间步骤每一步都有 Checkpoint 快照可完整回溯工作流的每一步状态变化人工介入无原生支持需要自己实现中断逻辑原生 Interrupt 机制支持工作流暂停、人工介入、恢复执行适用场景简单线性任务、单轮对话、无状态一次性任务复杂 Agent、多轮对话、有状态工作流、多智能体协作、需要人工介入的场景学习曲线低简单链式调用上手快中需要理解状态机、图结构的核心概念上限更高3.2 关键差异详解循环支持是核心分水岭LangChain 的 LCEL 是无环的线性链天生无法实现 Agent 最核心的「思考→工具调用→再思考」的循环逻辑。而 LangGraph 基于状态机天然支持循环这也是 LangChain 官方推出 LangGraph 的核心原因 ——LCEL 根本不适合构建 Agent。状态管理的本质区别LangChain 中上下文需要在链的每个步骤手动传递多轮对话的状态管理需要自己写代码实现而 LangGraph 中State 是全局共享的所有节点都可以读取和更新内置的 reducer 机制自动处理数据合并开发者只需要关注业务逻辑。生态定位的互补性LangGraph 不是替代 LangChain而是扩展了 LangChain 的能力边界。LangChain 提供了海量的组件LLM 集成、工具、向量库、文档加载器等LangGraph 负责把这些组件编排成复杂、健壮的 Agent 工作流两者是「原子组件」和「流程编排框架」的关系。四、什么时候应该使用 LangGraph4.1 优先使用 LangGraph 的 7 大场景构建任何类型的智能体 (Agent)只要你的场景需要 LLM 自主思考、调用工具、多步执行比如 ReAct 工具调用 Agent、客服 Agent、代码执行 Agent必须优先用 LangGraph。它原生支持 Agent 的循环逻辑是 LangChain 官方的标准方案比自己用 LCEL 封装要简单、健壮得多。需要多轮对话、会话持久化的场景比如智能客服、个人知识库问答、陪伴式机器人需要服务重启后用户的对话历史不丢失LangGraph 的 Checkpointer 一行代码就能实现会话持久化无需自己开发状态存储。复杂的非线性工作流比如 RAG 反思重写、文档审核→纠错→二次审核、多步骤分支判断、错误重试 / 降级兜底这些场景用 LCEL 很难实现用 LangGraph 的条件边、节点循环可以轻松搞定。需要人工介入的场景比如合同审核 Agent 执行到关键步骤需要人工确认、代码生成 Agent 需要人工审核代码后再执行LangGraph 的 Interrupt 原生支持工作流暂停和恢复无需自己开发状态暂停机制。多智能体协作场景比如产品经理 Agent 开发 Agent 测试 Agent 协作完成需求开发LangGraph 的子图、多节点并行执行可以轻松实现多智能体的分工和协作。需要高可观测性、可调试性的生产级应用生产环境中你需要知道 Agent 每一步做了什么、为什么出错LangGraph 的 Checkpoint 机制会保存每一步的完整状态可以完整回溯执行过程快速定位问题。需要断点续跑的长任务场景比如长文档总结、批量数据处理、自动化报告生成这些长任务如果中途中断LangGraph 可以从断点继续执行无需从头开始节省时间和 Token 成本。4.2 用 LangChain LCEL 就足够的场景简单的线性单轮任务比如文本翻译、文案生成、单轮摘要、简单的单轮 RAG提问→检索→生成回答没有循环、没有多轮对话、没有分支逻辑用 LCEL 一行代码就能实现无需引入 LangGraph。无状态的一次性任务执行一次就结束不需要保存状态、不需要多轮对话比如一次性的文档格式转换、数据清洗、单条数据的标签生成。快速原型验证想快速验证一个想法的效果用 LCEL 可以快速搭起一个最简原型验证通过后再用 LangGraph 重构为生产级的工作流。五、LangGraph 最佳实践结合你的踩坑经验State 设计优先先定义好 State 结构再开发节点和边。State 只保留工作流必须的字段避免冗余数据复杂字段一定要绑定 reducer避免手动处理数据合并出错。优先用内置组件工具调用优先用内置的ToolNode消息管理优先用add_messagesreducer持久化优先用官方的 Saver 实现不要自己重复造轮子避免踩坑。条件边一定要做边界校验条件路由函数中一定要对 State 的数据做类型校验和空值判断比如你之前踩过的「HumanMessage 没有 tool_calls」的坑就是没有先过滤消息类型导致的。开发用 MemorySaver生产用 Redis/PostgresSaver开发阶段用MemorySaver零配置快速迭代上线生产环境切换到 Redis/PostgresSaver保证数据持久化。节点职责单一一个节点只做一件事比如「检索节点」只做文档检索「生成节点」只做 LLM 调用不要把多个逻辑写在一个节点里方便调试和复用。