news 2026/2/25 9:07:29

收藏速看!LangGraph1.0入门宝典—— 核心概念与“点边”实战

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
收藏速看!LangGraph1.0入门宝典—— 核心概念与“点边”实战

对于刚入门大模型开发的程序员来说,LangGraph 绝对是值得深耕的框架——它不是简单的工具封装,而是一套能让你“编程式构建智能体”的强大体系。这篇文章作为 LangGraph1.0 系列的开篇,会用最直观的方式拆解核心概念,再通过完整代码带你吃透“状态、节点、边”这三大基石,小白也能轻松跟上,建议收藏慢慢练!

一、先搞懂:LangGraph 为什么这么香?

LangGraph 的核心设计理念可以概括为:它是专为智能体而生的“编程语言”。就像我们用 Python 写脚本时需要“变量、函数、流程控制”一样,LangGraph 也提供了对应的抽象组件,让你能用声明式的方式编排复杂的智能体工作流。

很多人刚接触时会混淆这些概念,这里直接用“编程语言类比表”帮你一次性理清楚:

编程语言核心要素LangGraph 对应概念大白话说明
数据 (Data)状态 (State)工作流里“流转的信息包”,所有节点都能读取和更新它
函数 (Function)节点 (Node)干活的“小工人”,接收状态后执行具体操作(调模型/跑工具)
控制逻辑 (Control Flow)边 (Edge)节点间的“指挥棒”,决定哪个节点先跑、哪个后跑,支持分支和并行
存储 (Storage)检查点/记忆 (Checkpointing)给状态“存个档”,就算程序断了也能接着跑,实现长期记忆
中断 (Interrupt)人在回路 (Human-in-the-Loop)关键步骤“喊停”,让人工审核后再继续,避免智能体乱决策

二、状态State和节点Node

2.1 状态State

LangGraph是一个状态图,状态可以理解为图的数据。在定义图之前首先应该定义一个状态作为初始值,在图的更新过程中状态值是可以被图更新并返回给用户的。

图本身是无状态的,在定义图时,首先要定义图将操作的状态。这个状态由图中的所有节点共享。状态通常是一个Python数据结构。定义状态的方式可以采用类型化字典。比如下面代码就定义了包含名为nList字符串列表的状态值。当图被调用时,状态会被初始化,LangGraph运行时会选择一个节点来执行,然后它会提供当前状态,运行节点,最后更新状态再到结束。

class State(TypedDict): nlist: List[str]

2.2 节点

在LangGraph中,节点的本质是一个函数。节点函数的输入参数是状态,输出则是对状态的更新。举例如下:

def node_a(state: State): ... return ({"nlist":[note]})

以上节点函数执行完成后会对字符串列表进行更新。State 可以被持久化(通过 Checkpointing 机制)。这意味着如果节点执行失败或工作流被中断,可以从上一个检查点恢复状态重新运行,保证了工作流的鲁棒性。

2.3 使用LangGraph构建简单图

下面笔者通过一个完整的代码示例,将 State 和 Node 的概念串联起来,构建一个最简单的单节点图。在开始之前,请大家确保已经安装了langgraph相关环境。如未安装,可以运行pip install -U langgraph命令。

  1. 导入依赖库
from langgraph.graph import START, END, StateGraph
  1. 创建图表的第一步是定义一个State类,类型是字典。它包含一个字符串类型的列表,状态可以是一个Python数据类也可以是一个Pydantic基类模型
class State(TypedDict): nList: List[str]
  1. 定义节点,节点实际上就是Python函数,该函数接收状态,打印信息然后返回状态的更新。
def node_a(state): print(f"noda_a接收到{state['nList']}") note = "Hello, 我是节点a" return(State(nList=[note]))
  1. 编译状态图,将节点加入到图中,定义为"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()

以上代码构建了一个如下的简单图。

  1. 接下来测试这个图,首先使用State初始化一个状态变量,然后通过.invoke方法运行图。可以看到节点A接收了初始状态,然后当调用图完成后返回了更新后的状态。
initial_state = State( nList=["Hello Node a, how are you?"] ) print(graph.invoke(initial_state))

通过以上案例大家应该可以理解在定义图和状态时,图中所有节点可以共享相同的状态。

三、边

在定义了“数据”(State)和“操作”(Node)之后,笔者接下来介绍边 (Edge)是 LangGraph 中定义工作流执行逻辑的核心。它决定了节点之间的连接关系与执行顺序,是实现串行、并行、条件分支等复杂逻辑的关键。

3.1 普通边

