深度理解LangGraph核心-Graph

m0_62949473

556人浏览 · 2026-05-21 12:20:39

m0_62949473 · 2026-05-21 12:20:39 发布

在学习LangGraph的Graph之前，我们必须先掌握3个核心前置知识，这些是看懂后续内容的基础，全程用大白话讲解，不搞专业术语堆砌，完全零基础也能吃透。

前置知识1：Python基础（必备中的必备）

我们后续所有代码都是用Python写的，以下3个知识点是高频用到的，必须先搞懂：

函数的定义与调用：函数就是“一段可以重复使用的代码块”，比如我们要做“把输入的字符串加后缀”这个操作，不用每次都写一遍代码，定义一个函数，每次调用它就行。示例： def add_suffix(s): # 定义函数，函数名add_suffix，参数s（输入的字符串） `` return s + "abc" # 函数执行的操作：给s加后缀"abc"，并返回结果调用：result = add_suffix("图灵") # 调用函数，传入参数"图灵" 运行结果：result的值就是"图灵abc"（这就是函数的返回值）。
字典（dict）的基本用法：字典是Python里用来“存储键值对”的数据结构，比如用字典存一个人的信息：person = {"name": "小明", "age": 18}，其中“name”“age”是“键（key）”，“小明”“18”是“值（value）”，我们可以通过key获取value，比如person["name"]就能拿到“小明”。
TypedDict的作用（重点）：普通字典没有“类型限制”，比如上面的person字典，我们可以随便给"age"赋值为字符串"18"，也能赋值为数字18，容易出错。 TypedDict就是“给字典规定好键和对应的值的类型”，强制我们按照规定的格式赋值，避免出错。比如： from typing import TypedDict ``class Person(TypedDict): # 定义一个TypedDict类，继承自TypedDict `` name: str # 规定key="name"，值必须是字符串（str） `` age: int # 规定key="age"，值必须是整数（int）这样一来，我们创建Person类型的字典时，就必须符合这个规则，比如p1 = {"name": "小明", "age": 18}（正确），如果写p2 = {"name": "小红", "age": "18"}（错误，age是字符串），运行时就会提示错误，帮我们规避问题。

前置知识2：有向无环图（DAG）是什么（Graph的核心本质）

大白话解释：有向 = 有方向（比如从A到B，不能反过来从B到A，除非专门设置）；无环 = 不能绕圈（比如A→B→C→A，这种绕圈的结构不允许）；图 = 由“点”和“线”组成。

举个生活中的例子：做一道菜“番茄炒蛋”的流程就是一个有向无环图：准备食材（点A）→ 打鸡蛋（点B）→ 切番茄（点C）→ 炒鸡蛋（点D）→ 炒番茄（点E）→ 混合翻炒（点F）→ 出锅（点G）这里的“点”（A-G）就是后续要讲的“Node（节点）”，“箭头”（→）就是“Edge（边）”，流程有方向（只能按顺序来），不绕圈（不会炒完鸡蛋又回去准备食材），这就是DAG的核心特点。

LangGraph的Graph就是用这种“有向无环图”的结构，串联起多个处理步骤（Node），让数据按规定的流程一步步处理，最终得到结果。

前置知识3：LangGraph相关依赖安装与核心概念铺垫

依赖安装：LangGraph是一个Python库，就像我们用“计算器”需要先安装计算器软件一样，用LangGraph需要先安装它的依赖包，后续所有代码都需要先执行安装命令（复制到Python环境中运行即可）。
核心概念铺垫：提前简单记一下这3个词，后续会详细拆解，现在先有个印象：
1. State（状态）：相当于“全局共享的容器”，所有节点（Node）都能读取这个容器里的数据，也能修改容器里的数据（比如番茄炒蛋流程中，“食材”就是一个State，准备食材节点往里面放番茄、鸡蛋，后续节点从里面拿食材来处理）。
2. Node（节点）：相当于“一个具体的处理步骤”（比如番茄炒蛋里的“打鸡蛋”“切番茄”），接收State里的数据，处理完后，把新的结果放回State。
3. Edge（边）：相当于“步骤之间的连接箭头”，规定数据从哪个Node流到下一个Node（比如“打鸡蛋”之后，数据流到“炒鸡蛋”节点）。

正文：深度理解LangGraph核心-Graph

在了解了LangGraph中如何构建Agent智能体之后，接下来就要进入LangGraph的重头戏——Graph了。Graph是LangGraph的核心，它以有向无环图（DAG）的方式来串联多个Agent，构建更复杂的Agent大模型应用，形成更复杂的工作流。并且提供了很多产品级的特性，保证这些应用可以更稳定高效的执行。

一、理解什么是Graph图

Graph是LangGraph的基本构建模块，它是一个有向无环图（DAG），用于描述任务之间的依赖关系（就像番茄炒蛋的流程，描述了“准备食材”依赖“无”，“炒鸡蛋”依赖“打鸡蛋”，以此类推）。

Graph主要包含三个基本的元素，我们结合“能懂的解释+代码案例”，逐一看明白：

核心元素1：State: 整个应用当中共享的一种数据结构

大白话：State就是“全局共享容器”，所有Node（处理步骤）都能“读”这个容器里的数据，也能“写”数据到这个容器里。比如我们后续的代码案例中，用户输入的内容、每个节点的处理结果，都会存在State里，供其他节点使用。

核心元素2：Node : 一个处理数据的节点

大白话：Node就是“具体的处理步骤”，本质是一个Python函数，这个函数的“输入”是State（从容器里拿数据），经过一些操作（比如给字符串加后缀）后，“输出”是更新后的State（把处理结果放回容器）。

核心元素3：Edge : 表示Node之间的依赖关系

大白话：Edge就是“步骤之间的连接箭头”，本质也是一个Python函数（简单场景下可以不用写复杂函数，直接指定连接关系），作用是“根据当前State里的数据，决定下一个执行哪个Node”。比如判断State里的数字是否大于5，如果大于5，就执行Node1，否则就结束流程。

入门案例：最简化的Graph用法（逐行拆解，必看）

下面我们用一个最简单的案例，完整演示Graph的构建、运行流程，每一行代码都加详细注释，同时讲解“运行过程”和“中间结果”，确保你能看懂每一步在做什么。

第一步：安装依赖（必做）

首先需要安装LangGraph库，复制下面的代码，在Python环境（比如Jupyter、PyCharm）中运行，安装完成后才能使用后续的所有功能。

# 安装LangGraph依赖，-U表示更新到最新版本
!pip install -U langgraph

第二步：导入所需的工具（相当于“准备工具”）

# 从typing模块导入TypedDict（用来定义有类型限制的字典，也就是State的基础）
from typing import TypedDict
# 从langgraph.constants导入START和END（Graph的默认入口和出口，相当于番茄炒蛋的“准备食材”和“出锅”）
from langgraph.constants import END, START
# 从langgraph.graph导入StateGraph（用来构建Graph图的核心工具）
from langgraph.graph import StateGraph

第三步：定义State（创建“共享容器”，规定容器里能放什么数据）

# 定义InputState（输入状态）：用来接收用户输入的数据
# 继承自TypedDict，规定里面只有一个key：user_input，值的类型是字符串（str）
class InputState(TypedDict):
    user_input: str

