# 大模型技术之LangGraph，从原型到生产：LangGraph工程化落地全解

# 第一章认识LangGraph：从线性链到图编排，Agent开发的新范式

随着大模型从「单次调用工具」转向「推理引擎」，我们需要构建的智能体(Agent)越来越复杂：多轮循环推理、多智能体协作、人工介入审核、长时间运行的后台任务……传统基于线性链式调用的框架，已经很难应对这些场景的状态管理、流程控制和容错需求。

LangGraph正是LangChain团队针对上述痛点推出的新一代智能体编排框架，它以「图」为核心抽象，为构建生产级、有状态、可持久的Agent应用提供了完整的解决方案。本章我们将从底层逻辑到能力边界，全面建立对LangGraph的认知。

# 1.1 从Agent开发的痛点说起

在LangGraph出现之前，大多数开发者基于LangChain的LCEL(LangChain表达式语言)构建Agent工作流。这种线性链式的方式在简单场景下足够高效，但随着业务复杂度提升，会暴露出三个核心问题：

状态管理混乱：线性链没有内置的全局状态机制，跨节点共享数据、留存对话历史、维护中间结果都需要开发者手动处理，多轮交互后极易出现状态不一致。
流程控制能力不足：复杂Agent往往需要循环推理、条件分支、并行执行，线性链只能通过hack方式实现循环，代码可读性和可维护性急剧下降。
容错与恢复困难：长时间运行的任务一旦遇到网络中断、程序崩溃，只能从头重新执行；需要人工审核的环节，无法优雅地暂停流程、介入修改后再继续。

这些问题的本质，是「线性执行模型」和「复杂有状态工作流」的不匹配。LangGraph的解法，是把Agent工作流建模为有状态的图结构，用图计算的思路来解决智能体编排问题。

# 1.2 LangGraph到底是什么

LangGraph是一个低级编排框架与运行时环境，专门用于构建、管理和部署长期运行的有状态智能体。它的核心思想非常简洁：把整个Agent工作流抽象成一张图，由三个基础元素构成：

节点(Nodes)：计算单元，可以是大模型调用、工具执行、自定义业务逻辑，所有具体的计算都发生在节点中。
边(Edges)：流程控制规则，定义节点之间的跳转逻辑，可以是固定跳转，也可以是基于状态的条件分支。
状态(State)：全局共享的数据结构，在图的整个执行过程中在节点间传递，所有节点的输出都会更新到状态中。

简单来说：节点负责干活，边决定下一步去哪，状态承载所有中间数据。三者组合，就能构建出支持循环、分支、并行的复杂工作流，同时内置了完整的状态管理机制。

LangGraph既可以独立使用，也能和LangChain、LangSmith无缝集成，它没有替代LangChain，而是补全了LangChain在复杂智能体编排上的能力短板。

# 1.3 LangGraph的核心生产级能力

相比普通的工作流引擎，LangGraph针对大模型Agent场景做了深度优化，提供了5项构建生产级应用的核心能力：

持久化执行(断点续跑) 每个执行步骤的状态都会自动保存为检查点。如果程序崩溃、服务重启，工作流可以从最后一个成功的步骤继续执行，无需从头开始，特别适合长时间运行的后台任务。
人机协作 可以在工作流的任意节点设置中断，暂停执行并将状态暴露给人工审核、修改，确认后再恢复运行。这是审批类工作流、人工增强型Agent的必备能力。
分层记忆管理 原生支持两种记忆：
- 短期记忆：基于检查点的线程级对话历史，支撑多轮对话交互；
- 长期记忆：跨会话的存储能力，可以保存用户偏好、知识库，支持向量语义检索。
全链路流式处理 不仅支持大模型输出的Token流式返回，还支持状态更新、自定义事件、子图内部数据的流式输出，能很好地优化前端用户体验。
生产级部署基础设施 底层基于并行图计算模型设计，支持水平扩展，同时提供了SQLite、PostgreSQL等多种持久化存储方案，满足从本地开发到线上生产的全阶段需求。

# 1.4 一文分清：LangChain vs LangGraph

很多开发者会混淆两者的定位，我们从5个维度做清晰对比，帮你快速判断该用哪个：

对比维度	LangGraph	LangChain
抽象级别	低级编排，提供细粒度的流程控制	高级封装，提供开箱即用的链与组件
状态管理	内置状态机与检查点机制，自动管理	无原生全局状态，需要开发者自行维护
执行模型	基于图的并行、循环、条件分支执行	线性链式执行，循环与分支能力弱
持久化能力	原生支持，一行代码开启断点续跑	无原生支持，需要额外开发实现
适用场景	复杂有状态智能体、多Agent系统、长流程任务	简单链式调用、快速原型开发、单轮交互应用

一句话总结：简单场景用LangChain快速搭建，复杂有状态的Agent用LangGraph精细编排。

# 1.5 底层架构：Pregel图计算模型

LangGraph的运行时基于Google提出的Pregel图计算算法，这也是它能稳定支持并行执行、状态一致性的核心原因。理解Pregel的执行逻辑，就能吃透LangGraph的运行原理。

整个执行过程以「超步(Super-step)」为单位循环迭代，每一个超步分为三个阶段：

规划(Plan)：根据当前状态和边的规则，确定本轮需要执行的所有节点；
执行(Execution)：并行运行所有选中的节点，每个节点读取状态、执行逻辑、生成输出；
更新(Update)：将所有节点的输出，按照规则合并更新到全局状态中。

这个循环会持续执行，直到没有新的节点需要运行，工作流结束。

在这个模型下，还有两个核心概念：

Actors(节点)：订阅状态通道、读写数据的计算单元，兼容LangChain的Runnable接口，可以直接复用LangChain生态的组件。
Channels(通道)：节点间通信的载体，也是状态的底层实现。常用的通道类型有：
- LastValue：存储最新值，对应默认的覆盖式状态更新；
- Topic：发布-订阅模式的通道，适合事件通知；
- BinaryOperatorAggregate：聚合型通道，适合列表追加、数值累加等场景。

# 1.6 哪些场景该用LangGraph

结合能力特点，LangGraph尤其适合以下5类场景：

复杂多智能体系统：多个Agent分工协作、需要调度流转的场景，比如研发Agent团队、客服多技能组调度。
需要长期记忆的应用：跨会话留存用户信息、个性化偏好的智能客服、个人助理类应用。
含人工审核的工作流：比如内容生成审核、财务审批、高危操作确认等需要人机协作的流程。
后台长任务+实时交互：比如文档批量处理、数据分析任务，支持后台运行、进度实时推送、中断后恢复。
高度定制化的智能体：需要精确控制执行流程、自定义状态逻辑的复杂Agent，不满足于开箱即用的标准Agent框架。

# 第二章快速入门：从零搭建第一个LangGraph智能体

上一章我们建立了对LangGraph的全局认知，本章我们直接进入实操环节，从环境搭建到运行第一个完整的图工作流，带你用最少的代码掌握LangGraph的基础开发范式。

# 2.1 环境准备

LangGraph基于Python开发，环境搭建非常轻量，仅需几步即可完成。

# 2.1.1 版本与依赖要求

Python版本：推荐使用 3.11 及以上，官方稳定测试版本为 Python 3.12
核心依赖：仅需安装 langgraph 基础包即可完成入门开发，持久化、向量存储等扩展能力可后续按需安装

# 2.1.2 安装步骤

# 方式1：使用 Conda 管理环境(推荐)

如果你习惯用 Conda 管理Python环境，可以按以下步骤操作：

## 1. 创建虚拟环境
conda create -n langgraph python=3.12
## 2. 激活环境
conda activate langgraph
## 3. 安装 LangGraph
pip install -U langgraph

1
2
3
4
5
6

# 方式2：使用原生 venv

如果不使用Conda，用Python自带的虚拟环境也可以：

## 1. 创建虚拟环境
python -m venv langgraph-env
## 2. 激活环境（Windows）
langgraph-env\Scripts\activate
## 2. 激活环境（Mac/Linux）
source langgraph-env/bin/activate
## 3. 安装 LangGraph
pip install -U langgraph

1
2
3
4
5
6
7
8

# 2.1.3 验证安装

安装完成后，执行以下命令验证是否安装成功：

pip show langgraph

正常输出版本号、作者等信息即代表安装完成。截至本书编写时，最新稳定版为 1.0.x 系列。

