LangGraph

版本：V1.1.0

第 1 章 LangGraph 概述 1.1 什么是 LangGraph LangGraph 是一个低级编排框架和运行时环境，用于构建、管理和部署长期运行的有状态智能体（agents）。核心理念是将 Agent 工作流建模为图（Graph），其中： ➢ 节点（Nodes）：代表计算单元，可以是 LLM 调用、工具执行或任何自定义逻辑 ➢ 边（Edges）：定义节点之间的转换逻辑，决定执行流程 ➢ 状态（State）：在整个图执行过程中共享和传递的数据

LangGraph 提供了构建生产级智能体应用的核心能力： ➢ 持久化执行：构建能够从故障中恢复并长时间运行的智能体 ➢ 人机协作：在任何时刻检查和修改智能体状态 ➢ 记忆管理：支持短期工作记忆和跨会话的长期记忆 ➢ 流式处理：专为流式工作流设计 ➢ 生产级部署：为有状态、长期运行的工作流提供可扩展的基础设施 LangGraph 官方源码地址及官网地址如下： ➢ GitHub 地址：https://github.com/langchain-ai/langgraph ➢ 官方文档：https://docs.langchain.com/oss/python/langgraph/overview 1.2 与 LangChain 区别

LangGraph 与 LangChain 的高级抽象不同，它提供了更细粒度的控制，让开发者能够精确控制智能体的执行流程，适合需要定制化编排的复杂应用场景。通过一句话概述 LangChain 和 LangGraph 的区别：当 LLM 应用需要「有状态、可循环、可分支的多步骤控制流（像一个可回溯的流程图/状态机）」时，LangChain 很难优雅完成，而必须用 LangGraph。以下表格以更细粒度的方式展示了两者的区别：特性 LangGraph LangChain 抽象级别低级，提供细粒度控制高级，开箱即用状态管理内置状态机和检查点需要自行管理状态执行模型基于图的并行执行线性链式执行持久化原生支持需要额外实现适用场景复杂、有状态的智能体应用开发简单的链式调用正是因为 LangGraph 和 LangChain 上述的区别，使得 LangGraph 在以下场景下，有了更加广泛的应用： ➢ 复杂的多智能体系统 ➢ 需要长期记忆的应用 ➢ 需要人工审核的工作流 ➢ 后台处理任务和实时交互 ➢ 需要精细控制的定制化智能体编排下面通过快速入门示例来具体讲解如何通过 LangGraph 构建一个应用。第 2 章快速入门 2.1 环境安装 1）创建环境 conda create –-name=langgraph_venv 3.12 conda activate langgraph_venv pip install langgraph==1.0.5 2）确认成功 pip show langgraph 2.2 简单示例使用 LangGraph 构建一个应用，可以分为如下步骤：

（1）定义状态；（2）定义节点函数；（3）通过状态创建 builder 实例，并添加节点和边；（4）对 builder 进行编译，得到可调用的 graph 对象；（5）调用 graph。接下来的例子，以一个基于 RAG 的问答系统为例，来做一个快速入门展示。该问答系统需要实现的功能如下：当接收用户问题之后，先分别进行联网搜索和基于知识库的检索，得到结果之后，使用大语言模型进行总结回答。具体代码如下所示： from typing import TypedDict from pydantic import BaseModel from langgraph.constants import START, END from langgraph.graph import StateGraph from langgraph.types import Send,Command class MyState(TypedDict): rag_result:str web_search_result:str query:str final_answer:str

graph = StateGraph(state_schema=MyState)

def rag_search_node(state:MyState): query = state["query"] rag_result = f"关于{query}的 rag_result" return {"rag_result":rag_result}

def web_search_node(state:MyState): query = state["query"] web_search_result = f"关于{query}的 web_search_result" return {"web_search_result":web_search_result}

def final_answer_node(state:MyState): rag_result = state["rag_result"] web_search_result = state["web_search_result"] final_answer = f"LLM基于{rag_result}和{web_search_result}的最终回复" return {"final_answer":final_answer}

graph.add_node(rag_search_node) graph.add_node(web_search_node)

graph.add_node(final_answer_node) graph.add_edge(START, "rag_search_node") graph.add_edge(START,"web_search_node") graph.add_edge("rag_search_node", "final_answer_node") graph.add_edge("web_search_node","final_answer_node") graph.add_edge("final_answer_node", END) compiled_graph = graph.compile() res = compiled_graph.invoke({"query":"如何使用 LangGraph"}) print(res) 编译好之后，可以通过 mermaid 查看图的结构。

pip install grandalf

graph_structure = compiled_graph.get_graph() res = graph_structure.draw_ascii() print(res) 通过该例子，可以快速对 LangGraph 编程流程有一个概念。第 3 章状态需要通过 LangGraph 构建一个应用程序时，第一步就是定义 State，因为 State 代表了整个图的结果目标，和图当中节点需要修改的数据目标。 State 作为 LangGraph 当中最重要的概念，有其专门的定义方式，和运行时的行为特点。 3.1 状态的定义 3.1.1 定义状态类 State Schema 可以通过三种方式定义： ➢ TypedDict（推荐方式）: TypedDict 是 Python 提供的一种类型提示工具，用于为字典（Dict）的键和值指定精确的类型信息。状态类继承 TypedDict，定义键的类型和 reducer 函数（具体参看 3.1.4 reducer 章节内容）即可；代码示例如下： from typing import TypedDict class MyStateFull(TypedDict): rag_result:str web_search_result:str final_answer:str query:str a_new_key:str ➢ Pydantic BaseModel：Pydantic 提供运行时数据校验，并支持静态类型检查工具进行类型推导。状态类通过继承 Pydantic 的 BaseModel，定义键的类型和 reducer 函数；

代码示例如下： from pydantic import BaseModel class MyStateFull(BaseModel): rag_result:str web_search_result:str final_answer:str query:str a_new_key:str ➢ Dataclass: dataclass 是 Python 标准库中的一个装饰器，用于自动生成常见特殊方法（如__init__、repr、__eq__等），从而简化主要用作数据容器的类的定义。状态类通过 dataclass 装饰器装饰后，定义键的类型和 reducer 函数即可。代码示例如下： from dataclasses import dataclass @dataclass class MyStateFullTwo(): rag_result:str web_search_result:str final_answer:str query:str a_new_key:str 3.1.2 输入输出数据隔离在 LangGraph 当中，可以精细管理输入到图中的状态键有哪些，以及输出的状态键有哪些。这是通过初始化 StateGraph 时，分别指定三个参数：state_schema、input_schema 和 output_schema 来实现的。其具体作用，可以参考如下图示：

1）state_schema 这是图的完整内部状态，包含了所有节点可能读写的字段，必须指定，不能为空。特点： ➢ 是图的"全局状态空间"； ➢ 所有节点都可以访问和写入这个 schema 中的任何字段。 2）input_schema 定义图接受什么输入，是 state_schema 的子集特点： ➢ 可选参数，如果不指定，默认等于 state_schema； ➢ 限制图的输入接口，只能传入这些字段； ➢ 是 state_schema 的子集或相等。 3）output_schema 定义图返回什么输出，是 state_schema 的子集特点： ➢ 可选参数，如果不指定，默认等于 state_schema； ➢ 限制图的输出接口，只返回这些字段； ➢ 是 state_schema 的子集或相等。代码如下所示： from typing import TypedDict

from langgraph.constants import START, END from langgraph.graph import StateGraph class MyStateFull(TypedDict): rag_result:str web_search_result:str final_answer:str query:str a_new_key:str