# 定义OutputState（输出状态）：用来存储Graph最终的输出结果
# 规定里面只有一个key：graph_output，值的类型是字符串（str）
class OutputState(TypedDict):
    graph_output: str

# 定义OverallState（全局状态）：整个Graph中所有Node都能访问、修改的状态
# 里面包含3个key，分别存储不同的数据，供不同Node使用
class OverallState(TypedDict):
    foo: str       # 存储node_1的处理结果
    user_input: str# 存储用户输入（从InputState继承过来）
    graph_output: str# 存储最终输出（供OutputState使用）

# 定义PrivateState（私有状态）：临时存储某个Node的处理结果，供后续Node使用
class PrivateState(TypedDict):
    bar: str       # 存储node_2的处理结果

讲解：这里定义了4个State，本质都是“有类型限制的字典”，目的是“规范数据格式”，避免出错。比如InputState只能接收user_input（用户输入），不能接收其他无关数据；OverallState是全局共享的，所有Node都能读写，而PrivateState是临时的，只供node_2和node_3使用。

第四步：定义Node（创建“处理步骤”，每个Node都是一个函数）

# 定义node_1（第一个处理步骤）
# 输入参数：state: InputState → 表示这个Node接收InputState类型的数据（也就是用户输入的user_input）
# 返回值：OverallState → 表示这个Node处理完后，返回更新后的全局状态
def node_1(state: InputState) -> OverallState:
    # 处理逻辑：给用户输入的user_input加后缀“>学院”，赋值给OverallState的foo键
    # 比如用户输入“图灵”，这里就会得到“图灵>学院”
    return {"foo": state["user_input"] + ">学院"}

# 定义node_2（第二个处理步骤）
# 输入参数：state: OverallState → 接收全局状态的数据（里面有node_1处理后的foo）
# 返回值：PrivateState → 返回临时状态，存储当前Node的处理结果
def node_2(state: OverallState) -> PrivateState:
    # 处理逻辑：取出OverallState里的foo，再加后缀“>非常”，赋值给PrivateState的bar键
    # 比如foo是“图灵>学院”，这里就会得到“图灵>学院>非常”
    return {"bar": state["foo"] + ">非常"}

# 定义node_3（第三个处理步骤）
# 输入参数：state: PrivateState → 接收临时状态的数据（里面有node_2处理后的bar）
# 返回值：OutputState → 返回最终的输出状态，供Graph输出结果
def node_3(state: PrivateState) -> OutputState:
    # 处理逻辑：取出PrivateState里的bar，再加后缀“>靠谱”，赋值给OutputState的graph_output
    # 比如bar是“图灵>学院>非常”，这里就会得到“图灵>学院>非常>靠谱”
    return {"graph_output": state["bar"] + ">靠谱"}

讲解：每个Node都是一个函数，核心逻辑是“读取上一个状态的数据 → 处理数据 → 输出新的状态数据”。node_1处理用户输入，node_2处理node_1的结果，node_3处理node_2的结果，形成一个连贯的流程。

第五步：构建Graph（把Node和Edge串联起来，形成流程）

# 1. 创建Graph构建器（builder），相当于“画流程图的画板”
# 参数说明：
# OverallState：全局状态（所有Node共享的数据容器）
# input=InputState：Graph的输入类型（接收用户输入的数据格式）
# output=OutputState：Graph的输出类型（最终返回的结果格式）
builder = StateGraph(OverallState, input=InputState, output=OutputState)

# 2. 向构建器中添加Node（把“处理步骤”放到画板上）
# 语法：builder.add_node("节点名称", 节点函数)
# 节点名称是字符串，唯一即可；节点函数就是我们上面定义的node_1、node_2、node_3
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

# 3. 向构建器中添加Edge（给步骤之间画箭头，规定流程方向）
# 语法：builder.add_edge(当前节点, 下一个节点)
builder.add_edge(START, "node_1")  # 从默认入口（START）开始，先执行node_1
builder.add_edge("node_1", "node_2")  # node_1执行完，执行node_2
builder.add_edge("node_2", "node_3")  # node_2执行完，执行node_3
builder.add_edge("node_3", END)  # node_3执行完，到默认出口（END），流程结束

# 4. 编译Graph（把画板上的流程“固化”，变成可执行的对象）
# 编译后得到graph对象，后续就能调用这个对象来执行流程了
graph = builder.compile()

讲解：这一步是“构建流程图”的核心，我们把3个Node放到“画板”上，用Edge规定它们的执行顺序：START → node_1 → node_2 → node_3 → END，和番茄炒蛋的流程完全一致，有方向、不绕圈。

第六步：调用Graph（执行流程，得到结果）

# 调用graph的invoke方法，传入用户输入（符合InputState的格式：{"user_input": "用户输入内容"}）
# 这里我们传入{"user_input": "图灵"}，看看流程如何执行
result = graph.invoke({"user_input": "图灵"})

# 打印最终结果
print("Graph执行结果：", result)

运行过程拆解（重点看，搞懂每一步的变化）

我们一步步看“图灵”这个输入，如何经过3个Node的处理，最终得到结果：

第一步：用户调用graph.invoke({"user_input": "图灵"})，输入数据进入START入口，根据Edge的规定，先执行node_1。
第二步：node_1接收InputState（{"user_input": "图灵"}），执行处理逻辑："图灵" + ">学院" → "图灵>学院"，返回OverallState（{"foo": "图灵>学院", "user_input": "图灵", "graph_output": None}）（graph_output此时还没赋值，是None）。
第三步：根据Edge（node_1 → node_2），node_2接收OverallState（里面的foo是"图灵>学院"），执行处理逻辑："图灵>学院" + ">非常" → "图灵>学院>非常"，返回PrivateState（{"bar": "图灵>学院>非常"}）。
第四步：根据Edge（node_2 → node_3），node_3接收PrivateState（里面的bar是"图灵>学院>非常"），执行处理逻辑："图灵>学院>非常" + ">靠谱" → "图灵>学院>非常>靠谱"，返回OutputState（{"graph_output": "图灵>学院>非常>靠谱"}）。
第五步：根据Edge（node_3 → END），流程结束，graph返回最终的OutputState，也就是我们打印的result。

最终运行结果（控制台输出）： Graph执行结果： {'graph_output': '图灵>学院>非常>靠谱'}

完全符合我们的预期，这就是一个最简单的Graph执行流程！

补充：可视化Graph（直观看到流程结构）

如果觉得文字描述的流程不够直观，我们可以用代码把Graph的结构画出来，生成一张图片，就能清晰看到Node和Edge的连接关系。

# 从IPython.display导入Image和display（用来显示图片）
from IPython.display import Image, display
# draw_mermaid_png()方法会生成Graph的流程图图片（mermaid是一种绘图语法）
# 调用graph.get_graph().draw_mermaid_png()，得到图片的二进制数据
image_data = graph.get_graph().draw_mermaid_png()
# 显示图片
display(Image(image_data))

运行结果：会生成一张图片，图片中显示的流程是：START → node_1 → node_2 → node_3 → END，和我们构建的流程完全一致，非常直观。

小结：一个Graph中，可以通过对Node和Edge的灵活组合，形成各种复杂的流程。接下来，我们就是要接入Agent，来完成各种复杂的任务。在构建复杂任务之前，我们先来仔细看看Graph中的这三个主要组件（State、Node、Edge），以及其他高级特性。

二、详细拆解Graph的核心组件（友好版）