说明：LangGraph 基础包已内置内存级检查点、状态管理等核心能力，本地开发和调试无需安装任何额外依赖。

# 2.2 第一个最简工作流：单节点对话

我们从最简单的场景入手：构建一个单节点的对话工作流，接收用户输入，模拟大模型返回回复。整个流程只有一个计算节点，从起始点进入，执行完成后直接结束。

# 2.2.1 完整代码示例

from langgraph.graph import StateGraph, MessagesState, START, END

## 1. 定义节点函数：模拟大模型调用
def mock_llm(state: MessagesState):
    """接收当前状态，返回更新后的消息"""
    return {"messages": [{"role": "ai", "content": "hello world"}]}

## 2. 初始化状态图，指定状态类型为 MessagesState
graph = StateGraph(MessagesState)

## 3. 添加节点与边，编排执行流程
graph.add_node("mock_llm", mock_llm)  ## 添加计算节点
graph.add_edge(START, "mock_llm")     ## 起始点 -> 大模型节点
graph.add_edge("mock_llm", END)       ## 大模型节点 -> 结束点

## 4. 编译图，生成可执行对象
app = graph.compile()

## 5. 传入初始状态，启动工作流
if __name__ == "__main__":
    result = app.invoke({"messages": [{"role": "user", "content": "hi!"}]})
    print("最终状态：", result)
    print("AI回复：", result["messages"][-1]["content"])

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

运行代码后，你会得到如下输出：

最终状态： {'messages': [{'role': 'user', 'content': 'hi!'}, {'role': 'ai', 'content': 'hello world'}]}
AI回复： hello world

1
2

# 2.2.2 逐行拆解代码

别看代码短，它已经完整覆盖了LangGraph开发的核心流程，我们逐部分拆解：

导入基础组件
- StateGraph：LangGraph的核心类，用于构建状态图
- MessagesState：官方内置的状态类型，自带消息列表追加的合并逻辑，对话类场景可以直接复用
- START / END：两个特殊节点，分别代表图的入口和出口
定义节点函数 节点本质上就是普通的Python函数，它遵循两个规则：
- 输入参数是图的当前状态(类型与图定义的State一致)
- 返回值是一个字典，键对应状态中的字段，值为要更新的内容
本例中mock_llm函数读取状态，返回一条AI消息，状态会自动合并这条新消息。
构建图结构
- StateGraph(MessagesState)：创建一个状态图实例，绑定状态类型
- add_node("节点名", 节点函数)：向图中添加计算节点，节点名是唯一标识
- add_edge(起点, 终点)：添加一条普通边，表示执行完起点节点后，无条件跳转到终点节点
编译图 compile() 方法会校验图结构的合法性，将定义好的图转换为可执行的运行时对象。后续所有执行、流式输出、持久化等能力，都基于编译后的对象调用。
调用执行 invoke(初始状态) 是最基础的同步执行方法，传入符合State格式的初始数据，工作流会从头到尾执行完毕，返回最终的完整状态。

# 2.3 LangGraph通用开发五步法

从上面的例子可以提炼出，所有LangGraph应用的开发都遵循一套固定的流程，掌握这套范式，后续再复杂的工作流也只是节点和边的扩展：

定义State(状态) 确定整个工作流需要共享的数据结构，这是图的“数据骨架”。所有节点都基于状态读写，状态的设计直接决定了工作流的灵活性。
实现Nodes(节点) 将业务逻辑封装为独立的节点函数，每个节点负责一项具体计算(大模型调用、工具执行、数据处理等)。节点之间通过状态传递数据，彼此解耦。
编排Edges(边) 通过普通边、条件边定义节点的执行顺序和跳转规则，搭建工作流的“控制流程”。简单的线性流程用普通边，复杂的分支、循环用条件边。
编译Graph(图) 调用compile()完成图的构建，同时可以在此处配置持久化存储、缓存、中断点等高级能力。
调用执行 传入初始状态，通过invoke()同步执行、stream()流式输出、ainvoke()异步执行等方式启动工作流，获取最终结果。

# 第三章 Graph API 核心详解：状态、节点与边，从零构建复杂工作流

第二章我们跑通了最简单节点工作流，掌握了LangGraph开发的五步法。本章我们深入Graph API的核心，逐一拆解State(状态)、Nodes(节点)、Edges(边) 三大基础组件，再拓展Send动态分支、Command一体化控制、运行时上下文、可视化、异步编程等高级能力。掌握本章内容，你就能独立编排绝大多数复杂的Agent工作流。

# 3.1 State：工作流的全局共享内存

State是整个图的“数据骨架”：它是贯穿所有节点的共享数据结构，所有节点读取输入、写入输出都基于State完成。

一个完整的State定义由两部分组成：

Schema(结构)：定义状态包含哪些字段、每个字段的类型，相当于数据模型；
Reducer(合并规则)：定义节点返回的新值，如何和状态中已有的旧值合并。每个字段可以独立配置自己的Reducer。

# 3.1.1 State Schema：定义数据结构

Schema通常用TypedDict或Pydantic模型定义，LangGraph提供了三种Schema维度，帮你做接口隔离和数据封装。

# 1. state_schema：全局状态空间

state_schema是图的完整内部状态，包含所有节点可能读写的字段，是必须指定的核心参数。

特点：所有节点都可以访问、写入其中的任意字段，是图的“全局变量空间”。
定义方式：

from typing_extensions import TypedDict

class OverallState(TypedDict):
    question: str
    answer: str
    intermediate_result: str

1
2
3
4
5
6

# 2. input_schema / output_schema：接口隔离

在实际开发中，我们往往不希望把全部内部状态暴露给调用方，这时就可以用input_schema和output_schema做接口约束：

input_schema：定义图接受的入参字段，必须是state_schema的子集。调用方只能传入这些字段，避免非法输入污染内部状态。
output_schema：定义图返回的出参字段，必须是state_schema的子集。执行结束后只返回指定字段，隐藏内部中间数据。

示例：

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

## 输入状态：只暴露question字段
class InputState(TypedDict):
    question: str

## 输出状态：只返回answer字段
class OutputState(TypedDict):
    answer: str

## 全局状态：包含所有内部字段
class OverallState(InputState, OutputState):
    intermediate_result: str

def answer_node(state: InputState):
    answer = "你好" if "你好" in state["question"] else "再见"
    return {"answer": answer, "intermediate_result": "处理完成"}

## 构建图时指定三种Schema
builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
builder.add_node("answer", answer_node)
builder.add_edge(START, "answer")
builder.add_edge("answer", END)
graph = builder.compile()

## 调用时只能传入question，最终只返回answer
result = graph.invoke({"question": "你好"})
print(result)  ## {'answer': '你好'}

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
28
29

# 3. 私有状态：节点间的局部数据

如果两个节点之间需要传递一些临时数据，但不希望写入全局状态被其他节点读取，可以用私有状态实现。

原理：上游节点的输出Schema包含私有字段，下游节点的输入Schema接收该字段，该字段不会写入全局State，仅在两个节点间传递。
适用场景：节点间的临时中间数据，不需要全局共享，避免状态膨胀。

# 3.1.2 Reducer：状态合并规则

Reducer是State最核心的设计：它决定了节点返回的新值，如何与状态中的旧值合并。状态中的每个字段都可以独立配置Reducer，尤其在多节点并行更新同一字段时，Reducer能保证数据一致性。

LangGraph常用的Reducer有以下几种：

# 1. 默认行为：覆盖更新

如果字段没有指定Reducer，默认执行覆盖更新：新值直接替换旧值。适合单值、不需要累加的字段。

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class DefaultState(TypedDict):
    count: int  ## 无注解，默认覆盖更新

def add_one(state: DefaultState):
    return {"count": 2}  ## 直接覆盖原值

builder = StateGraph(DefaultState)
builder.add_node("add", add_one)
builder.add_edge(START, "add")
builder.add_edge("add", END)
graph = builder.compile()

result = graph.invoke({"count": 1})
print(result["count"])  ## 输出 2，原值1被覆盖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# 2. add_messages：消息列表专用

add_messages是对话场景最常用的Reducer，专门用于处理消息列表：

自动将新消息追加到历史消息末尾，而不是覆盖；
自动兼容LangChain的HumanMessage、AIMessage等消息对象；
配合MessagesState内置类型可以直接使用。

