LangGraph 基础入门

---

目录

• 第 1 章 LangGraph 总览
  • 1.1 LangChain 与 LangGraph 的定位
  • 1.2 构成图的三要素
  • 1.3 图的运行过程：Superstep（超步）
  • 1.4 Graph API 与 Functional API
• 第 2 章 图的基础构建与运行
  • 2.1 第一个例子：两节点顺序执行
  • 2.2 代码逐行拆解
• 第 3 章 图的状态（State）管理
  • 3.1 定义状态的三种方式
  • 3.2 State Reducer（状态归约）
  • 3.3 节点中读写 State
  • 3.4 Multi Schema：四种状态类型
  • 3.5 预定义状态
• 结语

---

第 1 章 LangGraph 总览

LangGraph 是 LangChain 团队开源的 Agent 运行时，用来构建"有状态、可控、可持久化"的复杂 Agent 工作流。简言之，LangChain 提供易用的 Agent 高层抽象，LangGraph 提供可靠、可持久化的底层执行能力。

1.1 LangChain 与 LangGraph 的定位

LangGraph 的运行时底层基于自研的 Pregel 运行时，其核心思想借鉴了 Google 的 Pregel 计算模型，用于组织和执行复杂的图计算流程。更底层的编排框架与 Agent Runtime，负责复杂工作流和有状态 Agent 的执行，提供持久化、流式输出、Durable Execution、Human-in-the-loop 等运行时能力。create_agent 底层正是基于 LangGraph 实现的。

二者的定位对比如下：

对比维度	LangChain	LangGraph
定位	Agent 高层开发框架	底层编排框架 & Agent Runtime
核心入口	create_agent	StateGraph / @entrypoint
适用场景	结构直接的 Agent 应用	复杂工作流、持久化状态、长时间运行、人工介入
流程控制	Agent 循环自动管理	精细控制节点、边、条件分支
学习成本	较低	较高

LangChain 提供易于使用的 Agent 高层抽象，LangGraph 提供可靠、可持久化且可精细控制的底层执行能力。

大多数 Agent 项目从 create_agent 开始即可；需要复杂工作流编排、确定性步骤与 Agent 步骤混合、长时间运行或底层状态控制时，再引入 LangGraph。

1.2 构成图的三要素

LangGraph 运行时主要由三个基本要素构成：State（状态）、Node（节点）、Edge（边）。

要素	是什么	通俗理解
State	图运行过程中的共享数据结构，表示应用在某一时刻的状态快照	一份所有节点都能读写的"公共档案"，中间结果都存在这里
Node	具体的执行单元，通常是一个函数。读取当前 State，执行业务逻辑，返回对 State 的局部更新	流水线上的一个"工位"，干完活只提交自己改的那部分
Edge	定义节点之间的流转关系，决定一个节点执行完后下一步去哪	流水线上的"传送带"，可以固定，也可以根据状态做条件判断

这是一个最简单的运行图，拓扑结构为 START → node_1 → node_2 → END。

1.3 图的运行过程：Superstep（超步）

LangGraph 的图运行过程基于 Superstep（超步） 来组织和推进。一次图运行从开始到结束，就是一系列连续的 Superstep 串联而成。

每个 Superstep 分为三个阶段：

阶段详解

1. 计划/路由阶段：根据当前的 State 和 Edge 逻辑，确定本轮超步应该执行哪些节点。
2. 执行阶段：运行本轮被选中的节点。如果多个节点同时被触发，它们会并行执行。每个节点都基于本轮开始时的状态快照计算，输出各自的局部更新。一个节点产生的更新不会立即被其他节点读取到。
3. 状态更新/提交阶段：当本轮所有节点执行完成后，LangGraph 将所有节点的输出统一合并到 State 中，生成新的状态快照。这个新状态作为下一轮 Superstep 的输入。

1.4 Graph API 与 Functional API

LangGraph 提供了两种构建运行图的 API，它们共享同一底层运行时。

1.4.1 Graph API（图式 API）

采用声明式方式构建工作流，显式定义 State、Node 和 Edge，把业务流程组织成可视化的图结构。

适合场景：复杂分支、多节点共享状态、并行执行、结果汇聚、需要图结构辅助调试和团队协作。

1.4.2 Functional API（函数式 API）