前面我们通过简单案例了解了State、Node、Edge的基本用法，接下来我们深入拆解每个组件的细节、注意事项，结合更具体的代码案例，让你彻底掌握。

1、State 状态（全局共享容器，重中之重）

再次强调：State是所有节点共享的状态，它本质是一个“有类型限制的字典”（也可以是其他类型，后面会讲），包含了所有节点需要用到的状态数据，所有Node都能读取、修改它里面的数据。

下面我们详细讲解State的4个关键注意点，每个点都配代码案例和通俗解释：

注意点1：State的两种实现方式（TypedDict和Pydantic BaseModel）

State形式上，有两种常用的实现方式，本质上没有太多区别，都是“规范数据格式”，可以根据自己的习惯选择。

# 方式1：TypedDict（我们前面案例用的方式，简单易懂，适合快速定义）
from typing import TypedDict

class OverallState(TypedDict):
    a: str  # 规定key="a"，值为字符串

# 方式2：Pydantic BaseModel（更强大，支持数据校验、默认值等，适合复杂场景）
from pydantic import BaseModel

# 定义一个继承自BaseModel的State类，里面的属性就是State的key
class OverallState(BaseModel):
    a: str  # 规定key="a"，值为字符串（和TypedDict一样）
    # 还可以设置默认值，比如：b: int = 10（如果不赋值，默认是10）

讲解：两种方式都能实现“规范State格式”的目的，入门建议先用TypedDict，简单易上手；后续如果需要更复杂的功能（比如数据校验），再用Pydantic BaseModel。

注意点2：State的属性可以设置默认值（通过Node实现）

State中定义的属性，通常不需要指定默认值（比如前面的a: str），如果需要设置默认值，不能直接在State类中写（比如class OverallState(TypedDict): a: str = "hello"，这样是错误的），而是要通过“在START节点后，定义一个Node来设置默认值”。

from typing import TypedDict
from langgraph.graph import StateGraph
from langgraph.constants import START, END

# 定义State，a属性没有默认值
class OverallState(TypedDict):
    a: str

# 定义一个Node，用来给State的a属性设置默认值
def set_default(state: OverallState):
    # 如果State中没有a的值（或者a的值为None），就设置为"goodbye"
    if "a" not in state or state["a"] is None:
        return {"a": "goodbye"}
    # 如果已经有值，就返回原State（不修改）
    return state

# 构建Graph
builder = StateGraph(OverallState)
# 添加节点：先执行set_default（设置默认值），再结束
builder.add_node("set_default", set_default)
builder.add_edge(START, "set_default")
builder.add_edge("set_default", END)

# 编译并调用Graph（不传入a的值，测试默认值）
graph = builder.compile()
result = graph.invoke({})  # 传入空字典，没有a的值
print("默认值测试结果：", result)  # 输出：{'a': 'goodbye'}

运行过程：调用graph.invoke({})时，State中没有a的值，执行set_default节点，给a设置默认值"goodbye"，最终返回{"a": "goodbye"}。如果我们传入{"a": "hello"}，则set_default节点会返回原State，最终结果就是{"a": "hello"}。

注意点3：State的属性可以定义“更新规则”（重点，高频使用）

默认情况下，Node返回的State数据会“覆盖”原有的数据（比如原State的extra_field是5，Node返回{"extra_field": 10}，则extra_field会变成10）。但有时候我们需要“追加”数据（比如聊天记录，每次新增消息都要加到原有消息列表中），这时候就需要给State的属性定义“更新规则”。

LangGraph提供了两个常用的更新规则：add_messages（用于聊天消息列表）、add（用于普通列表），我们结合代码案例讲解：

# 导入所需工具
from langchain_core.messages import AnyMessage, AIMessage, HumanMessage
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages  # 用于消息列表的追加更新
from typing import Annotated, TypedDict
from operator import add  # 用于普通列表的追加更新

# 定义State，给不同属性设置更新规则
class State(TypedDict):
    # 1. messages：聊天消息列表，用add_messages规则，新增消息会追加到原有列表中
    # Annotated用来给属性添加“额外规则”，第一个参数是属性类型，第二个是更新规则
    messages: Annotated[list[AnyMessage], add_messages]
    # 2. list_field：普通整数列表，用add规则，新增列表会和原有列表合并（追加）
    list_field: Annotated[list[int], add]
    # 3. extra_field：普通整数，没有更新规则，默认是“覆盖”
    extra_field: int

# 定义node1：返回更新后的数据
def node1(state: State):
    # 新增一条AI消息（AIMessage是聊天消息的一种类型）
    new_message = AIMessage("Hello!")
    # 返回更新的数据：
    # messages：新增一条消息（会追加到原有messages列表）
    # list_field：新增[10]（会和原有list_field合并）
    # extra_field：设置为10（会覆盖原有值）
    return {"messages": [new_message], "list_field": [10], "extra_field": 10}

# 定义node2：继续返回更新后的数据
def node2(state: State):
    new_message = AIMessage("LangGraph!")
    return {"messages": [new_message], "list_field": [20], "extra_field": 20}

# 构建Graph，设置执行流程：START → node1 → node2 → END
graph = (StateGraph(State)
         .add_node("node1", node1)
         .add_node("node2", node2)
         .set_entry_point("node1")  # 等价于add_edge(START, "node1")，设置入口节点
         .add_edge("node1", "node2")
         .add_edge("node2", END)
         .compile())

# 调用Graph，传入初始数据
input_data = {
    "messages": [HumanMessage(content="Hi")],  # 初始消息：用户输入"Hi"
    "list_field": [1, 2, 3],  # 初始列表：[1,2,3]
    "extra_field": 5  # 初始值：5
}
result = graph.invoke(input_data)

# 打印最终结果，看看每个属性的变化
print("最终messages（追加后）：")
for message in result["messages"]:
    print(message.content)  # 输出：Hi、Hello!、LangGraph!（三条消息，追加成功）
print("最终list_field（合并后）：", result["list_field"])  # 输出：[1,2,3,10,20]（合并成功）
print("最终extra_field（覆盖后）：", result["extra_field"])  # 输出：20（被node2覆盖）

运行过程拆解（重点看更新规则的效果）：

初始input_data：messages=[Hi]，list_field=[1,2,3]，extra_field=5。
执行node1：返回messages=[Hello!]（追加）、list_field=[10]（合并）、extra_field=10（覆盖），此时State变为：
1. messages：[Hi, Hello!]
2. list_field：[1,2,3,10]
3. extra_field：10
执行node2：返回messages=[LangGraph!]（追加）、list_field=[20]（合并）、extra_field=20（覆盖），此时State变为：
1. messages：[Hi, Hello!, LangGraph!]
2. list_field：[1,2,3,10,20]
3. extra_field：20

小结：add_messages和add规则的核心是“追加/合并”，而没有规则的属性是“覆盖”，这在实际开发中非常常用（比如聊天机器人的消息记录，就需要用add_messages规则）。

注意点4：LangGraph提供的快捷State——MessagesState（简化聊天消息存储）

在LangGraph的应用中，State通常都需要保存聊天消息（比如用户消息、AI消息），为了简化开发，LangGraph专门提供了一个langgraph.graph.MessagesState，可以快速定义“用于存储聊天消息的State”，不用我们自己写TypedDict。