from typing import Annotated, List
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class ChatState(TypedDict):
    ## 用Annotated为messages字段指定add_messages reducer
    messages: Annotated[List, add_messages]

def chat_node(state: ChatState):
    return {"messages": [{"role": "ai", "content": "你好"}]}

builder = StateGraph(ChatState)
builder.add_node("chat", chat_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
graph = builder.compile()

result = graph.invoke({"messages": [{"role": "user", "content": "hi"}]})
print(len(result["messages"]))  ## 输出 2，消息被追加而非覆盖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 3. operator.add：追加与累加

operator.add是通用的追加型Reducer，支持三种数据类型：

列表追加：新列表元素追加到旧列表末尾；
数值累加：新值与旧值相加；
字符串拼接：新字符串拼接到旧字符串后面。

import operator
from typing import Annotated, List
from typing_extensions import TypedDict

class DemoState(TypedDict):
    data_list: Annotated[List[int], operator.add]  ## 列表追加
    total: Annotated[int, operator.add]            ## 数值累加
    text: Annotated[str, operator.add]             ## 字符串拼接

1
2
3
4
5
6
7
8

提示：并行节点更新同一字段时，operator.add会把所有节点的输出都合并进去，非常适合多分支结果汇总的场景。

# 4. operator.mul：数值相乘

operator.mul用于数值字段的相乘更新，但官方实现存在一个设计缺陷：初始化阶段会用类型默认值(0)与初始值相乘，导致结果始终为0。

解决方案：使用自定义Reducer，跳过初始化阶段的默认值计算。

# 5. 自定义Reducer

内置Reducer不满足需求时，可以自己编写Reducer函数。函数接收两个参数：current_value(当前状态中的旧值)和new_value(节点返回的新值)，返回合并后的值。

示例：实现字典合并，新值覆盖旧键、保留旧键中不存在的字段：

from typing import Annotated, Dict, Any
from typing_extensions import TypedDict

def dict_merge(current: Dict[str, Any], new: Dict[str, Any]) -> Dict[str, Any]:
    """合并字典：新值覆盖旧键，保留旧值独有键"""
    result = current.copy()
    result.update(new)
    return result

class CustomState(TypedDict):
    metadata: Annotated[Dict[str, Any], dict_merge]

1
2
3
4
5
6
7
8
9
10
11

# Reducer核心价值总结

控制更新策略：不同字段用不同合并逻辑，覆盖、追加、聚合按需选择；
解决并行冲突：多节点同时更新同一字段时，按规则合并，避免数据丢失；
灵活扩展：自定义Reducer可以适配任意业务的状态合并需求。

# 3.2 Nodes：计算逻辑的载体

节点是LangGraph中执行具体计算的单元，本质上就是普通的Python函数(支持同步和异步)。每个节点接收当前状态，执行业务逻辑，返回状态更新。

节点函数支持三个入参：

state：图的当前状态，必选；
config：运行配置对象，包含thread_id、标签等信息；
runtime：运行时上下文，包含存储、流写入器等对象。

# 3.2.1 START 与 END 特殊节点

LangGraph有两个内置的特殊节点，不需要自己实现：

START：图的入口节点，代表工作流的起点，用于指定第一个执行的节点；
END：图的终点节点，代表工作流结束，执行到该节点后工作流终止。

# 3.2.2 节点缓存

对于耗时较长的节点(比如复杂计算、重复的大模型调用)，可以开启节点缓存，相同输入再次执行时直接返回缓存结果，大幅提升性能。

使用方式分为两步：

编译图时指定缓存存储(如内存缓存)；
添加节点时配置缓存策略(TTL过期时间)。

import time
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy

class State(TypedDict):
    x: int
    result: int

def expensive_node(state: State):
    time.sleep(2)  ## 模拟耗时计算
    return {"result": state["x"] * 2}

builder = StateGraph(State)
## 为节点添加缓存策略，TTL=10秒
builder.add_node("calc", expensive_node, cache_policy=CachePolicy(ttl=10))
builder.set_entry_point("calc")
builder.set_finish_point("calc")

## 编译时启用内存缓存
graph = builder.compile(cache=InMemoryCache())

## 第一次调用耗时2秒，第二次调用直接命中缓存，瞬间返回
print(graph.invoke({"x": 5}))
print(graph.invoke({"x": 5}))

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

# 3.2.3 节点重试策略

调用外部API、数据库、大模型等不稳定的操作时，可以为节点配置重试策略，失败后自动重试，提升系统容错性。

通过retry_policy参数配置：

max_attempts：最大重试次数；
retry_on：指定哪些异常需要重试，支持传入异常类型或自定义判断函数。

默认重试策略会跳过ValueError、TypeError、SyntaxError等代码类异常，仅对网络类等临时性异常重试。

from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy
from typing_extensions import TypedDict

class State(TypedDict):
    result: str

def unstable_api(state: State):
    ## 模拟不稳定的API调用
    import random
    if random.random() < 0.7:
        raise Exception("网络超时")
    return {"result": "调用成功"}

builder = StateGraph(State)
## 配置最多重试5次
builder.add_node("api", unstable_api, retry_policy=RetryPolicy(max_attempts=5))
builder.add_edge(START, "api")
builder.add_edge("api", END)
graph = builder.compile()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 3.2.4 延迟节点执行

在多分支并行的场景中，经常需要一个汇总节点等待所有分支执行完成后再运行。这时可以给汇总节点设置defer=True，它会推迟到所有其他待执行任务完成后再启动。

典型场景：Map-Reduce中的Reduce节点、多分支结果聚合节点。

from langgraph.graph import StateGraph, START, END

## ... 省略状态和节点定义 ...
builder.add_node("summary", summary_node, defer=True)  ## 延迟执行

1
2
3
4

# 3.3 Edges：流程控制的核心

边定义了节点之间的跳转规则，决定了工作流的执行顺序。LangGraph提供了普通边、条件边两种核心类型，配合循环控制可以实现任意复杂的流程。

# 3.3.1 普通边：固定跳转

普通边是最基础的边，表示执行完源节点后，无条件跳转到目标节点。

## 语法：add_edge(源节点名, 目标节点名)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)

1
2
3
4

一个节点可以有多条出边，此时所有目标节点会在下一轮并行执行。

# 3.3.2 条件边：动态路由

条件边根据当前状态的值，动态决定下一个执行的节点，是实现分支逻辑的核心。

使用add_conditional_edges方法，包含三个参数：

源节点名；
路由函数：接收状态，返回下一个节点的名称；
路由映射(可选)：将路由函数的返回值映射到实际节点名，解耦逻辑与节点命名。

from typing import Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    value: int

def node_a(state: State):
    return {"value": state["value"] + 1}

def node_b(state: State):
    return {"value": state["value"] * 2}

def node_c(state: State):
    return {"value": state["value"] - 1}

## 路由函数：根据value奇偶性选择分支
def route(state: State) -> Literal["b", "c"]:
    if state["value"] % 2 == 0:
        return "b"
    return "c"

builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_node("c", node_c)
builder.add_edge(START, "a")
## 添加条件边：从a节点出发，根据route函数跳转
builder.add_conditional_edges(
    "a",
    route,
    {"b": "b", "c": "c"}  ## 路由映射
)
builder.add_edge("b", END)
builder.add_edge("c", END)

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
28
29
30
31
32
33
34
35

# 3.3.3 入口点与出口点

除了用START/END加边，也可以用set_entry_point和set_finish_point直接指定起止节点，效果等价：

builder.set_entry_point("node_a")
builder.set_finish_point("node_b")

1
2

# 3.3.4 循环与递归限制

结合条件边，我们可以轻松实现循环执行的工作流，比如Agent的“思考-行动”循环。同时为了防止死循环，LangGraph提供了递归限制。

# 实现循环

核心思路：节点A执行后，通过条件边判断是否满足终止条件，满足则跳转到END，不满足则跳回节点B继续循环。

# 递归限制

默认情况下，LangGraph最多执行25个超步(执行轮次)，超过会抛出GraphRecursionError。可以通过config中的recursion_limit自定义该值。

from langgraph.errors import GraphRecursionError

## 调用时设置递归限制为100
result = graph.invoke(initial_state, config={"recursion_limit": 100})

1
2
3
4

# 3.4 Send：动态分支与Map-Reduce模式

普通的边都是预先定义好节点数量的，但实际场景中经常需要动态生成任务：比如根据文档列表，为每篇文档生成一个处理任务，任务数量在运行时才能确定。

LangGraph通过Send对象支持这种动态Map-Reduce模式：从条件边返回一组Send对象，每个Send对应一个节点实例，所有实例并行执行。

Send接收两个参数：

目标节点名称；
传递给该节点的独立状态。

示例：根据主题列表，批量生成笑话

from typing import Annotated, List, Sequence
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send

class OverallState(TypedDict):
    subjects: List[str]
    jokes: Annotated[List[str], lambda x, y: x + y]

## 1. 生成主题列表的节点
def generate_subjects(state: OverallState):
    return {"subjects": ["猫", "狗", "程序员"]}

## 2. Map节点：为单个主题生成笑话(会被并行调用多次)
def make_joke(state: OverallState):
    subject = state.get("subject", "")
    joke = f"关于{subject}的笑话..."
    return {"jokes": [joke]}

## 3. 路由函数：为每个主题创建一个Send对象
def map_subjects(state: OverallState) -> Sequence[Send]:
    return [Send("make_joke", {"subject": s}) for s in state["subjects"]]

## 构建图
builder = StateGraph(OverallState)
builder.add_node("generate", generate_subjects)
builder.add_node("make_joke", make_joke)
builder.add_edge(START, "generate")
## 条件边返回Send列表，动态启动多个make_joke节点
builder.add_conditional_edges("generate", map_subjects)
builder.add_edge("make_joke", END)

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
28
29
30
31

执行流程：

先执行generate_subjects，生成主题列表；
条件边为每个主题创建一个Send，并行启动3个make_joke实例；
所有make_joke执行完成后，结果通过Reducer合并到状态中。

# 3.5 Command：节点内一体化控制流

通常我们用节点更新状态、用边控制流程，但很多时候状态更新和流程跳转是强关联的(比如处理完任务直接决定下一步去哪)。这时分开写节点和条件边会很冗余，Command对象就是为了解决这个问题：在节点返回值中同时包含状态更新和下一步跳转指令。

# 3.5.1 基本用法

节点返回Command对象，通过update参数指定状态更新，通过goto参数指定下一个节点。

示例：多智能体调度

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing_extensions import TypedDict

class AgentState(TypedDict):
    messages: list
    current_agent: str

def decision_agent(state: AgentState) -> Command[AgentState]:
    last_msg = state["messages"][-1]
    if "数学" in last_msg:
        return Command(
            update={"messages": ["路由到数学代理"], "current_agent": "math"},
            goto="math_agent"
        )
    elif "翻译" in last_msg:
        return Command(
            update={"messages": ["路由到翻译代理"], "current_agent": "translate"},
            goto="translate_agent"
        )
    else:
        return Command(update={"messages": ["任务完成"]}, goto=END)

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

# 3.5.2 Command vs 条件边

方案	适用场景	优势
Command	状态更新与流程跳转强关联，比如多Agent切换、工具执行后路由	代码内聚，一个函数同时处理逻辑与路由，减少冗余
条件边	仅路由、不需要更新状态，或路由逻辑独立于业务逻辑	职责分离，路由规则集中管理，适合复杂分支

# 3.5.3 父图导航

在子图中使用Command时，加上graph=Command.PARENT参数，可以直接从子图跳转到父图的指定节点，实现跨层级的流程控制，特别适合多智能体交接场景。

# 3.5.4 实战：工具内部更新状态

Command的典型应用是工具函数内部：工具执行完成后，直接更新状态并决定下一步流程，不需要额外的路由节点。比如客户支持场景中，查询完客户信息后自动跳转到客服代理节点。

# 3.6 Runtime Context：运行时依赖注入

状态应该只存业务数据，但实际开发中节点经常需要一些运行时配置(比如模型名称、数据库连接、API密钥)。如果把这些配置放进State，会污染业务数据，也不利于多环境切换。

LangGraph提供了context_schema机制，专门用于传递不属于状态的运行时依赖：

状态存业务数据，上下文存配置依赖，职责分离；
上下文在节点间共享，但不会被持久化到检查点；
支持类型注解，IDE可以自动补全。

使用方式：

用dataclass定义上下文结构；
创建图时指定context_schema；
节点通过runtime.context访问上下文；
调用时通过context参数传入具体值。

from dataclasses import dataclass
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime

## 1. 定义状态(业务数据)
class AgentState(TypedDict):
    response: str

## 2. 定义上下文(运行时配置)
@dataclass
class ContextSchema:
    model_name: str
    db_connection: str
    api_key: str

## 3. 节点通过runtime参数访问上下文
def process_node(state: AgentState, runtime: Runtime[ContextSchema]):
    model = runtime.context.model_name
    db = runtime.context.db_connection
    return {"response": f"使用{model}处理，连接{db}"}

## 4. 构建图时指定context_schema
builder = StateGraph(AgentState, context_schema=ContextSchema)
builder.add_node("process", process_node)
builder.add_edge(START, "process")
builder.add_edge("process", END)
graph = builder.compile()

## 5. 调用时传入上下文
context = ContextSchema(
    model_name="gpt-4o",
    db_connection="postgresql://localhost/db",
    api_key="sk-xxxx"
)
result = graph.invoke({}, context=context)

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
28
29
30
31
32
33
34
35
36

# 3.7 工作流可视化

复杂的工作流编写完成后，可以通过可视化快速校验结构、调试问题。LangGraph提供了两种常用的可视化方式。

# 1. Mermaid 流程图

调用draw_mermaid()方法可以生成Mermaid格式的流程图代码，复制到Mermaid在线编辑器或支持Mermaid的Markdown工具中即可渲染。

graph_structure = graph.get_graph()
mermaid_code = graph_structure.draw_mermaid()
print(mermaid_code)

1
2
3

如果在Jupyter Notebook环境，可以直接渲染为图片：

from IPython.display import Image
Image(graph.get_graph().draw_mermaid_png())

1
2

# 2. ASCII 文本图

安装grandalf依赖后，可以生成纯文本的ASCII流程图，直接在控制台打印，适合无GUI的服务器环境调试。

pip install grandalf

ascii_graph = graph.get_graph().draw_ascii()
print(ascii_graph)

1
2

# 3.8 异步编程：提升IO密集型性能

对于大量API调用、网络请求的IO密集型场景，同步执行会浪费大量等待时间。LangGraph原生支持异步编程，配合async/await可以显著提升并发性能。

# 核心用法

节点函数用async def定义；
执行图时用await graph.ainvoke()或await graph.astream()；
多个异步任务可以用asyncio.gather()并发执行。

性能对比：

同步执行4个耗时1秒的API调用，总耗时约4秒；
异步并发执行，总耗时约等于最慢的单个任务(1.5秒左右)，性能提升数倍。

提示：异步只对IO绑定任务有明显收益，如果是CPU密集型计算，异步并不会提升速度。

# 第四章高级特性：构建生产级Agent的核心能力

前三章我们基于Graph API搭建了基础工作流，实现了节点编排、条件分支、循环、动态任务分发等核心逻辑。但要把Agent落地到真实生产环境，仅靠流程编排远远不够：多轮对话需要保留上下文、长任务需要容错断点、高风险操作需要人工审核、长对话需要管控上下文窗口……

本章我们系统拆解LangGraph的核心高级特性，覆盖持久化、人机中断、流式输出、记忆管理、时间回溯、子图复用六大模块。掌握这些能力后，就能构建出完整可用的生产级智能体应用。

# 4.1 持久化：给工作流加上存档能力

# 4.1.1 为什么需要持久化

没有持久化的工作流是「一次性」的：程序重启、对话结束，所有状态就全部丢失。真实业务中我们普遍需要这些能力：

多轮对话：用户分多次交互，自动继承历史上下文，无需手动拼接消息
断点续跑：长任务执行中服务崩溃、网络中断，重启后从断点继续，避免重复计算
状态回溯：调试时查看每一步的状态快照，精准定位问题根源
多会话隔离：同时服务多个用户、多个任务，彼此状态互不干扰

LangGraph通过Checkpointer(检查点器) 机制实现持久化：工作流每执行完一个超步(Superstep)，就自动将完整状态写入存储，全程无需业务代码手动干预。

# 4.1.2 核心概念

Checkpointer(检查点器) 持久化的核心管理器，负责状态的写入、读取、查询、删除。采用可插拔设计，编译图时注入即可，业务节点零修改就能获得持久化能力。
Thread(会话线程) 会话隔离的最小单位，每个Thread对应一个独立的工作流实例，通过唯一的thread_id标识。不同用户、不同任务分配不同的thread_id，状态完全隔离。
Checkpoint(检查点快照) 某一时刻工作流的完整状态存档，包含当前全量业务状态、已执行节点轨迹、下一个待执行节点、版本号(用于乐观锁并发控制)。

# 4.1.3 内置Checkpointer选型

LangGraph官方提供多种开箱即用的实现，覆盖从开发到生产的全场景：

实现类	存储介质	适用场景	核心特点
MemorySaver	进程内存	本地开发、单元测试、Demo演示	零依赖、速度最快，程序重启后数据丢失
SQLiteSaver	本地SQLite文件	单机部署、轻量生产工具	单文件易迁移，支持事务，无需额外部署服务
PostgresSaver	PostgreSQL数据库	分布式生产环境	高可用、支持事务、易扩展，适合企业级应用
RedisSaver	Redis缓存	高并发低延迟场景	读写性能高，支持自动过期清理

# 4.1.4 基础使用示例

给已有工作流添加持久化仅需两步：创建Checkpointer实例，编译图时注入。

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver

## 1. 节点逻辑完全不变，业务代码零侵入
def chat_node(state: MessagesState):
    user_last_msg = state["messages"][-1].content
    reply = f"已收到：{user_last_msg}"
    return {"messages": [{"role": "ai", "content": reply}]}

## 2. 构建图结构
builder = StateGraph(MessagesState)
builder.add_node("chat", chat_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)

## 3. 创建内存检查点器，编译时注入
memory_saver = MemorySaver()
graph = builder.compile(checkpointer=memory_saver)

## 4. 配置线程ID，区分不同会话
config = {"configurable": {"thread_id": "user_001"}}

## 第一轮对话
result_1 = graph.invoke(
    {"messages": [{"role": "user", "content": "你好，我是张三"}]},
    config=config
)

## 第二轮对话：无需手动拼接历史，自动加载存档
result_2 = graph.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字？"}]},
    config=config
)

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
28
29
30
31
32
33

