当前位置： 首页 > news >正文

LangGraph底层API入门总结

news 2025/7/28 15:13:20

LangGraph底层API入门总结

一、LangGraph图结构基础

LangChain图结构概念说明：
在以图构建的框架中，任何可执行的功能都可以作为对话、代理或程序的启动点。这个启动点可以是大模型的 API 接口、基于大模型构建的 AI Agent，通过 LangChain 或其他技术建立的线性序列等等，即下图中的 “Start” 圆圈所示。无论哪种形式，它都首先处理用户的输入，并决定接下来要做什么。下图展示了在 LangGraph 概念下，最基本的一种代理模型：👇

图片描述
如上图所示，**在启动点定义的可运行功能会根据收到的输入决定是否进行检索以及如何响应。** 比如在执行过程中，如果需要检索信息，则可以利用搜索工具来实现，如`Web Search`（网络搜索）、`Query Database`（查询数据库）、`RAG`等获取必要的信息（图中的 "Action" 圆圈）。接下来，再使用一个大语言模型（“LLM”）处理工具提供的信息，结合用户最初传入的初始查询，生成最终的响应（图中的 "LLMs" 圆圈）。最终，这个响应被传递至终点节点（图中的 "End" 圆圈）。

这个流程就是在LangGraph框架中一个非常简单的代理构成形式。非常关键且我们必须清楚的概念是：**每个圆圈代表一个“节点”（Nodes），每个箭头表示一条“边”（Edges）。在 LangGraph 中，无论代理的构建是简单还是复杂，它最终都是由节点和边通过特定的组合形成的图。这样的构建形式形成的工作流原理就是：当每个节点完成工作后，通过边告诉下一步该做什么，所以也就得出了：LangGraph的底层图算法就是在使用消息传递来定义通用程序。当节点完成其操作时，它会沿着一条或多条边向其他节点发送消息。然后，这些接收节点执行其功能，将结果消息传递给下一组节点，然后该过程继续。如此循环往复。

这就是LangGraph底层架构设计中图算法的根本思想。

LangGraph框架是通过组合Nodes和Edges去创建复杂的循环工作流程，通过消息传递的方式串联所有的节点形成一个通路。那么维持消息能够及时的更新并向该去的地方传递，则依赖langGraph构建的State概念。 在LangGraph构建的流程中，每次执行都会启动一个状态，图中的节点在处理时会传递和修改该状态。这个状态不仅仅是一组静态数据，而是由每个节点的输出动态更新，然后影响循环内的后续操作。如下所示：👇

图片描述

1. 核心概念

  • StateGraph：LangGraph的核心类，用于创建状态图
  • 节点(Node)：图中的功能单元，通过add_node方法添加
  • 边(Edge)：节点间的连接，通过add_edge方法定义
  • 状态(State)：在节点间传递和修改的数据上下文
  • 检查点(Checkpointer)：用于保存和恢复对话状态

2. 图结构创建流程

  定义图时要做的第一件事是定义图的State。状态表示会随着图计算的进行而维护和更新的上下文或记忆。它用来确保图中的每个步骤都可以访问先前步骤的相关信息，从而可以根据整个过程中积累的数据进行动态决策。这个过程通过状态图StateGraph类实现，它是由LangGraph框架提供的核心类之一，专门用来创建state状态。

  构建state的方法非常简答。我们可以将图的状态设计为一个字典，用于在不同节点间共享和修改数据，然后使用StateGraph类进行图的实例化。代码如下：

from langgraph.graph import StateGraph# 使用 stategraph 接收一个字典
builder = StateGraph(dict)

这里需要注意的是，builder也是后面要用到的图构建器（Graph Builder）对象，用于逐步添加节点、边、控制流逻辑，最终编译成可执行的 LangGraph 图。而这个图构建器需要通过带入一个状态对象来创建。

  接下来，定义两个节点。addition节点是一个加法逻辑，接收当前状态StateGraph(dict)，将字典中x的值增加1，并返回新的状态。而subtraction节点是一个减法逻辑，接收从addition节点传来的状态StateGraph(dict)，从字典中的x值减去2，创建并返回一个新的键y。代码如下：

def addition(state):# 注意：这里接收到的是初始状态print(f"init_state: {state}")return {"x": state["x"] + 1}def subtraction(state):# 注意：这里接收到的是上一个节点的状态print(f"addition_state: {state}")return {"x": state["x"] - 2}

  然后，进行图结构的设计。具体来看，我们添加名为addition和subtraction的节点，并关联到上面定义的函数。设定图的起始节点为addition，并从addition到subtraction设置一条边，最后从subtraction到结束节点设置另一条边。代码如下：