# 导入MessagesState
from langgraph.graph import MessagesState
from langchain_core.messages import HumanMessage

# 方式1：直接使用MessagesState（默认只有一个messages属性，带add_messages规则）
# 等价于：class MessagesState(TypedDict): messages: Annotated[list[AnyMessage], add_messages]
state1 = MessagesState(messages=[HumanMessage(content="Hello LangGraph")])
print("MessagesState默认用法：", state1)

# 方式2：也可以对Messages进行序列化声明（两种写法都可以，效果一样）
# 写法1：用Message对象（HumanMessage、AIMessage等）
state2 = {"messages": [HumanMessage(content="message")]}
# 写法2：用字典格式（type指定消息类型，content指定消息内容）
state3 = {"messages": [{"type": "user", "content": "message"}]}  # type="user"对应HumanMessage
print("序列化写法1：", state2)
print("序列化写法2：", state3)

运行结果：三种写法的效果完全一致，都是创建了一个包含聊天消息的State，简化了我们定义State的代码，后续开发聊天相关的Graph时，直接用MessagesState即可。

2、Node 节点（处理步骤，核心执行单元）

Node是图中的“处理步骤”，本质是一个Python函数，核心作用是“接收State输入 → 处理数据 → 返回更新后的State输出”。下面我们详细讲解Node的5个关键注意点，结合代码案例和运行过程：

注意点1：Node的基本格式（必记）

在LangGraph中，Node通常是一个Python函数，它有两个参数（第二个可选），返回一个State对象：

from typing import TypedDict
from langchain_core.runnables import RunnableConfig  # 用于Node的配置参数

# 定义State
class State(TypedDict):
    number: int

# Node的基本格式（两个参数）
def my_node(
    state: State,  # 第一个参数：State对象，必选，接收当前的状态数据
    config: RunnableConfig = None  # 第二个参数：配置项，可选，包含节点运行的配置参数
) -> State:  # 返回值：更新后的State对象
    # 处理逻辑：给number加1
    return {"number": state["number"] + 1}

讲解：第一个参数state是必选的，所有Node都必须接收State作为输入；第二个参数config是可选的，用于传递节点运行的配置（比如用户ID、超时时间等），后面会详细讲解。

注意点2：Node的名称（唯一标识）

每个Node都有一个唯一的名称，通常是一个字符串，有两种设置方式：

手动设置：添加Node时，第一个参数就是节点名称，比如builder.add_node("node1", node_1)，节点名称是"node1"。
自动生成：如果没有手动设置名称，LangGraph会自动生成一个和函数名一样的名称，比如builder.add_node(node_1)，节点名称就是"node_1"。

注意：节点名称必须唯一，不能重复（比如不能有两个叫"node1"的节点），否则会报错。

注意点3：Node的缓存机制（提升运行速度，重点）

LangGraph对每个Node提供了缓存机制：只要Node的传入参数（state和config）相同，LangGraph就会优先从缓存中获取Node的执行结果，不用重新执行Node函数，从而提升运行速度（比如一个Node执行需要3秒，缓存后再次调用相同参数，瞬间就能拿到结果）。

下面用代码案例演示缓存机制，包含详细注释和运行过程：

import time  # 用于测试运行时间
from typing import TypedDict
from langchain_core.runnables import RunnableConfig
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.types import CachePolicy  # 用于设置缓存策略
from langgraph.cache.memory import InMemoryCache  # 内存缓存（临时缓存，重启程序后消失）

# 1. 定义State（存储数据）
class State(TypedDict):
    number: int  # 要处理的数字
    user_id: str  # 用户ID（用于区分不同用户的缓存）

# 2. 定义Node配置Schema（用于接收config参数）
class ConfigSchema(TypedDict):
    user_id: str  # 配置项：用户ID，从config中获取

# 3. 定义Node（执行需要3秒，方便测试缓存效果）
def node_1(state: State, config: RunnableConfig):
    time.sleep(3)  # 模拟Node执行耗时3秒
    # 从config中获取用户ID（config的结构是：{"configurable": {"user_id": "xxx"}}）
    user_id = config["configurable"]["user_id"]
    # 返回更新后的State：number加1，user_id设为从config中获取的值
    return {"number": state["number"] + 1, "user_id": user_id}

# 4. 构建Graph，设置缓存策略
builder = StateGraph(State, config_schema=ConfigSchema)  # 传入配置Schema
# 添加Node，设置缓存策略：CachePolicy(ttl=5) → 缓存有效期5秒（5秒内相同参数会走缓存）
builder.add_node("node1", node_1, cache_policy=CachePolicy(ttl=5))
# 设置流程：START → node1 → END
builder.add_edge(START, "node1")
builder.add_edge("node1", END)
# 编译Graph，指定缓存方式为InMemoryCache（内存缓存）
graph = builder.compile(cache=InMemoryCache())

# 5. 第一次调用Graph（无缓存，会执行node_1，耗时3秒）
print("第一次调用（无缓存）：")
start_time = time.time()  # 记录开始时间
result1 = graph.invoke(
    {"number": 5},  # 传入state
    config={"configurable": {"user_id": "123"}}  # 传入config（用户ID=123）
)
end_time = time.time()  # 记录结束时间
print("结果：", result1)  # 输出：{'number': 6, 'user_id': '123'}
print("耗时：", round(end_time - start_time, 1), "秒")  # 耗时约3.0秒

# 6. 第二次调用Graph（参数和第一次相同，走缓存，耗时接近0秒）
print("\n第二次调用（有缓存）：")
start_time = time.time()
result2 = graph.invoke(
    {"number": 5},  # 相同的state
    config={"configurable": {"user_id": "123"}}  # 相同的config
)
end_time = time.time()
print("结果：", result2)  # 输出：{'number': 6, 'user_id': '123'}（和第一次一样）
print("耗时：", round(end_time - start_time, 1), "秒")  # 耗时约0.0秒
print("缓存标识：", result2.get("__metadata__"))  # 输出：{'cached': True}（表示走了缓存）

# 7. 第三次调用（config不同，不缓存，耗时3秒）
print("\n第三次调用（config不同，无缓存）：")
start_time = time.time()
result3 = graph.invoke(
    {"number": 5},  # 相同的state
    config={"configurable": {"user_id": "456"}}  # 不同的config（用户ID=456）
)
end_time = time.time()
print("结果：", result3)  # 输出：{'number': 6, 'user_id': '456'}
print("耗时：", round(end_time - start_time, 1), "秒")  # 耗时约3.0秒

运行过程讲解：

第一次调用：state={"number":5}，config={"user_id":"123"}，无缓存，执行node_1（耗时3秒），返回结果，同时将结果缓存5秒。
第二次调用：state和config和第一次完全相同，且缓存未过期（5秒内），直接从缓存中获取结果，耗时接近0秒，结果中会有metadata: {"cached": True}，标识这是缓存结果。
第三次调用：state相同，但config不同（user_id=456），缓存不生效，重新执行node_1（耗时3秒），返回新的结果（user_id=456）。

小结：缓存机制的核心是“参数相同则复用结果”，适合那些“执行耗时久、参数重复率高”的Node，能大幅提升Graph的运行效率。

注意点4：Node的重试机制（应对执行失败，实用特性）

在实际开发中，Node可能会执行失败（比如网络问题、接口报错），LangGraph提供了重试机制，可以设置“失败后自动重试”，避免一次失败就导致整个Graph流程终止。