class InputSchema(TypedDict): query:str

class OutputSchema(TypedDict): final_answer:str

graph = StateGraph(state_schema=MyStateFull,input_schema=InputSchema,output_sch ema=OutputSchema)

def rag_search_node(state:MyStateFull): print(state) query = state["query"] rag_result = f"关于{query}的 rag_result" return {"rag_result":rag_result,"a_new_key":"a_new_key_value"}

def web_search_node(state:MyStateFull): print(state) query = state["query"] web_search_result = f"关于{query}的 web_search_result" return {"web_search_result":web_search_result}

def final_answer_node(state:MyStateFull): print('在 final_answer_node当中的 state',state) rag_result = state["rag_result"] web_search_result = state["web_search_result"] final_answer = f"LLM基于{rag_result}和{web_search_result}的最终回复" return {"final_answer":final_answer}

graph.add_edge("web_search_node","final_answer_node") graph.add_edge("final_answer_node", END) compiled_graph = graph.compile() res = compiled_graph.invoke({"query":"如何使用 LangGraph","a_new_key":"a_new_key_value"}) print('最终结果为',res) 3.1.3 节点间数据隔离 LangGraph 当中节点所接收的状态，还可以为一个非全局状态的私有状态。
示例代码如下： from typing import TypedDict from pydantic import BaseModel from langgraph.constants import START, END from langgraph.graph import StateGraph class MyState(TypedDict): query:str final_answer:str

class SearchState(TypedDict): rag_result:str web_search_result:str

class InputSchema(TypedDict): query:str

class OutputSchema(TypedDict): final_answer:str

graph = StateGraph(state_schema=MyState)

def rag_search_node(state:MyState): print(state) query = state["query"] rag_result = f"关于{query}的 rag_result" return {"rag_result":rag_result,"a_new_key":"a_new_key_value"}

def web_search_node(state:MyState): query = state["query"] web_search_result = f"关于{query}的 web_search_result" return {"web_search_result":web_search_result}

def final_answer_node(state:SearchState): print('在 final_answer_node当中的 state',state)

rag_result = state["rag_result"] 
web_search_result = state["web_search_result"] 
final_answer = f"LLM基于{rag_result}，和{web_search_result}的最终回复

" return {"final_answer":final_answer}

graph.add_node(rag_search_node) graph.add_node(web_search_node) graph.add_node(final_answer_node) graph.add_edge(START, "rag_search_node") graph.add_edge(START,"web_search_node") graph.add_edge("rag_search_node", "final_answer_node") graph.add_edge("web_search_node","final_answer_node") graph.add_edge("final_answer_node", END) compiled_graph = graph.compile() res = compiled_graph.invoke({"query":"如何使用 LangGraph"}) print('最终结果为',res) 3.1.4 Reducer 函数 Reducer 用于进行当前增量状态（节点输出的状态）和全局状态的合并。State 中的每个键都有其独立的 reducer 函数。每个 node 的返回值中的每个 key 与全局 state_schema 中对应的 key 进行合并更新，具体更新逻辑取决于每个 key 指定的 reducer 函数。 Reducer 常用函数有以下几种： ➢ 默认行为：未指定 Reducer 时使用覆盖更新 ➢ 内置 reducer 函数：例如 langgraph.graph.messages 当中的 add_messages 函数 ➢ 自定义 Reducer：支持用户自定义合并逻辑如果未明确指定 reducer 函数，则默认对该键的更新是覆盖行为。以下为默认覆盖行为： """ LangGraph Reducer函数演示 - 默认 Reducer（覆盖更新） """

from typing import List from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END

1. 默认 Reducer（覆盖更新）

class DefaultReducerState(TypedDict): foo: int bar: List[str]

def node_default_1(state: DefaultReducerState) -> dict: return {"foo": 2} def node_default_2(state: DefaultReducerState) -> dict: return {"bar": ["bye"]} def run_demo(): print("1. 默认 Reducer（覆盖更新）演示:") builder = StateGraph(DefaultReducerState) builder.add_node("node1", node_default_1) builder.add_node("node2", node_default_2) builder.add_edge(START, "node1") builder.add_edge("node1", "node2") builder.add_edge("node2", END) graph = builder.compile() result = graph.invoke({"foo": 1, "bar": ["hi"]}) print(f"初始状态: {{'foo': 1, 'bar': ['hi']}}") print(f"执行结果: {result}\n")

if name == "main": run_demo() 内置函数:operator.add：用于列表追加或数值累加 1）列表追加 """ LangGraph Reducer函数演示 - operator.add Reducer（列表追加） """

import operator from typing import Annotated, List from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END

3. operator.add Reducer（列表追加）

class ListAddState(TypedDict): data: Annotated[List[int], operator.add]

def producer_1(state: ListAddState) -> dict: return {"data": [1, 2]}

def producer_2(state: ListAddState) -> dict: return {"data": [3, 4]}

def run_demo():

print("operator.add Reducer（列表追加）演示:") 
builder = StateGraph(ListAddState) 
builder.add_node("producer1", producer_1) 
builder.add_node("producer2", producer_2) 
builder.add_edge(START, "producer1") 
builder.add_edge(START, "producer2")  # 并行执行 
builder.add_edge("producer1", END) 
builder.add_edge("producer2", END) 
graph = builder.compile() 
 
result = graph.invoke({"data": [0]}) 
print(f"初始状态: {{'data': [0]}}") 
print(f"执行结果: {result}\n")

if name == "main": run_demo() 2）字符串连接 """ LangGraph Reducer函数演示 - 字符串连接 Reducer """

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

6. 字符串连接 Reducer

class StringConcatState(TypedDict): text: Annotated[str, operator.add]

def add_text_1(state: StringConcatState) -> dict: return {"text": "Hello "}

def add_text_2(state: StringConcatState) -> dict: return {"text": "World!"}

def run_demo(): print(" 字符串连接 Reducer演示:") builder = StateGraph(StringConcatState) builder.add_node("add_text_1", add_text_1) builder.add_node("add_text_2", add_text_2)

builder.add_edge(START, "add_text_1") 
builder.add_edge(START, "add_text_2")  # 并行执行 
builder.add_edge("add_text_1", END) 
builder.add_edge("add_text_2", END) 
graph = builder.compile() 
 
result = graph.invoke({"text": "Say: "}) 
print(f"初始状态: {{'text': 'Say: '}}") 
print(f"执行结果: {result}\n")

if name == "main": run_demo() 3）数值累加

""" LangGraph Reducer函数演示 - 数值累加 Reducer """

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

7. 数值累加 Reducer

class NumberAddState(TypedDict): count: Annotated[int, operator.add]

def increment_1(state: NumberAddState) -> dict: return {"count": 5}

def increment_2(state: NumberAddState) -> dict: return {"count": 3}

def run_demo(): print(" 数值累加 Reducer演示:") builder = StateGraph(NumberAddState) builder.add_node("increment_1", increment_1) builder.add_node("increment_2", increment_2) builder.add_edge(START, "increment_1") builder.add_edge(START, "increment_2") # 并行执行

builder.add_edge("increment_1", END) 
builder.add_edge("increment_2", END) 
graph = builder.compile() 
 
result = graph.invoke({"count": 10}) 
print(f"初始状态: {{'count': 10}}") 
print(f"执行结果: {result}\n")

if name == "main": run_demo() 可以自定义 Reducer 函数，用于合并操作。自定义 Reducer 函数接收两个参数，用以将其合并，最终返回一个结果。例如此处定义一个 add_message 函数，用于将全局状态当中的 message和节点的 message 进行合并操作： from langgraph.graph import StateGraph from langchain_core.messages import ToolMessage,HumanMessage,AIMessage,BaseMessage from typing import Annotated, TypedDict,List from langgraph.constants import START

