基础入门
基本概念
图的基本要素
（1）State（状态）：LangGraph 运行过程中的共享数据结构，用于表示应用在某一时刻的状态快照。它承载了图运行所需的上下文信息、中间结果和后续节点需要读取的数据，是节点之间传递信息的核心载体。它与我们在学习 LangChain Agent 时使用的 State 是同一概念
（2）Node（节点）：LangGraph 中的具体执行单元，通常实现为一个函数。节点会读取当前 State，执行相应的业务逻辑，并返回对 State 的局部更新。节点本身并不直接修改全局状态，状态的合并与提交由运行时统一完成
（3）Edges（边）：用于定义节点之间的流转关系，决定一个节点执行完成后下一步应该进入哪个节点。Edge 可以是固定流转，也可以根据当前 State 进行条件判断，从而实现分支、循环等复杂控制流程
图运行过程
（1）计划 / 路由阶段（Plan / Routing）：根据当前的 State（状态） 和 Edge（边） 的逻辑，确定本轮超步中应该被执行的节点
（2）执行阶段（Execution）：运行本轮被选中的节点。如果本轮有多个节点同时被触发，它们会并行执行。每个节点都会基于本轮开始时的状态快照进行计算，并输出各自对状态的局部更新。在本阶段中，一个节点产生的更新不会立即被其他节点读取到
（3）状态更新/提交阶段（Update / Commit）：当本轮所有节点都执行完成后，LangGraph 会将它们的输出统一合并到 State 中，生成新的状态快照。这个新状态会作为下一轮 Superstep 的输入
API 类型
（1）Graph API
Graph API 采用声明式方式构建工作流。开发者需要显式定义 State、Node 和 Edge，将业务流程组织成一个可视化的图结构
当流程中存在较复杂的分支、多个节点之间共享状态、并行执行、结果汇聚，或者需要通过图结构帮助调试和团队协作时，更适合使用 Graph API。官方文档也明确建议，在需要复杂流程可视化、显式状态管理、多条件分支、并行路径以及团队协作时，优先选择 Graph API
（2）Functional API
Functional API 采用命令式方式构建工作流，更接近普通 Python 函数调用。开发者可以使用
@entrypoint定义工作流入口，使用@task定义可被检查点记录的任务，然后在函数内部使用普通的if/else、循环和函数调用来组织流程。官方文档指出，当已有过程式代码需要最小改造、流程主要是线性的、分支逻辑较简单、希望快速原型验证时，更适合使用 Functional API
| 对比项 | Graph API | Functional API |
|---|---|---|
| 编程风格 | 声明式图结构 | 命令式函数流程 |
| 核心抽象 | State、Node、Edge | entrypoint、task |
| 状态管理 | 显式定义全局 State | 更多依赖函数参数和返回值 |
| 流程表达 | 通过节点和边表达 | 通过普通 Python 控制流表达 |
| 可视化能力 | 强，天然适合画图和调试 | 弱，更像普通代码流程 |
| 适合场景 | 复杂工作流、多分支、多节点协作 | 简单流程、快速原型、已有代码改造 |
| 学习成本 | 相对更高 | 相对更低 |
基本使用
状态类型
全局状态 / 内部状态：图内部主要使用的状态，创建
StateGraph时传递给state_schema参数。它通常包含图运行过程中需要读写的大部分字段 输入状态：图对外接收输入时使用的状态，创建
StateGraph时传递给input_schema参数。它用于约束调用图时允许传入哪些字段 输出状态：图最终对外返回结果时使用的状态，创建
StateGraph时传递给output_schema参数。它用于约束图运行结束后只返回哪些字段 私有状态：图内部节点之间传递的临时状态，通常不作为图的输入，也不作为图的最终输出。它可以通过节点函数的入参类型注解声明，并在节点返回值中写入
示例代码
基本步骤：定义状态、定义节点、创建图，添加节点、添加边、获取图、运行图
使用 addEdge（）方法添加边，表示节点 A 指向节点 B 的边关系
调用图时，输入会按照 input_schema 进行约束，如果创建图时声明了 input_schema，那么外部输入会按照 input_schema 进行约束，如果没有声明 input_schema，则通常按照 state_schema 作为图的输入 Schema
状态定义有三种类型，分别是 TypedDict、Pydantic、dataclass，官方推荐使用 TypedDict
相比普通
dict，TypedDict可以提供更明确的字段约束和类型提示；相比dataclass，TypedDict更贴近 LangGraph 中状态的更新方式，因为节点通常返回的是表示“部分状态更新”的字典，而不是完整对象；相比Pydantic BaseModel，它又更加轻量，不会引入额外的数据校验开销。因此，在没有复杂校验需求的情况下，TypedDict 是定义 LangGraph State Schema 的首选方式
START不能省略。因为 START 不只是一个语义上的起点标记，它还用于告诉 LangGraph：图运行时应当从哪些节点开始执行，即下面这条边是启动图执行的关键
END可以省略，END 并不是运行阶段真正执行的节点，即指向 END 的边不会像指向普通节点的边那样触发一个真实的节点任务，它更多用于表达图结构中的终止语义：当前路径执行到这里即可结束
python
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class InputState(TypedDict):
username: str
class OutputState(TypedDict):
graph_output: str
class OverAllState(TypedDict):
nickname: str
username: str
graph_output: str
class PrivateState(TypedDict):
greeting: str
def node_1(state: InputState) -> OverAllState:
# 向全局状态写入数据
return {
"nickname": "Dear " + state["username"]
}
def node_2(state: OverAllState) -> PrivateState:
# 从全局状态读取数据，写入私有状态
return {
"greeting": state["nickname"] + ", 早上好~"
}
def node_3(state: PrivateState) -> OutputState:
# 从私有状态读取数据，写入输出状态
return {
"graph_output": state["greeting"] + " 很高兴认识你！"
}
builder = StateGraph(OverAllState,input_schema=InputState,output_schema=OutputState)
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_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)
graph = builder.compile()
print(graph.invoke({"username":"小黄"}))
from IPython.display import display
display(graph)
"""
运行结果如下
{'graph_output': 'Dear 小黄, 早上好~ 很高兴认识你！'}
"""
"""
图结构如下
┌───────────┐
│ __start__ │
└─────┬─────┘
│
▼
┌───────────┐
│ node_1 │
└─────┬─────┘
│
▼
┌───────────┐
│ node_2 │
└─────┬─────┘
│
▼
┌───────────┐
│ node_3 │
└─────┬─────┘
│
▼
┌───────────┐
│ __end__ │
└───────────┘
"""代码执行过程解析
（1）调用图时，InputState 限制外部只需传入 username
（2）node_1 读取 username，生成 nickname 并写入全局状态
（3）node_2 读取 nickname，生成临时字段 greeting，通过 PrivateState 传递给下一个节点
（4）node_3 读取 greeting，生成最终的 graph_output
（5）图执行结束后，根据 OutputState 对结果进行裁剪，只返回 graph_output
输入 username
↓
node_1 生成 nickname
↓
node_2 生成临时 greeting
↓
node_3 生成 graph_output
↓
输出 graph_output图结构可视化
（1）方式一：绘制 mermaid 并获取源码
python
raw_mermaid = graph.get_graph().draw_mermaid()
print(raw_mermaid)（2）方式二：绘制 mermaid 并转换为 png
python
from IPython.display import display, Image
png_bytes = graph.get_graph().draw_mermaid_png()
png = Image(png_bytes)
display(png)
# 保存为图片文件
png_filename = 'first_demo_graph.png'
with open(png_filename, "wb") as f:
f.write(png_bytes)（3）方式三：Jupyter 中的快捷用法
python
from IPython.display import display
display(graph)状态设计规范
（1）输入状态和输出状态通常应是全局状态的子集
（2）私有状态和全局状态应尽量避免字段重名
（3）节点函数应明确声明入参状态类型和返回状态类型
（4）节点函数中不应该访问入参状态类型中不存在的字段
（5）节点函数返回的字典应尽量和返回类型注解保持一致
状态记录约束
全局状态、输入状态、输出状态通常在创建
StateGraph时被记录 私有状态通常在调用
add_node()添加节点时，根据节点入参类型注解被记录 被记录后的状态字段，底层会成为图运行时可以读写的状态字段
状态访问约束
（1）调用图时，输入会按照
input_schema进行约束，如果创建图时声明了input_schema，那么外部输入会按照input_schema进行约束，如果没有声明input_schema，则通常按照state_schema作为图的输入 Schema，因此，input_schema的作用不是“只让第一个节点可见”，而是约束图的外部输入结构，此处的约束是指：按照schema裁剪输入，只保留schema中出现的状态字段 （2）节点接收到的状态会按照节点入参类型进行裁剪，每个节点能读取哪些字段，主要取决于该节点第一个参数的类型注解
（3）节点函数返回的是状态更新，即本节点想要更新的字段，不需要返回完整状态
（4）节点返回值的应用主要由字段名称和图中已记录的状态字段决定，节点返回的字典会根据字段名称写入对应状态字段，并按照该字段的
Reducer规则进行合并，需要注意的是，函数返回类型注解主要用于表达代码意图，不是严格的运行时写入边界，也就是说，如果某个字段已经被图记录为可用状态字段，那么节点即使没有在返回类型注解中声明该字段，也可能仍然可以返回并更新它，不过，为了代码清晰，仍然推荐让节点的返回值和返回类型注解保持一致 （5）最终输出会按照
output_schema进行裁剪，图运行完成后，最终返回给外部调用方的结果会按照output_schema进行裁剪，因此，output_schema的作用不是“只让最后一个节点可见”，而是约束图最终对外暴露哪些字段
MessageState
LangGraph 构建的计算图通常会和 LLM 结合使用，而 LLM 在运行过程中通常需要维护一组消息列表
为了提升开发效率，LangGraph 官方提供了一个预定义状态类型 MessageState，只有一个字段 messages，开发者可以直接继承该状态类型，并在其基础上扩展自定义状态字段。
python
from langchain_deepseek import ChatDeepSeek
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import MessagesState
from langchain.messages import HumanMessage
from dotenv import load_dotenv
load_dotenv(override=True)
model = ChatDeepSeek(
model='deepseek-v4-flash',
extra_body={
"thinking": {
"type": "disabled"
}
}
)
class OverAllState(MessagesState):
username: str
output: str
def node_a(state: OverAllState) -> OverAllState:
return {
"messages": [HumanMessage("你好，我是 " + state["username"])]
}
def llm_node(state: OverAllState) -> OverAllState:
res = model.invoke(state["messages"])
return {
"messages": [res],
"output": res.content
}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("llm_node", llm_node)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "llm_node")
builder.add_edge("llm_node", END)
graph = builder.compile()
response = graph.invoke({"username": "小黄"})
print(response)State Reducer
基本介绍
State Reducer 是 LangGraph 中用于合并状态更新的核心机制。在 LangGraph 的
StateGraph中，每个节点可以读取和写入共享状态，而 Reducer 定义了如何将多个节点对同一状态键的更新合并 Reducer 的核心特征如下
（1）函数签名：
(Value, Value) -> Value，接收当前值和更新值，返回合并后的新值 （2）注解定义：通过
Annotated[Type, reducer_function]为状态键指定 Reducer （3）默认行为：未指定 Reducer 的状态键使用覆盖策略（Last-Write-Wins）
函数定义
Reducer 本质上是一个二元合并函数，用于定义当同一个字段产生多个更新值时，LangGraph 应该如何将这些值合并为一个最终结果
该 Reducer 的作用是：当某个状态字段存在多次列表更新时，将这些列表内容追加合并，而不是直接覆盖原值
python
# left: 从最开始的位置合并到当前节点的位置的值
# right: 当前节点的值
# 返回值: 合并后的值
def my_reducer(left: list[str],right: list[str]) -> list[str]:
return left + right
# 1. node1 运行之后的值
left = ["start","node_1 运行完毕"]
right = ["node_2 运行完毕"]
# 2. 合并
merged = my_reducer(left,right)
print(merged)
# 输出：['start', 'node_1 运行完毕', 'node_2 运行完毕']内置函数
add 函数，用于将两个列表合并为一个列表，可在状态定义时将 Reducer 和状态字段关联
add_messages（left：Messages, right：Messages） 函数，在合并 left 与 right 时，不是简单地执行列表拼接，而是依据消息的 id 进行合并
若
right中的某条消息的id在left中不存在，则将该消息追加到结果列表末尾 若
right中的某条消息的id与left中已有消息的id相同，则使用right中的新消息替换left中的旧消息
python
# reducer 关联状态字段
from typing import TypedDict, Annotated
class OverAllState(TypedDict):
logs: Annotated[list[str], my_reducer]
cur_id: str
# add 函数
from operator import add
print(f"{add(1,2) = }")
print(f"{add([1,2], [3,4]) = }")
print(f"{add(['a','b'], ['c']) = }")
""""
输出内容如下
add(1,2) = 3
add([1,2], [3,4]) = [1, 2, 3, 4]
add(['a','b'], ['c']) = ['a', 'b', 'c']
""""
# add_messages 函数
from langgraph.graph.message import add_messages
from langchain.messages import HumanMessage, AIMessage, SystemMessage
left = [
SystemMessage(content="你是个善解人意的助手", id='1'),
HumanMessage(content="你好", id='2'),
AIMessage(content="你好~", id='3'),
]
right = [
HumanMessage(content="我是老王，你是小王", id='2'),
AIMessage(content="好的，我记住啦", id='3'),
HumanMessage(content="你是谁？", id='4'),
AIMessage(content="我是小王", id='5'),
]
merged = add_messages(left, right)
for msg in merged:
print(msg)
""""
输出内容如下
content='你是个善解人意的助手' additional_kwargs={} response_metadata={} id='1'
content='我是老王，你是小王' additional_kwargs={} response_metadata={} id='2'
content='好的，我记住啦' additional_kwargs={} response_metadata={} id='3' tool_calls=[] invalid_tool_calls=[]
content='你是谁？' additional_kwargs={} response_metadata={} id='4'
content='我是小王' additional_kwargs={} response_metadata={} id='5' tool_calls=[] invalid_tool_calls=[]
""""默认行为
如果某个 State 字段没有显式定义
Reducer，LangGraph 会使用默认的状态更新行为后一次更新值会覆盖该字段原有的状态值 换句话说，当节点返回的更新结果中包含某个字段时，如果该字段没有配置 Reducer，LangGraph 不会对新旧值进行追加、合并或累加，而是直接使用本次返回的新值替换原来的旧值
节点中访问 State
图节点中读取 State
state 表示当前节点执行时可以访问到的全局状态快照。节点可以通过读取 state 中的字段获取上游节点写入的数据，并基于这些数据完成当前节点的业务逻辑
python
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add
class OverAllState(TypedDict):
logs: Annotated[list[str], add]
id: str
def node_a(state: OverAllState):
for k, v in state.items():
print(f"k: {k}, v: {v}")
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", END)
graph = builder.compile()
result = graph.invoke({"logs": ["START"], "id": "start"})
""""
输出内容如下
k: logs, v: ['START']
k: id, v: start
""""图节点更新 state
对于节点没有返回的字段，LangGraph 会保留其原有状态值
对于节点返回的字段，LangGraph 会根据该字段是否配置了
Reducer来决定如何合并更新值 （1）如果字段配置了
Reducer，则使用对应的Reducer函数将旧值和新值合并； （2）如果该字段没有配置
Reducer，则按照默认规则使用节点返回的新值覆盖原值。
python
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add
class OverAllState(TypedDict):
# 使用 add 作为 Reducer 函数，将新旧 logs 列表合并
logs: Annotated[list[str], add]
id: str
def node_a(state: OverAllState):
for k, v in state.items():
print(f"k: {k}, v: {v}")
return {
"logs": ["node_a 更新状态"]
}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", END)
graph = builder.compile()
result = graph.invoke({"logs": ["START"], "id": "start"})
print('=' * 30, '-> result <-', '=' * 30)
print(result)
""""
输出内容如下
k: logs, v: ['START']
k: id, v: start
============================== -> result <- ==============================
{'logs': ['START', 'node_a 更新状态'], 'id': 'start'}
""""Overwrite 覆盖
在某些场景下，我们可能并不希望继续执行 Reducer 的聚合逻辑，而是希望本次更新直接覆盖旧值，这时可以使用 Overwrite
Overwrite 的作用是：告诉 LangGraph 本次状态更新不走该字段原本定义的 Reducer，而是直接用新值覆盖状态中的旧值
需要注意的是，Overwrite 只影响当前这一次更新，并不会修改状态字段本身的 Reducer 定义。后续节点如果继续正常返回该字段的更新值，仍然会按照原来的 Reducer 逻辑进行合并
python
from langgraph.graph import StateGraph, START, END
from langgraph.types import Overwrite
from typing import TypedDict, Annotated
from operator import add
class OverAllState(TypedDict):
logs: Annotated[list[str], add]
id: str
def node_a(state: OverAllState):
return {
"logs": ["node_a"],
"id": "node_a"
}
def node_b(state: OverAllState):
return {
"logs": Overwrite(["node_b"]),
"id": "node_b"
}
def node_c(state: OverAllState):
return {
"logs": ["node_c"],
"id": "node_c"
}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("node_c", node_c)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", "node_c")
builder.add_edge("node_c", END)
graph = builder.compile()
result = graph.invoke({"logs": ["START"], "id": "start"})
print('=' * 30, '-> result <-', '=' * 30)
print(result)
"""
输入内容如下
============================== -> result <- ==============================
{'logs': ['node_b', 'node_c'], 'id': 'node_c'}
"""执行结果解析
（1）node_b 返回更新时，用 Overwrite 包裹了 logs 字段的值，那么当前状态的 logs 会被 ["node_b"] 覆盖，因此最终输出的 logs 字段值变成了 ['node_b', 'node_c']
（2）id 字段按照默认行为，保留最后一次更新的值