Agent之LangGraph

米落枫 · 发表于 4 天前

作者：CSDN博客
一、LangGraph概念

1.1 什么是LangGraph

LangGraph图包含三部分：

节点（Nodes）

边（Edges）

状态（State）

LangGraph提供了构建生产级智能体应用的核心能力：

1.2 与LangChain的区别

有状态、可循环、可分支的多步骤控制流时

特性	LangGraph	LangChain
抽象级别	低级，提供细粒度控制	高级，开箱即用
状态管理	内置状态机和检查点	需要自行管理状态
执行模型	基于图的并行执行	线性链式执行
持久化	原生支持	需要额外实现
适用场景	复杂、有状态的智能体应用开发	简单的链式调用

LangGraph在以下场景下，应用更加广泛：

LangGraph官方源码地址及官网地址如下：

二、LangGraph入门案例

2.1 环境准备

创建环境：

conda create –-name=langgraph_venv 3.12
conda activate langgraph_venv
pip install langgraph==1.0.5
# 确认安装成功
pip show langgraph

复制代码

2.2 入门示例

LangGraph代码编写流程

代码示例

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_schema=MyAgentState)
# 添加节点
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)

复制代码

特殊说明

三、LangGraph状态

通过LangGraph构建应用程序时，第一步就是定义State，State代表了整个图中节点的状态数据，以及应用最终结果和目标。
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 BaseModel：Pydantic 提供运行时数据校验，并支持静态类型检查工具进行类型推导。状态类通过继承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:str

复制代码

3.2 输入输出隔离

在LangGraph 当中，可以精细管理输入到图中的状态键有哪些，以及输出的状态键有哪些。这是通过初始化StateGraph时，分别指定三个参数：state_schema、input_schema 和 output_schema 来实现的。

state_schema

input_schema

output_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_schema=MyAgentState,
input_schema=InputSchema,
output_schema=OutputSchema)
# 添加节点
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常用函数有以下几种：

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_schema=MyAgentState)
# 添加节点
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(content="xxxx")
return {"messages":ai_message}
def tool_node(state:MyAgentState):
tool_message = ToolMessage(content="xxxx", tool_call_id='xx')
return {"messages":tool_message}
# 创建builder
builder = StateGraph(state_schema=MyAgentState)
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的一个会话里，需要在多次调用当中保持上下文

2）场景二：图执行过程当中报错，不想重复已执行完节点，想要实现断点续传

node2

故障中恢复注意两点：

（1）invoke时传递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_schema=ResumeState)
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(
conn=sqlite3.connect("./resume_demo.db",
check_same_thread=False),
)
graph = builder.compile(checkpointer=checkpointer)
# 第一次调用：node2节点出错
# res = graph.invoke({},
# config={"configurable":
# {"thread_id":"2"}})
# print(res)
# 第二次调用：从node2恢复执行
# 要想从故障中恢复，需要以下两点：
# （1）invoke时传递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 (节点)

Channels（通道）

总结

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(checkpointer=InMemorySaver())
# 图调用
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函数（可以是同步的，也可以是异步的），它们接受以下参数：

   以上参数，会在运行过程中，自动被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"},config=config,context=context)
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支持基于节点输入对节点进行缓存。对于配置了缓存的节点，且缓存结果没有过期，以相同的输入再次调用节点时，可直接从缓存当中读取结果，不需要再进行节点计算。
使用缓存的方法如下：

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 = f"order_id_{user_id}"
return {"order_id": order_id}
# 创建builder
builder = StateGraph(state_schema=MyState)
# 添加节点
# 设置缓存策略
builder.add_node(get_order_id,
cache_policy=CachePolicy(ttl=3))
# 添加边
builder.add_edge(START,"get_order_id")
# 图编译
# 缓存存储方式
cache = InMemoryCache()
graph = builder.compile(cache=cache)
# 图执行
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_schema=MyAgentState)
builder.add_node(llm_node, retry_policy=RetryPolicy(
max_attempts=10, # 最多尝试 10 次（含首次）
# initial_interval=0.5, # 第一次重试前等待 0.5 秒
# backoff_factor=2.0, # 每次间隔翻倍：0.5 -> 1 -> 2 -> 4 ...
# max_interval=128.0, # 间隔上限 10 秒
# jitter=False, # 关闭随机抖动，方便观察间隔规律
# 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 Token，metadata）
messages模式代码实现：

load_dotenv()
llm = init_chat_model(
model="qwen-plus",
model_provider="openai",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
# 千问API Key
api_key=os.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_mode="messages",
)
# 流式输出两元组数据：(LLM Token，metadata）
for chunk,metadata in res:
# print(metadata)
print(chunk.content,end="")
node_stream_messages()

复制代码

4.6 人工审核节点

Agent在完成用户设定的任务时，有时候我们希望用户参与当中部分重要决策过程。LangGraph为此提供了一个非常方便的原语：interrupt。

需要注意的是

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
graph=builder.compile(checkpointer=checkpointer)
res = graph.invoke({"query":"什么是langgraph"},
config={"configurable":{"thread_id":"1"}})
print(res)
# {
# 'query': '什么是langgraph',
# '__interrupt__': [
# Interrupt(value={
# 'query': '什么是langgraph',
# 'msg': '是否调用大语言模型'
# },
# id='e070c7acab89ba244cd4eae235e4be01')
# ]
# }
# 从中断信息获取数据
to_review_data=res["__interrupt__"][0].value
print(to_review_data)
# 业务处理
# 审核通过，继续往后执行
res = graph.invoke(Command(resume=True),
config={"configurable":{"thread_id":"1"}})
print(res)

复制代码

五、边

5.1 边的本质

定义边，本质上是定义了，Pregel当中节点订阅的状态通道和节点执行之后，需要更新的状态通道。
边有几种关键类型：

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(
source="node_a",
path=condtional_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"递归错

复制代码

原文地址：https://blog.csdn.net/weixin_42796403/article/details/162883458

25位大厂高管转战生成式AI创业！吸金猛，不

Agent之LangGraph

收藏 | LangChain核心概念解析：2026最新版

关于我们

产品与服务

解决方案

产品与服务