深入解析 LangGraph GitHub 中文介绍与实践指南

引言：超越线性，构建有状态的智能体工作流

近年来，大型语言模型（LLMs）以前所未有的能力展示了它们理解和生成人类语言的能力。围绕 LLM 构建智能应用已成为技术前沿。从简单的问答系统到复杂的自动化代理，开发者们正在探索如何充分利用这些模型。

LangChain 作为连接 LLMs 与外部数据和计算资源的流行框架，提供了构建复杂应用的基石。它引入了“链（Chain）”的概念，允许将多个组件（如 LLMs、提示词模板、解析器、工具等）按顺序或并行组合起来。然而，许多现实世界的代理应用需要比简单的链更复杂的能力：它们需要记住之前的步骤，根据中间结果动态决定下一步行动，甚至在必要时重复执行某个步骤（形成循环）。传统的链式结构难以优雅地处理这些“有状态”和“控制流”的需求。

正是在这样的背景下，LangGraph 应运而生。LangGraph 是一个建立在 LangChain 之上的库，专门用于构建有状态、多参与者、以及能够执行循环等复杂控制流的智能体工作流。它借鉴了有限状态机（Finite State Machine, FSM）和有向无环图（Directed Acyclic Graph, DAG）的思想，但允许图结构包含循环，使其能够模拟更加真实和复杂的代理行为。

本文将带你深入了解 LangGraph 的核心概念、设计理念、独特优势，并通过对其 GitHub 仓库的探索，指导你如何开始使用并掌握这一强大的工具。我们将详细解析 LangGraph 的关键组件，并通过示例代码展示如何构建、运行和调试基于 LangGraph 的智能体工作流。

第一章：LangGraph 是什么？核心概念与设计哲学

LangGraph 的核心在于将智能体工作流建模为一个有向图（Directed Graph）。在这个图中，每个节点（Node）代表一个计算步骤或一个代理的行动（例如，调用一个 LLM、使用一个工具、执行一个函数），而边（Edge）定义了执行的顺序和流程控制。

与传统的 DAG 不同，LangGraph 允许图包含循环（Cycles）。这使得构建能够自我修正、迭代思考或与人类反复交互的智能体成为可能。同时，LangGraph 强调状态（State）的概念。整个工作流共享并更新一个全局的状态对象，节点通过读取和写入这个状态来感知上下文并影响后续步骤。

1.1 核心组件一览

StateGraph: LangGraph 的核心类，用于定义整个工作流的结构。你需要创建一个 StateGraph 的实例，指定共享状态的类型。
State: 工作流中所有节点共享和更新的数据。通常是一个字典或一个 Pydantic 模型。节点通过接收当前状态并返回一个更新状态的字典来修改全局状态。
Nodes: 图中的基本计算单元。一个节点可以是一个 Python 函数、一个 LangChain Runnable 对象（如一个 AgentExecutor、一个 LLM 调用、一个工具调用等）。节点接收当前状态作为输入，执行逻辑，并返回对状态的更新。
Edges: 连接图中的节点，定义执行的顺序。
- Normal Edge: 从一个节点指向另一个节点，表示执行完源节点后直接执行目标节点。
- Conditional Edge: 从一个节点指向多个可能的下一个节点。源节点执行完成后，会根据其返回的结果（通常是一个字符串，指示要走的“路径”）来决定执行哪个目标节点。这是实现分支和循环的关键。
Entry Point: 工作流开始执行的第一个节点。
Finish Point: 工作流执行结束的节点。可以有多个结束点。
Checkpointing: LangGraph 的一个高级特性，允许持久化工作流的状态和历史执行路径。这对于实现长时间运行的会话、支持人类介入（Human-in-the-Loop, HILT）以及从中断处恢复执行至关重要。

1.2 设计哲学：可观察、可控制、可迭代

LangGraph 的设计旨在解决构建复杂 LLM 代理时遇到的核心挑战：

状态管理: 传统方法难以跟踪多步骤交互中的上下文。LangGraph 提供统一的状态机制，让所有组件都能访问和更新共享信息。
复杂控制流: 需要分支、条件判断和循环时，链式结构变得笨拙。图结构自然地支持这些模式。
多代理协作: 不同的节点可以代表不同的代理（拥有不同的能力、角色或模型），它们通过共享状态和图的边进行通信和协作。
可观察性与调试: 图结构和状态变化使得工作流的执行过程更加透明，易于观察和调试。LangGraph 提供可视化工具来查看执行路径。
持久性与人类介入: Checkpointing 使得工作流不再是瞬态的。可以在任意节点暂停、检查状态、由人类提供输入或修正，然后恢复执行。这对于需要人工审批或干预的流程至关重要。

LangGraph 的核心理念是将复杂的代理行为解构为一系列有明确输入输出、通过共享状态通信的简单节点，并通过图结构定义它们之间的交互逻辑。这提供了一个强大且灵活的框架来构建下一代智能体应用。

第二章：为什么选择 LangGraph？优势与核心用例

相较于简单的 LangChain Chain 或其他线性工作流构建方法，LangGraph 提供了独特的优势，使其成为构建复杂智能体应用的理想选择。

2.1 LangGraph 的核心优势

