核心内容摘要
岁月鎏金，风华依旧——张柏芝的47年人生掠影
文章介绍了LangGraph
0作为构建智能体工作流的核心框架详细讲解了状态(State)、节点(Node)和边(Edge)三大核心概念。
通过代码示例展示了如何定义状态、创建节点函数、构建图结构以及处理并行执行和条件分支。
文章还介绍了Reducer机制用于解决并行执行时的状态更新冲突为读者提供了构建复杂AI工作流的基础知识。
前言
LangGraph 核心特性LangGraph 是一个极其灵活的框架其核心设计理念可以概括为它是一种用于构建智能体的“编程语言”。
如同任何编程语言都包含数据、函数、控制流等基本要素一样LangGraph 也提供了功能对等的抽象组件使开发者能够以声明式的方式编排智能体工作流。
其核心特性的对应关系如下编程语言要素LangGraph 对应概念说明数据 (Data)状态 (State)工作流执行过程中传递和更新的信息载体。
函数 (Function)节点 (Node)对状态进行操作如调用模型、执行工具的基本单元。
控制逻辑 (Control Flow)边 (Edge)决定节点间执行顺序的逻辑可以是条件分支或静态路径支持串行与并行。
存储 (Storage)检查点 / 记忆 (Checkpointing / Memory)状态的持久化机制允许工作流暂停、恢复或具有长期记忆。
中断 (Interrupt)人在回路 (Human-in-the-Loop)通过中断控制流在关键环节引入人工审核或干预的能力。
通过以上组件LangGraph 使得构建一个具备复杂逻辑、可持续运行且支持人机协作的智能体系统变得直观且高效。
状态State和节点Node
1 状态StateLangGraph是一个状态图状态可以理解为图的数据。
在定义图之前首先应该定义一个状态作为初始值在图的更新过程中状态值是可以被图更新并返回给用户的。
图本身是无状态的在定义图时首先要定义图将操作的状态。
这个状态由图中的所有节点共享。
状态通常是一个Python数据结构。
定义状态的方式可以采用类型化字典。
比如下面代码就定义了包含名为nList字符串列表的状态值。
当图被调用时状态会被初始化LangGraph运行时会选择一个节点来执行然后它会提供当前状态运行节点最后更新状态再到结束。
class State(TypedDict): nlist: List[str]
2 节点在LangGraph中节点的本质是一个函数。
节点函数的输入参数是状态输出则是对状态的更新。
举例如下def node_a(state: State): ... return ({nlist:[note]})以上节点函数执行完成后会对字符串列表进行更新。
State 可以被持久化通过 Checkpointing 机制。
这意味着如果节点执行失败或工作流被中断可以从上一个检查点恢复状态重新运行保证了工作流的鲁棒性。
3 使用LangGraph构建简单图下面笔者通过一个完整的代码示例将 State 和 Node 的概念串联起来构建一个最简单的单节点图。
在开始之前请大家确保已经安装了langgraph相关环境。
如未安装可以运行pip install -U langgraph命令。
导入依赖库from langgraph.graph import START, END, StateGraph创建图表的第一步是定义一个State类类型是字典。
它包含一个字符串类型的列表,状态可以是一个Python数据类也可以是一个Pydantic基类模型关于pydantic的使用大家可以参考笔者的文章: 深入浅出LangGraph AI Agent智能体开发教程六—LangGraph 底层API入门, 也可查看pydantic官方文档 这里不加赘述了class State(TypedDict): nList: List[str]定义节点节点实际上就是Python函数该函数接收状态打印信息然后返回状态的更新。
def node_a(state): print(fnoda_a接收到{state[nList]}) note Hello, 我是节点a return(State(nList[note]))编译状态图将节点加入到图中定义为a节点同时添加从START节点到a节点的边然后添加从a到END的边。
这里的START和END都是LangGraph定义的常量节点仅包含语义信息。
最后编译图。
builder StateGraph(State) builder.add_node(a, node_a) builder.add_edge(START, a) builder.add_edge(a, END) graph builder.compile()以上代码构建了一个如下的简单图。
接下来测试这个图首先使用State初始化一个状态变量然后通过.invoke方法运行图。
可以看到节点A接收了初始状态然后当调用图完成后返回了更新后的状态。
initial_state State( nList[Hello Node a, how are you?] ) print(graph.invoke(initial_state))通过以上案例大家应该可以理解在定义图和状态时图中所有节点可以共享相同的状态。
边在定义了“数据”State和“操作”Node之后笔者接下来介绍边。
边 (Edge)是 LangGraph 中定义工作流执行逻辑的核心。
它决定了节点之间的连接关系与执行顺序是实现串行、并行、条件分支等复杂逻辑的关键。
1 普通边普通边静态边用于连接两个节点指定了确定无误的执行顺序。
它是最简单的控制流定义了工作流中“下一步该执行谁”。
串行执行通过连续的普通边连接多个节点形成一个线性的执行链。
例如START - A - B - C - END。
并行执行一个节点可以同时拥有多个出边outgoing edges指向不同的后续节点。
LangGraph 运行时会在当前节点完成后同时触发所有后续节点执行。
2 条件边除普通边外LangGraph还有另一种条件边。
条件边允许工作流根据运行时状态动态决定下一步的执行路径。
它需要一个路由函数该函数接收当前状态并选择下一个要执行的节点。
下图左侧根据条件选择将状态传递给下一级左侧节点进行处理。
条件边还有一个特殊情况是MapReduce, 它可以创建可变数量的下游节点并且每个节点都传递了一个唯一值笔者之后也会讲到。
3 Reducer详解当一个节点拥有多个出边即触发了并行执行时多个后续节点可能同时更新 State 中的同一个字段。
此时就会产生冲突以谁的更新为准默认情况下节点的更新是覆盖式的。
如果节点B和节点C并行执行都尝试更新state[‘messages’]后完成的节点会覆盖先完成节点的结果导致数据丢失。
为了更精确的控制这时就需要用到Reducer归约器来解决解决此问题。
它是一个合并函数定义了当多个更新同时作用于同一个状态字段时应如何合并这些更新。
下面的状态定义中的operator.add就是一个Reducer它指定了列表状态应该是合并而不是覆写import operatorfrom typing import TypedDict, Annotatedclass State(TypedDict): nList: Annotated[list[str], operator.add]LangGraph 提供了一些开箱即用的 Reducer下表列出了最常见的几种Reducer 函数导入来源适用数据类型主要行为典型应用场景operator.addoperator(Python标准库)int,float数值累加计数器、计分器operator.extendoperator(Python标准库)List[T]列表扩展(list.extend)收集多项结果如搜索条目operator.or_operator(Python标准库)Set[T]集合取并集自动去重收集标签、唯一ID集合update_dictlanggraph.utilsDict合并字典新值覆盖旧键更新配置或元数据除了使用内置的Reducer外开发者甚至可以自定义reducer进行处理。
Reducer就是一个合并函数该函数接收当前值和新传入值作为参数并在函数中定义处理逻辑。
以下是一个自定义Reducer的代码示例该函数合并新值和旧值并返回去重后结果。
from typing import TypedDict, List, Annotated from langgraph.graph import START, END, StateGraph def deduplicate_merge(old_list: List[str], new_list: List[str]) - List[str]: 自定义Reducer合并列表并去重 combined old_list new_list return list(dict.fromkeys(combined)) # 保持顺序的去重 class MyState(TypedDict): unique_items: Annotated[List[str], deduplicate_merge] from typing import TypedDict, List, Annotated class State(TypedDict): unique_items: Annotated[List[str], deduplicate_merge] def node_a(state: State) - State: print(fAdding A to {state[unique_items]}) return State(unique_items[A]) def node_A_extra(state: State) - State: print(fAdding A to {state[unique_items]}) return State(unique_items[A]) builder StateGraph(State) builder.add_node(a, node_a) builder.add_node(a_extra, node_A_extra) builder.add_edge(START, a) builder.add_edge(a, a_extra) builder.add_edge(a_extra, END) graph builder.compile() initial_state State( unique_items [Initial String] ) print(graph.invoke(initial_state))上面代码的图结构如下所示按理说列表中应该有两个“A”但是因为添加时会去重所以最后列表中只有这一个“A”了。
最后的结果如下
4 探索LangGraph边的并行与数据共享笔者接下来将构建下图所示的工作流来帮助大家直观理解并行执行与全局状态共享。
第一步同样的需要定义状态不同的是这里的状态需要用reducer来指明列表的状态是累加而不是覆盖原来的值。
import operator from typing import TypedDict, List, Annotated class State(TypedDict): nList: Annotated[List[str], operator.add]下一步来定义节点函数每个节点都接收状态并返回对nList字段的更新。
def node_a(state: State) - State: print(fAdding A to {state[nList]}) return State(nList[A]) def node_b(state: State) - State: print(fAdding B to {state[nList]}) return State(nList[B]) def node_c(state: State) - State: print(fAdding C to {state[nList]}) return State(nList[C]) def node_bb(state: State) - State: print(fAdding BB to {state[nList]}) return State(nList[BB]) def node_cc(state: State) - State: print(fAdding CC to {state[nList]}) return State(nList[CC]) def node_d(state: State) - State: print(fAdding D to {state[nList]}) return State(nList[D])用状态实例化StateGraph然后添加定义好的节点并按照参考图中的边将不同的节点连接起来最后编译图。
builder StateGraph(State) builder.add_node(a, node_a) builder.add_node(b, node_b) builder.add_node(c, node_c) builder.add_node(bb, node_bb) builder.add_node(cc, node_cc) builder.add_node(d, node_d) builder.add_edge(START, a) builder.add_edge(a, b) builder.add_edge(a, c) builder.add_edge(b, bb) builder.add_edge(c, cc) builder.add_edge(bb, d) builder.add_edge(cc, d) builder.add_edge(d, END) graph builder.compile()提供一个初始状态来调用图表大家先来思考一下最后的运行结果。
initial_state State( nList [Initial String] ) print(graph.invoke(initial_state))并行与合并节点B和C在同一步骤中并行执行它们都接收到了来自节点A更新后的状态[‘Initial‘ ‘A’]。
它们的更新”B”和”C”通过Reducer被追加到了nList中。
全局状态共享节点BB和CC运行时它能“看到”的状态包含了其上游节点B的更新(‘B’)也包含了并行分支节点C的更新(‘C’)。
这是因为LangGraph的状态是全局共享的边只控制执行顺序不隔离数据。
5 条件边对于条件边笔者这里直接通过代码讲解使用条件边实现下图结构下图中的虚线表示条件边实线表示普通边。
定义状态from langgraph.graph import START, END, StateGraph import operator from typing import TypedDict, List, Annotated class State(TypedDict): nList: Annotated[List[str], operator.add]定义节点函数def node_a(state: State): return def node_b(state: State): return State(nList[B]) def node_c(state: State): return State(nList[C])定义图首先添加实线普通边builder StateGraph(State) builder.add_node(a, node_a) builder.add_node(b, node_b) builder.add_node(c, node_c) builder.add_edge(START, a) builder.add_edge(b, END) builder.add_edge(c, END) graph builder.compile()学习条件边的定义方法条件边需要定义条件路由函数该函数接受状态并返回一个值这个值代表想要分支到的下一节点。
笔者下面定义的函数接收图最近一次写入的状态然后返回一个值。
def conditional_edge(state: State) - Literal[b, c, END]: select state[nList][-1] if select b: return b elif select c: return c elif select q: return END else: return END在构建图时使用添加条件边的语法来添加条件边builder.add_conditional_edges(a, conditional_edge)测试条件边的逻辑。
下面代码笔者从用户那里获取输入作为初始输入状态然后用该状态调用图表user input(b, c or q to quit:) input_state State( nList[user] ) graph.invoke(input_state)可以看到当用户输入b, 会走到b节点在b节点中添加字符串’B’, 当用户输入c, 会走到c节点在c节点中添加字符串‘C’输入q 则会直接到结束节点这就是条件边的控制逻辑。
完整代码大家可以关注笔者的微信公众号大模型真好玩并私信:LangChain智能体开发免费获取。
以上就是今天的全部内容啦内容篇幅较长大家消化一下~
四、
总结本期分享笔者为大家介绍了LangGraph
0LangGraph
0是构建智能体的编程语言。
它通过状态State管理数据节点Node作为处理函数边Edge定义串行、并行或条件执行逻辑。
借助Reducer处理并行写入冲突并支持人在回路控制。
这使其成为搭建复杂、可编排AI工作流的强大底座。
本期分享着重介绍了LangGraph的节点和边的定义并用点和边构造了简单的图。
然而LangGraph不仅仅有如上所示的图的定义方法在下期分享笔者将为大家介绍更多点和边的构建方法以及介绍记忆Memory和人在回路Human In the loop大家敬请期待~AI大模型从0到精通全套学习大礼包我在一线互联网企业工作十余年里指导过不少同行后辈。
帮助很多人得到了学习和成长。
只要你是真心想学AI大模型我这份资料就可以无偿共享给你学习。
大模型行业确实也需要更多的有志之士加入进来我也真心希望帮助大家学好这门技术如果日后有什么学习上的问题欢迎找我交流有技术上面的问题我是很愿意去帮助大家的如果你也想通过学大模型技术去帮助就业和转行可以扫描下方链接大模型重磅福利入门进阶全套104G学习资源包免费分享
从入门到精通的全套视频教程包含提示词工程、RAG、Agent等技术点
AI大模型学习路线图还有视频解说全过程AI大模型学习路线
学习电子书籍和技术文档市面上的大模型书籍确实太多了这些是我精选出来的
大模型面试题目详解
这些资料真的有用吗?这份资料由我和鲁为民博士共同整理鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利同时还斩获了吴文俊人工智能科学技术奖。
目前我正在和鲁博士共同进行人工智能的研究。
所有的视频由智泊AI老师录制且资料与智泊AI共享相互补充。
这份学习大礼包应该算是现在最全面的大模型学习资料了。
资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。
智泊AI始终秉持着“让每个人平等享受到优质教育资源”的育人理念通过动态追踪大模型开发、数据标注伦理等前沿技术趋势构建起前沿课程智能实训精准就业的高效培养体系。
课堂上不光教理论还带着学员做了十多个真实项目。
学员要亲自上手搞数据清洗、模型调优这些硬核操作把课本知识变成真本事如果说你是以下人群中的其中一类都可以来智泊AI学习人工智能找到高薪工作一次小小的“投资”换来的是终身受益应届毕业生无工作经验但想要系统学习AI大模型技术期待通过实战项目掌握核心技术。
零基础转型非技术背景但关注AI应用场景计划通过低代码工具实现“AI行业”跨界。
业务赋能 突破瓶颈传统开发者Java/前端等学习Transformer架构与LangChain框架向AI全栈工程师转型。
获取方式有需要的小伙伴可以保存图片到wx扫描二v码免费领取【保证100%免费】