重试机制有两种设置方式，结合代码案例讲解：

from typing import TypedDict
from langgraph.graph import StateGraph
from langgraph.types import RetryPolicy  # 用于设置重试策略

# 定义State
class State(TypedDict):
    number: int

# 定义一个可能会失败的Node（模拟失败：当number=5时，抛出异常）
def node_1(state: State):
    if state["number"] == 5:
        raise Exception("模拟Node执行失败！")  # 抛出异常，模拟失败
    return {"number": state["number"] + 1}

# 构建Graph
builder = StateGraph(State)

# 方式1：针对单个Node指定重试策略（推荐）
# RetryPolicy(max_attempts=4) → 最多重试4次（加上第一次执行，总共尝试5次）
builder.add_node("node1", node_1, retry=RetryPolicy(max_attempts=4))

# 方式2：针对某一次Graph调用指定重试策略（全局生效）
# 后续调用graph.invoke时，通过config设置：config={"recursion_limit":25}（递归限制，也可控制重试）

# 设置流程
builder.add_edge(START, "node1")
builder.add_edge("node1", END)
graph = builder.compile()

# 测试重试机制（传入number=5，模拟失败）
try:
    result = graph.invoke({"number": 5})
except Exception as e:
    print("最终执行失败：", str(e))
    # 此时会尝试5次（1次执行+4次重试），都失败后才抛出异常

讲解：RetryPolicy(max_attempts=4)表示“最多重试4次”，如果Node执行失败，会自动重新执行，直到执行成功，或者达到重试次数上限。这种机制在对接外部接口、大模型调用时非常实用，能提升Graph的稳定性。

3、Edge 边（连接节点，控制流程方向）

Edge是Graph中“连接Node的箭头”，核心作用是“控制State的流动方向”，决定“当前Node执行完后，下一个执行哪个Node”。LangGraph提供了多种灵活的Edge构建方式，我们逐一看懂，结合代码案例：

方式1：普通Edge和EntryPoint（最基础，固定流程）

普通Edge就是“固定连接两个Node”，形成固定的流程（比如A→B→C），同时LangGraph提供了两个默认的Node：START（入口）和END（出口），用来作为Graph的起始和结束节点。

另外，我们也可以用set_entry_point和set_finish_point，手动指定Graph的入口和出口节点（等价于add_edge(START, 入口节点)和add_edge(出口节点, END)）。

from typing import TypedDict
from langgraph.graph import StateGraph
from langgraph.constants import START, END

class State(TypedDict):
    number: int

# 定义两个Node
def node1(state: State):
    return {"number": state["number"] + 1}

def node2(state: State):
    return {"number": state["number"] * 2}

# 构建Graph
builder = StateGraph(State)
builder.add_node("node1", node1)
builder.add_node("node2", node2)

# 方式1：普通Edge（固定流程）
builder.add_edge(START, "node1")  # 入口→node1
builder.add_edge("node1", "node2")  # node1→node2
builder.add_edge("node2", END)  # node2→出口

# 方式2：手动指定入口和出口（等价于上面的3条add_edge）
# builder.set_entry_point("node1")  # 指定入口节点为node1（替代add_edge(START, "node1")）
# builder.set_finish_point("node2")  # 指定出口节点为node2（替代add_edge("node2", END)）

graph = builder.compile()
result = graph.invoke({"number": 2})
print("普通Edge执行结果：", result)  # 流程：2→node1（3）→node2（6）→输出6

运行结果：{'number': 6}，流程固定为START→node1→node2→END，无论输入什么数据，都会按这个流程执行。

方式2：条件Edge和EntryPoint（动态流程，根据条件判断）

普通Edge是“固定流程”，而条件Edge是“动态流程”：根据当前State里的数据，判断下一个执行哪个Node（比如“如果数字大于5，执行node1；否则，结束流程”）。

实现方式：定义一个“路由函数”，函数的返回值就是“下一个要执行的Node名称”，然后用add_conditional_edges添加条件边。

from typing import TypedDict
from langchain_core.runnables import RunnableConfig
from langgraph.constants import START, END
from langgraph.graph import StateGraph

# 1. 定义State
class State(TypedDict):
    number: int

# 2. 定义Node（给number加1）
def node_1(state: State, config: RunnableConfig):
    return {"number": state["number"] + 1}

# 3. 定义路由函数（条件判断逻辑）
# 输入：当前State
# 输出：下一个要执行的Node名称（字符串），也可以返回END表示流程结束
def routing_func(state: State) -> str:
    # 条件判断：如果当前number < 5，继续执行node_1（循环加1）；否则，结束流程
    if state["number"] < 5:
        return "node_1"  # 返回下一个要执行的Node名称
    else:
        return END  # 返回END，流程终止

# 4. 构建Graph，添加条件Edge
builder = StateGraph(State)
builder.add_node("node_1", node_1)

# 添加条件Edge：当前节点是START，路由函数是routing_func，候选节点是["node_1", END]
# 语法：builder.add_conditional_edges(当前节点, 路由函数, 候选节点列表)
builder.add_conditional_edges(
    START,  # 从START入口开始，执行路由函数判断
    routing_func,  # 用于判断下一个节点的路由函数
    {"node_1": "node_1", END: END}  # 候选节点映射：路由函数返回值 → 实际节点
)

# 也可以简化候选节点映射（当路由函数返回值和节点名称一致时）
# builder.add_conditional_edges(START, routing_func, ["node_1", END])

graph = builder.compile()

# 测试条件Edge（传入初始number=2，看看流程如何执行）
result = graph.invoke({"number": 2})
print("条件Edge执行结果：", result)  # 最终输出：{'number': 5}

运行过程拆解（重点理解动态流程）：

初始输入：state={"number":2}，流程从START开始，执行路由函数routing_func。
第一次判断：2 < 5，路由函数返回"node_1"，执行node_1，number变为3。
node_1执行完后，会再次触发路由函数（条件Edge会在节点执行完后重新判断），第二次判断：3 < 5，继续执行node_1，number变为4。
第三次判断：4 < 5，继续执行node_1，number变为5。
第四次判断：5 不小于5，路由函数返回END，流程终止，返回最终state={"number":5}。

小结：条件Edge的核心是“路由函数”，通过判断State中的数据，动态决定流程走向，适合需要“分支逻辑”“循环逻辑”的场景（比如上面的循环加1，直到满足条件为止）。

方式3：多条条件分支（更复杂的动态流程）

实际开发中，流程可能有多个分支（比如“数字小于3，执行node1；大于等于3且小于6，执行node2；大于等于6，结束流程”），这时候可以通过路由函数返回不同的节点名称，实现多分支逻辑。

from typing import TypedDict
from langgraph.graph import StateGraph
from langgraph.constants import START, END

# 1. 定义State
class State(TypedDict):
    number: int

# 2. 定义两个Node，处理不同逻辑
def node_1(state: State):
    # 逻辑1：给number加2
    return {"number": state["number"] + 2}

def node_2(state: State):
    # 逻辑2：给number乘2
    return {"number": state["number"] * 2}

# 3. 定义多分支路由函数
def multi_routing(state: State) -> str:
    if state["number"] < 3:
        return "node_1"  # 分支1：执行node_1
    elif 3 <= state["number"] < 6:
        return "node_2"  # 分支2：执行node_2
    else:
        return END  # 分支3：结束流程