自定义 Reducer

def add_message(message_list_left:list,message_list_right:list): print("="*20) print("正在执行 add_message") print('左边的',message_list_left) print('右边的',message_list_right) print("="*20)

return message_list_left + message_list_right

class MyAgent(TypedDict):

# 在初始使用自定义的 reducer 
messages:Annotated[List[BaseMessage],add_message]

def tool_node(state:MyAgent): return {"messages":[ToolMessage(content="来自 tool_node的内容 ",tool_call_id="xxx")]}

def llm_node(state:MyAgent): return {"messages":[AIMessage(content="来自 llm_node的内容")]}

builder = StateGraph(state_schema=MyAgent)

builder.add_node(tool_node) builder.add_node(llm_node) builder.add_edge(START,"tool_node") builder.add_edge("tool_node","llm_node")

graph = builder.compile() graph.invoke({"messages":[HumanMessage(content="你好")]}) 3.2 状态的存储 3.2.1 实际场景当中的问题在前面的例子当中，我们看到，每次用户 invoke 时，LangGraph 都会初始化一个空状态，然后将用户传入的初始状态合并进来，再继续往下执行。这在一些一次性、简单任务过程中，没有什么问题。但是对于一些复杂任务，就会出现一些问题，考虑以下两个场景： 1）场景一：在 agent 的一个会话里，需要在多次调用当中保持上下文

在这个场景下，我们定义的状态为 messages 列表，第一次调用过程中，图会在 messages 列表当中写入值。而第二次调用，又会初始化一个新的 messages 列表，导致缺失了第一次调用的上下文。 2）场景二：图执行过程当中报错，不想重复已执行完节点，想要实现断点续传

在这个场景下，第一次调用过程中，图会在 Node2 节点因某些原因报错，但是我们想要拿到 Node1 节点所对应的值。并在修复完第二个节点之后，继续往下执行。 3.2.2 状态的存储前面所涉及到的问题，解决方案都是将第一次调用的结果，存储起来，然后在第二次调用时，从存储的结果当中取出状态，而不是让 LangGraph 去初始化空状态。由于同一个 agent 会存在多用户或者单用户多会话等场景，要实现状态存储，并且后续能够找到正确的状态，就需要在状态当中做一个 id 映射，写入时指定一个 id，读取时读取指定 id 的状态。首先需要将状态存储起来，LangGraph为我们提供了一个存储状态的方式：Checkpointer。 Checkpointer 也即检查点，它可以存储图执行过程当中所产生的状态，具体存储的位置，有内存、数据库等。Checkpointer 所有相关的实例化类都来自于 langgraph.checkpointer 包。 LangGraph 同样做了前述的 id 隔离，这个 id 在 langgraph 当中就称之为 thread_id。需要注意：Thread_id 和实际编码当中的线程没有联系，只是一种隔离状态空间的方式。实际编码进行状态隔离的步骤，如下所示：（1）创建 checkpointer 实例

（2）图编译时，传入 checkpointer 实例（3）图调用时，传递 thread_id：保证上下文空间，就传入相同的 thread_id，需要开辟新的上下文空间，就传入新的 thread_id。对于前面所讲到的场景一，要想在一个会话中保持上下文，只需要在多次调用时，传入同一个 thread_id即可；要想启动一个新会话，就传入一个新的 thread_id。这也就是所谓“Agent 短期记忆”。示例代码如下所示，重点关注内容参看带序号注释： def scenario_1_demo(): # 1、create_agent底层就是通过 LangGraph构建的图结构，此处直接通过 create_agent来举一个例子 from langchain.agents import create_agent from langchain.tools import tool from langchain_openai import ChatOpenAI

# 1、引入 checkpointer 
from langgraph.checkpoint.memory import InMemorySaver 

llm = ChatOpenAI(model="gpt-4o-mini") 

# 2、定义 checkpointer实例 
checkpointer = InMemorySaver() 
@tool 
def weather_tool(city:str,date:str)->str: 
    """查天气的工具""" 
    return f'{city}在{date}的天气是晴朗的' 
 
# 3、构建 Agent时引入 checkpointer 
agent = create_agent( 
    model=llm, 
    tools=[weather_tool], 
    checkpointer=checkpointer, 
) 