# START 和 END 是两个特殊的节点，分别表示图的开始和结束。
from langgraph.graph import START, END# 向图中添加两个节点
builder.add_node("addition", addition)
builder.add_node("subtraction", subtraction)# 构建节点之间的边
builder.add_edge(START, "addition")
builder.add_edge("addition", "subtraction")
builder.add_edge("subtraction", END)

  最后，执行图的编译。需要通过调用compile()方法将这些设置编译成一个可执行的图。代码如下所示：

graph = builder.compile()

  除了上述通过打印的方式查看构建图的结构，LangGraph还提供了多种内置的图形可视化方法，能够将任何Graph以图形的形式展示出来，帮助我们更好地理解节点之间的关系和流程的动态变化。可视化最大的好处是：直接从代码中生成图形化的表示，可以检查图的执行逻辑是否符合构建的预期。

  生成图结构的可视化非常直接，只需一行代码即可完成。具体代码如下：

# 这里需要下载依赖库：pip install pyppeteer ipython -i https://pypi.tuna.tsinghua.edu.cn/simple
from IPython.display import Image, displaydisplay(Image(graph.get_graph(xray=True).draw_mermaid_png()))

在这里插入图片描述
  当通过 builder.compile() 方法编译图后，编译后的 graph 对象提供了 invoke 方法，该方法用于启动图的执行。我们可以通过 invoke 方法传递一个初始状态（如 initial_state = {"x": 10}），这个状态将作为图执行的起始输入。代码如下：

# 定义一个初始化的状态
initial_state = {"x":10}graph.invoke(initial_state)

init_state: {‘x’: 10}
addition_state: {‘x’: 11}
{‘x’: 9}

将一个字典作为状态对象带入到图中，即可进行图的实际运行。LangGraph 的执行模型并不强制要求图中必须有 END 节点。只要执行路径在某个节点“无后继边”（即到达“终点”），那么该节点就被视为隐式终点。
  在图的执行过程中，每个节点的函数会被调用，并且接收到前一个节点返回的状态作为输入。每个函数处理完状态后，会输出一个新的状态，传递给下一个节点。
  上述代码执行过程中图的运行状态如下图所示：👇

这里需要注意的一个关键信息是：节点函数不需要返回整个状态，而是仅返回它们更新的部分。 也就是说：在每个节点的函数内部逻辑中，需要使用和更新哪些State中的参数中，只需要在return的时候指定即可，不必担心未在当前节点处理的State中的其他值会丢失，因为LangGraph的内部机制已经自动处理了状态的合并和维护。

# 定义一个初始化的状态
initial_state = {"x":10, "y": 9}graph.invoke(initial_state)

init_state: {‘x’: 10, ‘y’: 9}
addition_state: {‘x’: 11}
{‘x’: 9}

  总体来看，该图设置了一个简单的工作流程。其中值首先在第一个节点通过加法函数增加，然后在第二个节点通过减法函数减少。这一流程展示了节点如何通过图中的共享状态进行交互。需要注意的是，状态在任何给定时间只包含来自一个节点的更新信息。这意味着当节点处理状态时，它只能访问与其特定操作直接相关的数据，从而确保每个节点的逻辑是隔离和集中的。 使用字典作为状态模式非常简单，由于缺乏预定义的模式，节点可以在没有严格类型约束的情况下自由地读取和写入状态，这样的灵活性有利于动态数据处理。 然而，这也要求开发者在整个图的执行过程中保持对键和值的一致性管理。因为如果在任何节点中尝试访问State中不存在的键，会直接中断整个图的运行状态。

3. 借助Pydantic对象创建图

  Pydantic 是一个用于创建“数据模型”的 Python 库，它可以自动校验数据类型，并将字典数据转换为结构化对象。它就像是给字典加了一个“类型安全 + 自动验证”的外壳，是现代 Python 项目中最主流的“数据结构定义工具”。

state = {"x": 1, "y": "hello"}  # 没有类型限制state["x"] = "abc"  # 不小心改错也不会报错

通过使用 Pydantic，能够显著增强状态管理的类型安全性、可读性和扩展性。

from pydantic import BaseModelclass MyState(BaseModel):x: inty: str = "default"   # 设置默认值# 自动校验
state = MyState(x=1)
print(state.x)       # 输出 1
print(state.y)       # 输出 default# 错误类型会报错
# state = MyState(x="abc")  # ❌ 会抛出 ValidationError

1
default

然后我们可以在 StateGraph(MyState) 传入一个 Pydantic 模型：

