Agent 实战教程:LangGraph 核心概念节点、边以及状态详解

图结构(Graphs)

LangGraph 的核心是将智能体工作流建模为图结构。你可以通过三个关键组件来定义智能体的行为:

状态(State)

这是一个共享的数据结构,代表你应用程序的当前快照。它可以是任何数据类型,但通常使用共享状态模式来定义。

简单理解:就像是一个 “记事本”,记录着当前程序运行到什么状态了。

节点(Nodes)

这些是包含智能体逻辑的函数。它们接收当前状态作为输入,执行一些计算或操作,然后返回更新后的状态。

简单理解:就像是 “工作站”,负责干活,处理数据,完成具体任务。

边(Edges)

这些函数决定接下来要执行哪个节点,基于当前状态来判断。它们可以是条件分支或固定的转换路径。

简单理解:就像是 “交通指挥员”,告诉程序下一步该去哪里。

核心思想

通过组合节点和边,你可以创建复杂的、循环的工作流,让状态随时间不断演化。LangGraph 的真正优势来自于它如何管理这些状态。需要强调的是:节点和边本质上就是函数 - 它们可以包含大语言模型,也可以只是普通的代码。

简而言之:节点负责干活,边负责指路。

运行机制

LangGraph 的底层图算法使用消息传递来定义通用程序。当一个节点完成操作后,它会沿着一条或多条边向其他节点发送消息。接收消息的节点然后执行自己的函数,将结果消息传递给下一组节点,这个过程持续进行。这个设计灵感来自 Google 的 Pregel 系统,程序以离散的 “超步骤” 进行。

超步骤 (super-step) 的概念

  • 一个超步骤可以看作是对图节点的一次完整迭代

  • 并行运行的节点属于同一个超步骤

  • 顺序运行的节点属于不同的超步骤

执行流程

  1. 图执行开始时,所有节点都处于非活跃状态

  2. 当节点在任何传入边(或 “通道”)上收到新消息(状态)时,它变为活跃状态

  3. 活跃节点运行其函数并返回更新结果

  4. 在每个超步骤结束时,没有收到传入消息的节点会 “投票停止”,将自己标记为非活跃

  5. 当所有节点都是非活跃状态且没有消息在传输时,图执行终止

这就像是一个智能的流水线系统,每个工作站(节点)完成任务后,会自动将结果传递给下一个需要工作的站点,直到整个任务完成。

构建一个最简单的图结构

让我们构建一个包含 3 个节点和一个条件边的简单图。

状态(State)

首先,定义图的状态。

状态模式作为图中所有节点和边的输入模式。

我们使用 Python typing 模块中的 TypedDict 类作为模式,它为键提供类型提示。

1
2
3
4
5
6
7
8
9
10
11
12
13
import random
from typing import Literal
from typing import TypedDict

from langgraph.graph import StateGraph, END, START


# 定义状态 State
class State(TypedDict):
    graph_state: str



节点(Nodes)

节点就是 Python 函数。

第一个位置参数是状态,如上所定义。

因为状态是一个具有上述模式的 TypedDict,每个节点都可以通过state['graph_state']访问键graph_state

每个节点返回状态键graph_state的新值。

默认情况下,每个节点返回的新值将覆盖之前的状态值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 定义节点 None

def node_1(state: State):
    print("我在运行节点1的功能")
    return {"graph_state": state["graph_state"] + ",我的心情是:"}


def node_2(state: State):
    print("我在运行节点2的功能")
    return {"graph_state": state["graph_state"] + "开心的!"}


def node_3(state: State):
    print("我在运行节点3的功能")
    return {"graph_state": state["graph_state"] + "伤心的!"}def node_1(state):
    print("---节点 1---")
    return {"graph_state": state['graph_state'] + " 我是"}

def node_2(state):
    print("---节点 2---")
    return {"graph_state": state['graph_state'] + " 开心的!"}

def node_3(state):
    print("---节点 3---")
    return {"graph_state": state['graph_state'] + " 伤心的!"}

边(Edges)

边连接节点。

普通边:如果你想要总是从 node_1 跳转到 node_2,可以使用普通边。

条件边:如果你想要在节点之间进行可选路由,可以使用条件边。

