图结构（Graphs）
LangGraph 的核心是将智能体工作流建模为图结构 。你可以通过三个关键组件来定义智能体的行为：
状态（State）
这是一个共享的数据结构，代表你应用程序的当前快照。它可以是任何数据类型，但通常使用共享状态模式来定义。
简单理解：就像是一个"记事本"，记录着当前程序运行到什么状态了。
节点（Nodes）
这些是包含智能体逻辑的函数。它们接收当前状态作为输入，执行一些计算或操作，然后返回更新后的状态。
简单理解：就像是"工作站"，负责干活，处理数据，完成具体任务。
边（Edges）
这些函数决定接下来要执行哪个节点，基于当前状态来判断。它们可以是条件分支或固定的转换路径。
简单理解：就像是"交通指挥员"，告诉程序下一步该去哪里。
核心思想
通过组合节点和边，你可以创建复杂的、循环的工作流，让状态随时间不断演化。LangGraph 的真正优势来自于它如何管理这些状态。需要强调的是：节点和边本质上就是函数 - 它们可以包含大语言模型，也可以只是普通的代码。
简而言之：节点负责干活，边负责指路。
运行机制
LangGraph 的底层图算法使用消息传递来定义通用程序。当一个节点完成操作后，它会沿着一条或多条边向其他节点发送消息。接收消息的节点然后执行自己的函数，将结果消息传递给下一组节点，这个过程持续进行。这个设计灵感来自Google的Pregel系统，程序以离散的"超步骤"进行。
超步骤(super-step )的概念
- 一个超步骤可以看作是对图节点的一次完整迭代
- 并行运行的节点属于同一个超步骤
- 顺序运行的节点属于不同的超步骤
执行流程
- 图执行开始时，所有节点都处于
非活跃状态 - 当节点在任何传入边（或"通道"）上收到新消息（状态）时，它变为
活跃状态 - 活跃节点运行其函数并返回更新结果
- 在每个超步骤结束时，没有收到传入消息的节点会"投票停止"，将自己标记为
非活跃 - 当所有节点都是非活跃状态且没有消息在传输时，图执行终止
这就像是一个智能的流水线系统，每个工作站（节点）完成任务后，会自动将结果传递给下一个需要工作的站点，直到整个任务完成。
让我们构建一个包含3个节点和一个条件边的简单图。
状态（State）
首先，定义图的状态。
状态模式作为图中所有节点和边的输入模式。
我们使用Python typing模块中的TypedDict 类作为模式，它为键提供类型提示。
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的新值。
默认情况下，每个节点返回的新值将覆盖之前的状态值。
# 定义节点 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，可以使用普通边。
条件边 ：如果你想要在节点之间进行可选路由，可以使用条件边。
条件边被实现为函数，根据某些逻辑返回下一个要访问的节点。
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图表。
# 构建图：基于节点+边
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节点。
response = graph.invoke({"graph\_state": "你好，我是小明"})
print(response)
输出：
我在运行节点1的功能
我在运行节点2的功能
{'graph\_state': '你好，我是小明,我的心情是:开心的！'}
执行说明
invoke同步运行整个图。
它等待每个步骤完成后再移动到下一个步骤。
它在所有节点执行完毕后返回图的最终状态。
在这个例子中，它返回node_2完成后的状态：
{'graph\_state': '你好，我是小明,我的心情是:开心的！'}
添加微信，备注” LLM “进入大模型技术交流群
如果你觉得这篇文章对你有帮助，别忘了点个赞、送个喜欢
/ 作者：致Great
/ 作者：欢迎转载，标注来源即可
Agent实战教程：LangGraph中工作流与智能体区别与实现
Agent实战教程：LangGraph相关概念介绍以及快速入门
Deep Agents：用于复杂任务自动化的 AI 代理框架
Kimi K2智能体能力的技术突破：大规模数据合成 + 通用强化学习
构建AI Agent的完整实战指南：从邮件助手案例看6步落地方法
Context Engineering：从Prompt Engineering到上下文工程的演进
如何用LangGraph打造Web Research多智能体系统
Anthropic关于智能体的经验分享：如何构建高效的Agent？
Gemini开源项目DeepResearch：基于LangGraph的智能研究Agent技术原理与实现
xAI 把 Grok 的系统提示词全部公开了，我们看看DeepResearch的系统提示词怎么设计的?