普通边(静态边)用于连接两个节点,指定了确定无误的执行顺序。它是最简单的控制流,定义了工作流中“下一步该执行谁”。

  • 串行执行

    通过连续的普通边连接多个节点,形成一个线性的执行链。例如:START -> A -> B -> C -> END

  • 并行执行

    一个节点可以同时拥有多个出边(outgoing edges),指向不同的后续节点。LangGraph 运行时会在当前节点完成后,同时触发所有后续节点执行。

3.2 条件边

除普通边外,LangGraph还有另一种条件边。条件边允许工作流根据运行时状态动态决定下一步的执行路径。它需要一个路由函数,该函数接收当前状态,并选择下一个要执行的节点。下图左侧根据条件选择将状态传递给下一级左侧节点进行处理。条件边还有一个特殊情况是MapReduce, 它可以创建可变数量的下游节点,并且每个节点都传递了一个唯一值(笔者之后也会讲到)。

3.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(f"Adding 'A' to {state['unique_items']}") return State(unique_items=["A"]) def node_A_extra(state: State) -> State: print(f"Adding '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”了。

最后的结果如下:

3.4 探索LangGraph边的并行与数据共享

笔者接下来将构建下图所示的工作流,来帮助大家直观理解并行执行全局状态共享

  1. 第一步同样的需要定义状态,不同的是这里的状态需要用reducer来指明列表的状态是累加而不是覆盖原来的值。
import operator from typing import TypedDict, List, Annotated class State(TypedDict): nList: Annotated[List[str], operator.add]
  1. 下一步来定义节点函数,每个节点都接收状态,并返回对nList字段的更新。
def node_a(state: State) -> State: print(f"Adding 'A' to {state['nList']}") return State(nList=["A"]) def node_b(state: State) -> State: print(f"Adding 'B' to {state['nList']}") return State(nList=["B"]) def node_c(state: State) -> State: print(f"Adding 'C' to {state['nList']}") return State(nList=["C"]) def node_bb(state: State) -> State: print(f"Adding 'BB' to {state['nList']}") return State(nList=["BB"]) def node_cc(state: State) -> State: print(f"Adding 'CC' to {state['nList']}") return State(nList=["CC"]) def node_d(state: State) -> State: print(f"Adding 'D' to {state['nList']}") return State(nList=["D"])
  1. 用状态实例化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()
  1. 提供一个初始状态来调用图表,大家先来思考一下最后的运行结果。
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的状态是全局共享的,边只控制执行顺序,不隔离数据。

3.5 条件边

对于条件边笔者这里直接通过代码讲解,使用条件边实现下图结构,下图中的虚线表示条件边,实线表示普通边。

  1. 定义状态
from langgraph.graph import START, END, StateGraph import operator from typing import TypedDict, List, Annotated class State(TypedDict): nList: Annotated[List[str], operator.add]
  1. 定义节点函数
def node_a(state: State): return def node_b(state: State): return State(nList=['B']) def node_c(state: State): return State(nList=['C'])
  1. 定义图,首先添加实线普通边
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()
  1. 学习条件边的定义方法,条件边需要定义条件路由函数,该函数接受状态并返回一个值,这个值代表想要分支到的下一节点。笔者下面定义的函数接收图最近一次写入的状态,然后返回一个值。
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
  1. 在构建图时使用添加条件边的语法来添加条件边
builder.add_conditional_edges("a", conditional_edge)
  1. 测试条件边的逻辑。下面代码笔者从用户那里获取输入作为初始输入状态,然后用该状态调用图表:
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 1.0,LangGraph1.0是构建智能体的编程语言。它通过状态(State)管理数据,节点(Node)作为处理函数,边(Edge)定义串行、并行或条件执行逻辑。借助Reducer处理并行写入冲突,并支持人在回路控制。这使其成为搭建复杂、可编排AI工作流的强大底座

如何学习大模型 AI ?

由于新岗位的生产效率,要优于被取代岗位的生产效率,所以实际上整个社会的生产效率是提升的。

但是具体到个人,只能说是:

“最先掌握AI的人,将会比较晚掌握AI的人有竞争优势”。

这句话,放在计算机、互联网、移动互联网的开局时期,都是一样的道理。

我在一线科技企业深耕十二载,见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事,早已在效率与薪资上形成代际优势,我意识到有很多经验和知识值得分享给大家,也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套AI 大模型突围资料包

  • ✅ 从零到一的 AI 学习路径图
  • ✅ 大模型调优实战手册(附医疗/金融等大厂真实案例)
  • ✅ 百度/阿里专家闭门录播课
  • ✅ 大模型当下最新行业报告
  • ✅ 真实大厂面试真题
  • ✅ 2025 最新岗位需求图谱

所有资料 ⚡️ ,朋友们如果有需要《AI大模型入门+进阶学习资源包》下方扫码获取~

① 全套AI大模型应用开发视频教程

(包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点)

② 大模型系统化学习路线

作为学习AI大模型技术的新手,方向至关重要。 正确的学习路线可以为你节省时间,少走弯路;方向不对,努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划,带你从零基础入门到精通!

③ 大模型学习书籍&文档

学习AI大模型离不开书籍文档,我精选了一系列大模型技术的书籍和学习文档(电子版),它们由领域内的顶尖专家撰写,内容全面、深入、详尽,为你学习大模型提供坚实的理论基础。

④ AI大模型最新行业报告

2025最新行业报告,针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估,以了解哪些行业更适合引入大模型的技术和应用,以及在哪些方面可以发挥大模型的优势。

⑤ 大模型项目实战&配套源码

学以致用,在项目实战中检验和巩固你所学到的知识,同时为你找工作就业和职业发展打下坚实的基础。

⑥ 大模型大厂面试真题

面试不仅是技术的较量,更需要充分的准备。在你已经掌握了大模型技术之后,就需要开始准备面试,我精心整理了一份大模型面试题库,涵盖当前面试中可能遇到的各种技术问题,让你在面试中游刃有余

以上资料如何领取?

为什么大家都在学大模型?

最近科技巨头英特尔宣布裁员2万人,传统岗位不断缩减,但AI相关技术岗疯狂扩招,有3-5年经验,大厂薪资就能给到50K*20薪!

不出1年,“有AI项目经验”将成为投递简历的门槛。

风口之下,与其像“温水煮青蛙”一样坐等被行业淘汰,不如先人一步,掌握AI大模型原理+应用技术+项目实操经验,“顺风”翻盘!

这些资料真的有用吗?

这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理,现任上海殷泊信息科技CEO,其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证,服务航天科工、国家电网等1000+企业,以第一作者在IEEE Transactions发表论文50+篇,获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目,无论你是小白还是有些技术基础的技术人员,这份资料都绝对能帮助你提升薪资待遇,转行大模型岗位。

以上全套大模型资料如何领取?

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/2/22 2:10:13

小程序毕设项目推荐-基于微信小程序羽球快讯爱好者平台基于springboot+微信小程序的羽球快讯爱好者平台小程序【附源码+文档,调试定制服务】

博主介绍:✌️码农一枚 ,专注于大学生项目实战开发、讲解和毕业🚢文撰写修改等。全栈领域优质创作者,博客之星、掘金/华为云/阿里云/InfoQ等平台优质作者、专注于Java、小程序技术领域和毕业项目实战 ✌️技术范围:&am…

作者头像 李华
网站建设 2026/2/18 3:00:09

小程序毕设项目推荐-基于微信小程序的交通违法有奖曝光平台设计与实现基于springboot+微信小程序的的交通违法有奖曝光平台【附源码+文档,调试定制服务】

博主介绍:✌️码农一枚 ,专注于大学生项目实战开发、讲解和毕业🚢文撰写修改等。全栈领域优质创作者,博客之星、掘金/华为云/阿里云/InfoQ等平台优质作者、专注于Java、小程序技术领域和毕业项目实战 ✌️技术范围:&am…

作者头像 李华
网站建设 2026/2/18 6:24:44

精度下降预警机制设计,工业质检Agent稳定性保障的最后防线

第一章:精度下降预警机制设计,工业质检Agent稳定性保障的最后防线在高自动化产线中,质检Agent长期运行可能因模型漂移、环境噪声或设备老化导致识别精度隐性下降。若未及时发现,将引发批量误检,造成重大质量事故。为此…

作者头像 李华
网站建设 2026/2/6 9:26:16

真正能升上去的人,最会处理模糊地带

职场里有个残酷规律:越往上走,越没有“标准答案”。流程不清晰、责任边界不明确、资源分配模糊、方向时常变动……你以为这是混乱,其实这是组织在筛人。因为真正能升上去的人,都有一个共同点——他们最会处理模糊地带。 为什么&am…

作者头像 李华
网站建设 2026/2/20 20:20:19

算法---LeetCode 572. 另一棵树的子树

1. 题目 原题链接 给你两棵二叉树 root 和 subRoot 。检验 root 中是否包含和 subRoot 具有相同结构和节点值的子树。如果存在,返回 true ;否则,返回 false 。 二叉树 tree 的一棵子树包括 tree 的某个节点和这个节点的所有后代节点。tree…

作者头像 李华
网站建设 2026/2/23 9:09:54

力扣234.回文链表-反转后半链表

问题描述 给定一个单链表的头节点 head,判断该链表是否为回文链表。如果是,返回 true;否则,返回 false。 示例 : 输入: head [1,2,2,1] 输出: true输入: head [1,2] 输出: false 进阶: 你能否用 O(n)…

作者头像 李华