AI创想

标题: LangGraph概念全解 [打印本页]

---

作者: 米落枫    时间: 昨天 23:26
标题: LangGraph概念全解
作者：CSDN博客
LangGraph概念全解

智能体（Agent）

随着LLM能力不断提升，越来越多系统开始接入LLM，让LLM成为系统中的一个关键组件。大家目前也都习惯把这些应用系统叫做智能体系统（Agentic），究竟什么是智能体系统呢？一个智能体系统应该具备什么能力？总体来说就是：一个系统不再通过硬编码或者规则编码来控制工作流，而是让LLM来决定。

特斯拉的自动驾驶系统早期一直依赖基于规则的算法。整个系统从车辆摄像头获取视觉数据，识别车道标记、行人、车辆、交通信号及8个摄像头范围内的所有事物，再应用一系列规则，比如红灯停、绿灯行、保持在车道线标记正中、不越过双黄线闯入对面车道等等，特斯拉的工程师编写了数十万行C++代码来来应对各种复杂的场景。直到史洛夫找到马斯克提出用AI来解决自动驾驶问题，后来他们发现，直接用AI来实现自动驾驶后，不仅直接可以删除那30多万行代码，而且整体运行速度比之前还快了10倍。

整体来说，Agent目前主要体现在以下几个能力上：

工具调用（Tool calling）：一般Agent识别用户意图后，常常需要调用工具来执行，获取工具结果采取行动（Action taking）：识别用户意图，采取相应的行动，一般包括工具调用记忆（Memory）：让Agent拥有记忆能够使得在问题理解上更智能，不那么呆板计划（Planning）：智能体能够自己基于问题的复杂度进行规划，拆分成可执行的步骤。

‍
LangGraph的三个核心理念

可控制性（Controllability） ：
众所周知，LLM的输出不稳定，同样一个问题，大模型都可能采用不同的回复，甚至出现不同的理解，导致结果南辕北辙，如何通过框架设计，让大模型能够可控的输出是非常重要的一项框架能力。
人机协同（Human-in-the-Loop）
用过LLM的聊天模型可以发现一个问题，不管你的问题有多简略，大模型都会始终想办法去揣测你的意图，然后一定会给你一些回答。要让Agent像人一样，就应该也能够支持分析用户提问意图，当意图不清晰时，Agent应该能够进一步跟人类进行沟通，澄清，直到完全明白用户的问题后再去执行任务。LangGraph目前根据一些具体场景，支持了该模式。
流式输出（Streaming First）
LangGraph认为大模型的处理一般都是IO耗时的任务，所以能够实时看到大模型的输出是非常重要的一个特性。

‍
LangGraph中引入的主要概念

图（Graphs）

LangGraph的核心概念就是把Agent工作流以图的方式进行建模。
Graphs使用以下三个关键组件：

状态（State）
状态是一个共享的数据结构，通常是一个TypedDict或者Pydantic的BaseModel类型
节点（Nodes）
节点是一个Python函数，接受一个State作为输入，经过内部计算后，返回更新后的State
边（Edges）
边也是一个Python函数，基于当前State，决定下一步执行哪个/哪一些节点。

LangGraph的图，借鉴于Google的Pregel图计算系统。有兴趣可用了解下。
‍
状态图（StateGraph）

StateGraph是LangGraph主要使用的一个类，这是由用户定义的State对象参数化的。
‍
消息图（MessageGraph）

MessageGraph是Graph的一个特例，Graph的State类是一个消息列表，主要用于聊天型Agent。
‍
状态（State）

通常，定义一个StateGraph前，先要定义一个State。定义一个State一般需要定义它的Schema和reducer函数，reducer函数实现了如何更新状态图的方法。
图的schema会作为所有Node和Edge输入schema。State一般继承一个TypedDict或者Pytdantic类。
以下用一个代码例子来说明：

1. from typing import TypedDict, Annotated
2. from operator import add
4. # 定义一个节点函数defchange_bar(state):return{"bar":["bye"]}# 定义状态classState(TypedDict):
5.     foo:int
6.     bar:Annotated[list[str],add]#reducer函数为addfrom langgraph.graph import StateGraph,END
8. graph = StateGraph(State)
9. graph.add_node("change_bar",change_bar)
10. graph.set_entry_point("change_bar")
11. graph.add_edge("change_bar",END)
12. app = graph.compile()
14. app.invoke({"foo":2,"bar":["hi"]})# out：{'foo': 2, 'bar': ['hi', 'bye']}

复制代码

可用看到，初始State实例：{"foo":2,"bar":["hi"]}​经过change_bar节点后，bar的值追加了change_bar节点的返回值变成了：{'foo': 2, 'bar': ['hi', 'bye']}​。这就是reducer函数的作用。它可用指定State值的更新方式。
‍
消息状态（MessageState）

MessageState是State的一个特例，主要用于聊天模型时，把消息列表当作State在节点之间传递。
定义一个MessageState代码如下：

1. from langchain_core.messages import AnyMessage
2. from langgraph.graph.message import add_messages
3. from typing import Annotated,TypedDict
5. classMessageState(TypedDict):
6.     messages:Annotated[list[AnyMessage],add_messages]

复制代码

add_messages 方法除了把消息添加到messages列表，还会把OpenAI消息格式转换为标准的LangChain消息格式。
一般为了实现的灵活性，在图的多节点处理状态时，不仅仅只需要处理消息列表，可能还有别的数据，所以通常会去实现一个MessageState子类，在子类中加入别的数据，比如：

1. from langgraph.graph import MessagesState
3. classState(MessagesState):
4.     documents:list[str]# 加入文档列表

复制代码

‍
节点（Nodes）