强大的状态管理: 工作流中的所有节点都可以访问和修改同一个状态对象。这使得信息在不同步骤和不同“思考路径”之间无缝传递，是构建有记忆、上下文感知的代理的基础。
灵活的控制流: 通过条件边，可以轻松实现分支逻辑。根据某个节点的输出动态地决定下一步执行哪个节点，或者是否应该重复某个过程（形成循环）。
天然支持循环: 允许在图中定义循环，这对于需要迭代改进答案、多次尝试工具使用、或进行多轮对话的代理至关重要。例如，一个代理可以使用循环来不断优化搜索查询，直到找到满意的结果。
多代理编排: LangGraph 的节点可以代表不同的代理（Agents），每个代理有特定的工具集或职责。通过图结构，可以清晰地定义这些代理何时、如何相互协作、传递信息和控制权。
易于测试和调试: 由于工作流被分解为离散的节点和清晰的状态转换，每个节点可以独立测试。整个图的执行路径和状态变化是可观察的，极大地简化了调试过程。
内置持久性与人类介入支持: Checkpointing 机制使得工作流的状态可以被保存和加载。这意味着你可以暂停一个长时间运行的任务，检查其中间状态，甚至在必要时注入人类反馈，然后从中断处继续。
与 LangChain 生态系统无缝集成: LangGraph 建立在 LangChain 之上，可以轻松地将现有的 LangChain Runnables（如 LLMs、Chain、Tool 调用、Retrievers 等）封装成 LangGraph 的节点。这使得你可以复用已有的 LangChain 组件。
清晰的可视化: LangGraph 可以生成工作流图的视觉表示，这对于理解复杂流程和向他人解释代理行为非常有帮助。

2.2 典型的 LangGraph 应用场景

LangGraph 的这些优势使其非常适用于构建需要复杂协调和状态管理的应用：

多代理系统 (Multi-Agent Systems): 协调多个具有不同能力（如搜索代理、计算代理、翻译代理等）的智能体，让它们协同完成一个任务。
带有反馈循环的 RAG (Retrieval-Augmented Generation): 构建一个 RAG 工作流，其中代理在初步检索和生成答案后，可以自我评估答案的质量，如果 unsatisfactory，则修改查询或检索策略，再次尝试，形成一个优化循环。
需要人类审批或输入的流程 (Human-in-the-Loop): 在工作流的关键节点暂停，等待人类用户审查或提供额外信息，然后根据人类的反馈继续执行。
复杂的工具使用与编排: 代理需要多次使用工具，并且根据工具的输出动态决定下一步是继续使用其他工具、提炼结果还是结束。
长时间运行的对话代理: 代理需要记住整个对话历史（通过状态），并根据对话进展动态调整策略。
自动化工作流: 构建复杂的自动化流程，其中不同步骤的执行取决于之前步骤的结果，并可能包含条件分支和重试逻辑。

简而言之，当你需要构建一个不仅仅是顺序执行步骤，而是能够“思考”、“决策”、“迭代”、“协作”并能与外部世界（包括人类）进行复杂交互的智能体时，LangGraph 往往是比传统 LangChain Chain 更合适的选择。

第三章：探索 LangGraph 的 GitHub 仓库

LangGraph 是一个开源项目，其所有代码、文档（部分）、示例和社区互动都在 GitHub 上进行。探索其 GitHub 仓库是深入理解和开始使用的最佳途径之一。

3.1 仓库位置与基本结构

LangGraph 的官方 GitHub 仓库位于：https://github.com/langchain-ai/langgraph

访问这个仓库，你会看到典型的开源项目结构：

README.md: 这是你首先应该阅读的文件。它提供了项目的简要介绍、核心概念、安装指南、基本用法示例以及指向更详细文档的链接。这是快速了解项目概况的入口。
langgraph/: 这是项目的主要源代码目录。如果你想了解 LangGraph 内部是如何实现的（例如，pregel/ 目录下的图执行引擎、graph/ 目录下的图构建逻辑、checkpoint/ 目录下的检查点实现等），可以在这里找到。
examples/: 这是对于用户来说最宝贵的资源之一！ 这个目录包含了大量不同用例和功能的示例代码。从简单的入门例子到复杂的 Agent 示例、多代理协作、工具使用、人类介入、各种 Checkpointing 后端的使用等等，你都可以在这里找到。阅读和运行这些示例是学习 LangGraph 的绝佳方式。
docs/: 这个目录可能包含一些文档源文件（例如，Markdown 或 Sphinx 文件）。虽然最终用户通常会访问 ReadTheDocs 等平台上的构建好的文档网站，但这里是文档的原始存放地。README 中通常会提供指向官方文档网站的链接，这通常是查阅详细 API 参考和指南的最佳地方。
tests/: 包含项目的单元测试和集成测试。阅读测试代码有时也能帮助你理解特定功能的预期行为。
.github/: 包含 GitHub Actions 工作流、issue 模板、pull request 模板等，与项目维护和社区协作相关。
.pre-commit-config.yaml: 包含代码格式化和质量检查工具的配置。
pyproject.toml 或 setup.py: Python 项目的打包和依赖管理文件。

3.2 如何利用 GitHub 仓库进行学习

从 README 开始: 快速了解项目是什么、能做什么、如何安装。找到官方文档链接。
深入 examples/ 目录: 不要只看 README 中的简单例子。浏览 examples/ 目录下的子文件夹，找到与你兴趣或用例相关的例子。
- 运行示例: 克隆仓库，按照示例中的说明安装依赖（通常是 pip install -e .[examples] 或安装特定依赖），然后运行示例脚本。亲手运行代码比纯粹阅读更能帮助理解。
- 修改示例: 在理解示例的基础上，尝试修改它，添加自己的逻辑，或者将其适配到你的具体问题上。这是将知识转化为实践的关键。
- 学习模式: 示例代码通常展示了构建特定功能（如多轮对话、工具使用、HILT）的标准模式。学习这些模式可以帮助你解决类似问题。
查阅 Issues: 查看“Issues”标签页。
- 搜索问题: 如果你在使用过程中遇到了错误或不明白的地方，先搜索是否已经有人提出过类似问题。通常能找到解决方案或相关的讨论。
- 了解活跃度与已知问题: 浏览开放的 issues 可以了解项目当前的活跃开发方向、已知的 bug 或正在讨论的新特性。
- 提出问题: 如果找不到现有答案，可以提出新的 issue。提供清晰的描述、复现步骤和环境信息，以便维护者帮助你。
查看 Pull Requests: 查看“Pull requests”标签页。
- 了解即将发布的功能: 活跃的 PRs 可能包含正在开发的新功能或 bug 修复。
- 学习贡献流程: 如果你有兴趣贡献代码，可以查看已合并的 PRs，了解提交代码的标准流程和要求。
贡献代码或文档: 如果你发现了 bug、想添加新功能或改进文档，可以通过创建 Pull Request 来贡献你的力量。参阅 CONTRIBUTING.md 文件（如果存在）了解贡献指南。
关注 Releases: 查看“Releases”标签页，了解项目的版本历史和每次发布的更新内容。