关键说明：每次调用必须传入包含thread_id的配置，Checkpointer据此读写对应会话的存档；新输入的状态会通过对应Reducer与历史状态自动合并，消息字段默认追加，不会覆盖历史。

# 4.2 持久化执行：断点续跑的容错能力

持久化侧重「保存状态」，而持久化执行(Durable Execution) 侧重「基于已保存的状态可靠续跑」：即便工作流执行中遭遇系统崩溃、网络中断、服务重启，仍能从最后一个成功的检查点继续执行，无需从头重跑。

# 4.2.1 核心原则：确定性与幂等性

工作流恢复时，并不是从代码中断的那一行继续执行，而是从最近的检查点开始重跑节点。因此编写持久化工作流必须遵守两个原则：

封装非确定性操作：随机数生成、获取当前时间等操作，必须封装在节点/任务中。执行结果会随检查点一起保存，恢复时直接复用结果，不会产生不一致。
副作用操作幂等：API调用、文件写入、数据库操作等有副作用的逻辑，要设计成幂等的。避免恢复时重复执行导致重复发邮件、重复插入数据等问题。

# 4.2.2 三种持久性模式

LangGraph提供三种持久化等级，性能和可靠性呈反向关系，可根据业务场景选择：

exit 模式 仅在工作流完全结束(成功或失败)时才保存状态。性能最好，但中间状态不保存，无法中途中断或故障恢复，适合短平快的一次性任务。
async 模式(默认) 执行下一步的同时异步持久化当前状态。性能与可靠性的平衡之选，绝大多数场景推荐使用；极端情况下进程崩溃可能丢失最近一个检查点。
sync 模式 下一步执行前，同步完成当前状态的持久化。可靠性最高，但性能开销最大，适合金融、审批等高可靠要求的场景。