采用命令式方式构建工作流，更接近普通 Python 函数调用。用 @entrypoint 定义入口，用 @task 定义可被检查点记录的任务，内部使用普通的 if/else、循环和函数调用来组织流程。

适合场景：已有过程式代码需要最小改造、线性流程、简单分支、快速原型验证、局部任务持久化。

1.4.3 核心区别

对比项	Graph API	Functional API
编程风格	声明式图结构	命令式函数流程
核心抽象	State、Node、Edge	entrypoint、task
状态管理	显式定义全局 State	更多依赖函数参数和返回值
流程表达	通过节点和边表达	通过普通 Python 控制流表达
可视化能力	强，天然适合画图调试	弱，更像普通代码流程
学习成本	相对更高	相对更低

本文重点介绍 Graph API。

---

第 2 章 图的基础构建与运行

2.1 第一个例子：两节点顺序执行

先看一个最基础的例子：两个节点按顺序执行，边上是 START → node_1 → node_2 → END。

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add

class OverAllState(TypedDict):
    logs: Annotated[list[str], add]
    cur_id: str

def node_1(state: OverAllState) -> OverAllState:
    pre_id = state["cur_id"]
    return {
        "logs": ["node_1 运行完毕"],
        "cur_id": pre_id + ", node_1"
    }

def node_2(state: OverAllState) -> OverAllState:
    pre_id = state["cur_id"]
    return {
        "logs": ["node_2 运行完毕"],
        "cur_id": pre_id + ", node_2"
    }

builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)

graph = builder.compile()

print(graph.invoke({"cur_id": "start"}))

2.2 代码逐行拆解

2.2.1 导入

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add

• StateGraph：创建状态图的类
• START / END：图的起点和终点标记
• TypedDict：定义字典型状态结构
• Annotated：给类型附加额外信息（这里用来绑定 Reducer）
• add：Python 内置加法函数，充当 Reducer

2.2.2 定义全局状态

class OverAllState(TypedDict):
    logs: Annotated[list[str], add]
    cur_id: str

状态里有两个字段：

字段	类型	更新规则
logs	list[str]	用 add 合并，即追加
cur_id	str	默认规则，即覆盖

2.2.3 定义节点

def node_1(state: OverAllState) -> OverAllState:
    pre_id = state["cur_id"]
    return {
        "logs": ["node_1 运行完毕"],
        "cur_id": pre_id + ", node_1"
    }

节点函数接收一个参数 state（LangGraph 自动传入当前状态），返回一个字典（对状态的局部更新，不需要返回完整状态）。

• 从状态中读出 cur_id；
• 返回 logs 的新片段（会被 add 追加到原有 logs 后面）；
• 返回 cur_id 的新值（直接覆盖）。

2.2.4 添加节点

builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)

第一个参数是节点名（字符串），第二个参数是节点函数。

2.2.5 添加边

builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)

三条边分别表示：入口、顺序流转、出口。

2.2.6 编译与调用

graph = builder.compile()
print(graph.invoke({"cur_id": "start"}))

• compile()：把图编译成可执行对象
• invoke()：传入输入字典，运行整张图

状态变化过程：

阶段	logs	cur_id
输入	（空）	start
node_1 执行后	['node_1 运行完毕']	start, node_1
node_2 执行后	['node_1 运行完毕', 'node_2 运行完毕']	start, node_1, node_2

---

第 3 章 图的状态（State）管理

状态的定义实际上是在声明状态的 Schema。

3.1 定义状态的三种方式

3.1.1 方式一：TypedDict

其中，typedict能够精确描述这个字段有哪些键，并且每个键的类型是什么

class OverAllState(TypedDict):
    logs: Annotated[list[str], add]
    cur_id: str

特点：字段访问用 state["cur_id"]，利用字段下标的方式，结果也能更加的精确。

3.1.2 方式二：dataclass

from dataclasses import dataclass
@dataclass
class OverAllState:
    logs: Annotated[list[str], add]
    cur_id: str

输出与上面完全相同。区别：属性访问方式由 state['字段名'] 变为 state.字段名。

3.1.3 方式三：Pydantic

Pydantic 是一个数据验证框架，它利用 Python 的类型提示来：

• 自动验证：检查数据是否符合定义的规则
• 自动转换：将输入数据转换为正确的类型
• 清晰的错误信息：当验证失败时，提供详细的错误报告
• 序列化：轻松将数据转换为 JSON/字典

