LangGraph框架状态图&节点&边&Send

创想小编 · 发表于昨天 14:06

作者：CSDN博客
什么是LangGraph

Langgraph是一个用于构建循环、有状态的多智能体（Multi-agent）应用的框架，它的特点是：

持久执行

人机交互

综合记忆

使用LangSmith调试

简易部署

Langgraph与langchain区别

首先，LangGraph和LangChain是两个相关但不同的工具，都来自LangChain生态系统
LangChain

LangChain是一个用于构建大语言模型应用程序的框架

线性工作流

组件库

简单集成

抽象层

中间件：提供十多种预构建中间件和自定义中间件去进行Agent执行中的拦截

LangGraph

LangGraph是LangChain团队开发的更高级工具，专门用于构建复杂的多智能体系统：

图状工作流

状态管理

多智能体协作

复杂决策流

持久化

总而言之，Langchain的工作流程很简单，就是用户Query——向量检索——模型生成——格式化输出，对于那些流程明确的工作（比如文本翻译、问答），就适合langchain框架。
而langgraph更适合解决复杂的任务，langgraph原生支持持久化不需要像langchain一样专门存储历史记录，langgraph支持循环结构，允许模型在犯错的时候从上一步重新开始，同时langgraph支持多智能体协作，在一些重要决策前加入人工判断的步骤。
langgraph的基础知识

langgraph安装

pip install -U "langgraph"

复制代码

核心基础

State（状态）	整个工作流中被传输和存储的数据的数据结构
Node（结点）	工作流中执行的一个步骤（调用函数、工具）
Edge（边）	控制流程下一步应该去哪个结点
Graph（流程图）	将所有边和结点组织成一张流程图

LangGraph 的核心是将代理工作流程建模为图表，它有三个核心组件：

TypedDict

Pydantic的BaseModel，

state

Node

通过组合Node和Edge，您可以创建复杂的循环工作流，使其state随时间推移而演化。然而，真正的强大之处在于 LangGraph 对的管理发送state
状态图Stategraph

Stategraph是主要使用的python图形类，这个类提供了很多方法：

为什么叫“状态图”而不是“流程图”
LangGraph 不只是流程控制，还强调：

访问和修改状态

节点的跳转可以依据状态来判断

from typing import TypedDict
from langgraph.graph import StateGraph
# 定义状态结构
class MyState(TypedDict):
question: str
answer: str
# 定义节点函数
def search_node(state):
return {"answer": "这是答案"}
# 创建状态图
builder = StateGraph(state_schema=MyState)
# 添加一个节点
builder.add_node("search", search_node)
# 第一个要调用的节点
builder.set_entry_point("search")
# 要构建图，首先要定义状态，然后添加节点和边，最后进行编译，会进行基本的检查
graph = builder.compile()
# 执行图
result = graph.invoke({"question": "什么是状态图？"})
print(result["answer"]) # 输出：这是答案

复制代码

状态State

在使用 LangGraph 构建流程图之前，第一件事就是定义图的状态 State。这是整个图运行中用于共享和传递信息的核心机制。
LangGraph 中的 State 是图中所有节点（Node）之间传递数据的模式结构，可以类比为一个共享的上下文字典，它包含输入、输出、中间变量等
定义 State 时，需要包含两个部分：
Schema（模式）：指定 State 的字段结构（一般用TypedDict或Pydanic）

# langgraph推荐使用TypedDict
"""
1. TypedDict 是标准库的一部分（来自 typing 模块），零依赖，零性能开销而 Pydantic 会在每一步创建模型实例，会增加运行时负担
2. LangGraph 中的 State 实质就是一个字典（dict），而 TypedDict 就是“有类型注解的 dict”，与 LangGraph 的执行机制无缝对接，而 Pydantic 是类结构，需要 .dict() 转换，略显多余
"""
from typing import TypedDict
class State1(TypedDict):
user_input: str
# 使用 pydantic 可以进行参数校验和提供默认值
from pydantic import BaseModel
class State2(BaseModel):
question: str
result: str = ""

复制代码

多个模式（Multiple Schemas）：在大多数情况下，LangGraph 使用一个统一的 State 模式。但你也可以设置“输入模式”和“输出模式”分开