使用方式：

graph.stream(input_data, durability="sync")

# 4.2.3 恢复执行

两种常见恢复场景：

故障自动恢复：使用相同的thread_id，传入None作为输入重新调用，工作流会自动从最后一个成功的检查点继续执行。
人工中断恢复：通过Command(resume=...)传入人工输入，从中断点继续(详见4.4节)。

# 4.3 流式处理：提升交互体验的关键

大语言模型生成响应存在明显延迟，如果等完整结果生成再返回，用户体验很差。LangGraph内置了完善的流式传输体系，支持多种维度的流式输出。

# 4.3.1 五种流模式

模式	输出内容	适用场景
`values`	每一步执行后的完整状态	需要实时跟踪全量状态的场景
`updates`	每一步的状态增量更新	只需关注变化部分，减少传输量
`messages`	LLM生成的逐Token内容 + 元数据	对话类应用，实现打字机效果
`custom`	节点/工具内自定义发送的数据	进度提示、中间日志、业务事件
`debug`	全量调试信息(节点名、完整状态等)	开发调试阶段排障

# 4.3.2 基础用法

可以单模式使用，也可以多模式组合：

## 单模式：增量更新
for chunk in graph.stream(input_state, stream_mode="updates"):
    print(chunk)

## 多模式组合：同时接收状态更新和自定义数据
for mode, chunk in graph.stream(input_state, stream_mode=["updates", "custom"]):
    print(f"[{mode}]: {chunk}")

1
2
3
4
5
6
7

# 4.3.3 典型场景

LLM Token流式输出 使用messages模式，可以捕获工作流中任意节点、工具、子图里的LLM调用，逐Token返回，实现打字机效果：

for chunk, meta in graph.stream({"query": "写一篇短文"}, stream_mode="messages"):
    print(chunk.content, end="")

1
2

自定义进度流 在节点内部通过get_stream_writer()发送自定义数据，比如处理进度：

from langgraph.config import get_stream_writer

def process_node(state):
    writer = get_stream_writer()
    writer({"progress": "开始处理", "percent": 0})
    ## 执行步骤1
    writer({"progress": "步骤1完成", "percent": 30})
    ## 执行步骤2
    writer({"progress": "处理完成", "percent": 100})
    return {"result": "done"}

1
2
3
4
5
6
7
8
9
10

子图流式输出 调用.stream()时设置subgraphs=True，可以同时输出父图和所有嵌套子图的流式数据，输出中会携带命名空间区分层级。

# 4.4 中断机制：实现人机协作工作流

很多高风险场景不能让Agent完全自主执行：比如财务转账、内容发布、敏感操作，需要人工审核确认后再继续。LangGraph的中断(Interrupts) 机制就是为此设计：可以在工作流任意位置暂停，等待外部输入后再继续执行。

# 4.4.1 基础用法

在节点内调用interrupt()即可触发暂停，返回值就是后续恢复时传入的人工输入；恢复时使用Command(resume=...)传入数据。

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

class ApprovalState(TypedDict):
    amount: float
    status: str

def approval_node(state: ApprovalState):
    print(f"待审批金额：{state['amount']}")
    ## 中断执行，等待人工审批
    decision = interrupt({"action": "审批转账", "amount": state["amount"]})
    ## 恢复后decision就是人工传入的值
    return {"status": "已通过" if decision else "已拒绝"}

## 编译图，必须配合Checkpointer使用
builder = StateGraph(ApprovalState)
builder.add_node("approval", approval_node)
builder.add_edge(START, "approval")
builder.add_edge("approval", END)
graph = builder.compile(checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "approval_001"}}

## 1. 启动工作流，执行到interrupt处暂停
result = graph.invoke({"amount": 500, "status": "待审批"}, config)
print(result["__interrupt__"])  ## 查看中断信息

## 2. 人工审批通过，恢复执行
final = graph.invoke(Command(resume=True), config)
print(final["status"])  ## 已通过

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
28
29
30
31
32

# 4.4.2 常见应用场景

审批工作流：高风险操作前暂停，等待管理员批准
内容审阅编辑：LLM生成初稿后，人工修改润色再继续后续流程
工具调用校验：执行工具调用前暂停，让人确认参数无误再执行
表单输入验证：循环中断收集用户输入，校验不通过则重新询问

# 4.4.3 使用规则与避坑

