LangGraph 实战：如何用图结构解决大模型应用的“状态管理”难题？
LangGraph 实战：如何用图结构解决大模型应用的“状态管理”难题？
阅读建议： 本文约 11200 字，适合已使用过大语言模型（LLM）、LangChain 构建过简单 RAG/Agent 但遇到状态混乱、流程不可控、断点续传难等问题的开发者。我们将从痛点背景出发，剖析传统方法的缺陷，再引入 LangGraph 的核心概念，最后通过 3个完整的工业级场景代码实战（多轮对话检索增强 RAG、带工具调用的反思式 Agent、动态分支的多 Agent 协作系统），带你真正掌握 LangGraph 的核心逻辑与最佳实践。
引言
痛点引入：你是不是也在大模型应用里踩过这些“状态的坑”？
想象一下这个场景：你花了一周时间，用 LangChain 0.1.x 版本的 ConversationChain 搭配 ChromaDB 做了一个“法律合同多轮问答助手”——
- 第一轮对话还挺顺： 用户问“这份合同里的违约条款有哪些？”，系统检索出相关文本，LLM 准确提炼。
- 第二轮开始出问题： 用户追问“那如果甲方逾期付款30天，违约金怎么算？”——糟糕！ChromaDB 检索到的是另一份完全无关的房屋租赁合同的违约金，而不是之前那份供应链合同的！哦不对，不对，我明明把检索的 source 加在了 prompt 里？再看 prompt 历史，哦我的天，原来之前的多轮对话里用户提过“帮我找一份房屋租赁合同模板”，系统把那堆模板也扔进了上下文（
ConversationBufferMemory存的太多了？），导致第二轮检索虽然源是对的，但 LLM 在长上下文中混淆了“用户这次提问的主体”和“之前提到的模板”。 - 第三轮彻底崩溃： 你赶紧想加个“当前处理文档ID”的变量来控制检索，但是 ConversationChain 没有地方存自定义变量啊？哦对了，LangChain 有
ConversationSummaryMemory或者ConversationBufferWindowMemory，但自定义变量怎么办？那换成Agent吧，Agent 可以有工具调用，比如加一个“设置当前文档ID”和“获取当前文档ID”的工具，但工具调用的返回值只能塞回 LLM 的 prompt 里，状态不是全局共享的——如果 Agent 在调用工具A的时候更新了文档ID，但在调用工具B的时候工具B的代码拿不到这个新ID，只能再从 prompt 里解析（而且解析很容易出错！比如 LLM 返回的工具调用参数里文档ID写错了格式，比如写成123abc而不是 UUID）。 - 第四轮你心态崩了： 为了调试状态，你想在某个步骤（比如用户输入验证后）暂停流程，检查一下当前的所有变量（当前文档ID、检索到的 chunks、之前的关键信息摘要），但 LangChain 0.1.x 的链和 Agent 都是黑盒式的线性执行，一旦启动就只能等到结果或者报错，根本没法断点调试！更别说，如果用户在第二轮对话的时候网络断了，用户刷新页面，之前的状态（除了你存到 Redis 里的 prompt 片段）全丢了——你要从 Redis 里把所有 prompt 片段读出来，重新构建 ConversationChain/Agent，重新执行所有步骤？那如果之前的步骤涉及到调用外部 API（比如第三方支付验证、大规模数据库查询），重新执行成本太高了啊！
这还只是一个简单的多轮 RAG 场景，如果换成更复杂的多 Agent 协作（比如产品经理 Agent 写需求→设计师 Agent 生成 UI 稿原型描述→前端工程师 Agent 写代码→测试工程师 Agent 写测试用例→最后迭代），或者带反思的复杂推理任务（比如先让 LLM 生成一个数学题的解法，然后让另一个“反思者” Agent 检查解法的错误，然后修正，再检查，直到正确为止），状态管理的问题会指数级放大：
- 状态的一致性怎么保证？多个 Agent 同时更新同一个状态变量怎么办？
- 状态的持久化怎么实现？要持久化哪些状态？什么时候持久化？怎么恢复？
- 流程的可控性怎么提升？能不能根据当前的状态动态跳转分支？能不能在某个节点手动审核结果？
- 调试的便利性怎么增强？能不能可视化整个流程的执行路径和每个节点的状态变化？
解决方案概述：为什么是 LangGraph？
如果你之前被这些问题折磨过，那恭喜你——LangGraph 就是为解决这些问题而生的！
LangGraph 是 LangChain 团队在 2024 年初推出的、专门用于构建 有状态、可控、可调试、可持久化的复杂 LLM 应用（尤其是 Agent 和多 Agent 系统）的图结构框架。它的核心思想是：把 LLM 应用的逻辑拆分成一个个“节点（Node）”，把节点之间的跳转逻辑拆分成“边（Edge）”，把所有的状态集中到一个“全局状态字典（State）”里统一管理。
和传统的 LangChain 0.1.x 的线性 Chain/Agent 相比，LangGraph 有哪些革命性的优势呢？我们先列个核心对比表（后面会详细展开）：
| 维度 | LangChain 0.1.x 线性 Chain/Agent | LangGraph |
|---|---|---|
| 执行方式 | 严格线性（或有限的内置分支，比如 RouterChain） | 完全灵活的图结构（支持条件分支、循环、子图） |
| 状态管理 | 分散式（Chain 用 Memory、Agent 用 Prompt/临时变量） | 集中式全局状态字典（State Schema 定义类型，支持原子更新） |
| 断点调试 | 几乎不可能（黑盒式执行） | 原生支持（可以在任何节点暂停、检查、修改状态、手动恢复） |
| 状态持久化 | 需要自己实现（比如用 Redis 存 Memory） | 原生支持 Checkpointing（自动持久化状态、支持断点续传） |
| 多 Agent 协作 | 需要复杂的自定义 Chain/Agent 嵌套 | 原生支持（每个 Agent 是一个节点，边定义协作逻辑） |
| 动态分支/循环 | RouterChain 有限制，循环很难实现 | 条件边（Conditional Edge）轻松实现分支，循环边（Loop Back Edge）轻松实现循环 |
| 可视化执行路径 | 无原生支持 | 原生支持（可以画出图的结构，也可以画出某次执行的路径） |
最终效果展示（先剧透，后面会一步步实现）
在这篇文章里，我们会实现 3 个完整的工业级场景：
场景1：有状态的多轮对话检索增强 RAG（Stateful Multi-Turn RAG）
- 效果： 可以记住用户当前处理的文档 ID、当前的查询意图（比如“分析违约条款”还是“查看付款流程”）、之前检索到的关键 chunks，下次用户提问时自动过滤无关的文档、优化检索词。
- 可视化图结构：
场景2：带工具调用的反思式数学解题 Agent（Reflective Math Agent with Tools）
- 效果： 可以先调用计算器工具计算，然后让“反思者”LLM 检查解题步骤和结果，如果有错误就修正，再检查，直到正确为止（循环次数可配置）。
- 可视化图结构：
场景3：动态分支的多 Agent 协作产品需求评审系统（Dynamic Branch Multi-Agent Product Requirement Review System）
- 效果： 产品经理 Agent 先写需求，然后根据需求的类型（比如“新功能”还是“优化现有功能”）动态跳转分支：
- 如果是“新功能”：先跳转到市场分析 Agent，再跳转到技术可行性 Agent，再跳转到成本评估 Agent，最后跳转到产品总监 Agent 审核；
- 如果是“优化现有功能”：先跳转到用户反馈分析 Agent，再跳转到技术可行性 Agent，最后跳转到产品总监 Agent 审核；
- 可视化图结构（简化版，完整版后面会画）：
准备工作
环境/工具
在开始实战之前，我们需要准备好以下开发环境和工具：
| 工具/库名称 | 推荐版本 | 说明 |
|---|---|---|
| Python | 3.10 或 3.11 | LangGraph 的官方要求，避免版本兼容性问题 |
| LangChain | 0.2.x 或 0.3.x | LangGraph 是 LangChain 的子库，需要配套使用 |
| LangGraph | 0.2.x（最新稳定版） | 本文的核心库 |
| LangChain OpenAI | 0.2.x 或 0.3.x | 用于调用 OpenAI 的 GPT 模型和嵌入模型 |
| LangChain Chroma | 0.2.x 或 0.3.x | 用于场景1的向量数据库 |
| python-dotenv | 最新版 | 用于加载环境变量（比如 OpenAI API Key） |
| ipython | 最新版（可选） | 用于交互式调试，也可以用 VS Code 的 Jupyter Notebook |
安装步骤
-
创建虚拟环境（强烈推荐）：
# 用 conda 创建虚拟环境（推荐） conda create -n langgraph-demo python=3.11 -y conda activate langgraph-demo # 或者用 venv 创建虚拟环境 python -m venv langgraph-demo # Windows 激活虚拟环境 langgraph-demo\Scripts\activate # macOS/Linux 激活虚拟环境 source langgraph-demo/bin/activate -
安装依赖库：
pip install --upgrade langchain langchain-openai langgraph langchain-chroma python-dotenv ipython -
配置环境变量：
在项目根目录下创建一个.env文件，内容如下：# OpenAI API Key（必须配置，从 https://platform.openai.com/api-keys 获取） OPENAI_API_KEY=sk-your-openai-api-key-here # OpenAI API Base（可选，如果你用的是国内的镜像站） OPENAI_API_BASE=https://api.openai.com/v1 # OpenAI 模型名称（可选，默认是 gpt-4o-mini，便宜又好用） LLM_MODEL_NAME=gpt-4o-mini # OpenAI 嵌入模型名称（可选，默认是 text-embedding-3-small） EMBEDDING_MODEL_NAME=text-embedding-3-small
基础知识
在开始实战之前，你需要具备以下前置知识：
- 大语言模型（LLM）的基本原理： 知道什么是 prompt engineering、什么是工具调用（Function Calling/Tool Use）。
- LangChain 的基本用法： 知道什么是
ChatOpenAI、什么是Chroma向量数据库、什么是PromptTemplate。 - 图结构的基本概念： 知道什么是节点（Node）、边（Edge）、有向图（Directed Graph）。
如果你对这些知识不太熟悉，可以参考以下学习资源：
- LLM 基本原理与 Prompt Engineering： OpenAI 官方 Prompt Engineering 指南
- LangChain 基本用法： LangChain 0.2.x 官方文档
- 图结构的基本概念： 维基百科 - 图论
第一章：LangGraph 的核心概念与架构（打好基础，后面才能飞）
在开始实战之前，我们必须先搞清楚 LangGraph 的核心概念——只有把这些概念吃透了，后面写代码的时候才不会一头雾水，遇到问题也能快速定位和解决。
1.1 核心概念：从图的三要素到 LangGraph 的三要素
我们都知道，一个有向图由三个基本要素组成：节点（Vertex/Node）、边（Edge）、起点（Start Node）和终点（End Node）。LangGraph 也是基于这个基本原理构建的，但它为了适配 LLM 应用的场景，给这三个要素赋予了更具体的含义，还增加了一个最核心的要素：全局状态字典（State）。
我们先来看 LangGraph 的官方定义的核心概念列表（后面会逐一详细展开）：
| LangGraph 核心概念 | 说明 |
|---|---|
| State（状态） | 全局共享的字典/数据类，用于存储整个应用的所有状态（比如用户输入、LLM 输出、工具调用结果、自定义变量） |
| State Schema（状态模式） | 定义 State 的结构（字段名、字段类型、字段的更新方式）的类型提示或 TypedDict |
| Node（节点） | 图中的一个逻辑单元，接收当前的 State 作为输入，执行特定的逻辑（比如调用 LLM、调用工具、更新 State），然后返回更新后的 State（或者返回 State 的一部分） |
| Edge（边） | 连接两个节点的有向箭头，定义节点之间的跳转逻辑（比如“从节点A跳转到节点B”、“根据 State 的某个字段值跳转到不同的节点”） |
| Start Node（起点） | 图的入口节点，LangGraph 会自动为我们创建一个名为 START 的起点 |
| End Node（终点） | 图的出口节点，LangGraph 会自动为我们创建一个名为 END 的终点，也可以自定义多个终点 |
| Conditional Edge（条件边） | 根据 State 的某个字段值或某个自定义函数的返回值，动态跳转到不同节点的边 |
| Checkpoint（检查点） | 图在执行过程中某个时刻的 State 的快照，用于断点续传、调试和回滚 |
| Checkpointer（检查点管理器） | 用于管理 Checkpoint 的组件，LangGraph 内置了基于内存的 MemorySaver，也支持基于 Redis、SQLite 的 Checkpointer |
| Subgraph（子图） | 嵌入到另一个图中的图，用于模块化复杂的逻辑 |
| Compiled Graph（编译后的图） | 我们定义好节点和边之后，调用 compile() 方法编译出来的可执行的图对象 |
好，现在我们来逐一详细展开这些核心概念——这是本文最重要的部分之一，请务必认真阅读。
1.1.1 State（状态）：LangGraph 的“心脏”
如果说 LangChain 0.1.x 的 Chain/Agent 是“没有心脏的行尸走肉”（状态分散，难以管理），那么 State 就是 LangGraph 的“心脏”——所有的节点都围绕着 State 工作，所有的逻辑都依赖于 State，所有的结果都存储在 State 里。
什么是 State？
简单来说，State 就是一个全局共享的字典（或数据类），它贯穿整个图的执行过程，从起点到终点，所有的节点都可以读取它、修改它。
举个简单的例子，在我们的场景1（有状态的多轮对话 RAG）中，State 可能包含以下字段：
# 这是一个简化版的 State（用 Python 的字典表示）
state = {
"user_input": "那如果甲方逾期付款30天，违约金怎么算？", # 当前用户的输入
"current_doc_id": "123e4567-e89b-12d3-a456-426614174000", # 当前处理的文档ID
"query_intent": "供应链合同-违约条款-违约金计算", # 当前的查询意图
"retrieved_chunks": [{"page_content": "...", "metadata": {...}}], # 检索到的关键chunks
"llm_response": "根据合同第10.2条，甲方逾期付款30天的话，违约金是应付金额的0.5%/天...", # LLM的输出
"chat_history": [("用户之前的问题", "系统之前的回答")] # 整个对话的历史
}
State Schema（状态模式）：给 State 加上“紧箍咒”
在 Python 中，字典是动态的——你可以随时添加、删除、修改字段，这在简单的场景中很方便，但在复杂的 LLM 应用中，这可能会导致状态混乱（比如某个节点不小心把 current_doc_id 改成了 None，或者添加了一个拼写错误的字段 retreived_chunks）。
为了解决这个问题，LangGraph 要求我们定义一个 State Schema（状态模式）——也就是用 Python 的 TypedDict 或 Pydantic BaseModel 来定义 State 的结构（字段名、字段类型、字段的更新方式）。这样一来，LangGraph 就可以在编译和执行图的时候对 State 进行类型检查，避免状态混乱的问题。
在 LangGraph 中，最常用的 State Schema 定义方式是 TypedDict，因为它简单、轻量，而且和 LangGraph 的更新机制配合得很好。我们来看一下如何用 TypedDict 定义场景1的 State Schema：
from typing import TypedDict, List, Tuple, Annotated
from langchain_core.documents import Document
import operator
# 定义 State Schema
class State(TypedDict):
# 用户输入：每次用户输入都会覆盖这个字段
user_input: str
# 当前处理的文档ID：如果有的话就用，没有的话就是 None
current_doc_id: str | None
# 当前的查询意图：LLM 识别出来的，会覆盖之前的
query_intent: str | None
# 检索到的关键 chunks：每次检索都会追加到这个列表里（而不是覆盖）
# 注意：这里用了 Annotated 来指定字段的更新方式！！！这是 LangGraph 的一个核心特性
# operator.add 表示：如果新的 State 里有这个字段，就把新的值和旧的值相加（对于列表来说就是追加）
retrieved_chunks: Annotated[List[Document], operator.add]
# LLM 的输出：每次都会覆盖
llm_response: str | None
# 整个对话的历史：每次都会追加（用户输入是第一个元素，系统回答是第二个元素）
chat_history: Annotated[List[Tuple[str, str]], operator.add]
哦，对了！这里有一个 LangGraph 最核心、最容易被忽略但又最重要的特性：State 字段的更新方式（Reducer）。
在上面的代码中，我们用了 Annotated[List[Document], operator.add] 来定义 retrieved_chunks 字段——这里的 operator.add 就是这个字段的 Reducer（归约函数）。
State 字段的更新方式（Reducer）：怎么把新值合并到旧值里？
在 LangGraph 中，当一个节点执行完逻辑之后，它会返回一个更新字典（Update Dictionary）——这个字典里只包含需要更新的 State 字段，而不是整个 State。然后，LangGraph 会根据 State Schema 中定义的 Reducer（归约函数），把更新字典里的新值合并到旧的 State 里。
这一点和传统的编程方式完全不同——传统的编程方式是“修改全局变量”，而 LangGraph 是“返回更新字典，然后 LangGraph 自动根据 Reducer 合并”。这样做的好处是什么呢？
- 状态的一致性：所有的状态更新都是通过 LangGraph 统一管理的，避免了多个节点同时修改同一个变量导致的竞态条件（Race Condition）。
- 状态的可追溯性：每次状态更新都是一个增量更新，我们可以很容易地追溯到哪个节点在什么时候更新了哪个字段。
- 代码的简洁性：节点只需要返回需要更新的字段，不需要关心整个 State 的结构。
好，现在我们来详细讲一下 Reducer 的类型：
在 LangGraph 中，Reducer 可以分为以下几类：
1. 覆盖式 Reducer（默认）
如果我们在 State Schema 中没有用 Annotated 指定 Reducer，那么默认的 Reducer 就是覆盖式 Reducer——也就是说，如果更新字典里有这个字段，就用新值覆盖旧值；如果更新字典里没有这个字段，就保留旧值。
比如，在我们的 State Schema 中，user_input、current_doc_id、query_intent、llm_response 都是覆盖式 Reducer。
举个例子：
# 旧的 State
old_state = {
"user_input": "这份合同里的违约条款有哪些？",
"current_doc_id": "123e4567-e89b-12d3-a456-426614174000",
"query_intent": "供应链合同-违约条款-列举",
"llm_response": "根据合同第10条，违约条款包括...",
...
}
# 节点返回的更新字典（只包含需要更新的字段）
update_dict = {
"user_input": "那如果甲方逾期付款30天，违约金怎么算？", # 覆盖旧的 user_input
"query_intent": "供应链合同-违约条款-违约金计算", # 覆盖旧的 query_intent
"llm_response": None # 覆盖旧的 llm_response 为 None
}
# LangGraph 合并后的新 State
new_state = {
"user_input": "那如果甲方逾期付款30天，违约金怎么算？", # 被覆盖了
"current_doc_id": "123e4567-e89b-12d3-a456-426614174000", # 保留旧值（因为更新字典里没有）
"query_intent": "供应链合同-违约条款-违约金计算", # 被覆盖了
"llm_response": None, # 被覆盖了
... # 其他字段也保留旧值
}
2. 追加式 Reducer（operator.add 或自定义函数）
如果我们在 State Schema 中用 Annotated 指定了 Reducer 为 operator.add，那么对于列表类型的字段来说，就是追加式 Reducer——也就是说，如果更新字典里有这个字段，就把新值（列表）追加到旧值（列表）的后面；如果更新字典里没有这个字段，就保留旧值。
比如，在我们的 State Schema 中，retrieved_chunks 和 chat_history 都是追加式 Reducer（用了 operator.add）。
举个例子：
from langchain_core.documents import Document
import operator
# 旧的 State
old_state = {
"retrieved_chunks": [
Document(page_content="合同第10条：违约条款", metadata={"source": "供应链合同.pdf", "page": 10})
],
"chat_history": [
("这份合同里的违约条款有哪些？", "根据合同第10条，违约条款包括...")
],
...
}
# 节点返回的更新字典（只包含需要更新的字段）
new_chunk = Document(page_content="合同第10.2条：逾期付款违约金", metadata={"source": "供应链合同.pdf", "page": 10})
update_dict = {
"retrieved_chunks": [new_chunk], # 注意：这里是一个列表！因为 operator.add 是两个列表相加
"chat_history": [
("那如果甲方逾期付款30天，违约金怎么算？", None) # 注意：这里第二个元素是 None，后面 LLM 生成回答之后会更新
]
}
# LangGraph 合并后的新 State
new_state = {
"retrieved_chunks": [
Document(page_content="合同第10条：违约条款", metadata={"source": "供应链合同.pdf", "page": 10}),
Document(page_content="合同第10.2条：逾期付款违约金", metadata={"source": "供应链合同.pdf", "page": 10}) # 被追加了
],
"chat_history": [
("这份合同里的违约条款有哪些？", "根据合同第10条，违约条款包括..."),
("那如果甲方逾期付款30天，违约金怎么算？", None) # 被追加了
],
...
}
注意：在使用 operator.add 作为列表的 Reducer 时，更新字典里的字段值必须是一个列表——哪怕你只需要追加一个元素，也要把它放在列表里（比如 [new_chunk]）。
除了 operator.add，我们还可以自定义 Reducer 函数。比如，如果我们想把 retrieved_chunks 字段的最大长度限制在 5 个（避免 State 太大），我们可以自定义一个 Reducer 函数：
from typing import List
from langchain_core.documents import Document
# 自定义 Reducer 函数：限制列表的最大长度为 5
def limit_chunks(old_chunks: List[Document], new_chunks: List[Document]) -> List[Document]:
# 先合并旧的和新的
merged = old_chunks + new_chunks
# 然后只保留最后 5 个
return merged[-5:]
# 修改 State Schema，用自定义的 Reducer 函数
class State(TypedDict):
...
retrieved_chunks: Annotated[List[Document], limit_chunks]
...
自定义 Reducer 函数的规则是：
- 接收两个参数：第一个参数是旧的字段值，第二个参数是更新字典里的新字段值。
- 返回一个值：合并后的新字段值。
1.1.2 Node（节点）：LangGraph 的“手脚”
如果说 State 是 LangGraph 的“心脏”，那么 Node 就是 LangGraph 的“手脚”——它接收当前的 State 作为输入，执行特定的逻辑（比如调用 LLM、调用工具、验证用户输入、更新 State），然后返回更新字典（Update Dictionary）。
什么是 Node？
简单来说，Node 就是一个 Python 函数（或类的方法），它的签名必须符合 LangGraph 的要求。
LangGraph 要求 Node 函数的签名是什么样的呢？我们来看一下：
# Node 函数的标准签名
def node_function(state: State) -> dict:
# 1. 从 state 中读取需要的字段
user_input = state["user_input"]
current_doc_id = state["current_doc_id"]
...
# 2. 执行特定的逻辑（比如调用 LLM、调用工具、验证用户输入）
...
# 3. 返回更新字典（Update Dictionary）——只包含需要更新的字段
return {
"query_intent": "供应链合同-违约条款-违约金计算",
"retrieved_chunks": [new_chunk],
...
}
注意：
- Node 函数的第一个参数必须是 State（也就是我们定义的 State Schema 类型）。
- Node 函数的返回值必须是一个字典——也就是更新字典（Update Dictionary），只包含需要更新的 State 字段。
好，现在我们来看一下场景1中的“用户输入验证节点”的代码示例：
# 场景1：用户输入验证节点
def validate_user_input_node(state: State) -> dict:
"""
验证用户输入是否为空，如果为空就返回错误提示，否则就保留用户输入。
"""
# 1. 从 state 中读取用户输入
user_input = state["user_input"].strip()
# 2. 验证用户输入
if not user_input:
# 用户输入为空，返回错误提示（存储在 llm_response 字段里，后面会显示给用户）
return {
"llm_response": "请输入有效的问题或指令！",
"chat_history": [(user_input, "请输入有效的问题或指令！")] # 追加到对话历史
}
else:
# 用户输入有效，不需要更新任何字段（返回空字典即可）
return {}
哦，对了！如果 Node 函数不需要更新任何 State 字段，那么它可以返回空字典，或者返回 None——LangGraph 会忽略空字典和 None。
Node 的类型
在 LangGraph 中，Node 可以分为以下几类：
- 普通节点（Normal Node）：我们上面讲的就是普通节点——接收 State，返回更新字典。
- 工具节点（Tool Node）：专门用于调用工具的节点——LangGraph 内置了
ToolNode类，我们只需要把工具列表传给它即可，不需要自己写调用工具的逻辑。 - 条件节点（Conditional Node）：不执行任何逻辑，只用于根据 State 的某个字段值或某个自定义函数的返回值跳转到不同的节点——不过，在 LangGraph 中，我们通常用 Conditional Edge（条件边） 来实现条件跳转，而不是用条件节点（因为条件边更简洁、更灵活）。
- 子图节点（Subgraph Node）：把一个子图作为一个节点嵌入到另一个图中——用于模块化复杂的逻辑。
我们后面会在实战中详细讲 ToolNode 和子图节点。
1.1.3 Edge（边）：LangGraph 的“神经”
如果说 State 是 LangGraph 的“心脏”，Node 是 LangGraph 的“手脚”，那么 Edge 就是 LangGraph 的“神经”——它连接节点和节点，定义节点之间的跳转逻辑。
什么是 Edge？
简单来说，Edge 就是一个有向箭头，从一个节点（源节点）指向另一个节点（目标节点），表示“源节点执行完之后，跳转到目标节点”。
LangGraph 要求我们在定义图的时候，先添加所有的节点，然后再添加所有的边。
Edge 的类型
在 LangGraph 中，Edge 可以分为以下几类：
1. 普通边（Normal Edge）
普通边是最简单的边——它从一个固定的源节点指向一个固定的目标节点，表示“源节点执行完之后，无条件跳转到目标节点”。
我们用 graph.add_edge(source_node, target_node) 方法来添加普通边。
举个例子，在场景1中，“用户输入验证节点”执行完之后，无条件跳转到“当前文档ID检查节点”——这就是一条普通边：
# 先添加节点
graph.add_node("validate_user_input", validate_user_input_node)
graph.add_node("check_current_doc_id", check_current_doc_id_node)
# 再添加普通边：从 validate_user_input 跳转到 check_current_doc_id
graph.add_edge("validate_user_input", "check_current_doc_id")
2. 条件边（Conditional Edge）
条件边是 LangGraph 中最强大、最灵活的边——它从一个固定的源节点出发，根据 State 的某个字段值或某个自定义函数的返回值，动态跳转到不同的目标节点。
我们用 graph.add_conditional_edges(source_node, condition_function, edge_mapping) 方法来添加条件边。
其中：
source_node：源节点的名称（字符串）。condition_function：条件函数——它接收当前的 State 作为输入，返回一个字符串（目标节点的名称）。edge_mapping：边映射字典——可选参数，用于把条件函数的返回值映射成目标节点的名称（如果条件函数的返回值已经是目标节点的名称，那么这个参数可以省略）。
举个例子，在场景1中，“当前文档ID检查节点”执行完之后，我们需要根据 State 中的 current_doc_id 字段是否存在，跳转到不同的节点：
- 如果
current_doc_id存在（不是 None），就跳转到“查询意图识别与优化节点”； - 如果
current_doc_id不存在（是 None），就跳转到“请求用户提供文档ID节点”。
我们来看一下如何用条件边实现这个逻辑：
# 1. 先添加所有需要的节点
graph.add_node("check_current_doc_id", check_current_doc_id_node)
graph.add_node("identify_query_intent", identify_query_intent_node)
graph.add_node("ask_for_doc_id", ask_for_doc_id_node)
# 2. 定义条件函数
def check_current_doc_id_condition(state: State) -> str:
"""
条件函数：根据 current_doc_id 是否存在，返回目标节点的名称。
"""
current_doc_id = state["current_doc_id"]
if current_doc_id:
# current_doc_id 存在，跳转到 identify_query_intent 节点
return "identify_query_intent"
else:
# current_doc_id 不存在，跳转到 ask_for_doc_id 节点
return "ask_for_doc_id"
# 3. 添加条件边：从 check_current_doc_id 出发，根据条件函数的返回值跳转
# 注意：这里 edge_mapping 可以省略，因为条件函数的返回值已经是目标节点的名称
graph.add_conditional_edges(
source="check_current_doc_id",
path=check_current_doc_id_condition
)
再举一个稍微复杂一点的例子：在场景3（多 Agent 协作产品需求评审系统）中，“需求类型识别节点”执行完之后，我们需要根据 State 中的 requirement_type 字段的值，跳转到不同的节点：
- 如果
requirement_type是 “new_feature”，就跳转到 “market_analysis_agent” 节点； - 如果
requirement_type是 “optimize_existing_feature”，就跳转到 “user_feedback_analysis_agent” 节点； - 如果
requirement_type是其他值，就跳转到 “invalid_requirement_type” 节点。
我们来看一下如何用条件边和 edge_mapping 实现这个逻辑：
# 1. 先添加所有需要的节点
graph.add_node("identify_requirement_type", identify_requirement_type_node)
graph.add_node("market_analysis_agent", market_analysis_agent_node)
graph.add_node("user_feedback_analysis_agent", user_feedback_analysis_agent_node)
graph.add_node("invalid_requirement_type", invalid_requirement_type_node)
# 2. 定义条件函数（注意：这里返回的是 requirement_type 的值，而不是目标节点的名称）
def identify_requirement_type_condition(state: State) -> str:
"""
条件函数：返回 requirement_type 的值。
"""
return state["requirement_type"]
# 3. 定义 edge_mapping：把 requirement_type 的值映射成目标节点的名称
edge_mapping = {
"new_feature": "market_analysis_agent",
"optimize_existing_feature": "user_feedback_analysis_agent",
"other": "invalid_requirement_type"
}
# 4. 添加条件边：从 identify_requirement_type 出发，根据条件函数的返回值和 edge_mapping 跳转
graph.add_conditional_edges(
source="identify_requirement_type",
path=identify_requirement_type_condition,
path_map=edge_mapping
)
注意：如果条件函数的返回值不在 edge_mapping 的键中，那么 LangGraph 会抛出一个错误——所以，我们通常会在 edge_mapping 中添加一个 “other” 或 “default” 的键，用于处理未知的返回值。
3. 循环边（Loop Back Edge）
循环边其实就是一条普通边或条件边，从一个节点指向它自己，或者指向它之前的某个节点——用于实现循环逻辑（比如反思式 Agent 的“生成→检查→修正→生成”循环）。
举个例子，在场景2（带工具调用的反思式数学解题 Agent）中，“反思者 Agent 节点”执行完之后，我们需要根据 State 中的 is_correct 字段和 loop_count 字段的值，跳转到不同的节点：
- 如果
is_correct是 True，或者loop_count大于等于 5，就跳转到 END 节点； - 如果
is_correct是 False，并且loop_count小于 5，就跳转到“解题者 Agent 节点”（循环）。
我们来看一下如何用条件边实现这个循环逻辑：
# 1. 先添加所有需要的节点
graph.add_node("problem_solver_agent", problem_solver_agent_node)
graph.add_node("reflector_agent", reflector_agent_node)
# 2. 定义条件函数
def reflector_condition(state: State) -> str:
"""
条件函数：根据 is_correct 和 loop_count 的值，返回目标节点的名称。
"""
is_correct = state["is_correct"]
loop_count = state["loop_count"]
if is_correct or loop_count >= 5:
# 解题结果正确，或者循环次数超限，跳转到 END
return END
else:
# 解题结果错误，并且循环次数未超限，跳转到 problem_solver_agent（循环）
return "problem_solver_agent"
# 3. 添加条件边：从 reflector_agent 出发，根据条件函数的返回值跳转
graph.add_conditional_edges(
source="reflector_agent",
path=reflector_condition
)
1.1.4 Start Node（起点）和 End Node（终点）：LangGraph 的“入口”和“出口”
LangGraph 会自动为我们创建两个特殊的节点：
START节点：图的入口节点——所有的图都必须从START节点开始执行。END节点：图的出口节点——当图跳转到END节点时，执行就结束了。
我们需要做的是：
- 添加一条从
START节点到我们的第一个业务节点的普通边。 - 在适当的时机（比如业务逻辑执行完了），跳转到
END节点。
举个例子，在场景1中，我们的第一个业务节点是“用户输入验证节点”——所以我们需要添加一条从 START 到 validate_user_input 的普通边：
from langgraph.graph import StateGraph, START, END
# 创建图对象
graph = StateGraph(State)
# 添加业务节点
graph.add_node("validate_user_input", validate_user_input_node)
...
# 添加从 START 到第一个业务节点的普通边
graph.add_edge(START, "validate_user_input")
...
# 添加从最后一个业务节点到 END 的普通边
graph.add_edge("wait_for_next_input", END)
注意：START 和 END 是 LangGraph 内置的常量，我们不需要自己定义，直接导入使用即可。
1.1.5 Checkpoint（检查点）和 Checkpointer（检查点管理器）：LangGraph 的“时光机”
如果说前面的概念是 LangGraph 的“硬件”，那么 Checkpoint 和 Checkpointer 就是 LangGraph 的“软件增值服务”——它让我们可以：
- 断点续传：如果图在执行过程中因为网络中断、程序崩溃等原因停止了，我们可以从最近的 Checkpoint 恢复执行，不需要重新执行所有的步骤。
- 调试：我们可以在任何节点暂停执行，检查当前的 State，修改 State，然后手动恢复执行。
- 回滚：我们可以回滚到之前的某个 Checkpoint，重新执行。
- 状态持久化：我们可以把 Checkpoint 存储到磁盘、Redis、SQLite 等持久化存储中，即使程序重启了，状态也不会丢失。
什么是 Checkpoint？
简单来说，Checkpoint 就是图在执行过程中某个时刻的 State 的快照——它包含了当前的 State、当前执行到的节点、以及一些元数据（比如执行时间、执行ID等）。
什么是 Checkpointer？
Checkpointer 是用于管理 Checkpoint 的组件——它负责：
- 保存 Checkpoint：在图执行到某个节点之前或之后，自动保存当前的 State 到持久化存储中。
- 加载 Checkpoint：从持久化存储中加载之前的 Checkpoint，恢复图的执行。
- 列出 Checkpoint：列出某个执行ID的所有 Checkpoint。
- 删除 Checkpoint：删除某个执行ID的所有 Checkpoint。
LangGraph 内置了几种 Checkpointer：
MemorySaver：基于内存的 Checkpointer——Checkpoint 存储在程序的内存中，程序重启后就会丢失。适合用于开发和调试。SqliteSaver：基于 SQLite 的 Checkpointer——Checkpoint 存储在 SQLite 数据库文件中，程序重启后不会丢失。适合用于小型生产环境。PostgresSaver：基于 PostgreSQL 的 Checkpointer——Checkpoint 存储在 PostgreSQL 数据库中，适合用于大型生产环境。RedisSaver：基于 Redis 的 Checkpointer——Checkpoint 存储在 Redis 中，适合用于需要高性能的生产环境。
我们后面会在实战中详细讲如何使用 MemorySaver 和 SqliteSaver。
1.1.6 Compiled Graph（编译后的图）：LangGraph 的“可执行文件”
当我们定义好所有的节点、边、Checkpointer 之后，我们需要调用 graph.compile() 方法来**
更多推荐
1
0- 0
已为社区贡献35条内容
所有评论(0)