输出模式：只保留最终输出的字段（如final_answer）
1. from typing import TypedDict
2. from langgraph.graph import StateGraph
5. # 1. 定义输入、输出、图内部的状态结构
7. # 输入字段：用户的问题
8. class InputState(TypedDict):
9. question: str
12. # 中间状态：包括中间结果
13. class InternalState(TypedDict):
14. question: str
15. search_result: str
16. final_answer: str
19. # 输出字段：只想返回最终答案
20. class OutputState(TypedDict):
21. final_answer: str
24. # 2. 定义节点函数（中间节点用中间字段）
25. def search_node(state: InternalState) -> dict:
26. return {"search_result": f"搜索了：{state['question']}"}
29. def answer_node(state: InternalState) -> dict:
30. return {"final_answer": f"根据搜索结果：{state['search_result']}，这是答案"}
33. # 3. 创建 StateGraph，显式指定输入/输出 Schema
34. builder = StateGraph(state_schema=InternalState,
35. input_schema=InputState,
36. output_schema=OutputState)
38. # 4. 添加节点
39. builder.add_node("search", search_node)
40. builder.add_node("answer", answer_node)
42. # 5. 配置流程
43. builder.set_entry_point("search")
44. builder.add_edge("search", "answer")
46. # 6. 编译并执行图
47. app = builder.compile()
49. result = app.invoke({"question": "什么是LangGraph？"})
51. print(result) # {'final_answer': '根据搜索结果：搜索了：什么是LangGraph？，这是答案'}
复制代码
-> dict：返回值注解，表示该函数应该返回一个 dict 类型的对象

Reducer（归并函数）：在 LangGraph 中，所有节点返回的都是“局部更新结果”，Reducer 是用于合并多个节点输出更新的机制。 将每个节点返回的“局部状态更新”统一合并进全局的 State
在TypeDict 中，我们通过Annotated来给某个字段指定 Reducer 函数

Annotated[类型, Reducer函数] ：这里的 add 表示：每次节点返回这个字段时，都把新列表拼接到旧列表后面

1. from typing import Annotated
2. from typing_extensions import TypedDict
3. from operator import add
5. class State(TypedDict):
6. foo: int
7. bar: Annotated[list[str], add] # 每条消息是 {role, content}，会自动追加到列表末尾
复制代码
在长任务中，state会随着任务轮次的增加变得臃肿，容易导致解析变慢或者上下文溢出，为了解决这种现象，需要引入reducer机制

在图表中使用消息

在许多情况下，将之前的对话历史记录以消息列表的形式存储在图状态中会很有帮助。为此，我们可以向图状态添加一个键（通道），该键存储Message对象列表，并使用 Reducer 函数对其进行注释。Reducer 函数对于指示图如何Message在每次状态更新（例如，当节点发送更新时）时更新状态中的对象列表至关重要。如果您未指定 Reducer，则每次状态更新都会用最新提供的值覆盖消息列表。如果您只想将消息附加到现有列表中，可以使用operator。

from typing import TypedDict
from langgraph.graph import StateGraph
from typing import Annotated
import operator
# 定义状态结构如果定义的是list[dict]，会覆盖之前的数据
class ChatState(TypedDict):
messages: Annotated[list, operator.add] # 每条消息是 {role, content}，会自动追加到列表末尾
# 节点函数：添加用户问题
def user_input_node(state: ChatState) -> dict:
user_msg = {"role": "user", "content": "什么是LangGraph？"}
return {"messages": [user_msg]}
# 节点函数：添加助手回复
def assistant_node(state: ChatState) -> dict:
reply = {"role": "assistant", "content": "LangGraph 是一个有状态的图编排框架。"}
return {"messages": [reply]}
# 构建状态图
builder = StateGraph(state_schema=ChatState)
builder.add_node("user_input", user_input_node)
builder.add_node("assistant_reply", assistant_node)
builder.set_entry_point("user_input")
builder.add_edge("user_input", "assistant_reply")
graph = builder.compile()
result = graph.invoke({"messages": []})
print(result["messages"])

复制代码

MessagesState

MessagesState是langgraph中自带的状态模板，它能够自动存储对话历史并自动追加消息，不需要写typedict和reducer

from langgraph.graph import MessagesState
# 和上述代码不同会在State类中自动维护一个messages 字段，不需要显示创建
class State(MessagesState):
documents: list[str]

复制代码

节点Nodes

