一、LangGraph概念1.1 什么是LangGraphLangGraph 是一个低级编排框架和运行时环境用于构建、管理和部署长期运行的有状态智能体agents。核心理念是将 Agent 工作流建模为图Graph本质在执行流程过程中控制执行行为拦截中断LangGraph图包含三部分节点Nodes代表计算单元可以是 LLM 调用、工具执行或任何自定义逻辑边Edges定义节点之间的转换逻辑决定执行流程状态State在整个图执行过程中共享和传递的数据LangGraph提供了构建生产级智能体应用的核心能力持久化执行构建能够从故障中恢复并长时间运行的智能体人机协作在任何时刻检查和修改智能体状态记忆管理支持短期工作记忆和跨会话的长期记忆流式处理专为流式工作流设计生产级部署为有状态、长期运行的工作流提供可扩展的基础设施1.2 与LangChain的区别LangGraph提供了更细粒度的控制让开发者能够精确控制智能体的执行流程可以定制化编排的复杂应用场景。概述LangChain和LangGraph的区别当 LLM 应用需要有状态、可循环、可分支的多步骤控制流时LangChain 很难优雅完成选择 LangGraph。特性LangGraphLangChain抽象级别低级提供细粒度控制高级开箱即用状态管理内置状态机和检查点需要自行管理状态执行模型基于图的并行执行线性链式执行持久化原生支持需要额外实现适用场景复杂、有状态的智能体应用开发简单的链式调用LangGraph在以下场景下应用更加广泛复杂的多智能体系统需要长期记忆的应用需要人工审核的工作流后台处理任务和实时交互需要精细控制的定制化智能体编排LangGraph官方源码地址及官网地址如下GitHub地址https://github.com/langchain-ai/langgraph官方文档LangGraph overview - Docs by LangChain二、LangGraph入门案例2.1 环境准备创建环境conda create –-namelanggraph_venv 3.12 conda activate langgraph_venv pip install langgraph1.0.5 # 确认安装成功 pip show langgraph2.2 入门示例LangGraph代码编写流程第一步 定义状态第二步 定义节点第三步 构建builder添加节点和边第四步 编译图第五步 调用图代码示例from typing import TypedDict from langgraph.constants import START, END from langgraph.graph import StateGraph # 第一步 定义状态 {key:value} class MyAgentState(TypedDict): query:str rag_search:str web_search:str llm_answer:str # 第二步 定义节点 # rag搜索 {“query”.....} def rag_search_node(state:MyAgentState): # state字典根据query这个key获取问题 query state[query] # 模拟rag rag_search f根据用户问题{query}返回rag_search结果 return {rag_search:rag_search} # web搜索 {“query”.....} def web_search_node(state:MyAgentState): query state[query] web_search f根据用户问题{query}返回web_search结果 return {web_search: web_search} # llm整合数据节点 {“query”.....} def llm_node(state:MyAgentState): query state[query] rag_search state[rag_search] web_search state[web_search] # 模拟 llm_answer f用户问题:{query},LLM基于{rag_search}和{web_search}的最终回复 return {llm_answer: llm_answer} # 第三步 构建builder添加节点和边 builder StateGraph(state_schemaMyAgentState) # 添加节点 builder.add_node(rag_search_node) # rag_search_node:rag_search_node builder.add_node(web_search_node) builder.add_node(llm_node) # 添加边 builder.add_edge(START, rag_search_node) builder.add_edge(START, web_search_node) builder.add_edge(rag_search_node, llm_node) builder.add_edge(web_search_node, llm_node) builder.add_edge(llm_node, END) # 第四步 编译图 graph builder.compile() # 第五步 调用图 res graph.invoke({query:如何使用LangGraph}) print(res)特殊说明状态通过Dict返回state变量对node为只读不可通过修改变量的方式进行状态传递。通过Dict返回的状态LangGraph后台会自动进行状态收集与合并防止多并发场景对state进行修改导致的数据脏乱。添加节点向builder中添加节点时LangGraph后台会自动根据函数名生成{“函数名” : “函数名”} 的字典映射关系。添加边向builder中添加 边 时实际是从 LangGraph 维护的 Dict中获取对应的节点所以输入的是字符串。三、LangGraph状态通过LangGraph构建应用程序时第一步就是定义StateState代表了整个图中节点的状态数据以及应用最终结果和目标。3.1 状态的定义State Schema可以通过三种方式定义第一种 继承TypedDict类实现推荐# TypedDict是Python提供的一种类型提示工具用于为字典Dict的键和值指定精确的类型信息。 from typing import TypedDict class MyAgentState(TypedDict): query:str rag_search:str web_search:str llm_answer:str第二种 继承BaseModel类实现# Pydantic BaseModelPydantic 提供运行时数据校验并支持静态类型检查工具进行类型推导。状态类通过继承Pydantic的BaseModel定义键的类型和reducer函数 from pydantic import BaseModel class MyStateFull(BaseModel): rag_result:str web_search_result:str query:str第三种 使用python装饰器 dataclass# 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:str3.2 输入输出隔离在LangGraph 当中可以精细管理输入到图中的状态键有哪些以及输出的状态键有哪些。这是通过初始化StateGraph时分别指定三个参数state_schema、input_schema 和 output_schema 来实现的。state_schema这是图的全局状态空间所有节点都可以访问和写入这个 schema 中的任何字段必须指定不能为空。input_schema定义图接受什么输入是 state_schema 的子集。可选参数如果不指定默认等于 state_schema限制图的输入接口只能传入这些字段output_schema定义图返回什么输出是 state_schema 的子集。可选参数如果不指定默认等于 state_schema限制图的输出接口只返回这些字段from typing import TypedDict from langgraph.constants import START, END from langgraph.graph import StateGraph # 1 定义全局状态 class MyAgentState(TypedDict): query:str rag_search:str web_search:str llm_answer:str # 2 定义输入 和 输出状态类 class InputSchema(TypedDict): query:str class OutputSchema(TypedDict): llm_answer: str # 3 定义节点 def rag_search_node(state:MyAgentState): # state字典根据query这个key获取问题 query state[query] # 模拟rag rag_search f根据用户问题{query}返回rag_search结果 # 返回数据langgraph封装自动把返回数据更新到传输状态里面 return {rag_search:rag_search} def web_search_node(state:MyAgentState): query state[query] web_search f根据用户问题{query}返回web_search结果 return {web_search: web_search} def llm_node(state:MyAgentState): query state[query] rag_search state[rag_search] web_search state[web_search] # 模拟 llm_answer f用户问题:{query},LLM基于{rag_search}和{web_search}的最终回复 # 返回 return {llm_answer: llm_answer} # 4 创建builder添加节点和边 builder StateGraph(state_schemaMyAgentState, input_schemaInputSchema, output_schemaOutputSchema) # 添加节点 builder.add_node(rag_search_node) # rag_search_node:rag_search_node builder.add_node(web_search_node) builder.add_node(llm_node) # 添加边 builder.add_edge(START, rag_search_node) builder.add_edge(START, web_search_node) builder.add_edge(rag_search_node, llm_node) builder.add_edge(web_search_node, llm_node) builder.add_edge(llm_node, END) # 第四步 编译图 graph builder.compile() # 第五步 调用图 res graph.invoke({query:如何使用LangGraph,rag_search:xxxxxx}) print(res)3.3 Reducer函数Reducer用于进行当前增量状态节点输出的状态和全局状态的合并。State中的每个键都有其独立的reducer函数。每个node的返回值中的每个key与全局state_schema中对应的key进行合并更新具体更新逻辑取决于每个key指定的reducer函数。Reducer常用函数有以下几种默认行为未指定Reducer时使用覆盖更新内置reducer函数例如langgraph.graph.messages当中的add_messages函数自定义Reducer支持用户自定义合并逻辑from operator import add from typing import TypedDict, Annotated, List from langgraph.constants import START, END from langgraph.graph import StateGraph # 1 定义全局状态 class MyAgentState(TypedDict): 每个状态类key有默认reducer函数默认效果是覆盖新值覆盖旧值 效果通过一个key改变reducer逻辑把覆盖效果-》合并效果 query:str rag_search:str web_search:str llm_answer:str # reducer operator.add(1,2)3 operator.add([k1],[k2])[k1,k2]合并 test_key:Annotated[List[str],add] # 第二步 定义节点 def rag_search_node(state:MyAgentState): query state[query] # 模拟rag rag_search f根据用户问题{query}返回rag_search结果 # 返回数据langgraph封装自动把返回数据更新到传输状态里面 return {rag_search:rag_search,test_key:[test_key_rag]} def web_search_node(state:MyAgentState): query state[query] web_search f根据用户问题{query}返回web_search结果 return {web_search: web_search,test_key:[test_key_web]} def llm_node(state:MyAgentState): query state[query] rag_search state[rag_search] web_search state[web_search] llm_answer f用户问题:{query},LLM基于{rag_search}和{web_search}的最终回复 return {llm_answer: llm_answer} # 第三步 构建builder添加节点和边 builder StateGraph(state_schemaMyAgentState) # 添加节点 builder.add_node(rag_search_node) # rag_search_node:rag_search_node builder.add_node(web_search_node) builder.add_node(llm_node) # 添加边 builder.add_edge(START, rag_search_node) builder.add_edge(START, web_search_node) builder.add_edge(rag_search_node, llm_node) builder.add_edge(web_search_node, llm_node) builder.add_edge(llm_node, END) # 第四步 编译图 graph builder.compile() # 第五步 调用图 res graph.invoke({query:如何使用LangGraph}) print(res)自定义Reducer函数from typing import List, TypedDict, Annotated from langchain_core.messages import AnyMessage, AIMessage, ToolMessage, HumanMessage from langgraph.constants import START from langgraph.graph import StateGraph # 创建操作方法合并 AIMessage HuManMessage def add_message(message_list_left:List[AnyMessage], message_list_right:List[AnyMessage])-List[AnyMessage]: print(my_add_message_reducer函数的调用) print(message_list_left:, message_list_left) print(message_list_right:, message_list_right) new_message_list message_list_left message_list_right return new_message_list # 创建状态类 class MyAgentState(TypedDict): messages:Annotated[List[AnyMessage],add_message] # 创建节点 def llm_node(state:MyAgentState): ai_message AIMessage(contentxxxx) return {messages:ai_message} def tool_node(state:MyAgentState): tool_message ToolMessage(contentxxxx, tool_call_idxx) return {messages:tool_message} # 创建builder builder StateGraph(state_schemaMyAgentState) builder.add_node(llm_node) builder.add_node(tool_node) builder.add_edge(START,llm_node) builder.add_edge(llm_node, tool_node) graph builder.compile() res graph.invoke( {messages: [HumanMessage(content 你好)]}) print(res)3.4 状态存储在前面的例子当中我们看到每次用户invoke时LangGraph都会初始化一个空状态然后将用户传入的初始状态合并进来再继续往下执行。这在一些一次性、简单任务过程中没有什么问题。但是对于一些复杂任务就会出现一些问题考虑以下两个场景1场景一在agent的一个会话里需要在多次调用当中保持上下文使用checkpointer实现短期记忆和长期记忆2场景二图执行过程当中报错不想重复已执行完节点想要实现断点续传比如 node1 -》node2-》node3node2节点出错了从node2节点开始恢复因为node1已经执行了恢复时候从node2开始执行node1不再执行了故障中恢复注意两点1invoke时传递None作为初始参数2传入相同的thread_id。# 从状态中恢复执行 # 某个节点执行出错从出错节点恢复执行已经执行过不再执行 import sqlite3 from typing import TypedDict from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.constants import START from langgraph.graph import StateGraph # 创建状态 class ResumeState(TypedDict): key_1:str key_2:str key_3:str # 创建节点 def node_1(state:ResumeState): print(state) print(node1节点调用了) return {key_1:value_1} def node_2(state:ResumeState): print(state) print(node2节点调用了) #raise Exception(node2节点出错了....) return {key_2:value_2} def node_3(state:ResumeState): print(state) print(node3节点调用了) return {key_3:value_3} # 创建builder添加节点和边 builder StateGraph(state_schemaResumeState) 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_2, node_3) # 编译和执行 checkpointer SqliteSaver( connsqlite3.connect(./resume_demo.db, check_same_threadFalse), ) graph builder.compile(checkpointercheckpointer) # 第一次调用node2节点出错 # res graph.invoke({}, # config{configurable: # {thread_id:2}}) # print(res) # 第二次调用从node2恢复执行 # 要想从故障中恢复需要以下两点 # 1invoke时传递None作为初始参数 # 2传入相同的thread_id。 res graph.invoke(None, config{configurable: {thread_id:2}}) print(res)3.5 LangGraph底层运行算法在了解如何获取历史状态之前首先我们需要了解一下LangGraph底层算法—Pregel。Pregel 是LangGraph底层的一个类管理 LangGraph 应用程序的运行时runtime行为。也就是说整个图结构从开始到结束的迭代执行过程是由Pregel控制管理的。在这个类当中有两大主要组件Actors (节点)Actor即为在前面介绍的Node对应在LangGraph当中的类为PregelNode其订阅某些通道channels从通道当中读取数据或者写入数据。PregelNode也实现了LangChain的Runnable接口。Channels通道Channels用于actors之间通信每个通道都有一个值类型、一个更新类型和一个更新函数 —— 该函数接收一系列更新并修改存储的值。而在LangGraph图当中节点的执行来自于一个Pregel的超步SuperStep。而一个超步的执行逻辑分为如下三个过程1. Plan阶段确定在此步骤中要执行哪些Actors。例如在第一步中选择订阅特殊输入通道的Actors在后续步骤中选择订阅上一步骤中更新的通道的参与者2. Execute阶段并行执行所有选定的Actors直到所有参与者完成、其中一个失败或达到超时时间3. Update阶段用本步骤中Actors所写的值更新channels。由以上三个步骤组合而成的超步会重复执行直到没有Actors被选中执行或者达到最大步骤数为止。总结节点和通道订阅关系比如a节点到b节点a节点定义to:b通道from typing import TypedDict,Annotated,List from operator import add from langgraph.graph import StateGraph from langgraph.constants import START class MyState(TypedDict): aggregates:Annotated[List[str],add] def node_a(state:MyState): return {aggregates:[a]} def node_b(state:MyState): return {aggregates:[b]} def node_c(state:MyState): return {aggregates:[c]} def node_b_2(state:MyState): return {aggregates:[b_2]} def node_d(state:MyState): return {aggregates:[d]} builder StateGraph(MyState) builder.add_node(node_a) builder.add_node(node_b) builder.add_node(node_c) builder.add_node(node_b_2) builder.add_node(node_d) builder.add_edge(START,node_a) builder.add_edge(node_a,node_b) builder.add_edge(node_a,node_c) builder.add_edge(node_b,node_b_2) builder.add_edge(node_b_2,node_d) builder.add_edge(node_c,node_d) graph builder.compile() print(graph当中的nodes,graph.nodes) print(graph当中的通道channels,graph.channels) res graph.invoke({}) print(res)3.6 获取图执行的历史状态构建好的graph实例可以通过get_state() / get_state_history()方法传入想要获取的历史状态的thread_id即可拿到图执行过程中的历史状态get_state()方法获取到的是最近一个时间步的状态get_state_history()获取到的是图执行当中所有时间步的历史状态其输出值为一个迭代器按照时间步倒序排列。# .................... # 图编译 graph builder.compile(checkpointerInMemorySaver()) # 图调用 res graph.invoke({}, config{configurable: {thread_id:1}}) print(res) print(*50) # 获取图执行的历史状态 history_states graph.get_state_history(config{configurable:{thread_id:1}}) for state in history_states: print(state) print(*50) # 获取最近一次 state graph.get_state(config{configurable:{thread_id:1}}) print(state)四、节点4.1 节点的输入输出在LangGraph中一般来讲节点都是Python函数可以是同步的也可以是异步的它们接受以下参数state图的状态代表了具体的业务数据config一个RunnableConfig对象包含诸如thread_id之类的配置信息在调用图时也可以传递用户定义的其他配置runtime一个Runtime对象包含运行时context可自定context在调用图时传入即可以及其他信息如store和stream_writer等2以上参数会在运行过程中自动被LangGraph运行时注入。节点的输出为当前节点对状态的增量更新而不能将接收到的整个状态实例返回出去。LangGraph底层会将所有节点输出的状态都作为增量状态并尝试和当前的全局状态做一次合并操作。如果将整个状态都输出对于没有配置任何reducer的状态键langgraph底层无法完成合并会抛出异常对于配置了reducer的状态键如果节点输出了不属于该节点更新的状态也会导致数据产生问题。from typing import List, TypedDict from langchain_core.runnables import RunnableConfig from langgraph.constants import START from langgraph.graph import StateGraph from langgraph.runtime import Runtime # 定义状态 class CustomerState(TypedDict): query: str # 用户问题 response: str # 客服回复 log: List[str] # 处理日志 # 定义节点 # state状态 业务数据 # runtime上下文对象比如连接对象db # config配置信息比如thread_id def node_customer_service(state:CustomerState, runtime: Runtime, config: RunnableConfig): print(传入的状态为, state) print(传入的配置为, config) print(传入的runtime为, runtime) # 创建buildler builder StateGraph(CustomerState) builder.add_node(node_customer_service) builder.add_edge(START,node_customer_service) # 图编译 graph builder.compile() # 图执行 # 三个参数 input config context config { user_id:123_vip, configurable:{thread_id:1234} } context{ db:mysqldb, llm:llm_result } res graph.invoke({query:hello},configconfig,contextcontext) print(res)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。注END节点可不写LangGraph默认会自动添加。START节点对应的边必须进行指定。4.3 节点缓存LangGraph支持基于节点输入对节点进行缓存。对于配置了缓存的节点且缓存结果没有过期以相同的输入再次调用节点时可直接从缓存当中读取结果不需要再进行节点计算。使用缓存的方法如下在添加节点时候设置缓存策略包含缓存key信息和ttl过期时间值在图编译时候传递缓存存储方式import time from typing import TypedDict from langgraph.cache.memory import InMemoryCache from langgraph.constants import START from langgraph.graph import StateGraph from langgraph.types import CachePolicy # 需求 根据user_id生成order_id # 定义状态 class MyState(TypedDict): user_id:str order_id:str # 定义节点 def get_order_id(state:MyState): print(调用get_order_id节点.....) # 获取user_id user_id state[user_id] # 根据用户id生成order_id order_id forder_id_{user_id} return {order_id: order_id} # 创建builder builder StateGraph(state_schemaMyState) # 添加节点 # 设置缓存策略 builder.add_node(get_order_id, cache_policyCachePolicy(ttl3)) # 添加边 builder.add_edge(START,get_order_id) # 图编译 # 缓存存储方式 cache InMemoryCache() graph builder.compile(cachecache) # 图执行 res graph.invoke({user_id:123}) print(res) print(*50) res graph.invoke({user_id:123}) print(res) print(*50) time.sleep(5) res graph.invoke({user_id:123}) print(res)4.4 节点重试在很多使用场景中有些节点由于客观原因限制导致其执行过程是不稳定的。因此我们可能希望节点拥有自定义的重试策略例如在调用API、查询数据库或调用大语言模型LLM等情况下。为节点添加重试策略需要在add_node中设置retry_policy参数。retry_policy参数接受一个RetryPolicy命名元组对象。RetryPolicy对象有两个属性max_attempts和retry_on参数前者定义了总共重试次数后者定义了对于哪些异常类型进行重试。重试策略为初始重试时间为0.5秒幂指数级别更新重试间隔时长直到达到最大重试次数默认3次默认开启1s内的重试时长扰动。import datetime from typing import TypedDict from langgraph.constants import START from langgraph.graph import StateGraph from langgraph.types import RetryPolicy # 重试 attempt 0 class MyAgentState(TypedDict): llm_message: str def llm_node(state: MyAgentState): global attempt print(datetime.datetime.now()) attempt 1 if attempt 10: print(f第{attempt}次调用llm失败) raise ConnectionError(调用llm失败) print(f第{attempt}次调用llm成功) return {llm_message: 调用llm成功} builder StateGraph(state_schemaMyAgentState) builder.add_node(llm_node, retry_policyRetryPolicy( max_attempts10, # 最多尝试 10 次含首次 # initial_interval0.5, # 第一次重试前等待 0.5 秒 # backoff_factor2.0, # 每次间隔翻倍0.5 - 1 - 2 - 4 ... # max_interval128.0, # 间隔上限 10 秒 # jitterFalse, # 关闭随机抖动方便观察间隔规律 # retry_on(ConnectionError,), # 只对 ConnectionError 重试 )) builder.add_edge(START, llm_node) graph builder.compile() res graph.invoke({}) print(res)4.5 图内外数据传递默认情况下通过graph.invoke调用图时仅在整个图的执行过程都结束之后我们才能够拿到最终的状态那么如果想要在图的执行过程当中想要获取到图内部所产生的数据应该如何实现考虑Agent的一个场景首先我们希望在LLM输出时就能够拿到LLM所产生的token在前端进行展示其次如果Agent的流程较长我们希望能够拿到当前正在执行的节点或者流程是什么。由于LangGraph实现了langchain_core当中的Runnable接口其为我们提供了stream和astream方法也即流式输出方法通过流式输出就能解决前面所说到的问题。流式输出有下面常见模式1. values每一个执行后流式输出完整的状态2. messages在任何调用了LLM的节点当中流式输出两元组数据LLM Tokenmetadatamessages模式代码实现load_dotenv() llm init_chat_model( modelqwen-plus, model_provideropenai, base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, # 千问API Key api_keyos.getenv(OPENAI_API_KEY), ) def node_stream_messages(): # 状态 class MyAgentState(TypedDict): query:str llm_message:str # 节点 def llm_node(state:MyAgentState): query state[query] response llm.invoke(query) return {llm_message: response.content} # 创建builder builder StateGraph(MyAgentState) builder.add_node(llm_node) builder.add_edge(START, llm_node) # 编译 graph builder.compile() # 流式调用 res graph.stream( {query: 什么是langgraph}, stream_modemessages, ) # 流式输出两元组数据(LLM Tokenmetadata for chunk,metadata in res: # print(metadata) print(chunk.content,end) node_stream_messages()4.6 人工审核节点Agent在完成用户设定的任务时有时候我们希望用户参与当中部分重要决策过程。LangGraph为此提供了一个非常方便的原语interrupt。人工审核就是langchain的中间件流程核心拦截中断节点不执行返回中断信息根据中断信息判断决定是否执行后面节点人工审核流程 调用节点之前进行拦截中断通俗不执行这个节点返回中断信息根据中断信息判断决定是否执行后面节点。会有两次执行第一次返回中断信息审核通过继续往后执行。需要注意的是使用interrupt时在第二次调用graph.invoke(command)继续执行时有interrupt的函数会从函数起点往下执行如果在函数当中有更新数据、调用API接口等相关操作时会造成多次重复执行的情况。因此一定要保证这种类型操作的幂等性或者是将其封装到一个单独的节点当中避免在恢复时再次执行。from typing import TypedDict from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import StateGraph from langgraph.types import interrupt, Command # 状态 class MyAgentState(TypedDict): query:str llm_message:str rag_message:str # 节点 def rag_node(state:MyAgentState): print(rag node调用....) query state[query] return {query:query} def llm_node(state:MyAgentState): print(llm node调用....) query state[query] # 拦截中断 result interrupt({ query: query, msg: 是否调用大语言模型, }) # 模拟判断 if result: rag_message llm大语言模型调用了... else: rag_message 不可用调用大语言模型llm... return {rag_message: rag_message} builder StateGraph(MyAgentState) builder.add_node(rag_node) builder.add_node(llm_node) builder.add_edge(__start__, rag_node) builder.add_edge(rag_node, llm_node) checkpointer InMemorySaver() # 使用interrupt需要给graph配置一个checkpoint graphbuilder.compile(checkpointercheckpointer) res graph.invoke({query:什么是langgraph}, config{configurable:{thread_id:1}}) print(res) # { # query: 什么是langgraph, # __interrupt__: [ # Interrupt(value{ # query: 什么是langgraph, # msg: 是否调用大语言模型 # }, # ide070c7acab89ba244cd4eae235e4be01) # ] # } # 从中断信息获取数据 to_review_datares[__interrupt__][0].value print(to_review_data) # 业务处理 # 审核通过继续往后执行 res graph.invoke(Command(resumeTrue), config{configurable:{thread_id:1}}) print(res)五、边5.1 边的本质定义边本质上是定义了Pregel当中节点订阅的状态通道和节点执行之后需要更新的状态通道。边有几种关键类型Normal Edges: 普通边。直接从一个节点连接到下一个节点。Conditional Edges: 条件边。调用函数以确定接下来要前往哪个哪些节点。5.2 条件边条件边的本质是一个路由函数其根据当前状态动态决定下一个要执行的节点。from typing import TypedDict from langgraph.constants import END from langgraph.graph import StateGraph class MyAgentState(TypedDict): query: str number:int node:str rag_message: str def node_a(state:MyAgentState): print(正在执行node_a) return {node:node_a} def node_b(state:MyAgentState): print(正在执行node_b) return {node:node_b} #函数结合当前状态判断下一个节点 def condtional_function(state:MyAgentState): number state[number] if number % 2 0: return node_b else: return END builder StateGraph(MyAgentState) builder.add_node(node_a) builder.add_node(node_b) # START - node_a (条件)- node_b builder.add_edge(__start__, node_a) # node_a - node_b 添加条件边 builder.add_conditional_edges( sourcenode_a, pathcondtional_function ) graph builder.compile() graph.invoke({number:4})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递归错