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

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

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

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

# 1.1 从Agent开发的痛点说起

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

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

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

# 1.2 LangGraph到底是什么

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

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

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

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

# 1.3 LangGraph的核心生产级能力

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

  1. 持久化执行(断点续跑) 每个执行步骤的状态都会自动保存为检查点。如果程序崩溃、服务重启,工作流可以从最后一个成功的步骤继续执行,无需从头开始,特别适合长时间运行的后台任务。

  2. 人机协作 可以在工作流的任意节点设置中断,暂停执行并将状态暴露给人工审核、修改,确认后再恢复运行。这是审批类工作流、人工增强型Agent的必备能力。

  3. 分层记忆管理 原生支持两种记忆:

    • 短期记忆:基于检查点的线程级对话历史,支撑多轮对话交互;
    • 长期记忆:跨会话的存储能力,可以保存用户偏好、知识库,支持向量语义检索。
  4. 全链路流式处理 不仅支持大模型输出的Token流式返回,还支持状态更新、自定义事件、子图内部数据的流式输出,能很好地优化前端用户体验。

  5. 生产级部署基础设施 底层基于并行图计算模型设计,支持水平扩展,同时提供了SQLite、PostgreSQL等多种持久化存储方案,满足从本地开发到线上生产的全阶段需求。

# 1.4 一文分清:LangChain vs LangGraph

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

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

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

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

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

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

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

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

在这个模型下,还有两个核心概念:

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

# 1.6 哪些场景该用LangGraph

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

  1. 复杂多智能体系统:多个Agent分工协作、需要调度流转的场景,比如研发Agent团队、客服多技能组调度。
  2. 需要长期记忆的应用:跨会话留存用户信息、个性化偏好的智能客服、个人助理类应用。
  3. 含人工审核的工作流:比如内容生成审核、财务审批、高危操作确认等需要人机协作的流程。
  4. 后台长任务+实时交互:比如文档批量处理、数据分析任务,支持后台运行、进度实时推送、中断后恢复。
  5. 高度定制化的智能体:需要精确控制执行流程、自定义状态逻辑的复杂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

正常输出版本号、作者等信息即代表安装完成。截至本书编写时,最新稳定版为 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开发的核心流程,我们逐部分拆解:

  1. 导入基础组件

    • StateGraph:LangGraph的核心类,用于构建状态图
    • MessagesState:官方内置的状态类型,自带消息列表追加的合并逻辑,对话类场景可以直接复用
    • START / END:两个特殊节点,分别代表图的入口和出口
  2. 定义节点函数 节点本质上就是普通的Python函数,它遵循两个规则:

    • 输入参数是图的当前状态(类型与图定义的State一致)
    • 返回值是一个字典,键对应状态中的字段,值为要更新的内容

    本例中mock_llm函数读取状态,返回一条AI消息,状态会自动合并这条新消息。

  3. 构建图结构

    • StateGraph(MessagesState):创建一个状态图实例,绑定状态类型
    • add_node("节点名", 节点函数):向图中添加计算节点,节点名是唯一标识
    • add_edge(起点, 终点):添加一条普通边,表示执行完起点节点后,无条件跳转到终点节点
  4. 编译图 compile() 方法会校验图结构的合法性,将定义好的图转换为可执行的运行时对象。后续所有执行、流式输出、持久化等能力,都基于编译后的对象调用。

  5. 调用执行 invoke(初始状态) 是最基础的同步执行方法,传入符合State格式的初始数据,工作流会从头到尾执行完毕,返回最终的完整状态。

# 2.3 LangGraph通用开发五步法

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

  1. 定义State(状态) 确定整个工作流需要共享的数据结构,这是图的“数据骨架”。所有节点都基于状态读写,状态的设计直接决定了工作流的灵活性。

  2. 实现Nodes(节点) 将业务逻辑封装为独立的节点函数,每个节点负责一项具体计算(大模型调用、工具执行、数据处理等)。节点之间通过状态传递数据,彼此解耦。

  3. 编排Edges(边) 通过普通边、条件边定义节点的执行顺序和跳转规则,搭建工作流的“控制流程”。简单的线性流程用普通边,复杂的分支、循环用条件边。

  4. 编译Graph(图) 调用compile()完成图的构建,同时可以在此处配置持久化存储、缓存、中断点等高级能力。

  5. 调用执行 传入初始状态,通过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核心价值总结
  1. 控制更新策略:不同字段用不同合并逻辑,覆盖、追加、聚合按需选择;
  2. 解决并行冲突:多节点同时更新同一字段时,按规则合并,避免数据丢失;
  3. 灵活扩展:自定义Reducer可以适配任意业务的状态合并需求。