节点（Nodes）是图中执行逻辑的基本单位。每个节点表示一个函数步骤、处理阶段或子逻辑流程，多个节点通过边连接成有向图，组成一个完整的有状态计算流程。
LangGraph 中的节点就是你定义的一个函数（或 Runnable 对象），用于接收状态、执行逻辑，并返回更新后的状态

def my_node(state: dict) -> dict:
# 处理输入状态，并返回更新字段
return {"new_key": "new_value"}
# LangGraph 会自动用 reducer 把这些更新合并进全局状态。

复制代码

START节点

Node START是一个特殊节点，表示将用户输入发送到图的节点。引用此节点的主要目的是确定应首先调用哪些节点。

from langgraph.graph import START
graph.add_edge(START, "node_a")

复制代码

add_edge负责把两个节点连接起来：workflow.add_edge(上一个节点, 下一个节点)

End节点

Node End是一个特殊节点，表示终端节点。当需要指示哪些边在完成后没有操作时，可以引用此节点。

from langgraph.graph import END
graph.add_edge("node_a", END)

复制代码

并行运行节点

并行运行节点指的是同时启动多个不互相依赖的节点，而不是按顺序一个接一个地执行。
在langgraph中实现并行只需要将同一个起点的边指向多个不同的节点即可

# 当 start_node 运行结束时，A 和 B 会同时启动
workflow.add_edge("start_node", "node_A")
workflow.add_edge("start_node", "node_B")

复制代码

边Edge

Edge（边）是连接节点的通道，表示图中节点之间的执行跳转关系。你可以把它理解为「节点执行完之后，下一步去哪，是构成 LangGraph 流程图的核心
普通边

graph.add_edge("节点A", "节点B")

复制代码

条件边

条件边会根据state里的内容，判断下一步执行哪一个逻辑函数

def route_condition(state: MyState):
"""条件函数：只负责路由决策"""
if state["type"] == "a":
return "a"
elif state["type"] == "b":
return "b"
else:
return "default"
def node_a(state):
return {"result": "走了 A 分支"}
def node_b(state):
return {"result": "走了 B 分支"}
def node_default(state):
return {"result": "走了默认分支"}
graph.add_conditional_edges("judge_node", route_condition, {
"a": "a",
"b": "b",
"default": "default"
})

复制代码

条件边的格式
workflow.add_conditional_edges(
"起点节点",
routing_function, # 判断逻辑函数
{"path_a": "节点A", "path_b": "节点B"} # 路由映射表
)

条件入口点

根据初始输入的数据，动态决定首先运行哪个节点

from langgraph.graph import StateGraph, START, END
# 1. 定义路由函数
def route_start(state: State):
if state["question"].startswith("查资料"):
return "search_node"
else:
return "chat_node"
# 2. 构建图
builder = StateGraph(State)
builder.add_node("search_node", search_func)
builder.add_node("chat_node", chat_func)
# 3. 设置条件入口点：从 START 开始进行条件跳转
builder.add_conditional_edges(
START, # 起点是内置的 START
route_start, # 判断函数
{
"search_node": "search_node",
"chat_node": "chat_node"
}
)

复制代码

发送Send

默认情况下，Nodes和Edges是提前定义的，并在相同的共享状态下运行。但是，在某些情况下，确切的边无法提前知道，并且可能希望同时存在State的不同版本。
也就是说，普通的边只能沿着图预设的路线走；而send可以强制跳到任意节点并传输数据，支持运行时动态地分发任务

条件路由

并行处理

动态工作流

Map-Reduce

Map-Reduce就是send的工作原理，它是一种经典的并行计算模式，特别适合处理大规模数据
Map-Reduce 将复杂的数据处理任务分解为两个阶段：

Map（分发）：Send

Process（处理）

Reduce（聚合）

标准结构是 Send(目标节点名称, 要传递的状态数据)
Send("node_name", {"key": value})