# 4、用户第一次调用 
user1_res = agent.invoke({"messages":"北京 2026-01-14天气怎么样

"},config={"configurable":{"thread_id":"user_1_session_1"}}) print('第一次调用',user1_res['messages'][- 1],end="\n\n==============\n\n")

# 5、用户在同一个会话当中，第二次调用 

user1_res_2 = agent.invoke({"messages":"适合出去玩吗

"},config={"configurable":{"thread_id":"user_1_session_1"}}) print('第二次调用',user1_res_2['messages'][- 1],end="\n\n==============\n\n") 3.2.3 从状态中恢复执行上一小节当中所讲到的例子，checkpointer 是通过 InMemorySaver 进行实例化的，这个实例会将状态全部存储到内存当中，这仍然不能解决场景二的问题，场景二涉及的问题不再是多次调用之间，保持上下文不中断，而是如何进行故障恢复。在因为故障导致进程退出后，内存状态空间当中的内容，将全部丢失。在实际生产环境下，需要使用数据库做持久化状态存储。下面代码当中，Node2 节点表示在实际执行过程中，因为某些原因导致的错误。如果将相关的状态存储到数据库中，在第一次执行时，node2 前面节点的状态，会写入到数据库中，示例代码如下（关键点参见带序号注释）：

pip install langgraph-checkpoint-sqlite

def scenario_2_demo(): """ 场景二：如何从 checkpointer中恢复状态 """ from langgraph.graph import StateGraph from typing import TypedDict from langgraph.graph import START # 1、引入 sqlite3包和 SqliteSaver类 import sqlite3 from langgraph.checkpoint.sqlite import SqliteSaver # 2、构建 Connection对象 conn = sqlite3.connect(database="./sqlite_data/langgraph_sqlite", check_same_thread=False) # 3、通过 connection对象构建 checkpointer实例 memory = SqliteSaver(conn) class MyState(TypedDict): key_1: str key_2: str key_3: str

def node_1(state: MyState) -> MyState: 
    # 4、打印 node_1初始状态 
    print('node_1初始状态为',state) 
    return {"key_1":"value_1"} 
 
def node_2(state: MyState) -> MyState: 

    # 5、此处模拟生产环境下代码的 bug或者其他不可预见情况，导致的报错 
    raise Exception("node_2节点执行报错") 
    return {"key_2":"value_2"} 
 
def node_3(state: MyState) -> MyState: 
    return {"key_3":"value_3"} 

# 添加节点，添加边 
builder = StateGraph(MyState) 
builder.add_node(node_1) 
builder.add_node(node_2) 
builder.add_node(node_3) 
builder.add_edge(START, "node_1") 
builder.add_edge("node_1", "node_2") 
builder.add_edge("node_1", "node_3") 
config = { 
    "configurable":{ 
        "thread_id":"123" 
    } 
} 
# 6、compile时传入 checkpointer实例 
graph =builder.compile(checkpointer=memory) 
# 7、调用时传入 config 
graph.invoke({},config=config)

在 node2 节点问题处理完成后，要想从故障中恢复，需要以下两点：（1）invoke 时传递 None 作为初始参数；（2）传入相同的 thread_id。满足以上两点，即可让图从断点开始，以第一次调用的历史状态，继续往下执行，示例代码如下（关键点参见带序号注释）： def scenario_2_demo(): """ 场景一：如何从 checkpointer中恢复状态 """ from langgraph.graph import StateGraph from typing import TypedDict from langgraph.graph import START

import sqlite3 
from langgraph.checkpoint.sqlite import SqliteSaver 
 
conn = sqlite3.connect(database="./sqlite_data/langgraph_sqlite",

check_same_thread=False)

memory = SqliteSaver(conn) 
class MyState(TypedDict): 
    key_1: str 
    key_2: str 
    key_3: str 

def node_1(state: MyState) -> MyState: 
     
    print('node_1初始状态为',state) 
    return {"key_1":"value_1"} 
 
def node_2(state: MyState) -> MyState: 
    # 1、修复错误，注释掉报错代码 
    # raise Exception("node_2节点执行报错") 
    return {"key_2":"value_2"} 
 
def node_3(state: MyState) -> MyState: 
    return {"key_3":"value_3"} 

# 添加节点，添加边 
builder = StateGraph(MyState) 
builder.add_node(node_1) 
builder.add_node(node_2) 
builder.add_node(node_3) 
builder.add_edge(START, "node_1") 
builder.add_edge("node_1", "node_2") 
builder.add_edge("node_1", "node_3") 
config = { 
    "configurable":{ 
        "thread_id":"123" 
    } 
} 
 
graph = builder.compile(checkpointer=memory) 
# 2、调用时传入相同的 config，注意，从故障中恢复，此处传递 None即可 
graph.invoke(None,config=config)

3.3 获取历史状态前面我们了解到，构造 checkpointer 之后，langgraph 可以保证在多次调用之间，保证状态的连续，或者是通过状态进行恢复。不过，要想再深入了解状态，就需要知道状态的保存时机和如何获取到所有的历史状态。

3.3.1 LangGraph 底层运行算法在了解如何获取历史状态之前，首先我们需要了解一下 LangGraph 底层算法—Pregel。 Pregel 是 LangGraph 底层的一个类，管理 LangGraph 应用程序的运行时（runtime）行为。也就是说，整个图结构，从开始到结束的迭代执行过程，是由 Pregel 控制管理的。在这个类当中，有两大主要组件： 1）Actors (节点) Actor 即为在前面介绍的 Node，对应在 LangGraph 当中的类为 PregelNode，其订阅某些通道（channels），从通道当中读取数据，或者写入数据。PregelNode 也实现了 LangChain 的 Runnable 接口。 2）Channels（通道） Channels 用于 actors 之间通信，每个通道都有一个值类型、一个更新类型和一个更新函数 —— 该函数接收一系列更新并修改存储的值。正是因为这种设计，在 LangGraph 中通过边和节点构建的图，不是一个传统的有向无环图（DAG）。在传统的有向无环图下，节点的执行顺序，完全边的连接，某个节点只有在和它连接的所有的上游节点都执行完成了后，才会执行。而在 LangGraph 图当中，节点的执行来自于一个 Pregel 的超步（SuperStep）。而一个超步的执行逻辑，分为如下三个过程：（1）Plan 阶段：确定在此步骤中要执行哪些 Actors。例如，在第一步中，选择订阅特殊输入通道的 Actors；在后续步骤中，选择订阅上一步骤中更新的通道的参与者（2）Execute 阶段：并行执行所有选定的 Actors，直到所有参与者完成、其中一个失败或达到超时时间（3）Update 阶段：用本步骤中 Actors 所写的值更新 channels。由以上三个步骤组合而成的超步会重复执行，直到没有 Actors 被选中执行，或者达到最大步骤数为止。以如下的图为例：

图中展示了 builder.compile()内部逻辑：将逻辑图，转换成 Pregel 实例，而节点和边，分别转换成了 Actors 和 Channels 实例。针对于 Actors 实例，每个 Actor 实例都有其订阅和写入的 Channels，如下图所示：

整个图的执行流程，可以分为如下几个 Step，Step 索引从 0 开始：

整个执行流程如下：首先在 step 为-1 时，__START__节点执行（图中未标注），将相关状态写入到 branch:to:a 当中；在 Step0 当中，节点 a 执行，将状态写入到 branch:to:b 和 branch:to:c；在 Step1 当中，节点 b 和 c 执行，将状态写入到 branch:to:b_2 和 branch:to:d 当中；在 Step2，节点 b_2 和节点 d 执行，b_2 节点再次将状态写入到 branch:to:d 当中。在 Step3 当中，节点 d 再次执行，由于没有新的状态更新，图执行过程结束。示例代码如下所示： import operator from typing import Annotated, Any from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END

class State(TypedDict): """ 状态类型定义 """ aggregate: Annotated[list, operator.add]

def a(state: State,config): print(f'Adding "A" to {state["aggregate"]}') return {"aggregate": ["A"]}

def b(state: State,config): print(f'Adding "B" to {state["aggregate"]}') return {"aggregate": ["B"]}

def b_2(state: State,config): print(f'Adding "B_2" to {state["aggregate"]}') return {"aggregate": ["B_2"]}

def c(state: State,config): print(f'Adding "C" to {state["aggregate"]}') return {"aggregate": ["C"]}

def d(state: State,config): print(f'Adding "D" to {state["aggregate"]}') return {"aggregate": ["D"]}

builder = StateGraph(State)

builder.add_node("a", a) builder.add_node("b", b) builder.add_node("b_2", b_2) builder.add_node("c", c) builder.add_node("d", d)

builder.add_edge(START, "a") builder.add_edge("a", "b") builder.add_edge("a", "c") builder.add_edge("b", "b_2") builder.add_edge("b_2", "d") builder.add_edge("c", "d") builder.add_edge("d", END)

graph = builder.compile()

1、查看当前图的 channels

print('当前图的 channels为',graph.channels,end="\n\n")

2、查看当前图的节点

print('当前图的节点为',graph.nodes,end="\n\n")

3、查看 a节点的 trigger（当前节点的订阅）和 writers

print('当前图的节点 a的 triggers为',graph.nodes['a'].triggers,end="\n\n") print('当前图的节点 a的 writers为',graph.nodes['a'].writers,end="\n\n")

4、查看 d节点的 trigger和 writers

print('当前图的节点 d的 triggers为',graph.nodes['d'].triggers,end="\n\n") print('当前图的节点 d的 writers为',graph.nodes['d'].writers,end="\n\n")

5、执行图，查看执行结果

output_state = graph.invoke({"aggregate": []})
print('执行图后的状态为',output_state,end="\n\n")

3.3.2 获取图执行的历史状态理解了 Pregel 和 Step 的概念之后，现在可以来学习如何获取图执行的历史状态，以及如何看懂历史状态当中所存储的内容。构建好的 graph 实例，可以通过 get_state() / get_state_history()方法，传入想要获取的历史状态的 thread_id，即可拿到图执行过程中的历史状态，get_state()方法获取到的是最近一个时间步的状态，get_state_history()获取到的是图执行当中所有时间步的历史状态，其输出值为一个迭代器，按照时间步倒序排列。历史状态由 StateSnapShot 实例表示，也即状态快照，其中包含的信息如下：
名称类型描述 values dict[str, Any] | Any 当前状态信息 next tuple[str, ...] 本超步中每个任务要执行的节点名称 config RunnableConfig 用于获取此快照的配置。 metadata CheckpointMetadata 与此快照相关联的元数据 parent_config RunnableConfig 用于获取父快照（如果有的话）的配置。 interrupts tuple[Interrupt, ...] 此超步中发生且有待解决的中断。我们重点需要关注的是，在状态快照中，除了有当前状态值以外，还包含接下来需要执行的节点名称，这也为从故障中恢复提供了有用的信息，我们可以通过状态快照直接知道，从哪一个节点开始继续往下执行。示例代码如下所示： import operator from typing import Annotated, Any from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 connection = sqlite3.connect('checkpointer.db',check_same_thread=False) checkpointer = SqliteSaver(connection) class State(TypedDict): """ 状态类型定义

""" 
aggregate: Annotated[list, operator.add]

def a(state: State,config): print(f'Adding "A" to {state["aggregate"]}') return {"aggregate": ["A"]}

def b(state: State,config): print(f'Adding "B" to {state["aggregate"]}') return {"aggregate": ["B"]}

def b_2(state: State,config): print(f'Adding "B_2" to {state["aggregate"]}') return {"aggregate": ["B_2"]}

def c(state: State,config): print(f'Adding "C" to {state["aggregate"]}') return {"aggregate": ["C"]}

def d(state: State,config): print(f'Adding "D" to {state["aggregate"]}') return {"aggregate": ["D"]}

builder = StateGraph(State)

builder.add_node("a", a) builder.add_node("b", b) builder.add_node("b_2", b_2) builder.add_node("c", c) builder.add_node("d", d)

builder.add_edge(START, "a") builder.add_edge("a", "b") builder.add_edge("a", "c") builder.add_edge("b", "b_2") builder.add_edge("b_2", "d") builder.add_edge("c", "d") builder.add_edge("d", END)

graph = builder.compile(checkpointer=checkpointer) res = graph.invoke({},config={'configurable':{'thread_id':'1'}})

1、最终状态

print(res)

2、查看历史所有状态

all_states = graph.get_state_history(config={'configurable':{'thread_id':'1'}}) all_states_list = list(all_states) print("历史所有状态如下：\n") for state in all_states_list: print(state,end="\n"+"="*30+"\n")

print("\n\n最近一次状态如下：\n") last_state = graph.get_state(config={'configurable':{'thread_id':'1'}}) print(last_state) 第 4 章节点 4.1 节点的输入输出 4.1.1 节点输入在 LangGraph 中，一般来讲，节点都是 Python 函数（可以是同步的，也可以是异步的），它们接受以下参数： ➢ state：图的状态，代表了具体的业务数据； ➢ config：一个 RunnableConfig 对象，包含诸如 thread_id 之类的配置信息，在调用图时，也可以传递用户定义的其他配置； ➢ runtime：一个 Runtime 对象，包含运行时 context（可自定 context，在调用图时传入即可）以及其他信息，如 store 和 stream_writer 等；2 以上参数，会在运行过程中，自动被 LangGraph 运行时注入。节点定义 config 和 runtime 入参如下所示，在运行时，这两个参数 LangGraph 名称会通过关键字传参方式进行传参，因此两个参数位置可以调换。具体示例如下： import time from typing import TypedDict, Any, List from langchain_core.runnables import RunnableConfig from langgraph.graph import StateGraph, START, END from langgraph.runtime import Runtime

class MockLLM: def invoke(self, prompt: str): return f"AI Generated: Answer for '{prompt}'"

class MockDatabase: def get_user_info(self, user_id: str): return {"id": user_id, "role": "vip" if "vip" in user_id else "standard"}

class CustomerSupportState(TypedDict): query: str # 用户问题 response: str # 客服回复 log: List[str] # 处理日志

def node_customer_service(state: CustomerSupportState, config: RunnableConfig, runtime: Runtime) -> dict: """ 客服节点：展示如何从 config 中获取注入的依赖 (LLM, DB) 并使用 runtime 推送进度。 """

user_query = state["query"] 
 
# 1、从 runtime当中获取 context对象 
llm = runtime.context['llm'] 
db = runtime.context['db'] 
 

configurable = config.get("configurable", {}) 
user_id = configurable.get("user_id", "guest") 
 
print(f"\n[Node] 开始处理，User ID: {user_id}") 
 
# 2、验证依赖是否存在 
if not llm or not db: 
    return {"response": "System Error: Dependencies not injected!",

"log": ["Error"]}

# 3、使用 db对象查看用户角色 
user_info = db.get_user_info(user_id) 
user_tier = user_info.get("role") 
print(f"[Node] 从 DB 获取用户角色: {user_tier}") 

# 4、使用 runtime.writer 向图外输出自定义数据 --- 
writer = runtime.stream_writer 
writer({"status": "thinking", "message": f"正在调用 LLM 为

{user_tier} 用户生成回复..."})

time.sleep(0.5) 

# 5、根据用户角色构建不同的 Prompt，并模拟 LLM 调用 
prompt = f"User({user_tier}) asks: {user_query}" 
llm_response = llm.invoke(prompt) 
 
return { 
    "response": llm_response, 
    "log": [f"Processed by {llm.__class__.__name__}"] 
} 


return None

def run_demo(): builder = StateGraph(CustomerSupportState) builder.add_node("customer_service", node_customer_service) builder.add_edge(START, "customer_service") builder.add_edge("customer_service", END) graph = builder.compile()

# 初始化外部依赖 
my_llm = MockLLM() 
my_db = MockDatabase() 
 
print("==================================================") 
print("场景: 依赖注入演示") 
print("演示目标: 将 LLM 和 DB 对象通过 runtime.context 注入到节点中") 
print("==================================================") 
 
# 准备状态 
initial_state = {"query": "如何升级会员？"} 
 
# user_id为一个配置信息，将其放入到 config当中 
config = { 
    "configurable": { 
        "user_id": "vip_user_999", 
    } 
} 
 
print("[System] 开始运行图，并注入依赖对象...") 
 
context = { 
    "llm": my_llm, 
    "db": my_db 

} 
result = graph.invoke(initial_state,config=config,context=context) 
print(result)

if name == "main": run_demo() 4.1.2 节点输出节点的输出为当前节点对状态的增量更新，而不能将接收到的整个状态实例返回出去。 LangGraph 底层会将所有节点输出的状态，都作为增量状态，并尝试和当前的全局状态做一次合并操作。如果将整个状态都输出，对于没有配置任何 reducer 的状态键，langgraph 底层无法完成合并，会抛出异常；对于配置了 reducer 的状态键，如果节点输出了不属于该节点更新的状态，也会导致数据产生问题。以下演示错误返回了整个状态的问题： from typing import TypedDict from langgraph.constants import START,END from langgraph.graph import StateGraph

class MyState(TypedDict): query:str file_result:str web_result:str final_answer:str

def query_web(state:MyState)->dict: """ 做网络搜索，返回搜索结果 :return: """ # 1、错误演示：直接 return整个 state，而非当前节点增量修改的状态 query = state['query'] state['web_result'] = f'{query}的网络搜索结果' return state

def query_file(state:MyState)->dict: """ 做文件搜索，返回搜索结果 :return: """ query = state['query']

state['file_result'] = f'{query}的文件搜索结果' 
return state

def answer(state:MyState)->dict: """ 返回最终的答案 :return: """ web_result = state['web_result'] file_result = state['file_result'] final_answer = f'LLM基于{web_result}，{file_result} 的最终结果' state['final_answer'] = final_answer return state

graph = StateGraph(state_schema=MyState) graph.add_node(answer) graph.add_node(query_web) graph.add_node(query_file) graph.add_edge(START,'query_web') graph.add_edge(START,'query_file') graph.add_edge('query_web','answer') graph.add_edge('query_file','answer') compiled_graph = graph.compile() res = compiled_graph.invoke({"query":"什么是 Langgraph"}) print(res['final_answer']) 4.2 特殊节点 START 和 END 节点都是 LangGraph 当中的特殊节点。START 它代表着将用户输入发送到图中的节点。引用此节点的主要目的是确定应首先调用哪些节点。END 节点是一个特殊节点，代表终止节点。当想表示哪些边在完成后没有动作时，会引用这个节点。 START 和 END 节点本质上是一个字符串，如下所示： END = sys.intern("end") """The last (maybe virtual) node in graph-style Pregel.""" START = sys.intern("start") """The first (maybe virtual) node in graph-style Pregel.""" START 作为一个特殊节点，在图执行完 START 节点之后，也会创建一个 checkpoint，参考 3.3.2 节。 4.3 节点缓存

LangGraph 支持基于节点输入对节点进行缓存。对于配置了缓存的节点，且缓存结果没有过期，以相同的输入再次调用节点时，可直接从缓存当中读取结果，不需要再进行节点计算。使用缓存的方法如下：（1）编译图时指定缓存存储后端：langgraph.cache 包当中提供了内存级别的缓存后端， Redis 缓存后端，和 Sqlite 缓存后端的具体实现，需要构造相关实例，并在图编译时传入。（2）为节点指定缓存策略。每个缓存策略需配置： ➢ key_func：用于根据节点的输入生成缓存键，默认情况下是使用 pickle 对输入进行 hash 运算的结果。 ➢ ttl：缓存的生存时间（以秒为单位）。如果未指定，缓存将永不过期。 import time from typing_extensions import TypedDict from langgraph.graph import StateGraph from langgraph.cache.memory import InMemoryCache

from langgraph.types import CachePolicy from langgraph.constants import START from langgraph.checkpoint.memory import InMemorySaver

class State(TypedDict): x: int result: int

builder = StateGraph(State) checkpointer = InMemorySaver()

1、定义节点：模拟一些有耗时计算的节点

def expensive_node(state: State) -> dict[str, int]: # expensive computation print(f"expensive_node 被调用") time.sleep(5) print(f"expensive_node 计算完成") return {"result": state["x"] * 2}

2、添加节点时，为节点配置缓存策略，这里设置为 10秒缓存，缓存键的配置同样使用

默认的方式，可以在 CachePolicy当中传入 key_func配置不同的缓存键生成策略 builder.add_node("expensive_node", expensive_node, cache_policy=CachePolicy(ttl=10))

builder.add_edge(START, "expensive_node")

3、图编译时，传入缓存器，此处使用 InMemoryCache，也可以使用 RedisCache等

graph = builder.compile(cache=InMemoryCache(),checkpointer=checkpointer)

4、第一次调用，传入的状态为 5

print(graph.invoke({"x": 5},config={"configurable":{"thread_id":"1"}}))

5、第二次调用，传入相同的状态，由于缓存策略，会直接从缓存中返回结果，而不会

重新调用节点 print(graph.invoke({"x": 5},config={"configurable":{"thread_id":"2"}})) 4.4 节点重试在很多使用场景中，有些节点由于客观原因限制，导致其执行过程是不稳定的。因此，我们可能希望节点拥有自定义的重试策略，例如在调用 API、查询数据库或调用大语言模型（LLM）等情况下。为节点添加重试策略，需要在 add_node 中设置 retry_policy 参数。retry_policy 参数接受一个 RetryPolicy 命名元组对象。RetryPolicy 对象有两个属性：max_attempts 和 retry_on 参数，前者定义了总共重试次数，后者定义了对于哪些异常类型进行重试。示例代码如下： import random from typing import Dict, Any from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.types import RetryPolicy

定义状态

class State(TypedDict): result: str

模拟不稳定的 API调用，使用全局变量跟踪尝试次数

attempt_counter = 0

def unstable_api_call(state: State) -> Dict[str, Any]: """ 模拟一个不稳定的 API调用，有一定概率失败 """ global attempt_counter attempt_counter += 1 print(f"尝试调用 API，这是第 {attempt_counter} 次尝试")

# 模拟前几次尝试失败，最后一次成功 
if attempt_counter < 3: 
    raise Exception(f"模拟 API调用失败 (尝试 {attempt_counter})") 
else: 
    # 第三次尝试成功 
    return { 
        "result": f"API调用成功，经过 {attempt_counter} 次尝试" 
    }

模拟抛出 ValueError 的节点

def value_error_call(state: State) -> Dict[str, Any]: """ 模拟抛出 ValueError 的节点（不会被默认重试策略重试） """ print("调用会抛出 ValueError 的节点") raise ValueError("模拟 ValueError 异常")

def run_demo(): print("=== LangGraph 节点重试策略演示 ===\n")

# 重置全局计数器 
global attempt_counter 
attempt_counter = 0 
 
# 演示 1: 使用默认重试策略 
print("1. 使用默认重试策略:") 
print("   默认策略会对除特定异常外的所有异常进行重试") 
print("   不会重试的异常包括: ValueError, TypeError, ArithmeticError,

ImportError,") print(" LookupError, NameError, SyntaxError, RuntimeError,") print(" ReferenceError, StopIteration, StopAsyncIteration, OSError\n")

builder1 = StateGraph(State)

添加节点，使用默认重试策略，

builder1.add_node( 
    "unstable_call",  
    unstable_api_call,  
    retry_policy=RetryPolicy(max_attempts=5)  # 允许最多 5次尝试 
) 
 

builder1.add_edge(START, "unstable_call") 
builder1.add_edge("unstable_call", END) 
 
graph1 = builder1.compile() 
 
print("测试默认重试策略:") 
try: 
    result = graph1.invoke({"result": ""}) 
    print(f"最终结果: {result}\n") 
except Exception as e: 
    print(f"最终失败: {type(e).__name__}: {e}\n") 
 
# 重置全局计数器 
attempt_counter = 0 
# 演示 2: 不会重试的异常类型 
print("3. 测试不会重试的异常类型:") 
 
builder3 = StateGraph(State) 
 
# 添加节点，使用默认重试策略 
builder3.add_node( 
    "value_error_call",  
    value_error_call,  
    retry_policy=RetryPolicy(max_attempts=3) 
) 
 
builder3.add_edge(START, "value_error_call") 
builder3.add_edge("value_error_call", END) 
 
graph3 = builder3.compile() 
 
print("测试 ValueError（默认策略不会重试）:") 
try: 
    result = graph3.invoke({"result": ""}) 
    print(f"最终结果: {result}\n") 
except Exception as e: 
    print(f"最终失败: {type(e).__name__}: {e}\n")

if name == "main": run_demo() 4.5 图内外数据传递

默认情况下，通过 graph.invoke 调用图时，仅在整个图的执行过程都结束之后，我们才能够拿到最终的状态，那么如果想要在图的执行过程当中，想要获取到图内部所产生的数据，应该如何实现？考虑 Agent 的一个场景：首先，我们希望在 LLM 输出时，就能够拿到 LLM 所产生的 token，在前端进行展示；其次，如果 Agent 的流程较长，我们希望能够拿到，当前正在执行的节点或者流程是什么。由于 LangGraph 实现了 langchain_core 当中的 Runnable 接口，其为我们提供了 stream 和 astream 方法，也即流式输出方法，通过流式输出，就能解决前面所说到的问题。 Stream 方法提供了多种不同的模式，如下表所示：模式描述 values 每一个执行后，流式输出完整的状态 updates 图执行过程中，每一步执行后流式输出增量更新。如果在同一个步当中产生了多个增量更新，这些增量更新会分别流式输出。 custom 流式输出节点内部的自定义数据。 messages 在任何调用了 LLM 的节点当中，流式输出两元组数据：（LLM Token，metadata） debug 流式输出所有能输出的信息混合模式流模式传入列表，在列表当中添加多种不同的模式，可以得到多种流式输出具体代码如下所示： import time from typing import TypedDict, Annotated, List import operator from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.types import StreamWriter from langgraph.runtime import Runtime llm = ChatOpenAI(model="gpt-4o-mini")

--- 1. 定义状态 ---

class State(TypedDict): messages: Annotated[List[BaseMessage], operator.add] current_step: str

--- 2. 定义节点 ---

def node_input(state: State): """模拟接收用户输入""" return { "messages": [HumanMessage(content="请帮我写一段 Python代码")], "current_step": "input_received" }

def node_processing(state: State, runtime: Runtime): """ 模拟中间处理过程，并使用 writer 输出自定义流式数据这对应 stream_mode="custom" """ steps = ["正在分析意图...", "正在检索知识库...", "正在构建 Prompt..."] writer = runtime.stream_writer for i, step in enumerate(steps): time.sleep(0.5) # 模拟耗时操作

    # 使用 writer 发送自定义数据 (不影响图的状态) 
    # 这些数据只能通过 stream_mode="custom" 接收到 
    writer({ 
        "step_index": i + 1, 
        "description": step, 
        "timestamp": time.time() 
    }) 
     
return {"current_step": "processing_complete"}

def node_generation(state: State): """ 模拟 LLM 生成过程 LangGraph 的 stream_mode="messages" 会自动捕获 LLM 的流式输出 """

# 调用 LLM 
response = llm.invoke(state["messages"]) 
 
return { 
    "messages": [response], 
    "current_step": "generation_complete" 
}

--- 3. 构建图 ---

builder = StateGraph(State)

builder.add_node("input", node_input) builder.add_node("process", node_processing) builder.add_node("generate", node_generation)

builder.add_edge(START, "input") builder.add_edge("input", "process") builder.add_edge("process", "generate") builder.add_edge("generate", END)

graph = builder.compile()

--- 4. 演示不同的流式输出模式 ---

def run_demo(): initial_state = {"messages": [], "current_step": "start"}

print(f"\n{'='*20} 1. Mode: values (输出完整状态) {'='*20}") 
print("描述: 每执行完一个节点，输出当前的完整 State") 
for event in graph.stream(initial_state, stream_mode="values"): 
    # event 就是当前的 State 字典 
    print(f"State: keys={list(event.keys())},

step={event.get('current_step')}")

print(f"\n{'='*20} 2. Mode: updates (输出增量更新) {'='*20}") 
print("描述: 每执行完一个节点，输出该节点返回的增量数据") 
for event in graph.stream(initial_state, stream_mode="updates"): 
    # event 是一个字典，key是节点名，value是该节点的输出 
    handle_event(event) 
    print(f"Update: {event}") 

print(f"\n{'='*20} 3. Mode: custom (输出自定义数据) {'='*20}") 
print("描述: 仅输出节点内部通过 writer() 发送的数据") 
for event in graph.stream(initial_state, stream_mode="custom"): 
    # event 就是 writer() 中传入的对象 
    print(f"Custom Data: {event}") 

print(f"\n{'='*20} 4. Mode: messages (输出 LLM Token) {'='*20}") 
print("描述: 输出 LLM 生成的消息片段 (Token)") 
# 注意：FakeListChatModel 默认是一次性返回，但在 messages 模式下会被包装

成 chunk for chunk, metadata in graph.stream(initial_state, stream_mode="messages"): # chunk 是 BaseMessageChunk 对象 # metadata 包含 node 信息 node_name = metadata.get('langgraph_node', 'unknown')

    print(f"[{node_name}] Token: {chunk.content!r}") 
    # print(chunk.content,end="") 
    time.sleep(0.2) 
     

print(f"\n{'='*20} 5. Mode: debug (调试模式) {'='*20}") 
print("描述: 输出所有详细的执行信息，包括任务调度、输入输出等") 
# debug 模式输出量很大，这里只打印前几条示意 
count = 0 
for event in graph.stream(initial_state, stream_mode="debug"): 
    if count < 3: 
        print(f"Debug Event: {event['type']} - {event.get('payload',

{}).get('name')}") count += 1 print("... (省略后续 debug 信息)")

print(f"\n{'='*20} 6. Mixed Mode (混合模式) {'='*20}") 
print("描述: 同时获取 updates 和 custom 数据") 
# 返回的是 (mode, data) 元组 
for mode, data in graph.stream(initial_state,

stream_mode=["updates", "custom"]): if mode == "updates": print(f"[Updates] 来自节点 {list(data.keys())[0]}") elif mode == "custom": print(f"[Custom] {data['description']}")

if name == "main": run_demo() 4.6 人工审核节点 Agent 在完成用户设定的任务时，有时候我们希望用户参与当中部分重要决策过程。 LangGraph 为此提供了一个非常方便的原语：interrupt。其原理如下图所示：

可以在节点内部任意位置引入 interrupt，当节点执行到 interrupt 所在位置时，就会停止执行，并可以得到从图内传递出来的，需要用户审核的数据，在用户进行相关审核完成后，又可通过 graph.invoke，让图继续往下执行。具体示例如下： """ LangGraph interrupt 演示：让用户参与关键决策

场景：准备执行一笔“转账”，在真正执行前让用户审核/修改转账信息。 """

from future import annotations

from typing import Any from typing_extensions import TypedDict

from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import END, START, StateGraph from langgraph.types import Command, interrupt

class TransferState(TypedDict): recipient: str

amount: int 
memo: str 
approved: bool 
final_status: str

def review_transfer(state: TransferState) -> dict[str, Any]: print("\n[Node] review_transfer：生成待执行的转账请求") pending_transfer = { "recipient": state["recipient"], "amount": state["amount"], "memo": state["memo"], }

user_review = interrupt( 
    { 
        "title": "转账审核", 
        "pending_transfer": pending_transfer, 
        "instruction": "请返回 bool(是否批准) 或 dict(可改

recipient/amount/memo，并带 approved 字段)。", } )

approved = False 
updated_transfer = dict(pending_transfer) 
if isinstance(user_review, bool): 
    approved = user_review 
elif isinstance(user_review, dict): 
    approved = bool(user_review.get("approved", True)) 
    for k in ("recipient", "amount", "memo"): 
        if k in user_review: 
            updated_transfer[k] = user_review[k] 

print(f"[Node] review_transfer：用户审核结果 approved={approved}，

transfer={updated_transfer}") return { "approved": approved, "recipient": updated_transfer["recipient"], "amount": updated_transfer["amount"], "memo": updated_transfer["memo"], }

def execute_transfer(state: TransferState) -> dict[str, str]: if not state["approved"]: print("\n[Node] execute_transfer：用户未批准，取消转账")

    return {"final_status": "已取消：用户未批准转账"} 

print("\n[Node] execute_transfer：模拟执行转账") 
recipient = state["recipient"] 
amount = state["amount"] 
memo = state["memo"] 
return {"final_status": f"已转账：收款人={recipient}，金额={amount}，

备注={memo}"}

def build_graph(): builder = StateGraph(TransferState) builder.add_node("review_transfer", review_transfer) builder.add_node("execute_transfer", execute_transfer)

builder.add_edge(START, "review_transfer") 
builder.add_edge("review_transfer", "execute_transfer") 
builder.add_edge("execute_transfer", END) 

return builder.compile(checkpointer=InMemorySaver())

def run_demo(): graph = build_graph() config = {"configurable": {"thread_id": "interrupt-demo-1"}}

print("==================================================") 
print("场景：转账前人工审核（interrupt）") 
print("第一次 invoke 会停在 interrupt，并把审核数据带出来") 
print("第二次 invoke 使用 Command(resume=...) 把用户决策传回去继续执行

") print("==================================================")

initial_state: TransferState = { 
    "recipient": "Alice", 
    "amount": 100, 
    "memo": "午饭 AA", 
    "approved": False, 
    "final_status": "", 
} 

first = graph.invoke(initial_state, config=config) 
interrupt_payload = first["__interrupt__"][0].value 
print("\n[System] 图已暂停，等待用户审核。可给用户展示的数据如下：") 
print(interrupt_payload) 


user_resume_value = {"approved": True, "amount": 80, "memo": "改为 80

元（实际应付）"} final = graph.invoke(Command(resume=user_resume_value), config=config) print("\n[System] 已恢复执行，最终结果：") print(final["final_status"])

if name == "main": run_demo() 需要注意的是，使用 interrupt 时，在第二次调用 graph.invoke(command)继续执行时，有 interrupt 的函数，会从函数起点往下执行，如果在函数当中，有更新数据、调用 API 接口等相关操作时，会造成多次重复执行的情况。因此，一定要保证这种类型操作的幂等性，或者是将其封装到一个单独的节点当中，避免在恢复时再次执行。第 5 章边 5.1 边的本质定义边，本质上是定义了，Pregel 当中节点订阅的状态通道和节点执行之后，需要更新的状态通道。边有几种关键类型： ➢ Normal Edges: 普通边。直接从一个节点连接到下一个节点。 ➢ Conditional Edges: 条件边。调用函数以确定接下来要前往哪个（哪些）节点。 5.2 条件边条件边的本质是一个路由函数，其根据当前状态动态决定下一个要执行的节点。 from typing import Literal from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END

定义状态

class GraphState(TypedDict): value: int step: str

定义节点函数

def node_a(state: GraphState) -> dict: """节点 A""" print("执行节点 A") return {"value": state["value"], "step": "A执行完毕"}

def node_b(state: GraphState) -> dict: """节点 B""" print("执行节点 B") return {"value": state["value"] * 2, "step": "B执行完毕"}

def node_c(state: GraphState) -> dict: """节点 C""" print("执行节点 C") return {"value": state["value"] - 1, "step": "C执行完毕"}

条件边的路由函数

def route_condition(state: GraphState) -> Literal["node_b", "node_c"]: """根据 value值决定路由到哪个节点""" if state["value"] % 2 == 0: return "node_b_alias" # 偶数路由到节点 B else: return "node_c_alias" # 奇数路由到节点 C

def main(): """演示条件边""" print("=== 条件边演示 ===")

# 创建图 
builder = StateGraph(GraphState) 
 
# 添加节点 
builder.add_node("node_a", node_a) 
builder.add_node("node_b", node_b) 
builder.add_node("node_c", node_c) 
 
# 添加边 
builder.add_edge(START, "node_a")  # 入口点 
 
# 添加条件边 
builder.add_conditional_edges( 
    "node_a",  # 源节点 
    route_condition,  # 路由函数 
    {  # 路由映射 
        "node_b_alias": "node_b", 
        "node_c_alias": "node_c" 
    } 
) 
 

# 从 B和 C到结束 
builder.add_edge("node_b", END) 
builder.add_edge("node_c", END) 
 
# 编译图 
graph = builder.compile() 
 
# 执行图 - 偶数情况 
print("输入值为偶数:") 
result = graph.invoke({"value": 2}) 
print(f"执行结果: {result}") 
 
# 执行图 - 奇数情况 
print("\n输入值为奇数:") 
result = graph.invoke({"value": 1}) 
print(f"执行结果: {result}\n")

if name == "main": main() 5.3 可控循环通过条件边，我们可以构建带有循环结构的图，例如在典型的 REACT 模式下，工具调用和大模型总结生成结果形成了一个循环，如下图所示：

需要注意的是，这种带循环的图结构，有一个隐藏的问题：图执行过程当中，可能因为某些原因，导致一直在循环内循环往复执行，因此 LangGraph 也提供了一个强制使图的执行终止的递归限制参数。递归限制设定了图在抛出错误之前允许执行的超级步骤数量，默认值25，在graph.invoke 的 config 参数中指定。在经过指定数量的超级步骤后，图还没有自然停止执行时，LangGraph 会抛出异常 GraphRecursionError。示例代码如下所示：

from typing import Annotated, Dict, Literal from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.errors import GraphRecursionError

class LoopState(TypedDict): count: int result: str max_count: int

def node_a(state: LoopState) -> dict: """节点 a：处理逻辑并更新计数""" print(f"执行节点 a，当前计数: {state['count']}") return { 'count': state['count'] + 1, 'result': f"已处理{state['count']}次" }

def node_b(state: LoopState) -> dict: """节点 b：辅助处理""" print(f"执行节点 b，当前计数: {state['count']}") return { 'result': f"已处理{state['count']}次 - 辅助处理" }

def route(state: LoopState) -> Literal["b", END]: """条件路由函数：决定是继续循环还是终止""" # 终止条件：当计数达到最大值时终止 if state['count'] >= state['max_count']: print(f"满足终止条件，计数 {state['count']} >= {state['max_count']}，返回 END") return END else: print(f"未满足终止条件，计数 {state['count']} < {state['max_count']}，返回 b") return "b"

创建图

builder = StateGraph(LoopState)

添加节点

builder.add_node("a", node_a) builder.add_node("b", node_b)

添加边

builder.add_edge(START, "a") builder.add_conditional_edges("a", route) builder.add_edge("b", "a")

编译图

graph = builder.compile()

执行图

print("=== 开始执行工作流 ===") try: result = graph.invoke(input={ 'count': 0, 'result': '', 'max_count': 300 }, config={ 'recursion_limit': 6 # 设置递归限制 }) print("=== 执行结果 ===") print(result) except GraphRecursionError as e: print(f"递归错误: {e}")

LangGraphV1.1.0

pip install grandalf

1. 默认 Reducer（覆盖更新）

3. operator.add Reducer（列表追加）

6. 字符串连接 Reducer

7. 数值累加 Reducer

自定义 Reducer

pip install langgraph-checkpoint-sqlite

1、查看当前图的 channels

2、查看当前图的节点

3、查看 a节点的 trigger（当前节点的订阅）和 writers

4、查看 d节点的 trigger和 writers

5、执行图，查看执行结果

1、最终状态

2、查看历史所有状态

1、定义节点：模拟一些有耗时计算的节点

2、添加节点时，为节点配置缓存策略，这里设置为 10秒缓存，缓存键的配置同样使用

3、图编译时，传入缓存器，此处使用 InMemoryCache，也可以使用 RedisCache等

4、第一次调用，传入的状态为 5

5、第二次调用，传入相同的状态，由于缓存策略，会直接从缓存中返回结果，而不会

定义状态

模拟不稳定的 API调用，使用全局变量跟踪尝试次数

模拟抛出 ValueError 的节点

添加节点，使用默认重试策略，

--- 1. 定义状态 ---

--- 2. 定义节点 ---

--- 3. 构建图 ---

--- 4. 演示不同的流式输出模式 ---

定义状态

定义节点函数

条件边的路由函数

创建图

添加节点

添加边

编译图

执行图