# 4. 构建Graph，添加多分支条件Edge
builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)

# 添加条件Edge，候选节点包含node_1、node_2、END
builder.add_conditional_edges(
    START,
    multi_routing,
    ["node_1", "node_2", END]  # 候选节点列表，路由返回值对应节点名称
)

# 给node_1和node_2添加结束Edge（执行完后直接结束流程）
builder.add_edge("node_1", END)
builder.add_edge("node_2", END)

graph = builder.compile()

# 测试不同分支
print("分支1测试（number=2）：", graph.invoke({"number": 2}))  # 2→node1（4）→END，输出{'number': 4}
print("分支2测试（number=4）：", graph.invoke({"number": 4}))  # 4→node2（8）→END，输出{'number': 8}
print("分支3测试（number=6）：", graph.invoke({"number": 6}))  # 6≥6，直接END，输出{'number': 6}

运行过程拆解（分三个分支说明）：

分支1（number=2）：START→路由函数判断（2<3）→返回"node_1"→执行node_1（2+2=4）→node_1→END，最终结果{'number':4}。
分支2（number=4）：START→路由函数判断（3≤4<6）→返回"node_2"→执行node_2（4×2=8）→node_2→END，最终结果{'number':8}。
分支3（number=6）：START→路由函数判断（6≥6）→返回END→流程直接终止，返回初始number=6。

小结：多分支条件Edge的核心是“路由函数的多条件判断”，路由函数返回不同的节点名称，就能实现不同的流程分支，适合复杂的业务逻辑（比如“用户提问是事实类→执行查询节点，是创作类→执行生成节点，是闲聊类→执行对话节点”）。

方式4：并行Edge（同时执行多个Node，提升效率）

前面的Edge都是“串行执行”（一个Node执行完，再执行下一个），但实际开发中，有些Node之间没有依赖关系（比如“查询天气”和“查询新闻”，不需要先查天气再查新闻），这时候可以用并行Edge，让多个Node同时执行，大幅提升流程效率。

实现方式：使用builder.add_parallel_edges，指定“起始节点”和“多个并行节点”，这些并行节点会同时执行，全部执行完后，再进入下一个节点。

from typing import TypedDict
import time
from langgraph.graph import StateGraph
from langgraph.constants import START, END

# 1. 定义State（存储两个并行Node的结果）
class State(TypedDict):
    weather: str  # 存储查询天气的结果
    news: str     # 存储查询新闻的结果

# 2. 定义两个并行Node（模拟耗时操作，方便看出并行效果）
def get_weather(state: State):
    time.sleep(3)  # 模拟查询天气耗时3秒
    return {"weather": "今日晴，25℃，微风"}

def get_news(state: State):
    time.sleep(3)  # 模拟查询新闻耗时3秒
    return {"news": "今日热点：LangGraph新增并行Edge特性"}

# 3. 定义汇总Node（接收两个并行Node的结果，进行汇总）
def summary(state: State):
    # 汇总天气和新闻结果
    return {"weather": state["weather"], "news": state["news"], "summary": f"天气：{state["weather"]}；新闻：{state["news"]}"}

# 4. 构建Graph，添加并行Edge
builder = StateGraph(State)
# 添加三个Node：两个并行Node，一个汇总Node
builder.add_node("get_weather", get_weather)
builder.add_node("get_news", get_news)
builder.add_node("summary", summary)

# 添加并行Edge：从START开始，同时执行get_weather和get_news
builder.add_parallel_edges(
    START,  # 起始节点
    ["get_weather", "get_news"]  # 并行执行的Node列表
)

# 两个并行Node执行完后，都流向summary节点（汇总结果）
builder.add_edge("get_weather", "summary")
builder.add_edge("get_news", "summary")
# 汇总完后，流程结束
builder.add_edge("summary", END)

graph = builder.compile()

# 测试并行Edge（统计总耗时，看并行效果）
start_time = time.time()
result = graph.invoke({})  # 初始State为空，两个并行Node不需要输入
end_time = time.time()

print("并行Edge执行结果：", result)
print("总耗时：", round(end_time - start_time, 1), "秒")  # 耗时约3.0秒（两个Node同时执行，取最长耗时）

关键说明（必看）：

并行效果：两个Node各自耗时3秒，如果串行执行，总耗时会是6秒；并行执行后，总耗时约3秒，相当于“同时做两件事”，效率翻倍。
执行顺序：并行Node的执行顺序不固定（可能先执行get_weather，也可能先执行get_news），但都会在执行完后，统一流向summary节点。
适用场景：多个Node之间没有依赖关系（不需要一个Node的结果作为另一个Node的输入），比如“同时查询多个接口”“同时处理多个独立任务”。

小结：并行Edge是提升Graph运行效率的关键特性，核心是“无依赖节点同时执行”，适合需要处理多个独立任务的场景。

Edge小结（必记）

我们已经讲解了4种常用的Edge方式，整理成表格，方便大家快速区分和使用：

Edge方式	核心作用	适用场景	关键语法
普通Edge	固定流程，一个Node执行完→下一个Node	流程固定、无分支（如A→B→C）	builder.add_edge(当前节点, 下一个节点)
条件Edge	动态流程，根据State判断下一个Node	有分支、循环逻辑（如数字<5继续执行，否则结束）	builder.add_conditional_edges(当前节点, 路由函数, 候选节点)
多分支Edge	多条件判断，对应多个流程分支	复杂分支逻辑（如3个及以上分支）	同条件Edge，路由函数返回多个节点名称
并行Edge	多个无依赖Node同时执行，提升效率	多个独立任务（如同时查询天气、新闻）	builder.add_parallel_edges(起始节点, 并行Node列表)

三、Graph的高级特性（入门必备，提升开发效率）

掌握了State、Node、Edge的核心用法后，我们再学习几个Graph的高级特性，这些特性能帮我们解决实际开发中的复杂问题，让Graph更稳定、更灵活。

1、递归限制（防止流程无限循环）

在使用条件Edge实现循环逻辑时（比如前面的“循环加1直到number≥5”），如果路由函数写得有问题（比如把判断条件写成number>5，而初始number=10，会直接结束；但如果写成number<100，且没有终止条件，会无限循环），会导致Graph陷入无限执行，最终报错。

LangGraph提供了递归限制特性，可以设置“Graph最大执行步数”，超过步数就自动终止，避免无限循环。

from typing import TypedDict
from langgraph.graph import StateGraph
from langgraph.constants import START, END

class State(TypedDict):
    number: int

# 定义Node（加1）
def node_1(state: State):
    return {"number": state["number"] + 1}

# 有问题的路由函数（没有终止条件，会无限循环）
def bad_routing(state: State) -> str:
    return "node_1"  # 无论number是多少，都返回node_1，无限循环

# 构建Graph，设置递归限制
builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_conditional_edges(START, bad_routing, ["node_1"])

# 编译Graph时，设置递归限制：max_steps=10（最多执行10步，超过则终止）
graph = builder.compile(max_steps=10)

# 测试递归限制
try:
    result = graph.invoke({"number": 1})
    print("执行结果：", result)
except Exception as e:
    print("执行终止（触发递归限制）：", str(e))
    # 输出：执行终止（触发递归限制）：Recursion limit exceeded (max steps: 10)

