* LangGraph 基础入门：定位、核心概念与第一个可运行图

一、LangGraph总览LangGraph 运行时底层基于自研的 Pregel 运行时其核心思想借鉴了 Google Pregel 计算模型用于组织和执行复杂的图计算流程。1.0.LangGraph和LangChain的版本迭代及定位LangChain 1.x的定位发生了明显变化旧式 Chain、Retriever、Memory 等组件被迁移至 langchain-classicLCEL 保留为 langchain-core 的底层组合机制但不再是主要开发入口。LangChain 1.x现在聚焦于 Agent 开发核心入口是 create_agent围绕 Agent 提供模型、工具、消息、结构化输出、Middleware 和记忆管理等高层抽象。LangGraph是更底层的编排框架与 Agent Runtime负责复杂工作流和有状态 Agent 的执行提供持久化、流式输出、Durable Execution、Human-in-the-loop 等运行时能力。create_agent 底层基于 LangGraph 实现。二者的定位对比如下对比维度LangChainLangGraph定位Agent 高层开发框架底层编排框架 Agent Runtime核心入口create_agentStateGraph/entrypoint适用场景结构直接的 Agent 应用复杂工作流、持久化状态、长时间运行、人工介入流程控制Agent 循环自动管理精细控制节点、边、条件分支学习成本较低较高LangChain提供易于使用的 Agent 高层抽象LangGraph提供可靠、可持久化且可精细控制的底层执行能力。对于大多数 Agent 项目从 LangChain 的 create_agent 开始即可需要复杂工作流编排、确定性步骤与 Agent 步骤混合、长时间运行或底层状态控制时再引入 LangGraph。1.1. 构成图的基本要素LangGraph 运行时主要由三个基本要素构成State状态、Node节点 和 Edge边State状态LangGraph 运行过程中的共享数据结构用于表示应用在某一时刻的状态快照。它承载了图运行所需的上下文信息、中间结果和后续节点需要读取的数据是节点之间传递信息的核心载体。它与我们在学习 LangChain Agent 时使用的 State 是同一概念。Node节点LangGraph 中的具体执行单元通常实现为一个函数。节点会读取当前 State执行相应的业务逻辑并返回对 State 的局部更新。节点本身并不直接修改全局状态状态的合并与提交由运行时统一完成。Edges边用于定义节点之间的流转关系决定一个节点执行完成后下一步应该进入哪个节点。Edge 可以是固定流转也可以根据当前 State 进行条件判断从而实现分支、循环等复杂控制流程。下图展示了一个非常简单的运行图只有两个计算节点其拓扑结构为START -- node_1 -- node_2 -- END如下图所示1.2. 图运行过程LangGraph 的图运行过程基于 Superstep超步 来组织和推进。Superstep 可以理解为图运行过程中的一次”单步循环”。一次图运行过程从开始到结束就是由一系列连续的 Superstep 串联而成。下文以node_1执行完毕后的Superstep为例说明。每个 Superstep 通常可以分为三个阶段计划/路由阶段Plan / Routing根据当前的 State状态 和 Edge边 的逻辑确定本轮超步中应该被执行的节点。执行阶段Execution运行本轮被选中的节点。如果本轮有多个节点同时被触发它们会并行执行。每个节点都会基于本轮开始时的状态快照进行计算并输出各自对状态的局部更新。在本阶段中一个节点产生的更新不会立即被其他节点读取到。状态更新/提交阶段Update / Commit当本轮所有节点都执行完成后LangGraph 会将它们的输出统一合并到 State 中生成新的状态快照。这个新状态会作为下一轮 Superstep 的输入。完整的运行流程查看 index.html 双击在浏览器打开即可。其中和Checkpoint相关的部分学习相关内容后再看。1.3. Graph API vs Functional APILangGraph 提供了两种不同的 API 来构建运行图Graph API图式 API 和Functional API函数式 API。这两种 API 共享相同的底层运行时可以在同一应用程序中协同使用但它们针对不同的使用场景和开发偏好而设计。1.3.1. Graph APIGraph API 采用声明式方式构建工作流。开发者需要显式定义 State、Node 和 Edge将业务流程组织成一个可视化的图结构。当流程中存在较复杂的分支、多个节点之间共享状态、并行执行、结果汇聚或者需要通过图结构帮助调试和团队协作时更适合使用 Graph API。官方文档也明确建议在需要复杂流程可视化、显式状态管理、多条件分支、并行路径以及团队协作时优先选择 Graph API。总之Graph API 更适合构建结构清晰、节点关系复杂、需要长期维护的工作流。典型场景包括场景说明多节点复杂流程流程中存在多个处理节点需要清晰表达节点之间的关系条件分支较多根据 State 中的不同字段决定后续执行路径并行执行与结果汇聚多个节点并行运行之后汇总结果多组件共享状态多个节点都需要读写同一个全局 State需要图结构展示便于调试、讲解、文档化和团队协作1.3.2. Functional APIFunctional API 采用命令式方式构建工作流更接近普通 Python 函数调用。开发者可以使用entrypoint定义工作流入口使用task定义可被检查点记录的任务然后在函数内部使用普通的if/else、循环和函数调用来组织流程。官方文档指出当已有过程式代码需要最小改造、流程主要是线性的、分支逻辑较简单、希望快速原型验证时更适合使用 Functional API。可以这样理解Functional API 更适合在普通 Python 函数流程中以较低成本接入 LangGraph 的持久化、中断恢复和任务记录能力。典型场景包括场景说明现有代码改造原本已有函数式或过程式代码不希望重构成完整图结构线性流程主要是 A → B → C 的顺序执行简单分支只有少量if/else判断快速原型验证希望减少样板代码快速验证业务逻辑局部任务持久化希望某些函数作为独立 task 被检查点记录1.3.3. 二者的核心区别对比项Graph APIFunctional API编程风格声明式图结构命令式函数流程核心抽象State、Node、Edgeentrypoint、task状态管理显式定义全局 State更多依赖函数参数和返回值流程表达通过节点和边表达通过普通 Python 控制流表达可视化能力强天然适合画图和调试弱更像普通代码流程适合场景复杂工作流、多分支、多节点协作简单流程、快速原型、已有代码改造学习成本相对更高相对更低1.3.4. 选型建议从零构建或流程结构复杂用Graph API现有代码改造、快速原型验证或流程逻辑简单用Functional API。学习LangGraph建议优先掌握Graph API。因为后者更能体现LangGraph的核心思想通过State、Node、Edge显式描述一个可执行的计算图。本文介绍Graph API对Functional API感兴趣的同学自行查阅https://docs.langchain.com/oss/python/langgraph/functional-api二、图的基础构建与运行本节将从一个基础案例入手掌握图的基本创建、调用和可视化方法。图的构建与运行分为三个阶段定义状态图、编译状态图、调用状态图2.1. 定义状态图状态图的定义可以分为以下几个步骤1. 定义全局状态定义整个运行图共享的全局状态实际上除了全局状态LangGraph还支持其它类型的状态下文详述2. 创建状态图创建 StateGraph实例将状态和状态图绑定3. 定义图结构定义节点和边并添加到状态图中2.1.1. 定义全局状态全局状态是 LangGraph 运行图每个节点都可以访问的公共对象。classOverAllState(TypedDict):logs:Annotated[list[str],add]cur_id:str2.1.2. 创建状态图builderStateGraph(state_schemaOverAllState)2.1.3. 定义图结构2.1.3.1. 定义图节点defnode_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}2.1.3.2. 添加节点builder.add_node(node_1,node_1)builder.add_node(node_2,node_2)2.1.3.3. 添加边builder.add_edge(START,node_1)builder.add_edge(node_1,node_2)builder.add_edge(node_2,END)2.2. 编译状态图将上一步得到的状态图编译为可执行的运行图graphbuilder.compile()2.3. 调用状态图可以通过invoke调用计算图print(graph.invoke({cur_id:start}))2.4. 完整代码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}))运行结果如下{logs: [node_1 运行完毕, node_2 运行完毕], cur_id: start, node_1, node_2}2.5. 图结构可视化2.5.1. 绘制mermaid并获取源码mermaid是一种生成图表和流程图的文本标记语言它可以把简单的文本描述直接渲染成可视化图形。通过以下代码可以用mermaid语法表达图结构并打印mermaid源码raw_mermaidgraph.get_graph().draw_mermaid()print(raw_mermaid)输出如下--- config: flowchart: curve: linear --- graph TD; __start__([p__start__/p]):::first node_1(node_1) node_2(node_2) __end__([p__end__/p]):::last __start__ -- node_1; node_1 -- node_2; node_2 -- __end__; classDef default fill:#f2f0ff,line-height:1.2 classDef first fill-opacity:0 classDef last fill:#bfb6fcmermaidmermain输出文本从graph TD;开始的部分为mermaid源码。__start__node_1node_2__end__2.5.2. 绘制mermaid并转换为pngLanggraph也支持将mermaid转换为图片字节流png_bytesgraph.get_graph().draw_mermaid_png()draw_mermaid_png()底层会先调用draw_mermaid()得到mermaid语法的图表代码然后调用在线服务渲染为mermaid图片。默认的mermaid在线渲染服务链接为https://mermaid.ink由于网络问题可能存在渲染失败的情况通常重试即可成功。可以通过draw_mermaid_png()的base_url参数替换可用的mermaid在线渲染服务2.5.2.1. jupyter环境下直接展示图片在jupyter环境下可以通过IPython提供的功能将图片字节流直接展示为图片。fromIPython.displayimportdisplay,Image png_bytesgraph.get_graph().draw_mermaid_png()pngImage(png_bytes)display(png)运行结果如下所示2.5.2.2. 保存为图片文件也可以将图片字节流转换为PNG文件落盘png_bytesgraph.get_graph().draw_mermaid_png()png_filenamefirst_demo_graph.pngwithopen(png_filename,wb)asf:f.write(png_bytes)运行结束后代码文件同级目录下将会出现生成好的图片文件如下图所示。如果看不到文件可以刷新上级目录图片内容如下2.5.2.3. 快捷用法在Jupyter环境下可以直接使用display(graph)快速展示编译后的图结构fromIPython.displayimportdisplay display(graph)效果如下