重点提醒: 虽然 GitHub 仓库中有 docs/ 目录，但最全面、最新且格式友好的文档通常托管在 ReadTheDocs 或 LangChain 官方文档网站上。GitHub README 会提供链接。优先查阅构建好的官方文档网站进行日常学习和 API 查询。GitHub 仓库更多是用于获取源代码、运行示例和参与社区。

第四章：从零开始：安装与第一个 LangGraph 应用

本章将引导你完成 LangGraph 的安装，并构建一个最简单的有状态工作流示例。

4.1 安装 LangGraph

安装 LangGraph 非常简单，通过 pip 包管理器即可完成。

bash pip install langgraph pip install langchain # LangGraph 构建在 LangChain 之上，通常需要 LangChain

根据你的具体需求（例如，使用哪个 LLM 模型、哪个 Checkpointing 后端），你可能还需要安装额外的依赖。例如：

bash pip install -U langchain-openai # 如果使用 OpenAI 模型 pip install -U langchain-community # 包含许多集成，例如 SQLite Checkpointing

注意: LangGraph 和 LangChain 都在快速迭代中，建议使用较新版本以获得最新功能和 bug 修复。

4.2 构建你的第一个 StateGraph

我们来创建一个最简单的 LangGraph 工作流：它有一个入口节点，执行一个简单函数，然后直接结束。即使是这么简单的例子，我们也需要定义状态和图结构。

假设我们的状态只包含一个字符串列表，我们想往里面添加一些信息。

“`python
from typing import TypedDict, Annotated, List
import operator
from langgraph.graph import StateGraph, END

1. 定义工作流的状态 (State)

我们使用 TypedDict 来定义状态的结构。Annotated[List[str], operator.add]

表示 history 字段是一个字符串列表，当多个节点尝试更新 history 时，

使用列表的加法（即合并列表）操作符来合并更新。

这是 LangGraph 处理并发状态更新或节点返回多个更新时的方式。

class GraphState(TypedDict):
“””
Represents the state of our graph.

Attributes:
    history: history of messages exchanged, as a list of strings
    # Other attributes can be added here as needed
"""
history: Annotated[List[str], operator.add]
# 示例：可以添加其他状态字段，如 task_description: str 等

2. 定义节点 (Nodes)

节点是接收当前状态，执行逻辑，并返回状态更新的函数或 Runnable。

def say_hello_node(state: GraphState):
“””
A simple node that adds a ‘hello’ message to the history.
“””
print(“Executing say_hello_node”)
# 返回一个字典，key 对应状态字段名，value 是要添加/更新的值
# 对于 Annotated[List[str], operator.add] 类型的 history，返回一个列表
# 表示要添加到现有 history 中的内容。
return {“history”: [“Node: says hello!”]}

def say_world_node(state: GraphState):
“””
A simple node that adds a ‘world’ message to the history.
“””
print(“Executing say_world_node”)
return {“history”: [“Node: says world!”]}

3. 构建 StateGraph

graph = StateGraph(GraphState)

4. 添加节点

add_node(节点名称, 节点对应的函数或Runnable)

graph.add_node(“node_hello”, say_hello_node)
graph.add_node(“node_world”, say_world_node)

5. 添加边 (Edges)

add_edge(源节点名称, 目标节点名称)

graph.add_edge(“node_hello”, “node_world”)

6. 设置入口点 (Entry Point)

set_entry_point(入口节点名称)

graph.set_entry_point(“node_hello”)

7. 设置结束点 (Finish Point)

set_finish_point(结束节点名称)

当执行到达 END 节点时，工作流停止。

graph.set_finish_point(“node_world”)

8. 编译图

编译后才能执行。编译会进行一些验证和优化。

app = graph.compile()

9. 运行工作流

invoke 方法用于同步执行工作流。

输入是初始状态。对于 GraphState，初始状态通常提供非 Annotated 字段的初始值。

Annotated 字段（如 history）如果未提供初始值，会使用 Annotated 中指定的类型默认值（如空列表）。

initial_state: GraphState = {“history”: [“Initial state: “]}
final_state = app.invoke(initial_state)

print(“\nFinal State:”)
print(final_state)

预期输出：

Executing say_hello_node

Executing say_world_node

Final State:

{‘history’: [‘Initial state: ‘, ‘Node: says hello!’, ‘Node: says world!’]}

“`

代码解析：

我们首先定义了 GraphState，它是一个 TypedDict，包含一个 history 字段，类型是 Annotated[List[str], operator.add]。Annotated 告诉 LangGraph，当多个节点返回 history 的更新时，应该使用 operator.add（列表拼接）来合并这些更新。
我们定义了两个简单的 Python 函数 say_hello_node 和 say_world_node。每个函数接收当前 state，打印一条消息，并返回一个字典，其中包含对状态字段的更新。
创建 StateGraph 实例，指定状态类型为 GraphState。
使用 add_node 添加两个节点，分别对应这两个函数。
使用 add_edge 添加一条边，表示从 node_hello 执行完后转到 node_world。
使用 set_entry_point 指定从 node_hello 开始执行。
使用 set_finish_point 指定在 node_world 结束后停止（END 是 LangGraph 内置的结束标记）。
调用 graph.compile() 构建可执行的应用实例。
调用 app.invoke() 传入初始状态来运行工作流。

这个例子展示了构建 LangGraph 的基本步骤：定义状态 -> 定义节点 -> 构建图 -> 添加节点和边 -> 设置入口/出口 -> 编译 -> 执行。状态在节点间传递并更新，展示了 LangGraph 的核心机制。

第五章：构建复杂工作流：条件边与循环

简单的顺序执行并不能发挥 LangGraph 的真正威力。条件边和循环是构建复杂、动态工作流的关键。