# 3.2 Nodes:计算逻辑的载体

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

节点函数支持三个入参:

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

# 3.2.1 START 与 END 特殊节点

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

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

# 3.2.2 节点缓存

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

使用方式分为两步:

  1. 编译图时指定缓存存储(如内存缓存);
  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方法,包含三个参数:

  1. 源节点名;
  2. 路由函数:接收状态,返回下一个节点的名称;
  3. 路由映射(可选):将路由函数的返回值映射到实际节点名,解耦逻辑与节点命名。
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接收两个参数:

  1. 目标节点名称;
  2. 传递给该节点的独立状态。

示例:根据主题列表,批量生成笑话

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

执行流程:

  1. 先执行generate_subjects,生成主题列表;
  2. 条件边为每个主题创建一个Send,并行启动3个make_joke实例;
  3. 所有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可以自动补全。

使用方式:

  1. 用dataclass定义上下文结构;
  2. 创建图时指定context_schema;
  3. 节点通过runtime.context访问上下文;
  4. 调用时通过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
1
ascii_graph = graph.get_graph().draw_ascii()
print(ascii_graph)
1
2

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

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

# 核心用法

  1. 节点函数用async def定义;
  2. 执行图时用await graph.ainvoke()或await graph.astream();
  3. 多个异步任务可以用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 核心概念

  1. Checkpointer(检查点器) 持久化的核心管理器,负责状态的写入、读取、查询、删除。采用可插拔设计,编译图时注入即可,业务节点零修改就能获得持久化能力。

  2. Thread(会话线程) 会话隔离的最小单位,每个Thread对应一个独立的工作流实例,通过唯一的thread_id标识。不同用户、不同任务分配不同的thread_id,状态完全隔离。

  3. 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 核心原则:确定性与幂等性

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

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

# 4.2.2 三种持久性模式

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

  1. exit 模式 仅在工作流完全结束(成功或失败)时才保存状态。性能最好,但中间状态不保存,无法中途中断或故障恢复,适合短平快的一次性任务。

  2. async 模式(默认) 执行下一步的同时异步持久化当前状态。性能与可靠性的平衡之选,绝大多数场景推荐使用;极端情况下进程崩溃可能丢失最近一个检查点。

  3. sync 模式 下一步执行前,同步完成当前状态的持久化。可靠性最高,但性能开销最大,适合金融、审批等高可靠要求的场景。

使用方式:

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

# 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 典型场景

  1. LLM Token流式输出 使用messages模式,可以捕获工作流中任意节点、工具、子图里的LLM调用,逐Token返回,实现打字机效果:
for chunk, meta in graph.stream({"query": "写一篇短文"}, stream_mode="messages"):
    print(chunk.content, end="")
1
2
  1. 自定义进度流 在节点内部通过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
  1. 子图流式输出 调用.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 常见应用场景

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

# 4.4.3 使用规则与避坑

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

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

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

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

# 4.5.1 核心操作

  1. 查询历史检查点 通过get_state_history()获取指定线程的所有历史快照,按时间倒序排列:
history = list(graph.get_state_history(config))
print(f"共有{len(history)}个历史检查点")
1
2
  1. 从指定检查点恢复 选中某个历史检查点的配置,传入None作为输入,即可从该点继续执行:
## 取第2个历史检查点
checkpoint = history[1]
result = graph.invoke(None, checkpoint.config)
1
2
3
  1. 修改状态后重跑 可以先用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自动过期

基础使用流程:

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

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

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

# 4.7.1 两种使用方式

  1. 节点内调用子图(状态隔离) 在父图的节点函数内部手动调用子图,父子图状态完全隔离,需要自己做状态转换。适合子图和父图业务独立、不需要共享状态的场景。

  2. 子图直接作为节点(状态共享) 直接把编译好的子图当作节点添加到父图中,通过共享的状态键通信。适合多智能体协作、共享消息历史的场景,代码最简洁。

## 构建子图
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 调用、数据库查询、耗时计算。它有两个核心价值:

  1. 自动持久化:任务执行结果会存入检查点,工作流中断恢复后不会重复执行
  2. 天然支持并发:同时启动多个 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 等所有流模式。

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

  1. 通过参数注解注入 writer: StreamWriter
  2. 用 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 任务缓存:重复输入直接复用结果

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

配置方式:

  1. 给 @task 加 cache_policy 指定缓存有效期(TTL)
  2. 给 @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 节的示例。

最近更新: 2026/9/7
 |  Copyright © -Present Kelsen All Rights Reserved. | Author KelsenJing
大模型技术之LangGraph,从原型到生产:LangGraph工程化落地全解