中断是通过抛出异常实现的，使用时有几个必须遵守的规则：

不要把interrupt()包裹在try/except中，否则会捕获中断异常导致机制失效
中断之前的代码必须是幂等的：因为恢复时节点会从头重新执行，interrupt前的代码会再跑一遍
中断返回的值必须是JSON可序列化的，不能传递复杂对象
恢复时必须使用与中断发生时相同的thread_id

# 4.5 时间旅行：回溯历史节点重跑

调试Agent、优化工作流时，我们经常需要回到某个历史节点，修改状态后重新执行后续流程，对比不同分支的结果。LangGraph的时间旅行(Time Travel) 能力可以完美支持这个需求。

# 4.5.1 核心操作

查询历史检查点 通过get_state_history()获取指定线程的所有历史快照，按时间倒序排列：

history = list(graph.get_state_history(config))
print(f"共有{len(history)}个历史检查点")

1
2

从指定检查点恢复 选中某个历史检查点的配置，传入None作为输入，即可从该点继续执行：

## 取第2个历史检查点
checkpoint = history[1]
result = graph.invoke(None, checkpoint.config)

1
2
3

修改状态后重跑 可以先用update_state()修改检查点的状态，再恢复执行，实现修正后重跑：

## 修改历史检查点的状态
new_config = graph.update_state(
    checkpoint.config,
    values={"character": "会飞的龙"}  ## 修改角色设定
)
## 从修改后的检查点继续执行
result = graph.invoke(None, new_config)

1
2
3
4
5
6
7

注意：时间旅行不会覆盖原始历史记录，而是生成新的检查点分支，完整保留所有执行轨迹，方便对比。

# 4.5.2 适用场景

调试排障：回溯到出错前的节点，稳定复现问题
方案对比：同一前置状态下，测试不同参数/分支的效果
错误修正：人工修正错误的中间状态，重新跑后续流程

# 4.6 记忆管理：长短期记忆双体系

智能体的记忆分为两个层级：短期记忆负责单轮会话内的上下文延续，长期记忆负责跨会话的知识沉淀与用户画像存储。

# 4.6.1 短期记忆与上下文管理

短期记忆本质就是基于Checkpointer的会话状态持久化，同一个thread_id下的所有交互共享状态。但随着对话变长，消息会超出LLM的上下文窗口，需要做管控：

修剪消息：保留最近N条消息，丢弃最早的部分，适合对历史依赖不强的对话
删除消息：使用RemoveMessage精准删除指定消息，配合add_messages Reducer使用
总结消息：用LLM总结较早的对话历史，用摘要替代原始消息，兼顾信息留存和上下文长度

# 4.6.2 长期记忆(Store)

长期记忆跨线程生效，用于存储用户偏好、业务知识库、历史画像等跨会话数据，支持命名空间隔离和语义检索。

存储介质：支持内存、SQLite、PostgreSQL等多种后端
语义检索：配置Embedding模型后，可以通过向量相似度检索相关记忆
生存时间：支持TTL自动过期

基础使用流程：

编译图时注入store实例
节点函数通过参数接收store对象
使用store.put()写入记忆，store.get()精确读取，store.search()语义检索

# 4.7 子图：模块化复用与多智能体编排

当工作流越来越复杂，把所有节点写在一张图里会导致代码臃肿、难以维护。子图(Subgraphs) 可以把一组节点封装成独立模块，既可以复用代码，也方便多团队并行开发，更是多智能体系统的基础构建方式。

# 4.7.1 两种使用方式

节点内调用子图(状态隔离) 在父图的节点函数内部手动调用子图，父子图状态完全隔离，需要自己做状态转换。适合子图和父图业务独立、不需要共享状态的场景。
子图直接作为节点(状态共享) 直接把编译好的子图当作节点添加到父图中，通过共享的状态键通信。适合多智能体协作、共享消息历史的场景，代码最简洁。

## 构建子图
sub_builder = StateGraph(SharedState)
sub_builder.add_node("sub_node", sub_node)
subgraph = sub_builder.compile()

## 父图直接添加子图作为节点
parent_builder = StateGraph(ParentState)
parent_builder.add_node("agent_a", subgraph)

1
2
3
4
5
6
7
8

# 4.7.2 子图的价值

代码复用：通用流程封装成子图，多处业务直接引用
分治开发：不同团队负责不同子图，约定状态接口即可并行开发
多智能体：每个智能体封装为一个子图，通过父图编排协作流程

# 第五章更优雅的开发范式：Functional 函数式 API

前面我们用 Graph API 显式定义节点、边、状态，一步步搭建出了完整的 Agent 工作流。但对于很多业务场景，开发者更习惯用原生 Python 的分支、循环、函数调用来写逻辑——有没有办法不用强行套「图」的范式，也能用上 LangGraph 的持久化、记忆、流式等人机协同能力？

答案就是 Functional API(函数式 API)：它只需要两个装饰器，就能让几乎普通的 Python 函数获得完整的状态管理、持久化、中断恢复能力，写法更贴近日常开发习惯，学习成本更低，代码更简洁。

# 5.1 函数式 API 概述

# 5.1.1 核心定位

函数式 API 是 LangGraph 提供的第二种开发范式，主打「低侵入、轻量化」：你不需要学习节点、边、Reducer 这些图概念，用 if/for/while 等原生 Python 语法就能定义工作流，同时完整复用底层的 Pregel 运行时，和 Graph API 能力对等。

它特别适合快速原型开发、线性/简单分支类工作流、脚本类任务，以及不想重构现有代码、只想快速接入持久化能力的场景。

# 5.1.2 与 Graph API 的对比

两者底层共享同一套运行时，能力互通，但开发思路差异明显：

对比维度	Functional API	Graph API
控制流写法	原生 Python 语法，无需关心图结构	必须显式定义节点、边，按图范式编排
状态管理	无需声明 State 和 Reducer，状态随逻辑自然传递	需要定义 State Schema，配置 Reducer 管理更新
检查点粒度	按任务(Task)保存结果	每个超步(Superstep)生成检查点
可视化	运行时动态生成图，不支持静态可视化	可直接生成 Mermaid/ASCII 图，便于调试分享
适用场景	快速开发、线性流、简单分支、脚本化任务	复杂多智能体、强状态依赖、需要精细编排的生产级系统

两者并非互斥关系，可以在同一个项目中混合使用：复杂核心流用 Graph API 精细编排，外层调度和简单逻辑用函数式 API 快速实现。

# 5.2 核心构建块：两个装饰器搞定一切

函数式 API 的核心只有两个装饰器：@entrypoint(工作流入口)和 @task(任务单元)，掌握这两个就能完成 90% 的开发工作。

# 5.2.1 @entrypoint：定义工作流入口

用 @entrypoint 装饰的函数就是一个独立的工作流，装饰后会自动转换为 Pregel 实例，支持 invoke/stream/ainvoke 等所有执行方式，也能对接检查点、存储实现持久化和记忆。

基础用法

from langgraph.func import entrypoint
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def my_workflow(input_data: str) -> str:
    ## 这里写正常的 Python 业务逻辑
    processed = input_data.upper() + " [已处理]"
    return processed

## 执行方式和 Graph API 完全一致
result = my_workflow.invoke("hello world")
print(result)  ## HELLO WORLD [已处理]

1
2
3
4
5
6
7
8
9
10
11
12
13
14

可自动注入的运行时参数 在 entrypoint 函数中，你可以通过参数注解自动注入运行时上下文，无需手动传递：

previous：同线程上一次调用的保存状态，用于实现短期记忆
store：长期记忆存储对象(BaseStore)
writer：流写入器(StreamWriter)，用于发送自定义流式数据
config：运行时配置(RunnableConfig)
runtime：完整运行时上下文

示例：带短期记忆的累加器

from typing import Any
from langgraph.func import entrypoint
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def accumulate(n: int, *, previous: Any = None) -> int:
    ## 首次调用 previous 为 None，默认取 0
    previous = previous or 0
    return previous + n

config = {"configurable": {"thread_id": "acc_001"}}
print(accumulate.invoke(2, config=config))  ## 2
print(accumulate.invoke(3, config=config))  ## 5

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