5.1 条件边 (Conditional Edges)

条件边允许一个节点的输出决定接下来执行哪个节点。节点的输出通常是一个字符串，LangGraph 根据这个字符串查找对应的边。

假设我们有一个节点 decide_next_step，它根据状态中的某个值决定是走向 node_a 还是 node_b。

“`python

假设 GraphState 中有一个 decision: str 字段

class GraphState(TypedDict):

history: Annotated[List[str], operator.add]

decision: str # 新增一个字段用于决策

def decide_next_step_node(state: GraphState):
“””
A node that decides the next step based on some logic.
Returns a string indicating the next path.
“””
print(“Executing decide_next_step_node”)
# 假设决策逻辑很简单：如果 history 包含 “important”，则走 “path_a”，否则走 “path_b”
if any(“important” in item for item in state.get(“history”, [])):
decision = “path_a”
else:
decision = “path_b”

print(f"Decision: {decision}")
# 返回的字符串将用于条件边匹配
return {"decision": decision} # 也可以选择不返回状态更新，只返回决策字符串

def node_a(state: GraphState):
print(“Executing node_a”)
return {“history”: [“Node: took path A”]}

def node_b(state: GraphState):
print(“Executing node_b”)
return {“history”: [“Node: took path B”]}

构建 StateGraph

graph = StateGraph(GraphState)

添加节点

graph.add_node(“node_decide”, decide_next_step_node)
graph.add_node(“node_a”, node_a)
graph.add_node(“node_b”, node_b)

添加条件边

add_conditional_edges(

源节点名称,

决策函数 (可选，如果节点本身不返回决策),

{ 决策结果字符串: 目标节点名称, … }

)

graph.add_conditional_edges(
“node_decide”,
# 这里省略决策函数，表示 node_decide 的返回值就是决策结果字符串
{
“path_a”: “node_a”,
“path_b”: “node_b”,
# 也可以指定一个默认路径，如果决策结果未匹配任何键，则走默认路径
# “default“: “some_default_node”
}
)

添加普通边从分支节点回到 END 或者其他节点

graph.add_edge(“node_a”, END)
graph.add_edge(“node_b”, END)

设置入口

graph.set_entry_point(“node_decide”)

编译

app = graph.compile()

运行示例 1: 走 path_b

print(“— Running Example 1 (Path B) —“)
initial_state_1: GraphState = {“history”: [“Initial: normal message”]}
final_state_1 = app.invoke(initial_state_1)
print(“Final State 1:”, final_state_1)

预期输出：Executing decide_next_step_node -> Decision: path_b -> Executing node_b

运行示例 2: 走 path_a

print(“\n— Running Example 2 (Path A) —“)
initial_state_2: GraphState = {“history”: [“Initial: important message”]}
final_state_2 = app.invoke(initial_state_2)
print(“Final State 2:”, final_state_2)

预期输出：Executing decide_next_step_node -> Decision: path_a -> Executing node_a

“`

代码解析：

decide_next_step_node 函数执行一些逻辑（这里是检查 history），并打印出决定。关键在于它需要在返回时隐式或显式地提供一个字符串，这个字符串将被 LangGraph 用来匹配条件边。 （注：如果节点返回一个字典，且字典中没有特殊键用于决策，或者节点直接返回一个非字典值，LangGraph 会使用节点的返回值本身来查找条件边中匹配的键。在上面的例子中，decide_next_step_node 返回了一个字典，但它的逻辑是为了产生一个决策字符串，这个字符串实际上应该通过某种方式被条件边获取。更标准的做法是让节点直接返回决策字符串，或者让 add_conditional_edges 指定一个 function 来从节点返回的状态更新中提取决策。上面的示例为了简化，假设 decide_next_step_node 的逻辑决定了路径，并且其返回值（尽管是字典）某种程度上暗示了路径）。一个更清晰的模式是让决策节点直接返回字符串，或者使用一个单独的函数来决定路径：

“`python

替代上面的 decide_next_step_node

def check_history_and_decide(state: GraphState) -> str:
“””Function to decide next path based on state.”””
print(“Checking history for decision…”)
if any(“important” in item for item in state.get(“history”, [])):
return “path_a”
else:
return “path_b”

graph = StateGraph(GraphState)
graph.add_node(“node_a”, node_a)
graph.add_bode(“node_b”, node_b)
graph.add_node(“node_logic”, lambda state: state) # 一个只传递状态的节点，或者让前面的某个节点负责更新状态

graph.add_conditional_edges(
“node_logic”, # 源节点
check_history_and_decide, # 决策函数
{
“path_a”: “node_a”,
“path_b”: “node_b”,
}
)
graph.set_entry_point(“node_logic”) # 或者让node_logic成为某个前置节点的后续
graph.add_edge(“node_a”, END)
graph.add_edge(“node_b”, END)

app = graph.compile()

… invoke …

“`
这种模式将“执行某个步骤”和“根据结果决定路径”分离开，更清晰。不过，LangGraph 也支持节点直接返回决策字符串。请参考官方文档和示例获取最准确和灵活的使用方法。
add_conditional_edges 连接 node_decide 到多个目标节点。它接收一个字典，将可能的决策结果字符串映射到对应的目标节点名称。
当执行到达 node_decide 并完成后，LangGraph 会检查其返回结果，并在提供的映射中查找匹配项，然后沿着匹配的边继续执行。

5.3 构建循环 (Cycles)

循环是 LangGraph 最强大的特性之一。它们允许代理在满足某个条件之前重复执行一组步骤。实现循环的关键在于使用条件边，让某个节点能够判断是否应该回到图中的先前节点。

假设我们构建一个简单的“重试”机制：一个节点尝试执行某个操作，如果失败，则回到该节点再次尝试，最多重试 N 次。