from pydantic import BaseModel
class OverAllState(BaseModel):
    logs: Annotated[list[str], add]
    cur_id: str

输出同样一致。特点：字段访问方式和 dataclass 相同（state.cur_id），且带有严格的类型校验。

3.1.4 三种方式的校验行为对比

场景	TypedDict	dataclass	Pydantic
输入字段不匹配	把输入字段视为字典 Key，抛 KeyError	把输入字段视为类属性，抛 TypeError	对输入进行校验，抛 ValidationError
节点返回字段不匹配	状态更新被忽略	状态更新被忽略	状态更新被忽略

3.1.5 推荐用法

推荐优先使用 TypedDict：

大多数官方案例都用 TypedDict。没有复杂校验需求时，TypedDict 是首选。

3.2 State Reducer（状态归约）

3.2.1 什么是 State Reducer

State Reducer 是 LangGraph 中合并状态更新的核心机制。它定义了当多个节点对同一个状态字段产生多个更新值时，如何合并成最终结果。

核心特征：

• 函数签名：(Value, Value) -> Value，接收当前值和更新值，返回合并后的新值

• 注解定义：通过 Annotated[Type, reducer_function] 为状态键指定 Reducer

3.2.2 如何定义 Reducer

第一步：定义 Reducer 函数

def my_reducer(left: list[str], right: list[str]) -> list[str]:
    return left + right
left = ['a', 'b']
right = ['c']
print(my_reducer(left, right))

输出：['a', 'b', 'c']

第二步：把 Reducer 和状态字段关联

from typing import TypedDict, Annotated
class OverAllState(TypedDict):
    logs: Annotated[list[str], my_reducer]
    cur_id: str

这里 Annotated 的第一个参数是字段类型，第二个参数是 Reducer 函数。LangGraph 会利用 Annotated 携带的元数据，在状态合并时调用 my_reducer。

3.2.3 常用内置 Reducer

（1）operator.add

Python 内置加法函数，等价于 a + b：

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']

（2）add_messages（消息合并）

add_messages 是 LangGraph 专门用于合并消息列表的 Reducer，常用于维护对话历史。它不只是简单拼接，而是依据消息的 id 进行合并：

• right 中 id 在 left 中不存在 → 追加到末尾；
• right 中 id 与 left 中已有消息相同 → 用新消息替换旧消息。

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=[]

最容易理解的方式就是，如果left里面有id=1，right里面也有id=1,那么就按照right里面的content，同理要是right没有的话，那就按照对应的left里面的值即可，最终生成每一部分的content

3.2.4 默认行为：覆盖

如果字段没有定义 Reducer，LangGraph 使用默认覆盖规则：本次返回的新值直接替换旧值。

class OverAllState(TypedDict):
    logs: list[str]
    id: str
def node_a(state: OverAllState):
    return {
        "logs": ["node_a"],
        "id": "node_a"
    }
def node_b(state: OverAllState):
    return {
        "logs": ["node_b"],
        "id": "node_b"
    }
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)

graph = builder.compile()
result = graph.invoke({"logs": ["START"], "id": "start"})
print('=' * 30, '-> result <-', '=' * 30)
print(result)

输出：

============================== -> result <- ==============================
{'logs': ['node_b'], 'id': 'node_b'}

logs 初始值是 ["START"]，但因为没有配置 Reducer，最终被 node_b 的值覆盖成了 ['node_b']。

3.3 节点中读写 State

3.3.1 读取 State

节点函数的第一个参数通常是状态对象。节点执行时，LangGraph 会自动把当前状态传进来：

def node_a(state: OverAllState):
    for k, v in state.items():
        print(f"k: {k}, v: {v}")

3.3.2 更新 State

节点只需要返回本节点要更新的字段（局部更新）：

• 没有返回的字段 → 保持原值不变；
• 返回的字段 → 按是否配置 Reducer 决定"合并"还是"覆盖"。

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}")
    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'}

分析：

• logs 绑定了 add Reducer，所以 ["START"] + ["node_a 更新状态"]，最终是 ['START', 'node_a 更新状态']；代码中介绍的是以对应的list的形式进行追加的操作
• id 没有被 node_a 返回，保持原值 "start"。

3.3.3 Overwrite：绕过 Reducer

