
我挖掘了一个巨牛的 人工智能 学习网站通俗易懂风趣幽默忍不住分享一下给大家。点击跳转到网站1 Graph API 编码思路构建 Graph 图存在一套固定标准开发流程整体分为三大核心步骤定义全局 State 状态自定义 Node 节点添加节点与 Edges 流转边调用compile()方法编译图。编码流程状态定义图添加节点和边到图中定义节点5个函数(开始结束节点不用定义函数)编译图执行图在编译阶段框架会自动完成图结构合法性校验校验规则包含从入口START能够抵达所有业务节点所有业务节点最终都可以走到终止节点END不存在孤立无链路的节点、无无限死循环。重点区分LangGraph 编译 VS Java / C 传统编译LangGraph 语境中的「编译」和传统编程语言的编译逻辑完全不同。 LangGraph 的编译行为发生在程序运行初始化阶段本质是动态组装图结构、校验链路完整性不存在代码翻译、机器码生成行为 Java、C 的编译全部在开发构建阶段完成运行前就将高级代码翻译为字节码 / 机器码生成可独立运行的程序文件。特性LangGraph 的编译Java 的编译C 的编译核心本质图配置组装 链路结构校验高级语言翻译为 JVM 字节码高级语言翻译为 CPU 本地机器码执行阶段应用程序运行初始化阶段开发构建阶段程序运行前开发构建阶段程序运行前主要目的1. 生成可执行工作流对象 CompiledStateGraph2. 校验图链路完整、无孤立节点、无死循环1. 语法、语义静态检查2. Java 源码转为跨平台字节码1. 全量语法语义检查、代码性能优化2. 翻译为适配操作系统、CPU 架构的机器码3. 链接全部依赖模块生成独立可执行文件2 代码实现快递配送完整 LangGraph 实战2.1 步骤 1定义 State设置快递包裹全局信息搭建图应用的第一步是定义全局 StateState 作为整张图所有节点、边共享的数据载体。 LangGraph 支持两种 State 定义方式TypedDict、Pydantic 模型工程中优先选用TypedDict轻量化、运行性能更优。from typing import TypedDict, Annotated from operator import add # 1. 定义包裹全局状态模板 class PackageState(TypedDict): # 包裹基础固定信息 package_id: str # 包裹唯一编号 origin: str # 包裹始发站点 destination: str # 包裹目标城市 # 配送流转状态信息 status: str # 配送总状态待揽收/已揽收/运输中/派送中/已签收 history: Annotated[list[str], add] # 全流程流转记录自动追加不覆盖 total_distance: Annotated[int, add] # 累计运输里程数值自动累加 # 配送配置信息 priority: str # 配送等级普通 / 加急2.1.1 核心知识点State 更新机制 Reducers覆盖更新 VS 追加更新State 字段的更新逻辑由Annotated第二个参数 reducer 控制分为三种更新规则普通字段status: str默认覆盖更新节点返回新值直接替换原有数据list[str] operator.add列表追加更新新流转记录拼接在原有列表后方int operator.add数值累加更新多次运输里程自动叠加求和。# 覆盖更新每次返回直接替换旧值 status: str # 追加更新新增流转记录拼接到历史列表 history: Annotated[list[str], add] # 累加更新新里程与原有总里程相加 total_distance: Annotated[int, add]2.2 步骤 2定义 StateGraph 图构建器初始化工作流StateGraph是 LangGraph 内置的有状态图构造类基于有向图模型设计专门用来搭建多步骤、带持久状态的 AI 工作流。 通过该构建器我们可以像设计快递运输网络一样模块化拆分业务逻辑复杂多步骤 AI 应用也能清晰可控。初始化时必须传入提前定义好的PackageState绑定整张图统一的数据结构from langgraph.graph import StateGraph, START, END # 2. 初始化图构造器绑定包裹状态结构 delivery StateGraph(PackageState)注意当前仅完成图结构模板定义未编译的图无法执行任何业务逻辑。2.3 步骤 3定义 Nodes 节点创建各个配送站点LangGraph 中节点本质是同步 / 异步 Python 函数存在两条强制规范函数入参固定为全局 State无需返回完整 State 字典仅返回本轮发生变更的字段框架会自动根据 reducer 规则合并全局状态。# 揽收站节点完成包裹揽收 def receive_package(state: PackageState): 揽收站 origin state[origin] return { status: 已揽收, history: [f在{origin}揽收] } # 分拣中心节点根据目的地城市自动分流 def sort_package(state: PackageState): 分拣中心根据目的地分拣包裹 destination state[destination] if 北京 in destination: next_station 北京分拣中心 elif 上海 in destination: next_station 上海分拣中心 else: next_station 其他地区分拣中心 return { status: 已分拣, history: [f分拣至{next_station}] } # 标准配送节点普通陆运 def standard_delivery(state: PackageState): 标准陆运配送 return { status: 运输中, history: [标准陆运], total_distance: 500 } # 加急配送节点空运加急 def express_delivery(state: PackageState): 加急空运配送 return { status: 加急运输, history: [空运加急], total_distance: 800 } # 派送站节点末端派送完成签收 def final_delivery(state: PackageState): 末端派送站包裹签收完成 dest state[destination] return { status: 已签收, history: [f已送达{dest}] }2.4 步骤 4add_node 注册所有节点到图中调用add_node()方法将业务节点注册到 StateGraph 中方法接收两个参数第一个参数自定义节点名称字符串后续边流转时作为唯一标识第二个参数节点对应的执行函数。# 4. 将所有配送站点节点注册进图 delivery.add_node(揽收站, receive_package) delivery.add_node(分拣中心, sort_package) delivery.add_node(标准配送, standard_delivery) delivery.add_node(加急配送, express_delivery) delivery.add_node(派送站, final_delivery)2.5 步骤 5添加 Edges 流转边规划快递运输路线所有节点注册完成后需要通过边定义节点之间的流转顺序。LangGraph 包含两种核心边固定边、条件分支边。内置特殊节点说明START全局唯一入口节点初始输入状态统一流入该节点END工作流终止节点流转至该节点后图执行结束。1. 固定边 add_edge ()固定边用于无判断逻辑、顺序固定的流转场景语法add_edge(起始节点名, 目标节点名)。 本案例固定流转链路START → 揽收站 → 分拣中心# 固定流转路线入口 - 揽收站 - 分拣中心 delivery.add_edge(START, 揽收站) delivery.add_edge(揽收站, 分拣中心)2. 条件分支边 add_conditional_edges ()条件边会在离开指定节点时执行判断函数根据 State 数据动态选择下一跳节点存在两种常用写法。写法 1path_map 传入节点名称列表判断函数直接返回目标节点名字符串path_map 填写所有可能跳转的节点列表# 分拣完成后根据包裹优先级选择配送方式 def select_delivery(state: PackageState): 智能路由决策加急件走空运普通件走陆运 if state[priority] 加急: return 加急配送 else: return 标准配送 # 添加条件分支边分拣中心执行路由判断 delivery.add_conditional_edges( source分拣中心, pathselect_delivery, path_map[加急配送, 标准配送] )写法 2path_map 传入映射字典判断函数返回自定义标识字符串通过字典映射绑定真实节点逻辑解耦更清晰def select_delivery(state: PackageState): if state[priority] 加急: return 备注加急 else: return 无备注 delivery.add_conditional_edges( source分拣中心, pathselect_delivery, path_map{ 备注加急: 加急配送, 无备注: 标准配送 } )后续固定收尾路线无论走加急 / 标准配送最终都统一进入派送站完成后结束流程# 两条配送线路统一流向派送站 delivery.add_edge(标准配送, 派送站) delivery.add_edge(加急配送, 派送站) # 派送完成工作流结束 delivery.add_edge(派送站, END)2.6 步骤 6StateGraph 图编译生成可执行系统仅完成节点、边定义的构建器无法运行必须调用compile()编译图。 编译阶段会自动校验图链路完整性、不存在孤立节点与死循环编译完成后返回CompiledStateGraph对象该对象实现 Runnable 标准接口支持同步调用、流式输出、批量执行等能力。# 6. 编译图生成可执行的配送系统 delivery_system delivery.compile()2.2.7 完整测试调用 invoke 执行工作流使用invoke()方法传入初始化包裹状态即可完整执行整套快递配送流程invoke 输入为符合 PackageState 结构的字典执行完成返回最终完整状态。# 7. 测试两组不同类型包裹 test_packages [ { package_id: P001, origin: 北京, destination: 上海, priority: 普通, history: [], total_distance: 0 }, { package_id: P002, origin: 广州, destination: 乌鲁木齐, priority: 加急, history: [], total_distance: 0 } ] # 循环执行配送流程并打印结果 for package in test_packages: print(f\n 开始配送包裹: {package[package_id]} ) result delivery_system.invoke(package) print(包裹最终状态, result[status]) print(完整配送历史, result[history]) print(累计运输总里程, result[total_distance])程序运行输出结果成功运行可涵在学习时遇到的问题问题一# 揽收站节点 def receive_package(state: PackageState): 揽收站 # 进行状态扭转 # 状态值获取 print(---执行到揽收站节点) origin state[origin] return { status: 已揽收, history: [f在{origin}揽收] }为什么不把 return 的内容就放在函数体里面写还有可以不写 return 吗答案核心两点LangGraph 的 Node 必须 return不能省略且不能直接修改 state 就地生效一、先说结论不能不写 returnLangGraph 的节点函数有强制规则节点不能直接修改入参 state必须返回一个「局部更新字典」框架自动合并到全局 State。1. 为什么不能直接在函数里改 state不靠 return试写这种错误写法# 错误写法修改不会保存到全局状态 def receive_package(state: PackageState): print(---执行到揽收站节点) origin state[origin] # 就地修改传入的state字典 state[status] 已揽收 state[history] [f在{origin}揽收] # 没有return任何内容运行后会发现外层完整 Graph 的全局状态完全不会更新。原因 LangGraph 在每一次节点跳转时会做状态快照隔离传入函数的 state 只是一份临时副本。你在函数内部修改副本出函数后副本直接丢弃主流程的全局状态不受任何影响。LangGraph 设计逻辑进入节点复制一份当前全局 state 作为临时副本传给函数执行节点代码读取函数返回的更新字典根据你Annotated定义的合并规则覆盖 / 追加 / 累加把更新合并到真正的全局状态销毁临时副本进入下一个节点。只有 return 的字典才会被框架捕获、合并。无 return 没有任何更新全局状态保持原样。总结LangGraph 节点函数不认可函数内部对 state 的就地修改所有状态变更必须通过 return 返回更新字典框架捕获返回值后按照 TypedDict 定义的合并规则更新全局状态省略 return 将不会产生任何状态变化。问题二return 里面写什么仅写「需要更新的字段」return 只能返回字典字典里只放本轮节点执行完发生变化的字段。 作用告诉 LangGraph 哪些字段要更新框架自动合并到全局 PackageState。1. 哪些字段要放进 return 字典只要本节点的值和进入节点前不一样就写进去状态流转字段status待揽收→已揽收→已分拣带operator.add注解的累加 / 追加字段history、total_distance本次业务产生的新数据、新标识示例揽收节点只改 status、history只返回这两个 keyreturn { status: 已揽收, history: [f在{origin}揽收] }2. 绝对不用写进 return 的内容全程不变的固定信息package_id、origin、destination、priority包裹 ID、始发地不会中途修改不需要返回框架自动保留原值没有发生任何变化的字段 比如分拣节点不会改变 priority 加急类型不用写 priority3. 特殊情况节点无任何状态更新只是打印日志、查询数据不修改任何字段返回空字典def log_node(state: PackageState): print(当前包裹id, state[package_id]) return {}4、极简总结函数体内读取状态、打印日志、分支判断、调用工具、计算中间变量只做逻辑不持久化修改全局状态。return 字典内仅填写本次执行后发生变更的 State 字段固定不变的字段完全不用返回框架根据注解规则自动合并更新到全局状态。