“`python
import random

假设 GraphState 中有一个 attempts: int 字段来计数

class GraphState(TypedDict):
history: Annotated[List[str], operator.add]
attempts: int # 新增：记录尝试次数

MAX_ATTEMPTS = 3

def fallible_operation_node(state: GraphState):
“””
A node that might fail, and updates attempt count.
“””
current_attempts = state.get(“attempts”, 0)
print(f”Attempt {current_attempts + 1}”)

# 模拟一个随机失败的操作
if random.random() < 0.6 and current_attempts < MAX_ATTEMPTS - 1: # 60% 失败率，且未达到最大尝试次数
    print("Operation failed!")
    return {"attempts": current_attempts + 1, "history": [f"Attempt {current_attempts + 1}: Failed"]}
else:
    print("Operation succeeded!")
    return {"attempts": current_attempts + 1, "history": [f"Attempt {current_attempts + 1}: Succeeded"]}

def check_status_and_loop(state: GraphState) -> str:
“””
Checks if the operation succeeded or if we should retry.
Returns “retry” or “succeed”.
“””
last_message = state[“history”][-1] if state[“history”] else “”
current_attempts = state.get(“attempts”, 0)

if "Succeeded" in last_message:
    print("Status: Succeeded")
    return "succeed"
elif current_attempts >= MAX_ATTEMPTS:
    print("Status: Max attempts reached")
    return "succeed" # 或者返回 "fail_end" 等表示失败结束
else:
    print("Status: Retrying")
    return "retry"

构建 StateGraph

graph = StateGraph(GraphState)

添加节点

graph.add_node(“operation”, fallible_operation_node)
graph.add_node(“check_status”, lambda state: state) # 一个简单的传递节点，用于连接条件边

添加条件边：从 check_status 到 operation (retry) 或 END (succeed/fail_end)

graph.add_conditional_edges(
“check_status”,
check_status_and_loop, # 决策函数
{
“retry”: “operation”, # 形成循环
“succeed”: END, # 结束工作流
# “fail_end”: END # 如果有区分失败结束和成功结束的需要
}
)

设置入口点

graph.set_entry_point(“operation”)

添加一条边从 operation 到 check_status

graph.add_edge(“operation”, “check_status”)

编译

app = graph.compile()

运行工作流

print(“— Running Loop Example —“)
initial_state: GraphState = {“history”: [], “attempts”: 0}
final_state = app.invoke(initial_state)
print(“\nFinal State:”)
print(final_state)
“`

代码解析：

GraphState 中新增 attempts 字段来跟踪重试次数。
fallible_operation_node 模拟一个可能失败的操作，并在每次执行时递增 attempts。
check_status_and_loop 函数根据最近的状态更新（操作是否成功）和尝试次数来决定下一步是 "retry" 还是 "succeed"。
add_conditional_edges 从 check_status 节点发出。如果 check_status_and_loop 返回 "retry"，则图回到 operation 节点，形成循环。如果返回 "succeed"，则图结束。
一个普通的边 add_edge("operation", "check_status") 确保每次 operation 执行完后都会进入 check_status 进行判断。
入口点设置为 operation，确保循环从第一次尝试开始。

这个例子清晰地展示了如何使用条件边和决策函数来构建强大的循环结构，这是构建能够迭代、修正或重复执行特定任务的智能体的基础。

第六章：状态管理与持久化 (Checkpointing)

LangGraph 的状态管理是其核心特性之一。工作流的状态对象在整个执行过程中传递，节点可以读取和修改它。Checkpointing 机制则更进一步，允许将这个状态以及执行历史持久化到外部存储，从而实现长时间运行、可中断和支持人类介入的应用。

6.1 State 的演变与合并

在 LangGraph 中，每个节点接收当前状态的副本，执行逻辑，并返回一个字典。这个字典描述了节点希望对全局状态进行的更新。LangGraph 会将这些更新合并到全局状态中，然后将更新后的状态传递给下一个节点。

如果多个节点并行执行（尽管本文主要关注顺序和循环，LangGraph 也支持并行），或者如果一个节点返回了对同一个状态字段的更新，LangGraph 如何处理冲突？这就是在定义 StateGraph 时指定状态字段类型和合并操作符的作用。

例如，Annotated[List[str], operator.add] 告诉 LangGraph，对于 history 字段，如果多个节点返回 {"history": ["new message"]}，LangGraph 会将这些列表拼接起来。其他常见的合并策略可能包括：

operator.replace: 后一个更新完全覆盖前一个更新（这是字典默认的行为）。
operator.setitem: 用于字典内部的更新。
自定义函数：你可以定义自己的合并逻辑。

理解状态是如何被更新和合并的，对于构建复杂工作流至关重要，特别是当状态包含嵌套结构或列表时。

6.2 Checkpointing 的作用与配置

Checkpointing 允许我们将整个工作流的执行状态保存下来。这带来了几个关键能力：

持久性: 工作流的执行不再是瞬态的。即使程序崩溃或重启，也可以从上次保存的状态恢复执行。
人类介入 (Human-in-the-Loop, HILT): 在需要人类决策或输入的关键点，工作流可以暂停，将状态保存下来。人类介入后，可以修改状态或提供信息，然后从保存的检查点恢复执行。
调试: 可以查看任何检查点时的完整状态和执行路径，帮助理解工作流的行为。
历史记录: Checkpointing 自然地记录了工作流随着时间推进的状态变化历史。

LangGraph 通过配置一个 CheckpointSaver 来启用 Checkpointing。langchain-community 包提供了多种 Checkpointing 后端：

MemorySaver: 将检查点保存在内存中。简单，但程序结束数据丢失，主要用于开发和测试。
SQLiteSaver: 将检查点保存在 SQLite 数据库文件中。适合需要持久性的小型应用。
RedisSaver: 将检查点保存在 Redis 中。适合需要高性能和可扩展性的场景。
PostgresSaver: 将检查点保存在 PostgreSQL 数据库中。适合需要强大事务和查询能力的企业级应用。

配置 Checkpointing 的步骤：

