最近收到很多朋友反馈之前出的帖子不好和官方的demo结合起来看 虽然很全面但是利于喜欢通过官网快速学习的朋友快速入手感谢建议 的确对于部分开发测试者还是喜欢通过官网api来学习新技术不容易遗漏知识点那今天就以官网的graphAPI的学习路径来总结下知识点和代码demo 更方便专业的你们来多一条快速入手路径。今天可能些不完 欢迎大家收藏、关注 不迷路。 喜欢的哥哥姐姐帮三连啊下面是我从官网拷来的 目录图 地址在这 官网地址文章目录LangGraphGraphsStateGraphCompiling your graph代码实例StateSchemaMultiple schemasReducersReducer argumentsDefault reducerCustom reducersOverwriteWorking with messages in graph stateWhy use messages?Using messages in your graphSerializationMessagesStateNodesRe-execution and idempotencyUsing tasks in nodesSTART nodeEND nodeNode cachingEdgesNormal edgesConditional edgesEntry pointConditional entry pointSendCommandReturn from nodesupdate and gotographInput to invoke or streamresumeReturn from toolsGraph migrationsRuntime contextRecursion limitAccessing and handling the recursion counterHow it worksAccessing the current step counterProactive recursion handlingProactive vs reactive approachesOther available metadataVisualizationObservability and TracingLearn moreLangGraphGraphsLangGraph 运行时底层基于自研的 Pregel 运行时其核心思想借鉴了 Google Pregel计算模型用于组织和执行复杂的图计算流程。StateGraph状态图 StateGraph类是主要使用的图类。它由用户定义的状态对象进行参数化。下图展示了一个非常简单的运行图只有两个计算节点其拓扑结构为START -- node_1 -- node_2 -- END如下图所示Compiling your graph使用如下方式编译状态图以运行整个图流程graphgraph_builder.compile(...)LangGraph 的图运行过程基于 Superstep超步 来组织和推进。Superstep可以理解为图运行过程中的一次”单步循环”。一次图运行过程从开始到结束就是由一系列连续的 Superstep 串联而成。下文以node_1执行完毕后的Superstep为例说明。每个 Superstep 通常可以分为三个阶段计划/路由阶段Plan / Routing根据当前的 State状态 和 Edge边 的逻辑确定本轮超步中应该被执行的节点。执行阶段Execution运行本轮被选中的节点。如果本轮有多个节点同时被触发它们会并行执行。每个节点都会基于本轮开始时的状态快照进行计算并输出各自对状态的局部更新。在本阶段中一个节点产生的更新不会立即被其他节点读取到。状态更新/提交阶段Update / Commit当本轮所有节点都执行完成后LangGraph 会将它们的输出统一合并到 State 中生成新的状态快照。这个新状态会作为下一轮 Superstep 的输入。代码实例fromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDict,AnnotatedfromoperatorimportaddclassOverAllState(TypedDict):logs:Annotated[list[str],add]cur_id:strdefnode_1(state:OverAllState)-OverAllState:pre_idstate[cur_id]return{logs:[node_1 运行完毕],cur_id:pre_id, node_1}defnode_2(state:OverAllState)-OverAllState:pre_idstate[cur_id]return{logs:[node_2 运行完毕],cur_id:pre_id, node_2}builderStateGraph(state_schemaOverAllState)builder.add_node(node_1,node_1)builder.add_node(node_2,node_2)builder.add_edge(START,node_1)builder.add_edge(node_1,node_2)builder.add_edge(node_2,END)graphbuilder.compile()print(graph.invoke({cur_id:start}))State图的状态State管理状态的定义实际上是在声明状态的Schema后者是状态字段的完整描述。Schema官方推荐了三种定义Schema的方式TypedDict、dataclass、PydanticTypedDict#!/usr/bin/env python# -*- coding: utf-8 -*- Date : 2026/7/20 File : demo_TypeDict_state.py Author : cc Description : fromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDict,AnnotatedfromoperatorimportaddclassOverState(TypedDict):logs:Annotated[list[str],add]# 日志cur_id:str# 当前运行的节点iddefnode1(state:OverState)-OverState:pre_cur_idstate[cur_id]return{cur_id:pre_cur_id_node1,logs:[node1 进入并运行完毕]}defnode2(state:OverState)-OverState:pre_cur_idstate[cur_id]return{cur_id:pre_cur_id_node2,logs:[node2 进入并运行完毕]}defnode3(state:OverState)-OverState:pre_cur_idstate[cur_id]return{cur_id:pre_cur_id_node3,logs:[node3 进入并运行完毕]}graph_builderStateGraph(OverState)graph_builder.add_node(node2)graph_builder.add_node(node3)graph_builder.add_node(node1)graph_builder.add_edge(START,node1)graph_builder.add_edge(node1,node2)graph_builder.add_edge(node2,node3)graph_builder.add_edge(node3,END)graphgraph_builder.compile()resultgraph.invoke({cur_id:start})print(result)graph_plotgraph.get_graph().draw_png()withopen(demo_TypeDict_state.png,wb)asf:f.write(graph_plot)Pydantic#!/usr/bin/env python# -*- coding: utf-8 -*- Date : 2026/7/21 File : demo_pydantic_state.py Author : liwei68 Description : frompydanticimportBaseModel,Fieldfromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDict,AnnotatedfromoperatorimportaddclassOverState(BaseModel):logs:Annotated[list[str],add]Field(description日志)cur_id:strField(description当前运行的节点id)defnode1(state:OverState)-OverState:pre_cur_idstate.cur_idreturnOverState(cur_idpre_cur_id_node1,logs[node1 进入并运行完毕])defnode2(state:OverState)-OverState:pre_cur_idstate.cur_idreturnOverState(cur_idpre_cur_id_node2,logs[node2 进入并运行完毕])defnode3(state:OverState)-OverState:pre_cur_idstate.cur_idreturnOverState(cur_idpre_cur_id_node3,logs[node3 进入并运行完毕])graph_builderStateGraph(OverState)graph_builder.add_node(node2)graph_builder.add_node(node3)graph_builder.add_node(node1)graph_builder.add_edge(START,node1)graph_builder.add_edge(node1,node2)graph_builder.add_edge(node2,node3)graph_builder.add_edge(node3,END)graphgraph_builder.compile()resultgraph.invoke({cur_id:start})print(result)graph_plotgraph.get_graph().draw_png()withopen(demo_pydantic_state.png,wb)asf:f.write(graph_plot)resultgraph.invoke({cur_id:start})dataclassfromlanggraph.graphimportStateGraph,START,ENDfromtypingimportAnnotatedfromdataclassesimportdataclassfromoperatorimportadddataclassclassOverAllState:logs:Annotated[list[str],add]cur_id:strdefnode_1(state:OverAllState)-OverAllState:pre_idstate.cur_idreturn{logs:[node_1 运行完毕],cur_id:pre_id, node_1}defnode_2(state:OverAllState)-OverAllState:pre_idstate.cur_idreturn{logs:[node_2 运行完毕],cur_id:pre_id, node_2}builderStateGraph(state_schemaOverAllState)builder.add_node(node_1,node_1)builder.add_node(node_2,node_2)builder.add_edge(START,node_1)builder.add_edge(node_1,node_2)builder.add_edge(node_2,END)graphbuilder.compile()print(graph.invoke({cur_id:start}))Multiple schemasReducersReducer argumentsDefault reducerCustom reducersOverwriteWorking with messages in graph stateWhy use messages?Using messages in your graphSerializationMessagesStateNodesRe-execution and idempotencyUsing tasks in nodesSTART nodeEND nodeNode cachingEdgesNormal edgesConditional edgesEntry pointConditional entry pointSendCommandReturn from nodesupdate and gotographInput to invoke or streamresumeReturn from toolsGraph migrationsRuntime contextRecursion limitAccessing and handling the recursion counterHow it worksAccessing the current step counterProactive recursion handlingProactive vs reactive approachesOther available metadataVisualizationObservability and TracingLearn more