from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END# ✅ 1. 定义结构化状态模型
class CalcState(BaseModel):x: int# ✅ 2. 定义节点函数，接收并返回 CalcState
def addition(state: CalcState) -> CalcState:print(f"[addition] 初始状态: {state}")return CalcState(x=state.x + 1)def subtraction(state: CalcState) -> CalcState:print(f"[subtraction] 接收到状态: {state}")return CalcState(x=state.x - 2)# ✅ 3. 构建图
builder = StateGraph(CalcState)builder.add_node("addition", addition)
builder.add_node("subtraction", subtraction)builder.add_edge(START, "addition")
builder.add_edge("addition", "subtraction")
builder.add_edge("subtraction", END)graph = builder.compile()# ✅ 4. 执行图：传入结构化状态对象
initial_state = CalcState(x=10)
final_state = graph.invoke(initial_state)# ✅ 5. 打印最终结果
print("\n[最终结果] ->", final_state)

[addition] 初始状态: x=10
[subtraction] 接收到状态: x=11
[最终结果] -> {‘x’: 9}

但是需要注意的是，无论输入端输入什么结构的对象，最终图计算返回结果是一个字典类型对象。

4.创建条件分支图

在本示例中，我们基于 LangGraph 框架构建了一个简单的有状态条件分支图，用于演示如何使用结构化状态（通过 Pydantic 模型定义）在多步骤的决策流程中进行状态传递与条件控制。
名为 MyState 的 Pydantic 模型，用于描述图中每个节点共享的上下文状态信息。该状态包含两个字段：x 表示输入数值，result 表示最终处理结果。

from typing import Optional
from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END# ✅ 定义结构化状态
class MyState(BaseModel):x: intresult: Optional[str] = None# ✅ 定义各节点处理逻辑（接受 MyState，返回 MyState）
def check_x(state: MyState) -> MyState:print(f"[check_x] Received state: {state}")return statedef is_even(state: MyState) -> bool:return state.x % 2 == 0def handle_even(state: MyState) -> MyState:print("[handle_even] x 是偶数")return MyState(x=state.x, result="even")def handle_odd(state: MyState) -> MyState:print("[handle_odd] x 是奇数")return MyState(x=state.x, result="odd")# ✅ 构建图
builder = StateGraph(MyState)builder.add_node("check_x", check_x)
builder.add_node("handle_even", handle_even)
builder.add_node("handle_odd", handle_odd)# ✅ 添加条件分支
builder.add_conditional_edges("check_x", is_even, {True: "handle_even",False: "handle_odd"
})# ✅ 衔接起始和结束
builder.add_edge(START, "check_x")
builder.add_edge("handle_even", END)
builder.add_edge("handle_odd", END)# ✅ 编译图
graph = builder.compile()# ✅ 执行测试
print("\n✅ 测试 x=4（偶数）")
graph.invoke(MyState(x=4))print("\n✅ 测试 x=3（奇数）")
graph.invoke(MyState(x=3))

✅ 测试 x=4（偶数）
[check_x] Received state: x=4 result=None
[handle_even] x 是偶数
✅ 测试 x=3（奇数）
[check_x] Received state: x=3 result=None
[handle_odd] x 是奇数
{‘x’: 3, ‘result’: ‘odd’}

from IPython.display import display, Image
display(Image(graph.get_graph(xray=True).draw_mermaid_png(max_retries=5, retry_delay=2.0)))

在这里插入图片描述

5.创建条件循环图

from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END# ✅ 1. 定义结构化状态模型
class LoopState(BaseModel):x: int# ✅ 2. 定义节点逻辑
def increment(state: LoopState) -> LoopState:print(f"[increment] 当前 x = {state.x}")return LoopState(x=state.x + 1)def is_done(state: LoopState) -> bool:return state.x > 10# ✅ 3. 构建图
builder = StateGraph(LoopState)
builder.add_node("increment", increment)# ✅ 4. 设置循环控制：is_done 为 True 则结束，否则继续
builder.add_conditional_edges("increment", is_done, {True: END,False: "increment"
})builder.add_edge(START, "increment")
graph = builder.compile()# ✅ 5. 测试执行
print("\n✅ 执行循环直到 x > 10")
final_state = graph.invoke(LoopState(x=6))
print(f"[最终结果] -> x = {final_state['x']}")

✅ 执行循环直到 x > 10
[increment] 当前 x = 6
[increment] 当前 x = 7
[increment] 当前 x = 8
[increment] 当前 x = 9
[increment] 当前 x = 10
[最终结果] -> x = 11

display(Image(graph.get_graph(xray=True).draw_mermaid_png()))

在这里插入图片描述

5.2.创建条件循环图2

