免费获取学习方案
ARTICLE DETAIL

资讯详情

深耕编程基础知识与建站技术分享的一线实战洞察。

LangGraph-AI智能体开发框架 - LangGraph 入门案例1 : 智能快递配送系统

LangGraph-AI智能体开发框架 - LangGraph 入门案例1 : 智能快递配送系统 目录一、智能快递配送系统Graph API 编码思路智能快递配送系统图讲解智能快递配送流程:二、编码编码流程三、代码编写步骤1定义State设置快递的包裹信息步骤2定义节点 Nodes创建配送站点步骤3定义 StateGraph图成立快递公司步骤4添加节点 Nodes建设配送站点步骤5添加边 Edges规划运输路线步骤6StateGraph图编译从公司创建到运行步骤6测试运行结果:四、相关问题一、智能快递配送系统Graph API 编码思路构建 Graph 图首先需要【定义状态】然后【定义并添加节点和边】最后【编译】它。编译提供了对图形结构的一些基本检查没有孤立节点等。LangGraph 所谓的 “编译” 与传统意义上的语言编译完全不同LangGraph 编译本质是在运行时动态构建和验证一个复杂的图而非翻译代码。C 的编译是 “完整编译” 或 “静态编译” 的典范。它追求在程序运行之前就将所有代码 “解决” 完毕生成一个独立、高效、可直接被操作系统调用的 “成品”。它比 Java 的编译更彻底 (直接到机器码而非中间码)也更底层 (紧密绑定操作系统和 CPU 架构)。“编译” 对比表格智能快递配送系统图讲解下面用这张图的快递公司的业务类比 LangGraph 构建 Graph 图的完整流程:整体流程分为三大步骤定义状态 → 定义并添加节点和边 → 编译。定义状态相当于定义快递包裹的数据结构包裹 id、始发地、目的地、流转记录这些信息就是 Graph 的 State所有站点都共用这份包裹信息。定义并添加节点和边节点就是一个个快递站点揽收站、分拣中心、配送站每一个站点对应 LangGraph 里面的一个 node 节点函数每个站点只负责自己那一部分业务逻辑。边就是站点之间的运输路线规定包裹跑完 A 站点之后接下来去往哪一个站点。把揽收站、分拣中心、配送站通过运输路线边串联起来。图里的孤立节点只新建了一个揽收站点但是没有给它配置任何进出的运输路线没有边把它接入整个业务网络。编译 (compile)类比 “注册快递公司”。 编译并不是翻译代码而是做两件核心工作把已经写好的节点、边组装起来生成一个可以直接调用运行的工作流对象。做图结构校验检测是否存在孤立节点。像图中这个只创建、没有连线的揽收站编译的时候就会检测出来并报错防止出现有节点但是流程无法走到的问题。智能快递配送流程:这张图是智能快递配送完整的流程图。整个流程从 START 起点开始首先完成包裹信息的初始化也就是给 state 赋值包裹 id、始发地、目的地、优先级等全部数据。流程第一步进入揽收站节点完成包裹揽收操作更新包裹的状态与流转记录。揽收完成之后流转到主分拣中心这里作为默认中转节点。主分拣中心读取包裹 state 里面的目的地字段做条件判断执行路由分发。根据目的地归属把包裹分流到分支分拣中心 A东部、分支分拣中心 B西部、分支分拣中心 C南部其中的某一个分支。每一个分支分拣中心拿到包裹之后会读取包裹的 priority 优先级。如果包裹标记为加急就选择飞机运输普通包裹则走陆运运输。不同分支的飞机、陆运两条运输路径最终全部汇聚流向派送站节点。到达派送站之后执行最终配送业务逻辑更新包裹状态为已签收记录流转日志。全部业务处理完毕流转到 END整个快递配送流程正式结束。这两张图体现了 LangGraph 条件边的能力同样一个节点执行完毕之后可以根据 state 内部的业务数据动态选择下一跳去往哪一个节点不再是固定一条直线流程。多个不同分支路径最后又可以汇聚到同一个节点继续往下执行。二、编码编码流程整套 LangGraph 编码一共分为 5 个步骤和上面的流程图一一对应。第一步定义状态。我们先定义好 PackageState用来存放包裹全部业务数据整张图所有节点共享这一份状态数据包裹 id、目的地、priority 优先级、流转历史都保存在这里。后续所有节点函数读取、修改的数据都来自这个状态。第二步定义节点一共 5 个业务函数。这里要注意流程图里的 START、END 是 LangGraph 内置的起点、结束标记不需要我们自己编写函数框架内部已经实现。我们需要手写函数的业务节点分别是揽收站、分拣中心、加急配送、标准配送、派送站一共 5 个每一个节点就对应一个 Python 函数。节点和函数的关系一个业务节点绑定一个函数。函数是这个节点真正要执行的业务逻辑。当流程流转到这个节点的时候就会自动调用绑定的函数传入当前 state函数执行完成返回状态更新字典LangGraph 自动合并更新全局状态。第三步定义图添加节点和边到图中。实例化 StateGraph 图构建器把上面写好的 5 个函数通过 add_node 注册到图给每个函数起节点名字。再通过边配置流转逻辑START 指向揽收站揽收站固定流向分拣中心分拣中心这里配置条件边读取 state 里面 priority 字段如果是加急就走到加急配送节点如果是普通就走到标准配送节点加急配送、标准配送两个分支最后都汇聚到派送站派送站再指向 END。这一步就是把流程图上的连线全部用代码描述出来。第四步编译图。调用 compile() 方法完成图的组装与结构校验检查有没有孤立节点、连线是否合法输出一个可以调用执行的图对象。编译不会运行业务逻辑只是把图结构准备就绪。第五步执行图。调用 graph.invoke()传入初始化的包裹 state图就会从 START 开始按照我们配置好的节点、边自动依次调用各个节点绑定的函数走完整个配送流程直到走到 END流程终止。重点总结 START 和 END 只是流程标记代表流程的入口和出口不是业务节点没有业务逻辑不需要开发者定义函数。所有真正要干活的业务步骤都必须自己写函数再通过add_node把函数注册为图里面的节点。三、代码编写步骤1定义State设置快递的包裹信息定义图时要做的第一件事是定义图的状态。状态将是图中所有节点和边的输入可以是 TypedDict 或 Pydantic 模型Pydantic 的性能不如 TypedDict。如下所示首先看导入部分。第一行 import operator导入 Python 内置 operator 模块这里主要使用operator.add 用来告诉 LangGraph 使用 operator.add 的字段做追加合并而不是直接覆盖。第二行 from typing import TypedDict, AnnotatedTypedDict 用来定义带类型注解的字典结构也就是我们的状态Annotated 是给字段附加额外元信息在这里附加合并规则。后面两行导入 LangGraph 组件START、END 是图内置的起点、结束标记StateGraph 是构建状态图的核心类。接下来定义 PackageState(TypedDict)这个类就是整张图的状态模板图运行全程所有节点函数拿到的 state 都遵循这个结构。所有节点都可以读取 state 里面的字段节点返回的更新字典会和全局 state 做合并。package_id、origin、destination 这三个普通字符串字段对应包裹编号、始发站点、目的站点。这类普通字段节点返回更新字典时会直接覆盖原值。比如节点返回 {package_id:PKG002}就直接把原来的 package_id 替换掉。status 字段是字符串用来记录包裹当前配送状态可选值为待揽收、已揽收、运输中、派送中、已签收。同样属于覆盖更新节点返回新 status直接替换旧的 status。然后是 history 字段这里做了关键处理。如果直接写 history: list[str]节点返回新列表的时候 LangGraph 会直接把旧的 history 整个覆盖掉旧流转记录直接丢失。而使用 Annotated[list[str], operator.add]就是给这个字段指定合并策略当多个节点返回新的列表片段执行列表相加也就是追加。比如节点返回 {history:[在西安揽收]}不会覆盖旧列表而是把新字符串追加到原有列表后面完整保存全部流转日志。total_distance: Annotated[int, operator.add] 是整型字段同样附加 operator.add。它的合并行为是数值相加。节点返回 {total_distance:300}不会直接把总里程设置成 300而是在原来 total_distance 数值基础上做加法累加里程。最后 priority 字符串字段标记包裹是普通或者加急用于后续分拣中心做条件分支判断同样是覆盖更新。这里重点区分两种更新行为。普通字段默认是覆盖更新新值直接替换旧值。被 Annotated 加上 operator.add 修饰的字段会执行追加、累加合并适合日志列表、累计数值这类场景不需要我们在节点函数内部手动去拼接、累加LangGraph 底层自动完成合并。步骤2定义节点 Nodes创建配送站点接着我们可以定义各个配送站点 (节点)。在 LangGraph 中节点就是一个 Python函数 (同步或异步)。注意节点接收状态作为参数。节点不需要返回整个状态模式只需一个更新。我们一共有 5 个节点也就是 5 个函数如下所示首先看节点函数通用规则。LangGraph 里面每一个业务节点本质就是普通 Python 函数。函数的入参是 state: PackageState代表当前整张图最新的状态对象我们可以从 state 里面读取包裹的全部信息。函数不需要返回完整的全部 state只需要返回发生变更的字段字典。LangGraph 会自动把返回的更新字典和全局 state 做合并。第一个函数 receive_package对应流程图里的揽收站节点。函数内部第一处是打印语句 print(---执行到揽收站节点)这是调试用的输出可以直观看到当前走到了哪一个节点。origin state[origin]从传入的 state 字典取出始发站点拿到包裹从哪里发出这个业务数据。然后 return 返回更新字典。status: 已揽收status 是普通覆盖更新字段会直接把全局 state 的 status 修改为 “已揽收”。history: [f在{origin}揽收]history 字段之前定义过Annotated[list[str], operator.add]不会直接覆盖旧列表而是把这个新的字符串元素追加到原来 history 列表末尾实现流转日志累加。第二个函数 sort_package对应流程图的分拣中心节点。同样先打印日志方便调试观察节点执行。 destination state[destination]从 state 读取包裹的目的地。接下来做业务判断判断目的地字符串里面包含什么城市关键词。如果目的地包含北京就赋值next 北京分拣中心 如果包含上海next 上海分拣中心 其余所有情况走到 else 分支next 其他地区分拣中心。这里只是在函数内部算出了下一个分拣中心的名字注意这个函数本身不能直接控制流程跳转。函数只负责修改 state 数据真正决定下一个走哪个节点是后面代码里的条件边逻辑。return 返回更新字典。 status: 已分拣把全局状态的配送状态更新为已分拣。history: [f分拣至{next}]把本次分拣去向记录为一条新日志依靠 operator.add 自动追加到 history 列表保存完整流转记录。补充一个关键点这两个节点函数只负责读取 state、产出状态变更。节点函数不会写任何跳转代码流程走向由 add_edge、add_conditional_edges 在外部定义业务逻辑和流程控制相互分离开。这是剩下的 3 个业务节点函数同样遵循 LangGraph 节点函数统一规则接收当前 state 作为入参只返回需要更新的字段字典框架自动合并更新全局状态。第一个 final_delivery对应流程图的派送站节点代表包裹最后一步派送签收。函数从 state 读取 destination 拿到包裹目的地。返回字典中 status 被覆盖更新为 “已签收”代表整个配送业务完成。history 传入一条新日志 [已送达{state[destination]}]依靠 operator.add 将这条记录追加到历史列表末尾记录完成派送这件事。这个节点执行完之后流程就会走向 END 结束。第二个 standard_delivery 标准配送节点对应普通优先级包裹走陆运的业务。status 覆盖更新为 “运输中”标记包裹正在运输。history增加一条 [标准陆运]记录本次是陆运运输。 total_distance 返回数值 500。这个字段定义的时候标注了 Annotated[int, operator.add]不会直接把总里程改成 500而是在原有 total_distance 数值基础上加 500。第三个 express_delivery 加急配送节点对应加急包裹走空运的业务。status覆盖更新为 “加急运输”区分普通运输状态。history追加 [空运加急] 日志标记本次是空运加急。total_distance 返回数值 800同样触发 operator.add在原有里程基础上累加 800。这里重点区分两个里程数值的含义。500 和 800 不是设置最终总里程而是本次运输环节新增的里程LangGraph 自动做加法。同时要再次强调这两个配送函数本身不会自己做分支判断。到底执行标准配送还是加急配送不由函数内部 if 判断决定而是后面代码写条件边读取 state 里面 priority 字段动态选择调用哪一个函数。函数只负责业务逻辑流程路由交给边来控制实现业务逻辑和流程控制解耦。至此 5 个业务节点函数全部讲完receive_package 揽收站、sort_package 分拣中心、standard_delivery 标准配送、express_delivery 加急配送、final_delivery 派送站。START 与 END是框架内置标记不需要编写函数。步骤3定义 StateGraph图成立快递公司StateGraph 是一个有状态图计算框架它基于有向图DirectedGraph模型构建专门设计用于处理多步骤、有状态的工作流程。StateGraph 用来将复杂的工作流程可视化、模块化让开发者能够像设计快递配送网络一样设计软件系统。通过这种思维方式即使是复杂的多步骤AI应用也变得清晰可控。我们需要使用 langgraph.graph.state.StateGraph 来定义。StateGraph 仅是一个构建器类可以使用 State 来构建。如下所示这是编码流程的第三步实例化 StateGraph 对象类比于创建一家快递公司。StateGraph 是 LangGraph 提供的状态图核心类实例化的时候必须传入我们前面定义好的状态类PackageState。传入 PackageState 代表这整张图运行的时候全程都会使用这一套状态结构所有节点读取、修改的 state都必须遵守 PackageState 里面定义的字段、类型、合并规则。变量名 delivery 就是我们这个图实例后续所有注册节点、添加边的操作都要操作这个 delivery 对象。这里泛型部分 [Any, None, Any, Any] 是类型提示属于 IDE 静态类型注解不影响程序运行。执行完这一行代码只是完成了图对象的初始化。此时图里面是空的还没有任何节点也没有任何连线。就好比快递公司只是注册完成还没有把揽收站、分拣中心这些站点入驻进来下一步就要调用 add_node 把 5 个业务函数注册成为图里面的节点。步骤4添加节点 Nodes建设配送站点接下来我们需要将各节点组织进图中即添加节点到 StateGraph 中。我们使用 add_node() 将新节点添加到 StateGraph。这一步是向已经实例化完成的 delivery 图对象中注册业务节点类比给快递公司入驻各个业务站点。add_node() 是 StateGraph 提供的方法它接收两个参数。第一个参数是节点名字是字符串这是图流程内部识别节点的唯一标识后面配置边、配置跳转逻辑都要用这个字符串名字。第二个参数传入我们之前写好的业务函数对象我们只需要把函数本身交给图图运行流转到这个节点的时候会自动调用该函数。第一行 delivery.add_node(揽收站, receive_package) 把字符串名字 “揽收站” 和函数 receive_package 绑定。当流程走到 “揽收站” 这个节点框架自动调用 receive_package(state)传入当前全局状态。 剩下四行逻辑完全一致分别注册分拣中心、派送站、标准配送、加急配送正好对应我们 5 个业务函数。执行完这 5 行 add_node 之后图里面已经拥有全部业务节点但是节点和节点之间没有任何连线不知道执行顺序。就好比所有站点都建好但公路还没有修包裹不知道该从哪个站点去往哪个站点。下一步就要添加边 add_edge 以及条件边定义节点之间流转的路线。步骤5添加边 Edges规划运输路线各节点站点准备好后则需要为快递运输规划路线。实际上这就是为图定义边。边有几种关键类型普通边 / 固定边Normal Edges直接从一个节点转到下一个节点。条件边Conditional Edges调用函数来确定下一步要转到哪个节点。例如设置最简单运输路线快递由揽收站接收下一站固定为分拣中心最后到派送中进行派送。这就是固定边。如下图:再例如我们可以根据以下条件判断快递如何运输包裹是加急件→ 走空运线路包裹不是加急件→ 走标准线路 这就是条件边。这就是条件边。要说明的是在 LangGraph 中START 节点是一个特殊节点表示将用户输入发送到图形的节点。引用此节点的主要目的是确定应该首先调用哪些节点。END 节点是一个表示终端节点的特殊节点。当想要指示哪些边在完成后没有后续动作时将引用此节点。条件入口点Conditional Entry Point调用一个函数来确定在用户输入到达时首先调用哪个节点。下面我们来看 LangGraph 提供的添加固定边和条件边的两个函数:1. 使用add_edge()向图中添加从开始节点或起始节点列表到结束节点的固定边。add_edge() 方法常用参数说明普通固定边 add_edge就是写死的单向公路A 节点执行完毕一定走到 B 节点没有别的选择。2. 使用add_conditional_edges()向图中添加从起始节点到任意数量的目标节点的条件边。add_conditional_edges() 方法常用参数说明条件边 add_conditional_edges相当于岔路口节点执行完之后会执行 path 传入的判断函数读取当前 state 状态动态决定下一站去哪一个节点。现在调用 add_edge、add_conditional_edges 来修建公路定义节点之间流转的运输路线。delivery.add_edge(START, 揽收站)这是一条固定边。START 是 LangGraph 内置的特殊入口节点代表整个图的起点这条语句含义图一启动第一个执行的业务节点就是揽收站。delivery.add_edge(揽收站, 分拣中心)普通固定边。揽收站业务逻辑执行完毕之后没有任何分支判断一定会流转到分拣中心节点。接下来定义select_delivery 路由函数专门用来做分支判断。这个函数接收全局状态 state只能读取 state 里面的数据不允许修改 state只负责做路由判断。读取 state 当中的 priority 优先级字段如果是加急返回标记字符串备注加急否则返回标记字符串无备注。注意返回的这两个字符串并不是节点名称只是中间标记需要交给 path_map 做映射。delivery.add_conditional_edges() 用来添加条件边也就是分岔路口。source分拣中心代表当分拣中心节点全部执行完成之后就触发这个条件逻辑执行 select_delivery 路由函数。第二个参数传入 select_delivery 函数对象函数后面不能写括号交给框架在合适时机自动调用。path_map 字典完成标记字符串向真实节点名称的映射 路由函数返回备注加急就跳转到加急配送节点 路由函数返回无备注就跳转到标准配送节点。delivery.add_edge(加急配送, 派送站)固定边加急配送节点执行完成流转到派送站。delivery.add_edge(标准配送, 派送站)固定边标准配送节点执行完成同样流转到派送站。两条不同业务分支在这里汇合统一流入派送站节点。delivery.add_edge(派送站, END)派送站执行结束流转到内置END节点。END是图的终止标记走到这里整个工作流直接结束返回最终的 state 状态。步骤6StateGraph图编译从公司创建到运行在步骤2中我们仅是构建出 StateGraph还无法直接用于执行。LangGraph要求必须先编译图然后才能使用它。编译提供了对图结构的一些基本检查这会验证从START到所有节点的可达性从所有节点到END的可达性没有孤立节点或死循环使用 compile() 方法即可编译图。该方法将 StateGraph 编译为 CompiledStateGraph 对象。编译后的图实现了 Runnable 接口可以异步调用、流式传输、批处理和运行。调用 .compile() 就是对这份图纸做编译做校验、组装生成一个可以真正执行的可运行图实例赋值给变量 delivery_system。compile() 编译方法内部会做语法校验检查引用的节点是否全部存在、边的起点终点是否合法、有没有出现循环错误、START 和 END 链路是否连通。如果配置写错编译阶段就直接抛出异常不会等到运行时报错。把节点、边、条件路由函数全部组装成内部可执行的数据结构。delivery_system 就是编译完成之后得到的可运行图实例后续真正调用执行工作流就使用这个变量。步骤6测试前面我们已经完成图的编译得到可运行对象 delivery_system。编译后的图可以反复多次运行每次运行只需要传入一份初始状态就会完整走完整套业务流程。test_packages 是测试用的包裹列表里面准备了两份包裹的初始状态字典。每一个字典就是传给 LangGraph 的初始 PackageState。P001 是普通包裹P002 是加急包裹。字段全部初始化history 为空total_distance 初始值等于 0。工作流启动之后各个节点内部会读取并且修改这份状态。接下来写 for 循环遍历 test_packages 列表依次处理每一个包裹。不需要再次调用 compile编译好的图实例支持重复执行。print(f\n配送包裹: {package[package_id]}) 做控制台打印输出用来区分两次不同包裹的运行日志方便我们看控制台输出分辨哪一段输出对应哪一个包裹。result delivery_system.invoke(package) 是整段代码最核心的一行。invoke 是同步执行方法把包裹初始状态传进去。程序就会从 START 起点顺着我们定义好的节点、普通边、条件边一步步流转依次执行揽收站、分拣中心等各个业务节点。节点内部会修改 state 当中的各个字段。当流转走到 END 节点之后整个图停止运行把更新完成的完整状态返回赋值给 result。后面三行 print 就是打印图执行完成之后返回的最终状态。 result[status] 拿到包裹最终业务状态由各个业务节点负责更新。result[history] 拿到配送历史数组每经过一个站点节点就会往数组追加记录可以看到包裹完整走过哪些业务节点。 result[total_distance] 打印全程累计的配送里程。运行结果:程序运行后会先后跑两次图。处理 P001 普通包裹会走普通配送分支链路是START → 揽收站 → 分拣中心 → 标准配送 → 派送站 → END控制台打印出这个包裹的最终状态、站点历史、总里程。处理 P002 加急包裹会走加急配送分支链路是START → 揽收站 → 分拣中心 → 加急配送 → 派送站 → END控制台打印加急包裹对应的一套输出。到此我们已经构建出了一个图式的智能快递配送系统来理解 LangGraph 图的基本能力与用法核心概念回顾State包裹信息卡(记录所有状态Nodes配送站点(执行具体操作Edges运输路线控制流转顺序Reducers信息更新规则如何记录变更四、相关问题Q1这个快递案例没有接入 ChatGPT、DeepSeek 这类大模型为什么 LangGraph 图还能完整跑通整个快递派送流程LangGraph 本质是通用状态工作流编排框架它不是大模型的专属附属工具。它的核心职责是管理状态、定义节点流转逻辑、处理分支条件、控制执行流程。节点里面的业务逻辑可以是任意 Python 代码不一定非得调用大模型。这个案例里每个节点函数写的都是普通 Python 逻辑做状态字段修改、追加历史记录、累加里程不需要 LLM 参与整张图照样可以正常执行。Q2那 LangGraph 本来不都是要和大模型结合使用吗这是大家很常见的误解。LangGraph 有两大使用场景。第一种也是大家见得最多的场景AI Agent 场景节点内部调用大模型让 LLM 做思考、做工具选择、做决策用来实现智能代理这是 LangGraph 出圈的用法。第二种通用业务工作流场景节点只是普通业务函数不调用大模型就像我们这个快递案例用来实现带状态、带分支的业务流水线。大模型只是节点里面可以选用的其中一种业务组件不是 LangGraph 运行的强制必需品。Q3那这个不带大模型的案例学它意义是什么这个案例是用来吃透 LangGraph 底层基础能力。我们在这里学会了 State 状态定义、add_node 注册节点、普通边、条件分支边、compile 编译、invoke 执行这些底层核心机制。搞懂这套基础之后后续只需要把其中某一个节点内部的普通 Python 函数替换成调用 DeepSeek/ChatGPT 的代码就可以直接变成 AI Agent。如果直接一上来就学带大模型的 Agent会把「图框架本身的流程逻辑」和「大模型的输出」混在一起出问题排错的时候分不清 bug 是来自大模型还是图流转逻辑写错。先用纯 Python 业务跑通案例可以把 LangGraph 自身的机制先学明白。Q4什么时候才需要把大模型引入 LangGraph当业务流程里面需要模型自主做决策、理解自然语言、选择工具、不确定下一步该走哪个分支的时候才需要在节点内部调用 LLM。像快递这个案例分拣判断加急/普通判断规则写死在路由函数里规则固定不需要 AI 理解如果改造一下把 priority 优先级不再手动传入交给大模型读取包裹备注文本让 LLM 自己判断该走加急还是标准配送这个时候就需要在节点内部引入大模型。
返回列表