一、LangGraph 是什么LangGraph 是 LangChain 团队推出的面向大模型复杂流程的有状态有向图编排框架底层基于可迭代状态机实现原生支持分支、循环、状态持久、人工介入解决传统线性 Chain 无法处理长流程、自主 Agent 的问题。1.1 诞生背景根据 LangChain 2026 年发布的《State of Agent Engineering》报告超过 60% 的生产级 Agent 故障可追溯到状态管理失败Agent 在工作流中丢失上下文、重复执行步骤、或因崩溃无法恢复。LangGraph 正是为解决这一核心痛点而生。发布时间2024 年由 LangChain Inc. 开源GitHub Stars截至 2026 年 5 月已突破 32,000企业用户Klarna、Uber、LinkedIn、AppFolio 等 20 企业生产使用底层引擎受 Google Pregel 系统启发基于消息传递的图计算模型1.2 核心定位LangGraph 不是 LangChain 的替代品而是其底层执行引擎层级角色说明Deep Agents套件层开箱即用智能体套件LangChain组件层通用 LLM 应用开发框架LangGraph运行时层Agent 流程运行时本篇核心2026 年起LangChain 的 Agent 已全部运行在 LangGraph 运行时之上二者是分层依赖而非竞争关系。1.3 核心价值图结构精准管控节点、边、流转完全自定义流程控制权最高状态持久化每一步执行后自动快照支持断点续跑、跨会话记忆循环与分支原生支持循环Cycle实现多轮 Agent条件边实现动态路由人工介入HITL流程可暂停、可审核、可回滚满足企业治理需求模型无关兼容 OpenAI、Anthropic、Mistral 等所有 LangChain 支持的 LLM二、核心概念State、Nodes、Edges 三元组LangGraph 的核心是将 Agent 工作流建模为图由三个关键组件定义节点Nodes完成工作 → 边Edges指示下一步 → 状态State贯穿全局2.1 State状态全局共享数据容器State 是整张图唯一的全局数据容器所有节点共享读写是数据流转的核心载体。定义方式使用TypedDict约束字段名和数据类型配合AnnotatedReducer归约函数定义字段合并规则。fromtypingimportTypedDict,AnnotatedimportoperatorclassGraphState(TypedDict):query:str# 普通字段新值覆盖旧值messages:Annotated[list[str],operator.add]# 列表字段追加合并step:int两种更新规则默认覆盖新值直接替换旧值普通字段列表追加Annotated[list, operator.add]多轮对话标准用法2.2 Nodes节点最小执行单元Node 是图中最小执行单元代表一段独立逻辑调用 LLM、执行工具、数据处理、规则判断。本质普通 Python 函数入参当前全局State出参局部更新字典只返回需要修改的字段不用返回完整 State特性节点本身无状态纯计算逻辑数据全部托管在全局 Statedefllm_node(state:GraphState)-dict:# 接收当前状态执行逻辑responsellm.invoke(state[query])# 只返回需要更新的字段return{messages:[response]}2.3 Edges边定义执行流向Edge 定义节点间的执行流向是实现线性 / 分支 / 循环的关键边类型API作用适用场景普通边add_edge(A, B)固定跳转A → B线性串行流程条件边add_conditional_edges(node, router_fn)动态路由根据 State 选择下一节点分支、循环入口点set_entry_point(node)等价于add_edge(START, node)定义图起点结束点set_finish_point(node)等价于add_edge(node, END)定义图终点两个特殊节点START流程唯一入口END流程终止节点2.4 StateGraph状态图实例由State Node Edge组装而成的可执行对象fromlanggraph.graphimportStateGraph,START,END graphStateGraph(GraphState)graph.add_node(llm,llm_node)graph.add_edge(START,llm)graph.add_edge(llm,END)# 编译后才能执行appgraph.compile()生命周期定义结构 → 编译compile→ 执行invoke / stream三、底层执行机制迭代式状态机LangGraph 底层不是一次性执行的脚本而是循环驱动的迭代式状态机。3.1 整体执行流程1. 初始化阶段接收入参初始化全局 State若开启 Checkpointer优先从持久化存储恢复 2. 主调度循环核心 ├─ 取出当前待执行节点 ├─ 执行节点函数得到「局部更新字典」 ├─ 根据字段绑定的 Reducer 规则合并新旧 State ├─ 匹配当前节点的 Edge普通边 / 条件边计算下一节点 └─ 若下一节点是 END终止循环否则切换指针进入下一轮迭代 3. 结束阶段返回最终 State开启 Checkpointer 则保存最终快照3.2 超步Super-step概念LangGraph 底层图算法基于消息传递模型程序以离散的超步执行超步 图节点的一次迭代并行运行的节点属于同一个超步顺序运行的节点属于不同超步节点接收到新消息状态时变为active执行完后投票halt当所有节点都inactive时图执行结束四、状态管理Reducer 与消息系统4.1 Reducer归约函数配合Annotated使用定义同字段多份更新内容的合并策略fromtypingimportAnnotatedfromlanggraph.graphimportadd_messagesclassAgentState(TypedDict):# 使用 add_messages reducer智能追加支持去重、覆盖、删除messages:Annotated[list,add_messages]step:inttool_output:str|None内置 Reduceroperator.add列表拼接、add_messages消息智能合并自定义 Reducer可实现去重、保留最新、条件合并等逻辑4.2 MessagesState内置消息状态类LangGraph 提供内置的MessagesState自带add_messagesreducer专门管理对话历史fromlanggraph.graphimportMessagesState# 等价于手写# class MessagesState(TypedDict):# messages: Annotated[list, add_messages]graphStateGraph(MessagesState)4.3 消息类型体系消息类型作用示例HumanMessage用户输入“你好”AIMessageAI 回复含tool_calls“我需要查询天气”SystemMessage系统指令“你是专业助手”ToolMessage工具执行结果“北京今天 25℃ 晴”五、Checkpointer状态持久化中间件Checkpointer 是 LangGraph 实现生产级可靠性的核心组件每一步执行后自动快照全局 State。5.1 为什么需要 Checkpointer能力说明断点续跑服务崩溃重启后从最近 checkpoint 恢复无需重跑跨会话记忆通过thread_id实现多会话隔离与历史回溯时间旅行调试查看任意历史状态支持 replay 重放人工介入流程暂停后外部可读取 / 修改 State 再恢复5.2 三种 Checkpointer 选型# 1. 内存版仅开发测试fromlanggraph.checkpoint.memoryimportInMemorySaver checkpointerInMemorySaver()# 2. SQLite 版本地生产fromlanggraph.checkpoint.sqliteimportSqliteSaver checkpointerSqliteSaver.from_conn_string(checkpoints.db)# 3. Postgres 版生产推荐支持高并发fromlanggraph.checkpoint.postgresimportPostgresSaver checkpointerPostgresSaver.from_conn_string(postgresql://...)生产建议MemorySaver仅供本地开发SqliteSaver在高并发下写入性能差生产环境必须使用PostgresSaver。5.3 使用示例# 编译时传入 checkpointerappgraph.compile(checkpointercheckpointer)# 通过 thread_id 隔离会话config{configurable:{thread_id:user-123}}# 执行resultapp.invoke({messages:[HumanMessage(你好)]},config)# 查看当前状态snapshotapp.get_state(config)print(snapshot.values)# 查看历史状态forstateinapp.get_state_history(config):print(state.config[configurable][checkpoint_id],state.values)六、Human-in-the-Loop人工介入LangGraph 原生支持流程暂停机制满足企业级审批、审核需求。6.1 实现方式编译时通过interrupt_before/interrupt_after指定暂停点appgraph.compile(checkpointerMemorySaver(),interrupt_before[dangerous_tool]# 执行危险工具前暂停)# 第一次调用执行到危险工具前自动暂停app.invoke({input:删除数据库},config)# 人工审核后恢复执行app.invoke(None,config)# 传 None 表示从断点继续6.2 典型应用场景高风险操作审批删除数据、发送邮件、资金转账前人工确认内容审核AI 生成内容发布前人工校对任务回滚发现问题后回到历史 checkpoint 重新执行逐步调试开发阶段逐步检查每个节点的输出七、第一个 LangGraph 程序ReAct Agent以下示例展示一个完整的 ReAct Agent支持工具调用循环fromtypingimportAnnotatedfromtyping_extensionsimportTypedDictfromlangchain_openaiimportChatOpenAIfromlangchain_core.toolsimporttoolfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graphimportadd_messagesfromlanggraph.prebuiltimportToolNode,tools_conditionfromlanggraph.checkpoint.memoryimportMemorySaverfromdotenvimportload_dotenvimportos load_dotenv()# 1. 定义状态classAgentState(TypedDict):messages:Annotated[list,add_messages]# 2. 定义工具tooldefsearch(query:str)-str:搜索网络信息returnf搜索结果{query}的相关信息tools[search]tool_nodeToolNode(tools)# 3. 初始化模型并绑定工具llmChatOpenAI(modelgpt-4o-mini)llm_with_toolsllm.bind_tools(tools)# 4. 定义节点defcall_model(state:AgentState):responsellm_with_tools.invoke(state[messages])return{messages:[response]}# 5. 构建图graphStateGraph(AgentState)graph.add_node(agent,call_model)graph.add_node(tools,tool_node)graph.add_edge(START,agent)# 条件边判断是否需要调用工具graph.add_conditional_edges(agent,tools_condition,# 返回 tools 或 END{tools:tools,END:END})graph.add_edge(tools,agent)# 工具执行完回到 agent形成循环# 6. 编译带记忆appgraph.compile(checkpointerMemorySaver())# 7. 执行config{configurable:{thread_id:thread-1}}resultapp.invoke({messages:[{role:user,content:帮我搜索 LangGraph 是什么}]},config)print(result[messages][-1].content)执行流程START → agent → 判断是否调用工具 ├─ 是 → tools → agent循环 └─ 否 → END八、常见图模式8.1 顺序模式线性流水线graph.add_edge(START,node_a)graph.add_edge(node_a,node_b)graph.add_edge(node_b,node_c)graph.add_edge(node_c,END)适用场景链式思考、数据处理流水线。8.2 分支模式条件路由defrouter(state):ifstate[success]:returncontinuereturnretrygraph.add_conditional_edges(node_a,router,{continue:node_b,retry:node_c})适用场景成功 / 失败分支、错误重试。8.3 循环模式ReAct Loopgraph.add_edge(START,agent)graph.add_conditional_edges(agent,tools_condition)graph.add_edge(tools,agent)# 回到 agent 形成循环适用场景Agent 反复调用工具直到完成任务。8.4 多 Agent 协同Supervisor 模式defsupervisor(state):decisionllm.invoke(判断下一个该谁处理)return{next:decision.next_agent}graph.add_node(supervisor,supervisor)graph.add_node(researcher,researcher_graph)graph.add_node(writer,writer_graph)graph.add_conditional_edges(supervisor,lambdas:s[next],{researcher:researcher,writer:writer,DONE:END})graph.add_edge(researcher,supervisor)# 子 Agent 完成后回到 Supervisorgraph.add_edge(writer,supervisor)适用场景项目全流程规划、多维度决策分析。九、流式输出LangGraph 支持多种流式模式便于实时展示执行进度# 流式输出节点更新forchunkinapp.stream({messages:[...]},config):fornode,updateinchunk.items():print(f[{node}]{update})# 流式输出 LLM Tokenasyncforeventinapp.astream_events(input,config,versionv2):ifevent[event]on_chat_model_stream:print(event[data][chunk].content,end,flushTrue)流式模式对比stream_mode说明适用场景values每个节点执行后的完整状态调试查看全局updates仅变更的字段更高效生产环境messagesLLM Token 流实时对话debug详细执行信息深度调试十、最佳实践与避坑指南10.1 DO推荐做法只返回变更的字段节点函数只 return 需要修改的 key不要返回完整 State使用 Annotated reducer列表字段务必用Annotated[list, add_messages]避免覆盖丢失历史始终设置入口点用set_entry_point或add_edge(START, node)明确起点编译一次多次调用compile()耗资源invoke可重复执行生产用 PostgresSaverMemorySaver仅供开发生产环境必须用持久化存储10.2 DON’T常见陷阱不要原地修改 State永远返回新 dict不要state[messages].append(...)不要用可变默认值TypedDict字段不要用list []作为默认值条件边内不要做 I/O路由函数应为纯函数不要在其中调用 API 或读写文件危险工具别忘了 interrupt删除、发送等操作务必加interrupt_before不要遗忘 thread_id多用户场景必须通过thread_id隔离会话状态总结LangGraph 通过图结构 状态机模型解决了传统 Chain 无法处理的循环、分支、持久化问题是构建生产级 AI Agent 的核心运行时。掌握 State / Node / Edge 三元组配合 Checkpointer 持久化即可构建出可靠、可审计、可恢复的企业级智能体系统。