作者：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类。
以下用一个代码例子来说明：- from typing import TypedDict, Annotated
- from operator import add
- # 定义一个节点函数defchange_bar(state):return{"bar":["bye"]}# 定义状态classState(TypedDict):
- foo:int
- bar:Annotated[list[str],add]#reducer函数为addfrom langgraph.graph import StateGraph,END
- graph = StateGraph(State)
- graph.add_node("change_bar",change_bar)
- graph.set_entry_point("change_bar")
- graph.add_edge("change_bar",END)
- app = graph.compile()
- app.invoke({"foo":2,"bar":["hi"]})# out：{'foo': 2, 'bar': ['hi', 'bye']}
消息状态（MessageState）
MessageState是State的一个特例，主要用于聊天模型时，把消息列表当作State在节点之间传递。
定义一个MessageState代码如下：- from langchain_core.messages import AnyMessage
- from langgraph.graph.message import add_messages
- from typing import Annotated,TypedDict
- classMessageState(TypedDict):
- messages:Annotated[list[AnyMessage],add_messages]
一般为了实现的灵活性，在图的多节点处理状态时，不仅仅只需要处理消息列表，可能还有别的数据，所以通常会去实现一个MessageState子类，在子类中加入别的数据，比如：- from langgraph.graph import MessagesState
- classState(MessagesState):
- documents:list[str]# 加入文档列表
节点（Nodes）
节点通常就是一个Python函数，这个函数通常接收两个参数，第一个位置参数是共享的state，第二个是一个config，比如用户id，线程id会在这里传入。
每个节点的返回必须是一个dict类型
添加节点时，如果没有显示指定节点名称，会把函数的名称当作节点名称。
节点类实现了两个特殊的虚拟节点，即START节点和END节点。
使用START节点和上述graph.set_entry_point("change_bar")效果等同。- from langchain_core.runnables import RunnableConfig
- from langgraph.graph import StateGraph,END
- 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):
- state["node2_out"]="Hello,world!"return state
- bulider.add_node("node1",node1)
- bulider.add_node("node2",node2)
- bulider.add_edge("node1","node2")
- bulider.add_edge("node2",END)
- bulider.set_entry_point("node1")
- app = bulider.compile()
- app.invoke({"input":"HanMeiMei"},{"configurable":{"user_id":1}})#out：{'node1_out': 'Hello,HanMeiMei!', 'node2_out': 'Hello,world!'}
图的另外一个重要概念就是边（Edges），边决定了整个图如何从一个节点流向下一个节点。边包含如下几种类型：
普通边：总是从一个固定的开始节点流向下一个固定的节点，上述例子中bulider.add_edge("node1","node2")就是创建了一个普通边
条件边：根据一个条件函数，一个Map参数，决定一个节点执行完后，要通过哪些输出边（outgoing edges），输出边可用被并行执行。类似包容网关。
同时，routing_function默认应该返回后续节点名称或者节点名称列表- graph.add_conditional_edges("node_a", routing_function)
- graph.add_conditional_edges("node_a", routing_function,{True:"node_b",False:"node_c"})
条件节点：用户输入提交后，根据条件选择执行哪个节点
跟条件边类似，只是开始节点固定为虚拟的START节点，目标节点使用一个函数，然后再指定返回值与节点名称之间的MAP来动态实现下一个执行节点- graph.add_conditional_edges(START, routing_function,{True:"node_b",False:"node_c"})
Send
默认情况下，节点和边都是提前定义好的，并基于同一个共享State运行，但是有些场景下，无法提前确定边，甚至一个图内会出现State的不同版本。一个常见的例子就是Map-reduce模式，第一个节点根据入参动态生成一个对象列表。
为了支持这种场景，LangGraph支持通过在定义条件边时，引入Send对象，Send对象接收两个参数，第一个是节点名称，第二个是节点对应的State- defcontinue_to_jokes(state: OverallState):return[Send("generate_joke",{"subject": s})for s in state['subjects']]
- graph.add_conditional_edges("node_a", continue_to_jokes)
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类：- classConfigSchema(TypedDict):
- llm:str
- graph = StateGraph(State, config_schema=ConfigSchema)
- config ={"configurable":{"llm":"anthropic"}}
- graph.invoke(inputs, config=config)
- defnode_a(state, config):
- llm_type = config.get("configurable",{}).get("llm","openai")
- llm = get_llm(llm_type)
断点（Breakpoints）
可以在节点执行之前（interrupt_before）和节点执行之后（interrupt_after）设置执行断点，这样就可以让AI跑到一个关键选择点时，由用户进行选择确认，之后让程序继续往下执行，从而实现HIL。
在compile时传入对应的断点参数interrupt_before或者interrupt_after，记得使用断点时，一定要使用Checkpointer，否则无法恢复执行。- graph = builder.compile(checkpointer=memory, interrupt_before=["step_3"])
- graph.invoke(None, config=config)
流式输出（Streaming）
streaming支持两种模式：
values模式：图的每个步骤之后流式传输状态的完整值。updates模式：图的每个步骤之后，以增量更新的方式流式传输state值
原文地址：https://blog.csdn.net/sinat_35438568/article/details/140638180 |
广告