AI创想

标题: Agent之LangGraph [打印本页]

作者: 米落枫 时间: 4 天前
标题: Agent之LangGraph
作者：CSDN博客
一、LangGraph概念

1.1 什么是LangGraph

(, 下载次数: 13)

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模式下，工具调用和大模型总结生成结果形成了一个循环，如下图所示：

(, 下载次数: 15)

      需要注意的是，这种带循环的图结构，有一个隐藏的问题：图执行过程当中，可能因为某些原因，导致一直在循环内循环往复执行，因此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

欢迎光临 AI创想 (https://llms-ai.com/)