entrypoint.final：解耦返回值与保存值 默认情况下，previous 拿到的是上一次函数的返回值。如果你希望「给调用方返回 A，但给下一次调用保存 B」，可以使用 entrypoint.final：

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
def calc(n: int, *, previous: int = 0) -> entrypoint.final[int, int]:
    ## value：返回给调用方的值；save：存到检查点、下次 previous 拿到的值
    return entrypoint.final(value=previous, save=previous + n)

1
2
3
4
5
6

# 5.2.2 @task：定义可持久化的任务单元

@task 用来装饰一个独立的计算单元，比如 API 调用、数据库查询、耗时计算。它有两个核心价值：

自动持久化：任务执行结果会存入检查点，工作流中断恢复后不会重复执行
天然支持并发：同时启动多个 task 再统一等待结果，即可实现并行执行

注意：@task 只能在 @entrypoint 内部、其他 task 内部调用，不能直接在顶层代码执行。调用后立即返回 future 对象，用 .result() 同步等待结果，异步场景用 await。

基础示例

from langgraph.func import entrypoint, task
import time

@task
def slow_double(x: int) -> int:
    time.sleep(1)  ## 模拟耗时计算/API调用
    return x * 2

@entrypoint()
def workflow(x: int) -> int:
    ## 调用 task，立即返回 future，不阻塞
    future = slow_double(x)
    ## 等待结果返回
    result = future.result()
    return result

print(workflow.invoke(5))  ## 10

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

典型适用场景

长耗时操作：保存结果到检查点，避免恢复时重复计算
随机性/副作用操作：封装后保证工作流恢复的确定性
IO 密集型任务：支持并发执行，提升吞吐量
需要重试、缓存能力的外部调用
需要 LangSmith 可观测性追踪的操作

# 5.3 三大设计原则：写对工作流的前提

函数式 API 写起来简单，但要让持久化、中断恢复正常工作，必须遵守三个核心原则，否则会出现恢复后结果不一致、副作用重复执行等问题。

# 5.3.1 序列化原则

入口点的输入、输出，以及所有 @task 的返回值，必须是 JSON 可序列化的(字典、列表、字符串、数字、布尔值等)。

原因：检查点需要将状态序列化后持久化存储，不可序列化的对象(文件句柄、自定义类实例、连接对象等)会导致持久化失败。
最佳实践：尽量用 Python 原生数据类型传递数据，复杂对象提前转换为字典结构。

# 5.3.2 确定性原则

非确定性操作(随机数生成、获取当前时间、随机采样等)必须封装在 @task 内部。

原因：工作流中断恢复后，会重跑入口点函数到中断位置。如果随机逻辑直接写在 entrypoint 里，恢复时会重新生成值，导致前后结果不一致；封装在 task 中后，结果会被持久化，恢复时直接复用旧结果。

错误写法 ❌

@entrypoint(checkpointer=checkpointer)
def bad_workflow():
    import random
    val = random.randint(1, 100)  ## 恢复时会重新生成，结果不确定
    interrupt("等待审核")
    return val

1
2
3
4
5
6

正确写法 ✅

@task
def get_random() -> int:
    import random
    return random.randint(1, 100)

@entrypoint(checkpointer=checkpointer)
def good_workflow():
    val = get_random().result()  ## 结果被持久化，恢复时复用
    interrupt("等待审核")
    return val

1
2
3
4
5
6
7
8
9
10

# 5.3.3 幂等性原则

有副作用的操作(发邮件、写数据库、调用外部扣费接口)必须封装在 @task 中，且尽量设计为幂等操作。

原因：如果任务执行到一半失败，恢复时会重新执行该任务。如果操作不幂等，就会出现重复发邮件、重复插入数据、重复扣费的问题。
最佳实践：给外部调用增加幂等键，执行前先校验是否已经执行过。

# 5.4 并行执行：轻松提升 IO 密集场景性能

函数式 API 实现并发非常自然：同时启动多个 task，再依次调用 .result() 等待结果，它们就会并行执行。特别适合同时调用多个 LLM API、查询多个外部接口的场景，不用手动写协程池。

示例：并行生成多主题内容

from typing import List
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver
import time

@task
def generate_paragraph(topic: str) -> str:
    time.sleep(1)  ## 模拟调用大模型
    return f"《{topic}》：这是一段关于{topic}的生成内容..."

@entrypoint(checkpointer=InMemorySaver())
def parallel_workflow(topics: List[str]) -> str:
    ## 同时启动所有任务，不阻塞
    futures = [generate_paragraph(topic) for topic in topics]
    ## 等待所有任务完成并收集结果
    paragraphs = [f.result() for f in futures]
    return "\n\n".join(paragraphs)

## 3个任务并行，总耗时约1秒，而非串行的3秒
result = parallel_workflow.invoke(["大模型", "向量数据库", "Agent"])

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 5.5 混合编程：在函数式 API 中调用 Graph

函数式与 Graph API 底层运行时互通，你可以把已经用 StateGraph 写好的复杂工作流，直接放到函数式入口里调用，实现「复杂核心用 Graph 精细编排，外层调度用函数式快速开发」的组合。

示例：

from langgraph.graph import StateGraph, START, END
from langgraph.func import entrypoint
from typing_extensions import TypedDict

## 1. 先用 Graph API 定义核心计算流
class CalcState(TypedDict):
    value: int

def double(state: CalcState):
    return {"value": state["value"] * 2}

graph_builder = StateGraph(CalcState)
graph_builder.add_node("double", double)
graph_builder.add_edge(START, "double")
graph_builder.add_edge("double", END)
calc_graph = graph_builder.compile()

## 2. 在函数式入口里直接调用
@entrypoint()
def wrapper_workflow(x: int) -> int:
    graph_result = calc_graph.invoke({"value": x})
    return graph_result["value"] + 10

print(wrapper_workflow.invoke(5))  ## 5*2 + 10 = 20

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

# 5.6 嵌套工作流：调用其他入口点

你可以在一个 entrypoint 内部调用另一个 entrypoint，把通用逻辑封装成可复用的子工作流，实现模块化开发，甚至支持递归调用。

示例：复用子工作流

from langgraph.func import entrypoint

## 子工作流：通用乘法计算
@entrypoint()
def multiply(inputs: dict) -> int:
    return inputs["a"] * inputs["b"]

## 主工作流：调用子工作流完成复杂计算
@entrypoint()
def main_workflow(x: int, y: int) -> int:
    product = multiply.invoke({"a": x, "b": y})
    return product + 10

print(main_workflow.invoke(3, 4))  ## 3*4 + 10 = 22

1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 5.7 流式传输

函数式 API 完全复用 Graph API 的流式体系，支持 values、updates、custom、messages 等所有流模式。

发送自定义流数据有两种方式：

通过参数注解注入 writer: StreamWriter
用 get_stream_writer() 函数在代码任意位置获取

示例：发送进度流

from langgraph.func import entrypoint
from langgraph.types import StreamWriter
from langgraph.checkpoint.memory import InMemorySaver

@entrypoint(checkpointer=InMemorySaver())
def progress_workflow(x: int, writer: StreamWriter) -> int:
    writer("任务开始执行")
    writer(f"正在处理数据: {x}")
    result = x * 2
    writer(f"中间计算结果: {result}")
    writer("任务执行完成")
    return result