某些场景下，我们希望本次更新不走 Reducer、直接覆盖。此时可以用 Overwrite 包裹值：

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)

核心就是虽然是追加状态但是，但是我在node_b这个节点里面进行重写操作了，所以由原来的 “logs”: [“node_a”]变成了"logs": ([“node_b”])，然后再进行追加操作，变成了{‘logs’: [‘node_b’, ‘node_c’]，然后id没有reducer直接执行默认覆盖操作就行了。

输出：

============================== -> result <- ==============================
{'logs': ['node_b', 'node_c'], 'id': 'node_c'}

分析：

• 如果不使用 Overwrite，logs 应该累积为 ['START', 'node_a', 'node_b', 'node_c']；
• 但 node_b 用 Overwrite(["node_b"]) 把当前 logs 整体覆盖为 ['node_b']；
• 之后 node_c 正常追加，得到 ['node_b', 'node_c']。

3.4 Multi Schema：四种状态类型

LangGraph 支持在一个图中使用多个状态 Schema，用来区分图的外部输入、外部输出、内部共享状态以及节点间的临时状态。

3.4.1 四种状态类型总览

状态类型	作用	在哪声明
全局状态	图内部主要使用的状态，图运行中读写的大部分字段	StateGraph(state_schema=...)
输入状态	约束图对外接收哪些字段	StateGraph(input_schema=...)
输出状态	约束图最终只返回哪些字段	StateGraph(output_schema=...)
私有状态	节点之间传递的临时状态	节点函数入参的类型注解

3.4.1 案例

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":"小黄"}))

输出：

{ "graph_output": "Dear 小黄, 早上好~ 很高兴认识你！" }

数据流转过程：

• InputState：只声明 username，约束外部只能传这个字段；
• OverAllState：图内部的主状态，username（来自输入）、nickname（node_1 写）、graph_output（node_3 写）；
• PrivateState：node_2 和 node_3 之间传递 greeting 临时数据；
• OutputState：只声明 graph_output，所以最终结果只返回这一个字段（username、nickname、greeting 都被裁剪掉了）。

3.5 预定义状态

3.5.1 MessagesState

LangGraph 官方提供了一个预定义状态类型 MessagesState，可以直接继承并扩展。

源码如下：

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

它只有一个字段 messages，类型为消息列表，绑定了 add_messages Reducer。

示例：MessagesState + LLM

#此处省略了对应导入的模块部分
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)

首先解释一下难点部分，下面这一部分介绍的是继承了messagestate里面默认的messages，所以在最后输出的时候有三个属性，分别是messages，username和output，这个是里面规定的overAllstate的属性值有多少最终就输出多少。

同时messages绑定了 add_messages Reducer，可以进行添加操作，自带追加合并，invoke要返回的必须要是完整的state里面的快照，所以也会有username返回回去。

class OverAllState(MessagesState):
    username: str
    output: str

输出：

{
    "messages": [
        HumanMessage(content="你好，我是 小黄", ...),
        AIMessage(content="你好呀，小黄！😊 我是DeepSeek，很高兴认识你！...", ...),
    ],
    "username": "小黄",
    "output": "你好呀，小黄！😊 我是DeepSeek，很高兴认识你！..."
}

3.5.2 AgentState

AgentState 是 LangChain Agent 内部使用的状态类型，完整类名是 langchain.agents.middleware.types.AgentState。

源码（简化）：

class AgentState(TypedDict, Generic[ResponseT]):
    messages: Required[Annotated[list[AnyMessage], add_messages]]
    jump_to: NotRequired[Annotated[JumpTo | None, EphemeralValue, PrivateStateAttr]]
    structured_response: NotRequired[Annotated[ResponseT, OmitFromInput]]

主要字段：

字段	作用
messages	Agent 运行过程中的消息列表，使用 add_messages 作为 Reducer
jump_to	Agent 中间件体系使用的内部控制字段，表示跳转意图（普通 StateGraph 中不会自动触发跳转，跳转应使用 Command(goto=...)）
structured_response	存放 Agent 的结构化输出，OmitFromInput 表示不作为外部输入暴露。

结语

到这里，你就理解了一些langgraph基础逻辑，后续我会继续更新langgrpah控制流详解和fastApi进阶内容，希望我们共同进步。

祝你在 LangGraph 的道路上越走越顺！