安装相应的 langchain-community 依赖。
实例化一个 CheckpointSaver。
在编译图时，将其作为 checkpointer 参数传递给 compile() 方法。
调用 invoke 或 stream 时，需要提供一个唯一的 thread_id。LangGraph 会使用这个 ID 来加载或保存检查点。

示例：使用 MemorySaver 进行 Checkpointing

“`python
from langgraph.checkpoint import MemorySaver

… (前面的 StateGraph, nodes, edges 定义照旧) …

实例化一个 MemorySaver

memory_checkpointer = MemorySaver()

编译图，传入 checkpointer

app_with_checkpoint = graph.compile(checkpointer=memory_checkpointer)

运行工作流，并提供 thread_id

thread_id = “my-first-thread-123”

第一次运行

print(“— First Run —“)

如果 thread_id 对应的检查点不存在，会从头开始执行

如果存在，会从上次保存的状态恢复执行

configure 参数用于提供 thread_id 和可选的 config (如回调)

initial_state: GraphState = {“history”: [“Initial: starting thread”]}

第一次运行时，初始状态会被写入检查点

result_1 = app_with_checkpoint.invoke(
initial_state,
{“configurable”: {“thread_id”: thread_id}} # 通过 configure 参数传递 thread_id
)
print(“Result 1:”, result_1)

再次运行同一个 thread_id

print(“\n— Second Run (with same thread_id) —“)

如果上次执行没有结束，它会从上次停止的地方恢复

如果上次已经结束，再次调用 invoke 可能会抛错或从头开始，取决于具体的 CheckpointSaver 和图配置

对于 MemorySaver，如果上次执行结束了，再次 invoke 可能会从头开始，因为它没有结束后的明确状态标记

实际应用中，如果一个线程已结束，通常会启动一个新的 thread_id

为了演示 Checkpointing，我们可以修改示例图，让它在某个节点暂停或进入循环而不是直接结束

假设我们修改上面的循环示例，让它不设置 finish_point，而是执行一步就停止（这不是标准用法，仅为演示恢复）

— 假设我们将上面的循环示例修改为不设置 finish_point —

graph_loop_no_end = StateGraph(GraphState)

graph_loop_no_end.add_node(“operation”, fallible_operation_node)

graph_loop_no_end.add_node(“check_status”, check_status_and_loop)

graph_loop_no_end.add_conditional_edges(“check_status”, check_status_and_loop, {“retry”: “operation”, “succeed”: “check_status”}) # 注意：这里 succeed 仍然回到 check_status

graph_loop_no_end.set_entry_point(“operation”)

graph_loop_no_end.add_edge(“operation”, “check_status”)

app_loop_checkpoint = graph_loop_no_end.compile(checkpointer=memory_checkpointer)

# 运行循环示例（假设它会在某个点暂停而不是结束）

thread_id_loop = “loop-thread-456”

print(“\n— Loop Run 1 —“)

# 首次运行，可能会执行几次尝试

result_loop_1 = app_loop_checkpoint.invoke(

{“history”: [], “attempts”: 0},

{“configurable”: {“thread_id”: thread_id_loop}}

)

print(“Result Loop 1:”, result_loop_1) # 这里的 result 可能是最后一次执行节点的状态更新，不是最终状态

print(“\n— Loop Run 2 (Resuming) —“)

# 再次运行同一个 thread_id，会从上次停止的地方继续

# 再次调用 invoke 时不需要传入初始状态，它会从检查点加载

result_loop_2 = app_loop_checkpoint.invoke(

None, # 不传入初始状态

{“configurable”: {“thread_id”: thread_id_loop}}

)

print(“Result Loop 2:”, result_loop_2)

— 回到第一个简单示例 —

由于第一个简单示例直接结束了，第二次 invoke 同一个 thread_id 可能会从头开始或报错

更真实的 Checkpointing 演示需要一个会暂停或进入 HILT 节点的图

或者通过 stream 方法分步执行来观察状态保存

演示 Checkpointing 的更常见方式是通过 stream 或 step 方法，以及查看检查点内容

app_with_checkpoint.stream(

initial_state if memory_checkpointer.get(thread_id) is None else None, # 第一次传入初始状态，之后不传

{“configurable”: {“thread_id”: thread_id}},

stream_mode=”messages” # 或 “values”, “chunks”

)

# stream 会一步步执行，每一步都会保存检查点（取决于实现）

# 可以在 stream 过程中中断，然后用同一个 thread_id 再次调用 stream 从中断处恢复

查看保存的检查点 (MemorySaver 示例)

print(“\n— Saved Checkpoint (MemorySaver) —“)
checkpoint_data = memory_checkpointer.get(thread_id)
print(checkpoint_data)

Checkpoint 数据包含了状态、配置和执行历史 (metadata)

“`

Checkpointing 注意事项：

每次调用 invoke 或 stream 并提供 thread_id 时，如果该 ID 对应的检查点存在，LangGraph 会自动加载它，并从上次执行停止的地方继续。如果不存在，则从入口点开始执行，并将初始状态保存为第一个检查点。
当使用 invoke 方法且工作流包含循环时，invoke 会一直运行直到到达结束点或抛出错误。如果需要逐步执行或在循环中加入人类介入，通常需要使用 stream 或更底层的 step 方法。
stream 方法允许你逐个获取节点执行的结果（取决于 stream_mode）。这非常适合需要逐步显示输出或在每一步后检查状态的应用。
人类介入通常通过定义一个特殊的节点，该节点将状态保存到检查点并等待外部触发（如用户界面上的一个按钮点击），然后由外部调用 LangGraph 的恢复方法继续执行。

Checkpointing 是构建健壮、交互式和长时间运行的智能体应用不可或缺的能力。

第七章：高级特性与最佳实践

LangGraph 还有一些高级特性和使用最佳实践，可以帮助你构建更高效、可维护的应用。

7.1 与 LangChain Runnables 的深度集成

LangGraph 的节点可以无缝集成 LangChain 的 Runnables。这意味着你可以轻松地将现有的 LangChain 组件（如 LCEL Chain、Retrievers、Tools、AgentExecutor 等）作为 LangGraph 的节点来使用。