讲解：max_steps=10表示Graph最多执行10步（每执行一个Node算一步），超过10步就会抛出异常，终止流程，避免无限循环占用资源。实际开发中，建议根据业务逻辑设置合理的max_steps。

2、Graph的中断与恢复（实用特性，应对复杂场景）

实际开发中，有些流程可能需要“中断”（比如需要用户输入确认、需要等待外部接口返回），中断后，后续可以“恢复”流程，继续执行剩下的步骤，LangGraph支持这种中断与恢复的特性。

核心原理：Graph执行时，会生成一个“检查点”（checkpoint），记录当前的State和执行进度；中断后，通过这个检查点，就能恢复到中断前的状态，继续执行。

from typing import TypedDict
from langgraph.graph import StateGraph
from langgraph.constants import START, END

# 定义State
class State(TypedDict):
    number: int
    need_confirm: bool  # 标记是否需要用户确认（用于中断）

# 定义Node1：判断是否需要中断（需要用户确认）
def node_1(state: State):
    if state["number"] < 3:
        # number<3，需要用户确认，设置need_confirm=True（触发中断）
        return {"need_confirm": True}
    else:
        return {"need_confirm": False}

# 定义Node2：继续执行的逻辑（加1）
def node_2(state: State):
    return {"number": state["number"] + 1, "need_confirm": False}

# 定义路由函数（判断是否中断）
def routing(state: State) -> str:
    if state["need_confirm"]:
        # 返回None，表示中断流程，等待外部触发恢复
        return None
    else:
        return END

# 构建Graph
builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", routing, ["node_2", END])
builder.add_edge("node_2", END)
graph = builder.compile()

# 第一次执行：触发中断
print("第一次执行（触发中断）：")
iterator = graph.stream({"number": 2})  # 用stream方法，获取执行迭代器
for step in iterator:
    print("当前步骤：", step)
    # 保存检查点（用于恢复）
    checkpoint = iterator.get_checkpoint()
    print("中断检查点：", checkpoint)
    break  # 触发中断后，停止执行

# 恢复流程（用户确认后，继续执行）
print("\n恢复流程：")
# 传入检查点，继续执行剩下的流程
result = graph.stream({"number": 2, "need_confirm": False}, checkpoint=checkpoint)
for step in result:
    print("恢复后步骤：", step)
print("最终结果：", step)  # 输出：{'number': 3, 'need_confirm': False}

讲解：中断与恢复的核心是“检查点”，当路由函数返回None时，流程中断，保存检查点；后续通过传入检查点，就能恢复到中断前的状态，继续执行流程。适合需要用户交互、等待外部事件的场景（比如“生成文案后，需要用户确认，确认后再执行下一步”）。

3、Graph的序列化（保存与加载，复用流程）

我们构建的Graph，可以序列化为JSON格式保存到本地，后续需要使用时，直接加载JSON文件，不用重新写代码构建Graph，大幅提升开发效率（比如多个项目复用同一个流程）。

from typing import TypedDict
from langgraph.graph import StateGraph
from langgraph.constants import START, END
import json

# 1. 构建一个简单的Graph
class State(TypedDict):
    number: int

def node_1(state: State):
    return {"number": state["number"] + 1}

builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", END)
graph = builder.compile()

# 2. 序列化Graph（保存为JSON文件）
graph_json = graph.to_json()  # 转为JSON字符串
with open("my_graph.json", "w", encoding="utf-8") as f:
    json.dump(graph_json, f, ensure_ascii=False, indent=4)
print("Graph已序列化保存到my_graph.json")

# 3. 加载序列化的Graph（从JSON文件加载）
from langgraph.graph import load_graph_from_json

with open("my_graph.json", "r", encoding="utf-8") as f:
    loaded_graph_json = json.load(f)

# 加载Graph，需要传入State类和Node函数（因为JSON只保存结构，不保存函数逻辑）
loaded_graph = load_graph_from_json(
    loaded_graph_json,
    state=State,
    nodes={"node_1": node_1}  # 映射节点名称和对应的函数
)

# 测试加载后的Graph
result = loaded_graph.invoke({"number": 5})
print("加载后的Graph执行结果：", result)  # 输出：{'number': 6}

关键说明：序列化保存的是Graph的“结构”（Node、Edge的连接关系），而Node的函数逻辑需要单独保存，加载时再传入，这样既能复用流程结构，又能灵活修改Node的逻辑。

四、子图（Graph的复用与嵌套，构建复杂工作流）

在LangGraph中，一个Graph不仅可以单独使用，还能作为一个“大型Node”，嵌入到另一个Graph（父图）中，这种嵌套使用的Graph就称为「子图」。子图的核心价值是「复用流程」，尤其适合大型项目或多Agent协作场景——比如一个团队开发好单个Agent的Graph流程后，其他团队可直接将其作为子图，整合到更复杂的整体工作流中，无需重复开发。

子图的使用方式和普通Node几乎一致，唯一需要注意的是：当父图执行到子图对应的Node时，本质上是调用了一次subgraph.invoke(state)，子图会接收父图的State，执行自身流程后，将更新后的State返回给父图，继续执行父图后续步骤。

子图实战案例（必看，逐行拆解）

下面通过完整代码案例，演示子图的定义、嵌套和执行过程，结合注释和运行结果，搞懂子图的核心用法：

from operator import add
from typing import TypedDict, Annotated
from langgraph.constants import END, START
from langgraph.graph import StateGraph

# 1. 定义全局State（子图和父图使用相同的State，确保数据互通）
class State(TypedDict):
    # messages为列表，使用add规则，新增内容会追加到原有列表
    messages: Annotated[list[str], add]

# 2. 定义子图（SubGraph）：作为父图中的一个Node使用
# 子图的Node：处理逻辑，返回更新后的State
def sub_node_1(state: State) -> State:
    # 子图节点的逻辑：向messages中添加一条子图的响应信息
    return {"messages": ["response from subgraph"]}

# 构建子图（和普通Graph构建方式完全一致）
subgraph_builder = StateGraph(State)
subgraph_builder.add_node("sub_node_1", sub_node_1)  # 给子图添加Node
subgraph_builder.add_edge(START, "sub_node_1")       # 子图入口→sub_node_1
subgraph_builder.add_edge("sub_node_1", END)         # sub_node_1→子图出口
subgraph = subgraph_builder.compile()                # 编译子图，得到可执行对象

# 3. 定义父图（Parent Graph）：将子图作为一个Node嵌入
builder = StateGraph(State)
# 关键步骤：将子图subgraph作为一个Node添加到父图中，节点名称为"subgraph_node"
builder.add_node("subgraph_node", subgraph)
# 父图流程：入口→子图节点→出口
builder.add_edge(START, "subgraph_node")
builder.add_edge("subgraph_node", END)
graph = builder.compile()  # 编译父图

# 4. 调用父图，观察执行结果
result = graph.invoke({"messages": ["hello subgraph"]})
print("父图执行结果：", result)
# 输出：{'messages': ['hello subgraph', 'hello subgraph', 'response from subgraph']}

结果讲解（重点理解）：

为什么"hello subgraph"会出现两次？核心原因是「两次调用invoke」：