from pydantic import BaseModel
from typing import Optional
from langgraph.graph import StateGraph, START, END# ✅ 1. 定义状态模型
class BranchLoopState(BaseModel):x: intdone: Optional[bool] = False# ✅ 2. 定义各节点逻辑
def check_x(state: BranchLoopState) -> BranchLoopState:print(f"[check_x] 当前 x = {state.x}")return statedef is_even(state: BranchLoopState) -> bool:return state.x % 2 == 0def increment(state: BranchLoopState) -> BranchLoopState:print(f"[increment] x 是偶数，执行 +1 → {state.x + 1}")return BranchLoopState(x=state.x + 1)def done(state: BranchLoopState) -> BranchLoopState:print(f"[done] x 是奇数，流程结束")return BranchLoopState(x=state.x, done=True)# ✅ 3. 构建图
builder = StateGraph(BranchLoopState)builder.add_node("check_x", check_x)
builder.add_node("increment", increment)
builder.add_node("done_node", done)builder.add_conditional_edges("check_x", is_even, {True: "increment",False: "done_node"
})# ✅ 4. 循环逻辑：偶数 → increment → check_x
builder.add_edge("increment", "check_x")# ✅ 5. 起始与终点
builder.add_edge(START, "check_x")
builder.add_edge("done_node", END)graph = builder.compile()# ✅ 6. 测试执行
print("\n✅ 初始 x=6（偶数，进入循环）")
final_state1 = graph.invoke(BranchLoopState(x=6))
print("[最终结果1] ->", final_state1)print("\n✅ 初始 x=3（奇数，直接 done）")
final_state2 = graph.invoke(BranchLoopState(x=3))
print("[最终结果2] ->", final_state2)

✅ 初始 x=6（偶数，进入循环）
[check_x] 当前 x = 6
[increment] x 是偶数，执行 +1 → 7
[check_x] 当前 x = 7
[done] x 是奇数，流程结束
[最终结果1] -> {‘x’: 7, ‘done’: True}
✅ 初始 x=3（奇数，直接 done）
[check_x] 当前 x = 3
[done] x 是奇数，流程结束
[最终结果2] -> {‘x’: 3, ‘done’: True}

display(Image(graph.get_graph(xray=True).draw_mermaid_png()))

在这里插入图片描述

二、多轮对话实现

1. 消息管理

使用add_messages处理消息序列：

from langgraph.graph.message import add_messagesclass State(TypedDict):messages: Annotated[list[BaseMessage], add_messages]graph_builder = StateGraph(State)

2. 对话记忆

使用MemorySaver保存对话状态：

from langgraph.checkpoint.memory import MemorySavermemory = MemorySaver()
graph = builder.compile(checkpointer=memory)# 使用thread_id维护对话上下文
thread_config = {"configurable": {"thread_id": "abc123"}}
state = graph.invoke(inputs, config=thread_config)

三、流式输出

LangGraph支持通过stream和astream方法实现流式输出：

# 同步流式输出
for chunk in graph.stream(inputs):print(chunk)# 异步流式输出
async for chunk in graph.astream(inputs):print(chunk)

六、核心API总结

API描述
StateGraph创建状态图的核心类
add_node添加节点到图中
add_edge添加边连接节点
add_conditional_edges添加条件分支边
compile编译图为可执行对象
invoke执行图并返回最终状态
stream/astream流式执行图
checkpointer配置状态保存与恢复

参考：赋范空间大模型技术社区

http://www.lryc.cn/news/601878.html

相关文章：

  • OpenLayers 综合案例-地图绘制
  • 十字链表以及实现
  • SpringAI入门及浅实践，实战 Spring‎ AI 调用大模型、提示词工程、对话记忆、Adv‎isor 的使用
  • 第五章 中央处理器（CPU）知识体系与考法总结
  • 【第六节】方法与事件处理器
  • Gradle#Plugin
  • Windows---动态链接库Dynamic Link Library（.dll）
  • 2025.7.27总结—新励成
  • Kubernetes 核心组件解析
  • HCIE学习之路：MSTP实现负载均衡实验
  • 【INT范围提取字符串数字为正数】2022-8-29
  • Leetcode 3628. Maximum Number of Subsequences After One Inserting
  • rust- 定义模块以控制作用域和隐私
  • 握手未来，PostgreSQL认证专家
  • 【I】题目解析
  • Spring AI 学习笔记
  • 小架构step系列27：Hibernate提供的validator
  • 「mysql」Mac osx彻底删除mysql
  • Java面试宝典：MySQL性能优化
  • uart通信
  • JVM类加载机制全流程详解
  • 从MySQL的information_schema系统数据库中获取表的元数据信息
  • MySQL - 索引(B+树)
  • Cgroup 控制组学习(三)在容器中使用 CGroups
  • MySQL - 主从复制与读写分离
  • Cline与Cursor深度实战指南：AI编程助手的革命性应用
  • 基于CNN图像特征提取流程（简化版）
  • Linux实战：从零搭建基于LNMP+NFS+DNS的WordPress博客系统
  • Flink窗口：解锁流计算的秘密武器
  • QT---概览