节点通常就是一个Python函数，这个函数通常接收两个参数，第一个位置参数是共享的state，第二个是一个config，比如用户id，线程id会在这里传入。
每个节点的返回必须是一个dict类型
添加节点时，如果没有显示指定节点名称，会把函数的名称当作节点名称。
节点类实现了两个特殊的虚拟节点，即START节点和END节点。
使用START节点和上述graph.set_entry_point("change_bar")​效果等同。

1. from langchain_core.runnables import RunnableConfig
2. from langgraph.graph import StateGraph,END
4. bulider = StateGraph(dict)defnode1(state:dict,config:RunnableConfig):print("In node: ", config["configurable"]["user_id"])return{"node1_out":f"Hello,{state['input']}!"}defnode2(state:dict):
5.     state["node2_out"]="Hello,world!"return state
7. bulider.add_node("node1",node1)
8. bulider.add_node("node2",node2)
9. bulider.add_edge("node1","node2")
10. bulider.add_edge("node2",END)
11. bulider.set_entry_point("node1")
13. app = bulider.compile()
15. app.invoke({"input":"HanMeiMei"},{"configurable":{"user_id":1}})#out：{'node1_out': 'Hello,HanMeiMei!', 'node2_out': 'Hello,world!'}

复制代码

边（Edges）

图的另外一个重要概念就是边（Edges），边决定了整个图如何从一个节点流向下一个节点。边包含如下几种类型：

普通边：总是从一个固定的开始节点流向下一个固定的节点，上述例子中bulider.add_edge("node1","node2")​就是创建了一个普通边
条件边：根据一个条件函数，一个Map参数，决定一个节点执行完后，要通过哪些输出边（outgoing edges），输出边可用被并行执行。类似包容网关。
同时，routing_function默认应该返回后续节点名称或者节点名称列表
1. graph.add_conditional_edges("node_a", routing_function)
复制代码
当然你也可以指定一个Map参数来做输出值和节点名称的对应关系
1. graph.add_conditional_edges("node_a", routing_function,{True:"node_b",False:"node_c"})
复制代码
入口节点：用户输入提交后，指定先执行哪个节点作为图的入口节点。
条件节点：用户输入提交后，根据条件选择执行哪个节点
跟条件边类似，只是开始节点固定为虚拟的START节点，目标节点使用一个函数，然后再指定返回值与节点名称之间的MAP来动态实现下一个执行节点
1. graph.add_conditional_edges(START, routing_function,{True:"node_b",False:"node_c"})
复制代码

‍
Send

默认情况下，节点和边都是提前定义好的，并基于同一个共享State运行，但是有些场景下，无法提前确定边，甚至一个图内会出现State的不同版本。一个常见的例子就是Map-reduce模式，第一个节点根据入参动态生成一个对象列表。
为了支持这种场景，LangGraph支持通过在定义条件边时，引入Send对象，Send对象接收两个参数，第一个是节点名称，第二个是节点对应的State

1. defcontinue_to_jokes(state: OverallState):return[Send("generate_joke",{"subject": s})for s in state['subjects']]
3. graph.add_conditional_edges("node_a", continue_to_jokes)

复制代码

检查点（Checkpointer）

LangGraph有个使用checkpointer实现的内置的持久化层。实现检查点的好处是：

实现上述核心理念之一的：Human-in-the-Loop(HIL) ，有了这个checkpointer之后，用户能够随时中断、修改、恢复图的运行。实现记忆能力。可以使用checkpointer创建一个线程，然后在每个图执行步骤后，保存线程状态数据。任何后续消息都可以发送到该检查点，该检查点将保留其对先前消息的记忆，从而实现记忆能力。

‍
线程（Threads）

LangGraph通过线程实现了多租户之间或用户的多轮会话之间的数据隔离。所以在调用图形时，需要指定具体的thread_id​
‍
检查点状态（Checkpointer state）

系统保存的检查点状态包含两个属性：

values：当前节点的state内容next：接下来要执行的节点，一个python元组结构

获取状态：graph.get_state(config)​
获取状态历史：graph.get_state_history(config)​
更新状态：graph.update_state(config, {"foo": 2, "bar": ["b"]})​
‍
配置（configuration）

创建Graph时，可以指定一个自定义的schema类：

1. classConfigSchema(TypedDict):
2.     llm:str
4. graph = StateGraph(State, config_schema=ConfigSchema)

复制代码

然后把这个类的对象传递给图

1. config ={"configurable":{"llm":"anthropic"}}
3. graph.invoke(inputs, config=config)

复制代码

之后可以节点中使用配置

1. defnode_a(state, config):
2.     llm_type = config.get("configurable",{}).get("llm","openai")
3.     llm = get_llm(llm_type)

复制代码

‍
断点（Breakpoints）

可以在节点执行之前（interrupt_before）和节点执行之后（interrupt_after）设置执行断点，这样就可以让AI跑到一个关键选择点时，由用户进行选择确认，之后让程序继续往下执行，从而实现HIL。
​

(, 下载次数: 0)

上传

点击文件名下载附件

​
在compile时传入对应的断点参数interrupt_before或者interrupt_after，记得使用断点时，一定要使用Checkpointer，否则无法恢复执行。

1. graph = builder.compile(checkpointer=memory, interrupt_before=["step_3"])

复制代码

恢复执行时，不需要再次传入state，直接赋值None就行

1. graph.invoke(None, config=config)

复制代码

‍
流式输出（Streaming）

streaming支持两种模式：

values模式：图的每个步骤之后流式传输状态的完整值。updates模式：图的每个步骤之后，以增量更新的方式流式传输state值

‍

原文地址：https://blog.csdn.net/sinat_35438568/article/details/140638180

---

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

Powered by Discuz! X3.4