父图调用graph.invoke({"messages": ["hello subgraph"]})，将初始State传入父图，此时messages为["hello subgraph"]。
父图执行到"subgraph_node"（子图节点）时，会自动调用subgraph.invoke(state)，将父图的State传入子图。
由于State的messages属性使用了add更新规则，子图接收State时，会先将原有messages（["hello subgraph"]）追加一次（相当于子图的初始State），再执行sub_node_1，添加"response from subgraph"。
子图执行完毕后，将更新后的State返回给父图，最终形成三次追加的结果。

小结：子图的核心是「流程复用」，使用时只需确保子图和父图的State格式一致（数据互通），嵌入方式和普通Node完全相同，大幅降低复杂流程的开发成本。

五、图的Stream支持（分步查看执行过程，调试更高效）

和大模型调用类似，Graph除了用invoke()方法直接获取最终结果，还支持stream()（同步流式调用）和astream()（异步流式调用）方法。两者的核心区别的是：

大模型的流式调用：依次返回模型响应的Token（逐字/逐句输出）。
Graph的流式调用：依次返回State的「数据处理步骤」，能清晰看到每一步Node的执行结果，方便调试和跟踪流程。

1、Stream基础用法（同步流式调用）

使用graph.stream()方法，配合stream_mode参数，可指定流式输出的内容格式，最常用的是debug模式（查看详细执行过程）。

# 沿用上面子图的案例，演示Stream用法
# 流式调用父图，stream_mode="debug"（输出最详细的执行信息）
for chunk in graph.stream({"messages": ["hello subgraph"]}, stream_mode="debug"):
    print("流式输出：", chunk)
    print("-" * 50)

# 部分输出结果（关键信息）：
# 流式输出： {'subgraph_node': {'messages': ['hello subgraph', 'response from subgraph']}}
# --------------------------------------------------
# 流式输出： {'type': 'task', 'timestamp': '...', 'step': 1, 'payload': {'id': '...', 'name': 'subgraph_node', 'input': {'messages': ['hello subgraph']}, 'triggers': ('branch:to:subgraph_node',)}}
# --------------------------------------------------
# 流式输出： {'type': 'task_result', 'timestamp': '...', 'step': 1, 'payload': {'id': '...', 'name': 'subgraph_node', 'error': None, 'result': [('messages', ['hello subgraph', 'response from subgraph'])], 'interrupts': []}}

2、Stream Mode的5种类型（重点掌握常用款）

LangGraph提供5种stream_mode，对应不同的输出需求，重点掌握前3种即可：

values（常用）：每一步执行完毕后，流式输出「完整的State」，能看到当前State的全部数据。示例：执行完子图后，输出完整的messages列表。
updates（常用）：每一步执行完毕后，只输出「State的更新部分」，不输出完整State，简洁高效。示例：只输出子图新增的["response from subgraph"]，不重复输出原有内容。
debug（调试用）：输出最详细的执行信息，包括步骤、时间戳、节点名称、输入输出，适合排查流程问题（用得较少）。
messages（LLM专用）：仅当Graph中调用了大模型时，流式输出大模型的Token和元数据；未调用大模型时，无输出。
custom（自定义）：可在Node内部自定义流式输出内容，需通过get_stream_writer()方法实现。

3、Custom Stream Mode（自定义流式输出，实用调试技巧）

当需要在Node执行过程中，输出自定义信息（比如调试Node的执行状态），可使用custom模式，通过get_stream_writer()获取流对象，写入自定义内容。

from typing import TypedDict
from langgraph.config import get_stream_writer
from langgraph.graph import StateGraph, START, END

# 定义State
class State(TypedDict):
    query: str
    answer: str

# 定义Node，在Node内部自定义流式输出
def node(state: State):
    # 获取StreamWriter对象，用于写入自定义流式内容
    writer = get_stream_writer()
    # 写入自定义数据（可任意定义键值对）
    writer({"自定义调试信息": "Node开始执行", "当前query": state["query"]})
    # Node的核心处理逻辑
    return {"answer": "这是Node的处理结果", "query": state["query"]}

# 构建Graph
graph = (
    StateGraph(State)
    .add_node("node", node)  # 添加Node
    .add_edge(START, "node") # 入口→Node
    .add_edge("node", END)   # Node→出口
    .compile()
)

# 流式调用，stream_mode="custom"
inputs = {"query": "什么是LangGraph的Stream？"}
for chunk in graph.stream(inputs, stream_mode="custom"):
    print("自定义流式输出：", chunk)

# 输出结果：{'自定义调试信息': 'Node开始执行', '当前query': '什么是LangGraph的Stream？'}

4、补充：禁止流式输出（实用配置）

在LangChain中，构建LLM对象时，可通过disable_streaming=True属性，禁止大模型的流式输出（若Graph中调用了LLM，可配合使用）：

# 示例：禁止LLM流式输出（需导入对应LLM类）
from langchain_openai import ChatOpenAI

# disable_streaming=True：禁止大模型流式输出，直接返回完整结果
llm = ChatOpenAI(model="gpt-3.5-turbo", disable_streaming=True)

小结：Stream支持的核心价值是「分步调试」，通过不同的stream_mode，可灵活查看Graph的执行过程，自定义模式更能满足个性化调试需求，大幅提升开发效率。

六、Graph核心知识点总结（必背，快速回顾）

通过前面的讲解，我们已经掌握了LangGraph中Graph的核心用法，这里整理成核心知识点，方便大家快速回顾、巩固记忆：

Graph的本质：以有向无环图（DAG）为基础，串联Node，通过Edge控制流程，实现复杂任务的自动化执行。
三大核心元素： State：全局共享容器，规范数据格式，所有Node可读写，支持TypedDict和Pydantic两种实现方式，有追加、覆盖两种更新规则。
Node：核心执行单元，本质是Python函数，接收State输入，处理后返回更新后的State，支持缓存、重试机制。
Edge：控制流程方向，有普通、条件、多分支、并行四种方式，对应不同的流程场景。
核心流程：定义State→定义Node→构建Graph（添加Node和Edge）→编译Graph→调用Graph（invoke/stream）→获取结果。
高级特性：递归限制：设置max_steps，防止流程无限循环。
中断与恢复：通过检查点（checkpoint），实现流程的中断与后续恢复。
序列化：将Graph结构保存为JSON，后续可直接加载复用，提升开发效率。
子图：将一个Graph作为Node嵌入另一个Graph，实现流程复用，适合复杂工作流和多Agent协作。
Stream支持：通过stream()/astream()方法，分步查看执行过程，支持5种stream_mode，方便调试。
避坑点：Node名称必须唯一，否则会报错。
State设置默认值，不能直接在TypedDict中写，需通过Node实现。
条件Edge的路由函数，返回值必须是Node名称或END，否则会中断流程。
并行Edge仅适用于无依赖的Node，有依赖的Node需用串行Edge。
子图与父图需使用相同格式的State，否则会出现数据互通问题。
Stream的messages模式，仅在Graph调用LLM时才有输出。

至此，LangGraph核心——Graph的内容已全部讲解完毕。通过前面的前置知识、入门案例、核心组件拆解、高级特性（含子图、Stream支持）讲解，相信也能彻底掌握Graph的用法。接下来，我们就可以结合Agent，用Graph构建更复杂的大模型应用（比如多Agent协作、复杂任务流程自动化）啦！

通过前面的讲解，我们已经掌握了LangGraph中Graph的核心用法，这里整理成核心知识点，方便大家快速回顾、巩固记忆：