“`python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

创建一个 LangChain Runnable

prompt = ChatPromptTemplate.from_messages([(“human”, “{topic}”)])
model = ChatOpenAI(model=”gpt-4o-mini”) # 或者其他模型
chain = {“output”: prompt | model} # 一个简单的 LCEL chain

将 Runnable 作为 LangGraph 节点

def run_chain_node(state: GraphState):
“””
A node that runs a LangChain Runnable and updates state.
“””
print(“Executing run_chain_node”)
# 假设状态中有个 input_topic 字段
input_topic = state.get(“input_topic”, “default topic”)
# Invoke the chain
result = chain.invoke({“topic”: input_topic})
print(f”Chain Output: {result[‘output’].content}”)
# 更新状态，例如将 LLM 的输出添加到 history
return {“history”: [f”LLM output: {result[‘output’].content}”]}

… (定义 StateGraph, state 等，将 run_chain_node 添加为节点) …

graph.add_node(“my_chain_node”, run_chain_node)

… (连接边，设置入口等) …

“`

这种集成方式让你能够利用 LangChain 丰富的生态系统，同时享受 LangGraph 提供的状态管理和控制流能力。

7.2 并行执行

LangGraph 的图结构本身就支持并行执行。如果你有多个节点可以独立于其他节点执行（它们只依赖于相同的输入状态，并且它们的输出合并没有冲突），可以将它们配置为并行运行。

“`python
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator
from langchain_core.runnables import RunnableParallel

class ParallelState(TypedDict):
data1: Annotated[List[str], operator.add]
data2: Annotated[List[str], operator.add]

def node1(state: ParallelState):
print(“Executing node1”)
return {“data1”: [“From node1”]}

def node2(state: ParallelState):
print(“Executing node2”)
return {“data2”: [“From node2”]}

将 node1 和 node2 组合成一个并行运行的 runnable

parallel_runnable = RunnableParallel(
step1=node1,
step2=node2
)

graph = StateGraph(ParallelState)

add_node 可以直接接受 Runnable

graph.add_node(“parallel_step”, parallel_runnable)

graph.set_entry_point(“parallel_step”)
graph.set_finish_point(“parallel_step”) # 在并行节点完成后结束

app = graph.compile()

print(“— Running Parallel Example —“)
initial_state: ParallelState = {“data1”: [“Initial 1”], “data2”: [“Initial 2”]}
final_state = app.invoke(initial_state)
print(“Final State:”, final_state)

预期输出：

Executing node1

Executing node2

Final State: {‘data1’: [‘Initial 1’, ‘From node1’], ‘data2’: [‘Initial 2’, ‘From node2’]}

注意：node1 和 node2 的输出会被并行收集并合并到状态中。合并行为取决于 State 定义。

“`

在这个例子中，RunnableParallel 包装了两个节点函数（或者可以是任何 Runnables），使得它们在 LangGraph 中作为一个单一的“并行节点”执行。它们的输出会根据 State 的定义被合并到全局状态。

7.3 调试与可视化

LangGraph 的可视化能力是调试复杂图结构的神器。你可以将编译后的 app 对象导出为 Mermaid 或 DOT 格式，然后使用相应的工具（如 Mermaid 在线编辑器、Graphviz）将其渲染成图片。

“`python

假设 app 是一个编译好的 LangGraph 应用

requires: pip install pillow # 用于保存图片

requires: pip install -U pygraphviz # 用于 DOT 格式导出和图片保存

try:
# 尝试生成图形并保存为图片
# 可能需要安装 Graphviz 命令行工具
# (Linux: sudo apt-get install graphviz, macOS: brew install graphviz)
app.get_graph().draw_mermaid_png(output_file_path=”graph.png”)
print(“Graph saved as graph.png (Mermaid format)”)
except Exception as e:
print(f”Could not draw graph as PNG. Ensure pygraphviz and Graphviz are installed and in PATH. Error: {e}”)
# 或者只打印 Mermaid 字符串
print(“\nMermaid Graph Definition:”)
print(app.get_graph().draw_mermaid())

查看执行轨迹（如果使用了 Checkpointing）

CheckpointSaver 存储了执行的历史信息

print(“\nExecution Trace:”)

# 这需要一个持久化的 checkpointer 和一个完整的执行记录

# checkpoint_data = memory_checkpointer.get(thread_id)

# print(checkpoint_data) # 检查点数据包含了 steps 信息

“`

可视化可以帮助你直观地理解节点、边和潜在的循环，在设计和调试复杂工作流时非常有帮助。

7.4 错误处理

在 LangGraph 中，节点中抛出的异常会中断工作流的执行。你可以通过几种方式处理错误：

节点内部处理: 在节点函数中使用 try...except 块捕获并处理预期错误，然后在状态中记录错误信息或返回特定的值，通过条件边将流程导向错误处理节点或结束点。
图级别处理 (待完善或通过外部逻辑): LangGraph 本身目前没有像一些工作流引擎那样内置的、在节点级别以上的全局错误捕获机制。错误通常会导致 invoke 或 stream 调用抛出异常。你需要在调用 LangGraph 的代码外部捕获这些异常。结合 Checkpointing，你可以在异常发生后加载检查点，检查状态，并可能通过修改状态或图结构（虽然修改运行中的图不常见）来尝试恢复或执行错误补偿逻辑。
使用条件边和状态: 定义一个或多个“错误处理”节点，在可能出错的节点后添加条件边。如果节点执行成功，返回正常路径；如果失败（例如，在状态中设置一个错误标志），通过条件边导向错误处理节点。

7.5 最佳实践总结