条件边被实现为函数,根据某些逻辑返回下一个要访问的节点。

1
2
3
4
5
6
7
8
9
10

def decide_mood(state: State) -> Literal["node_2", "node_3"]:
    # 用户的输入决定了下一个访问节点
    user_input = state["graph_state"]

    if random.random() < 0.5:
        return "node_2"
    return "node_3"


图构建(Graph Construction)

现在,我们从上面定义的组件构建图。

StateGraph 类是我们可以使用的图类。

首先,我们用上面定义的 State 类初始化一个 StateGraph。

然后,添加节点和边。

我们使用 START 节点(一个将用户输入发送到图的特殊节点)来指示图的起始位置。

END 节点是表示终端节点的特殊节点。

最后,我们编译图以对图结构执行一些基本检查。

我们可以将图可视化为 Mermaid 图表。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 构建图:基于节点+边
builder = StateGraph(State)
# 添加节点
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

# 添加边
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)

# 编译图
graph = builder.compile()
# 显示工作流
png_data = graph.get_graph().draw_mermaid_png()
with open("graph.png""wb"as f:
    f.write(png_data)

图的调用(Graph Invocation)

编译后的图实现了可运行协议。

这提供了执行 LangChain 组件的标准方式。

invoke是此接口中的标准方法之一。

输入是一个字典{"graph_state": "你好,我是Lance。"},它为我们的图状态字典设置初始值。

当调用invoke时,图从 START 节点开始执行。

它按顺序通过定义的节点(node_1、node_2、node_3)进行。

条件边将使用 50/50 决策规则从节点 1 跳转到节点 2 或 3。

每个节点函数接收当前状态并返回一个新值,该值覆盖图状态。

执行持续到到达 END 节点。

1
2
3
response = graph.invoke({"graph_state""你好,我是小明"})
print(response)

输出:

1
2
3
4
我在运行节点1的功能
我在运行节点2的功能
{'graph_state': '你好,我是小明,我的心情是:开心的!'}

执行说明

invoke同步运行整个图。

它等待每个步骤完成后再移动到下一个步骤。

它在所有节点执行完毕后返回图的最终状态。

在这个例子中,它返回 node_2 完成后的状态:

{'graph_state': '你好,我是小明,我的心情是:开心的!'}

/ 作者:致 Great

/ 作者:欢迎转载,标注来源即可

Agent 实战教程:LangGraph 中工作流与智能体区别与实现

Agent 实战教程:LangGraph 相关概念介绍以及快速入门

Deep Agents:用于复杂任务自动化的 AI 代理框架

一图概览 2024 年到 2025 年 AI Agent 的发展趋势

警惕 AI 智能体构建误区:生产级系统的实战经验分享

AI 代理的上下文工程:构建 Manus 的经验教训

基于 Gemini API 进行大模型函数调用的指南与经验总结

Kimi K2 智能体能力的技术突破:大规模数据合成 + 通用强化学习

构建 AI Agent 的完整实战指南:从邮件助手案例看 6 步落地方法

企业级 AI 智能体系统的 5 种核心工作流模式

Context Engineering:从 Prompt Engineering 到上下文工程的演进

如何用 LangGraph 打造 Web Research 多智能体系统

智能体框架:11 个顶级 AI Agent 框架!

Anthropic 关于智能体的经验分享:如何构建高效的 Agent?

AI 智能体框架对比表

Gemini 开源项目 DeepResearch:基于 LangGraph 的智能研究 Agent 技术原理与实现

智能体卷疯了,又一款 Agent 框架开源了 Lemon AI

大模型 Agent 就是文字艺术吗?

xAI 把 Grok 的系统提示词全部公开了,我们看看 DeepResearch 的系统提示词怎么设计的?

OpenAI API JSON 格式指南与 json_repair 错误修复

txtai:全能 AI 框架

Suna - 开源智能体助手

谷歌的 A2A 到底是什么东西?

如何在 Agent 中设置 Memory

体验智能体构建过程:从零开始构建 Agent

AI 代理是大模型实现可扩展智能自动化的关键

Agent 系列教程 01 - 什么是 Agent?当今为什么这么重要?