## 流式读取输出
for mode, chunk in progress_workflow.stream(
    10, 
    stream_mode=["custom", "values"]
):
    print(f"[{mode}]: {chunk}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# 5.8 重试策略：自动处理临时故障

对于 API 调用、数据库查询这类容易出现临时网络故障的任务，可以给 @task 配置重试策略，失败后自动按规则重试，大幅提升工作流的鲁棒性。

通过 RetryPolicy 配置重试规则，支持指定重试异常类型，也支持自定义重试条件函数。

示例：

from langgraph.func import task
from langgraph.types import RetryPolicy

## 只对连接错误和超时错误重试，最多重试3次
api_retry = RetryPolicy(
    retry_on=(ConnectionError, TimeoutError),
    max_attempts=3
)

@task(retry_policy=api_retry)
def call_external_api(data: str) -> dict:
    ## 模拟可能失败的外部接口调用
    import random
    if random.random() < 0.7:
        raise ConnectionError("网络临时波动")
    return {"status": "success", "data": data}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

自定义重试条件：

def should_retry(e: Exception) -> bool:
    ## 只有包含 "temporary" 的错误才重试
    return "temporary" in str(e).lower()

custom_retry = RetryPolicy(retry_on=should_retry)

1
2
3
4
5

# 5.9 任务缓存：重复输入直接复用结果

对于计算量大、输入重复率高的任务，可以开启缓存，相同输入的任务第二次调用直接返回缓存结果，大幅提升性能。

配置方式：

给 @task 加 cache_policy 指定缓存有效期(TTL)
给 @entrypoint 传入缓存存储(如 InMemoryCache)

示例：

from langgraph.func import entrypoint, task
from langgraph.types import CachePolicy
from langgraph.cache.memory import InMemoryCache
import time

## 缓存有效期 120 秒
@task(cache_policy=CachePolicy(ttl=120))
def heavy_compute(x: int) -> int:
    time.sleep(2)  ## 模拟耗时计算
    return x ** 2

@entrypoint(cache=InMemoryCache())
def cache_workflow(x: int) -> int:
    ## 两次调用相同输入，第二次直接走缓存
    r1 = heavy_compute(x).result()
    r2 = heavy_compute(x).result()
    return r1 + r2

## 总耗时约 2 秒，而非串行的 4 秒
print(cache_workflow.invoke(10))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 5.10 人机协同：中断与恢复

函数式 API 的人机协同逻辑和 Graph API 完全一致：用 interrupt() 暂停工作流并返回数据给外部，再用 Command(resume=...) 恢复执行。因为是函数式写法，中断可以放在代码的任意位置，比 Graph API 更灵活自然。

示例：内容审核工作流

from langgraph.func import entrypoint, task
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

@task
def generate_article(topic: str) -> str:
    return f"关于{topic}的完整文章正文..."

@entrypoint(checkpointer=InMemorySaver())
def review_workflow(topic: str) -> dict:
    article = generate_article(topic).result()
    ## 中断执行，等待人工审核
    is_approved = interrupt({
        "article": article,
        "action": "请审核文章，返回 True 表示通过，False 表示驳回"
    })
    return {
        "article": article,
        "approved": is_approved
    }

## 1. 启动工作流，触发中断
config = {"configurable": {"thread_id": "review_001"}}
result = review_workflow.invoke("大模型应用开发", config=config)
print(result["__interrupt__"])  ## 拿到中断的文章内容

## 2. 人工审核后，恢复执行
final_result = review_workflow.invoke(Command(resume=True), config=config)
print(final_result["approved"])  ## True

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
28
29

# 5.11 记忆能力

函数式 API 同样支持短期记忆和长期记忆，用法比 Graph API 更简洁。

# 5.11.1 短期记忆

通过 previous 参数获取同线程上一次调用的保存值，天然实现多轮对话、状态累加等能力，本质就是利用检查点按 thread_id 隔离状态，跨调用共享。5.2.1 节的累加器就是典型的短期记忆实现。

# 5.11.2 长期记忆

通过注入 store 参数访问长期存储，跨线程、跨会话保存用户信息，实现用户画像、偏好记忆等能力。它和 Graph API 的 Store 底层是同一套实现，支持内存、SQLite 等后端，也支持向量嵌入实现语义检索。

# 5.12 常见陷阱与避坑指南

函数式 API 写起来简单，但有两个非常容易踩的隐蔽坑，开发时必须规避。

# 5.12.1 副作用不要直接写在入口点

所有有副作用的操作(写文件、发邮件、插入数据库、调用扣费接口)，必须封装在 @task 里。如果直接写在 entrypoint 函数体中，工作流中断恢复时会重新执行入口点代码，导致副作用重复执行。

错误写法 ❌

@entrypoint(checkpointer=checkpointer)
def bad_workflow():
    ## 错误：写文件直接写在入口点，恢复时会重复写入
    with open("result.txt", "w", encoding="utf-8") as f:
        f.write("计算结果")
    interrupt("等待确认")

1
2
3
4
5
6

正确写法 ✅

@task
def write_result_file():
    with open("result.txt", "w", encoding="utf-8") as f:
        f.write("计算结果")
    return "写入完成"

@entrypoint(checkpointer=checkpointer)
def good_workflow():
    write_result_file().result()  ## 结果被持久化，恢复时不会重跑
    interrupt("等待确认")

1
2
3
4
5
6
7
8
9
10

# 5.12.2 非确定性逻辑不要裸露

随机数、当前时间、环境变量这类每次运行结果可能不同的逻辑，一定要封装在 @task 内部，否则恢复时会产生不一致的结果，导致工作流逻辑错乱，具体可参考 5.3.2 节的示例。

← 大模型技术之LangChain系统实战 · 从入门到RAG与Agent全栈通关大模型智能体实战 · Coze与Dify从入门到企业级部署 →

# 大模型技术之LangGraph，从原型到生产：LangGraph工程化落地全解

# 第一章 认识LangGraph：从线性链到图编排，Agent开发的新范式

# 1.1 从Agent开发的痛点说起

# 1.2 LangGraph到底是什么

# 1.3 LangGraph的核心生产级能力

# 1.4 一文分清：LangChain vs LangGraph

# 1.5 底层架构：Pregel图计算模型

# 1.6 哪些场景该用LangGraph

# 第二章 快速入门：从零搭建第一个LangGraph智能体

# 2.1 环境准备

# 2.1.1 版本与依赖要求

# 2.1.2 安装步骤

# 方式1：使用 Conda 管理环境(推荐)

# 方式2：使用原生 venv

# 2.1.3 验证安装

# 2.2 第一个最简工作流：单节点对话

# 2.2.1 完整代码示例

# 2.2.2 逐行拆解代码

# 2.3 LangGraph通用开发五步法

# 第三章 Graph API 核心详解：状态、节点与边，从零构建复杂工作流

# 3.1 State：工作流的全局共享内存

# 3.1.1 State Schema：定义数据结构

# 1. state_schema：全局状态空间

# 2. input_schema / output_schema：接口隔离

# 3. 私有状态：节点间的局部数据

# 3.1.2 Reducer：状态合并规则

# 1. 默认行为：覆盖更新

# 2. add_messages：消息列表专用

# 3. operator.add：追加与累加

# 4. operator.mul：数值相乘

# 5. 自定义Reducer

# Reducer核心价值总结

# 3.2 Nodes：计算逻辑的载体

# 3.2.1 START 与 END 特殊节点

# 3.2.2 节点缓存

# 3.2.3 节点重试策略

# 3.2.4 延迟节点执行

# 3.3 Edges：流程控制的核心

# 3.3.1 普通边：固定跳转

# 3.3.2 条件边：动态路由

# 3.3.3 入口点与出口点

# 3.3.4 循环与递归限制

# 实现循环

# 递归限制

# 3.4 Send：动态分支与Map-Reduce模式

# 3.5 Command：节点内一体化控制流

# 3.5.1 基本用法

# 3.5.2 Command vs 条件边

# 3.5.3 父图导航

# 3.5.4 实战：工具内部更新状态

# 3.6 Runtime Context：运行时依赖注入

# 3.7 工作流可视化

# 1. Mermaid 流程图

# 2. ASCII 文本图

# 3.8 异步编程：提升IO密集型性能

# 核心用法

# 第四章 高级特性：构建生产级Agent的核心能力

# 4.1 持久化：给工作流加上存档能力

# 4.1.1 为什么需要持久化

# 4.1.2 核心概念

# 4.1.3 内置Checkpointer选型

# 4.1.4 基础使用示例

# 4.2 持久化执行：断点续跑的容错能力

# 4.2.1 核心原则：确定性与幂等性

# 4.2.2 三种持久性模式

# 4.2.3 恢复执行

# 4.3 流式处理：提升交互体验的关键

# 4.3.1 五种流模式

# 4.3.2 基础用法

# 4.3.3 典型场景

# 4.4 中断机制：实现人机协作工作流

# 4.4.1 基础用法

# 4.4.2 常见应用场景

# 4.4.3 使用规则与避坑

# 4.5 时间旅行：回溯历史节点重跑

# 4.5.1 核心操作

# 4.5.2 适用场景

# 4.6 记忆管理：长短期记忆双体系

# 4.6.1 短期记忆与上下文管理

# 4.6.2 长期记忆(Store)

# 第一章认识LangGraph：从线性链到图编排，Agent开发的新范式

# 第二章快速入门：从零搭建第一个LangGraph智能体

# 第四章高级特性：构建生产级Agent的核心能力

# 第五章更优雅的开发范式：Functional 函数式 API