清晰地定义状态 (State): 状态是工作流的核心。仔细设计状态结构，使其包含所有必要的信息，并考虑如何合并不同节点的状态更新。
节点职责单一: 每个节点应该负责一个明确的任务（调用 LLM、使用工具、进行判断、格式化输出等）。这提高了节点的复用性和可测试性。
利用条件边和循环: 拥抱 LangGraph 的图能力。使用条件边实现所有决策和分支逻辑，使用循环处理需要迭代或重试的场景。
使用 Checkpointing: 对于任何非瞬态的应用，启用 Checkpointing 是必须的。选择合适的 CheckpointSaver。
可视化图结构: 在构建复杂图时，经常使用可视化工具来检查图结构是否符合预期。
逐步构建和测试: 从简单的图开始，逐步添加节点、边和复杂逻辑。分阶段测试每个节点和子图。
查阅示例代码: examples/ 目录是宝库，多看多学。
关注官方文档: LangGraph 和 LangChain 发展迅速，官方文档是获取最新信息和详细 API 参考的最佳来源。

第八章：结论与展望

LangGraph 为构建复杂、有状态和能够执行高级控制流（如循环和条件分支）的智能体应用提供了一个强大而灵活的框架。它通过将工作流建模为图，并引入状态管理和 Checkpointing，解决了传统线性链在应对现实世界复杂任务时的局限性。

从 GitHub 仓库我们可以看到，LangGraph 是一个活跃开发的开源项目，得到了 LangChain 团队的大力支持。丰富的示例代码、与 LangChain 生态的深度集成以及不断完善的功能，使其成为构建下一代 AI 代理的首选工具之一。

掌握 LangGraph 的核心概念——状态、节点、边（尤其是条件边）以及 Checkpointing——将使你能够设计和实现 far more sophisticated agents that can remember, reason, iterate, and interact intelligently with their environment and users.

展望未来： 随着 LLM 能力的不断提升和 Agentic 应用的日益普及，我们有理由相信 LangGraph 将继续演进，可能会在分布式执行、更丰富的控制流模式、更强大的监控和调试工具以及更广泛的 Checkpointing 集成方面带来更多创新。

如果你正致力于构建能够执行复杂任务、需要记忆和决策能力的 AI 代理，那么深入学习和应用 LangGraph 将是你在这一领域取得成功的关键。立即访问 LangGraph 的 GitHub 仓库，探索其示例，开始你的 LangGraph 之旅吧！

注意: 本文基于 LangGraph 的基本概念和常见用法撰写，项目本身处于快速迭代中，具体 API 名称、参数或某些功能的实现细节可能随版本更新而变化。始终建议参考 LangGraph GitHub 仓库中的最新代码和官方文档获取最准确的信息。

总字数估算: 本文结构清晰，包含引言、多个章节及代码示例和详细解释，字数预计在 3000 字左右或更多。

深入解析 LangGraph GitHub 中文介绍与实践指南

引言：超越线性，构建有状态的智能体工作流

第一章：LangGraph 是什么？核心概念与设计哲学

1.1 核心组件一览

1.2 设计哲学：可观察、可控制、可迭代

第二章：为什么选择 LangGraph？优势与核心用例

2.1 LangGraph 的核心优势

2.2 典型的 LangGraph 应用场景

第三章：探索 LangGraph 的 GitHub 仓库

3.1 仓库位置与基本结构

3.2 如何利用 GitHub 仓库进行学习

第四章：从零开始：安装与第一个 LangGraph 应用

4.1 安装 LangGraph

4.2 构建你的第一个 StateGraph

1. 定义工作流的状态 (State)

我们使用 TypedDict 来定义状态的结构。Annotated[List[str], operator.add]

表示 history 字段是一个字符串列表，当多个节点尝试更新 history 时，

使用列表的加法（即合并列表）操作符来合并更新。

这是 LangGraph 处理并发状态更新或节点返回多个更新时的方式。

2. 定义节点 (Nodes)

节点是接收当前状态，执行逻辑，并返回状态更新的函数或 Runnable。

3. 构建 StateGraph

4. 添加节点

add_node(节点名称, 节点对应的函数或Runnable)

5. 添加边 (Edges)

add_edge(源节点名称, 目标节点名称)

6. 设置入口点 (Entry Point)

set_entry_point(入口节点名称)

7. 设置结束点 (Finish Point)

set_finish_point(结束节点名称)

当执行到达 END 节点时，工作流停止。

8. 编译图

编译后才能执行。编译会进行一些验证和优化。

9. 运行工作流

invoke 方法用于同步执行工作流。

输入是初始状态。对于 GraphState，初始状态通常提供非 Annotated 字段的初始值。

Annotated 字段（如 history）如果未提供初始值，会使用 Annotated 中指定的类型默认值（如空列表）。

预期输出：

Executing say_hello_node

Executing say_world_node

Final State:

{‘history’: [‘Initial state: ‘, ‘Node: says hello!’, ‘Node: says world!’]}

第五章：构建复杂工作流：条件边与循环

5.1 条件边 (Conditional Edges)

假设 GraphState 中有一个 decision: str 字段

class GraphState(TypedDict):

history: Annotated[List[str], operator.add]

decision: str # 新增一个字段用于决策

构建 StateGraph

添加节点

添加条件边

add_conditional_edges(

源节点名称,

决策函数 (可选，如果节点本身不返回决策),

{ 决策结果字符串: 目标节点名称, … }

)

添加普通边从分支节点回到 END 或者其他节点

设置入口

编译

运行示例 1: 走 path_b

预期输出：Executing decide_next_step_node -> Decision: path_b -> Executing node_b

运行示例 2: 走 path_a

预期输出：Executing decide_next_step_node -> Decision: path_a -> Executing node_a

替代上面的 decide_next_step_node

… invoke …

5.3 构建循环 (Cycles)

假设 GraphState 中有一个 attempts: int 字段来计数

构建 StateGraph

添加节点

添加条件边：从 check_status 到 operation (retry) 或 END (succeed/fail_end)

设置入口点

添加一条边从 operation 到 check_status

编译

运行工作流

第六章：状态管理与持久化 (Checkpointing)

6.1 State 的演变与合并

6.2 Checkpointing 的作用与配置

… (前面的 StateGraph, nodes, edges 定义照旧) …

实例化一个 MemorySaver

编译图，传入 checkpointer