"""
LangGraph Map-Reduce 简单案例：数字求和
把一堆数字分给多个worker算平方，然后把结果加起来
"""
from typing import Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.constants import Send
from typing import TypedDict, List
# 状态定义
class State(TypedDict):
numbers: List[int] # 输入的数字
results: Annotated[list[int], operator.add] # worker的结果
final_sum: int # 最终求和
# 1. Map阶段：分发数字
def split_numbers(state: State):
"""把数字分发给不同的worker"""
numbers = state["numbers"]
print(f"分发数字: {numbers}")
# 每个数字发给一个worker
return [Send("worker", {"number": num}) for num in numbers]
# 2. Worker阶段：计算平方
def calculate_square(state: State):
"""每个worker计算一个数字的平方"""
number = state["number"]
square = number * number
print(f"Worker: {number}² = {square}")
return {"results": [square]}
# 3. Reduce阶段：求和
def sum_results(state: State):
"""把所有结果加起来"""
results = state.get("results", [])
total = sum(results)
print(f"求和: {results} = {total}")
return {"final_sum": total}
# 构建图
def create_simple_graph():
graph = StateGraph(State)
# 添加节点
graph.add_node("splitter", lambda s: s) # 分发器
graph.add_node("worker", calculate_square) # 工作节点
graph.add_node("summer", sum_results) # 求和器
# 连接节点
graph.add_edge(START, "splitter")
graph.add_conditional_edges("splitter", split_numbers, ["worker"]) # Map阶段
graph.add_edge("worker", "summer") # Worker完成后求和
graph.add_edge("summer", END)

复制代码

注意这里，graph.add_node("worker",calculate_square)
第一个字符串参数是节点在图中的标识符
第二个参数是具体要执行的函数

这里解释两个点

分发器

数据分流（路由）

匿名函数

接收数据然后分发给每个worker节点。

Map分发过程

Send

路由映射表

split_numbers

n个Send

n个worker节点

命令Command

Command 是 LangGraph 中用于控制图执行流程、更新图状态，并支持人机交互、工具调用的标准化对象。
核心作用：

第一，更新图的运行状态；
第二，控制图的执行流向（指定下一个或多个执行节点）；
第三，衔接中断恢复、工具调用、人机交互等场景。
command参数

update：应用状态更新（类似于从节点返回更新）。
goto：导航到特定节点（类似于条件边，以后无需再stategraph里连线）。
graph：在从子图导航时定位到父图。
resume：在中断后提供一个值以继续执行。

在节点函数中返回command时，必须添加返回类型注释，其中包含节点路由到的节点名称列表，例如command[Literal ["my_other_node"]]。这对于图形渲染是必需的，它告诉 LangGraph 当前节点可以导航到my_other_node。

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.types import Command, Send
from typing import Literal
class MyState(TypedDict):
type: str
text: str
result: str
def judge_node(state: MyState) -> Command[Literal["a", "b", "default"]]:
"""条件函数：使用Command进行路由和状态更新"""
if state["type"] == "a":
return Command(update={"text": "走了 A 分支"}, goto="a")
elif state["type"] == "b":
return Command(update={"text": "走了 B 分支"}, goto="b")
else:
return Command(update={"text": "走了默认分支"}, goto="default")
def node_a(state):
return {"result": f"A节点处理: {state['text']}"}
def node_b(state):
return {"result": f"B节点处理: {state['text']}"}
def node_default(state):
return {"result": f"默认节点处理: {state['text']}"}
# 构建图
graph = StateGraph(state_schema=MyState)
graph.add_node("judge_node", judge_node)
graph.add_node("a", node_a)
graph.add_node("b", node_b)
graph.add_node("default", node_default)
graph.set_entry_point("judge_node")
# 添加结束边
graph.add_edge("a", END)
graph.add_edge("b", END)
graph.add_edge("default", END)
app = graph.compile()
# 测试
print("测试 A:", app.invoke({"type": "a", "text": "", "result": ""}))
print("测试 B:", app.invoke({"type": "b", "text": "", "result": ""}))
print("测试其他:", app.invoke({"type": "default", "text": "", "result": ""}))

复制代码

在这里，我们使用了Command对象过后就不需要写builder.add_node和builder.add_conditional_edges函数了
什么时候应该使用Command而不是条件边

如果你想在节点内部既要改状态state，又要实现跳转，那么就必须使用Command，在条件边中跳转和状态修改分散在两个函数里面，而command可以把这两个逻辑聚在一起。

原文地址：https://blog.csdn.net/jiaranran8/article/details/160506504

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

LangGraph框架状态图&节点&边&Send

学习LangChain-基础篇-认识LangChain

关于我们

产品与服务

解决方案

产品与服务

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

LangGraph框架 状态图&节点&边&Send

学习LangChain-基础篇-认识LangChain

关于我们

产品与服务

解决方案

产品与服务

LangGraph框架状态图&节点&边&Send