Skip to main content

langgraph-core

下面我来为你清晰、系统地讲解 LangGraph 的核心概念 与 运行原理。

LangGraph 是 LangChain 生态中用于构建 有状态、多角色、循环交互 的智能体（Agent）和图工作流的库。它的核心设计灵感来自图计算与状态机，让开发者能够精确控制大模型应用中的每一步逻辑与流转。

一、核心概念​

1. State（状态）​

  • 所有节点共享的持久化数据结构（类似字典或 Pydantic 模型）。
  • 每个节点执行后会更新状态。
  • 例如：消息列表、中间结果、计数器、当前步骤等。

2. Nodes（节点）​

  • 图中执行具体工作的单元（函数或可运行对象）。
  • 接收当前 State，返回要更新到 State 的字段。
  • 常见节点：
    • 调用大模型
    • 工具执行
    • 数据预处理/后处理
    • 条件判断

3. Edges（边）​

  • 连接节点，定义下一个要执行哪个节点。
  • 分为：
    • 普通边：无条件转移
    • 条件边：根据当前 State 动态决定下一个节点（比如判断是否需要调用工具）

4. Graph（图）​

  • 由节点和边组成的执行蓝图。
  • 可包含循环（这是 LangGraph 相对于普通 DAG 的核心优势）。
  • 有明确的入口点（Entry Point）和可能的终点（End）。

5. Checkpointer（检查点器）​

  • 持久化状态的机制（内存、文件、Redis、数据库等）。
  • 支持中断与恢复，实现“人机交互”或“长时间运行任务”。
  • 也是实现多轮对话记忆和回溯的基础。

6. 编译（Compile）​

  • 将定义好的图编译为可运行对象。
  • 编译时会做结构检查（如连通性、循环合法性）。
  • 编译后的图可以像普通 Runnable 一样调用：.invoke(), .stream() 等。

二、运行原理​

整体执行流程​

初始 State → 入口节点 → 更新 State → 选下一条边 → 下个节点 → ... → 到达 END → 返回最终 State

具体分为以下几个关键阶段：

1. 图构建阶段​

  • 用户定义 State 结构。
  • 添加节点（add_node）。
  • 添加边（add_edge, add_conditional_edges）。
  • 设置入口点（set_entry_point）。

2. 编译阶段​

  • 验证图结构（无悬空节点、合法循环条件）。
  • 绑定 Checkpointer（如果提供）。
  • 生成内部执行器。

3. 执行阶段（以 .invoke() 为例）​

  1. 接收输入 → 初始化 State（或从 Checkpointer 加载）。
  2. 从入口节点开始：
    • 调用当前节点的函数（传入当前 State）。
    • 节点返回更新字典 → 合并到 State。
  3. 决定下一个节点：
    • 如果是普通边 → 直接走目标节点。
    • 如果是条件边 → 执行路由函数 → 得到目标节点名（或 END）。
  4. 循环或终止：
    • 若目标节点非 END，重复步骤 2-3。
    • 若到达 END，将最终 State 写入 Checkpointer（可选），并返回给调用者。

4. 流式执行（.stream()）​

  • 在每个节点执行后 yield 中间状态。
  • 常用于展示思考链（Chain-of-Thought）或逐步输出。

三、关键机制详解​

1. 状态更新策略​

  • 节点返回的字典与当前 State 合并，而不是覆盖全部。
  • 合并规则可自定义（例如列表是追加还是替换）。
  • 这让不同节点只关注自己需要的字段，互不干扰。

2. 循环与递归限制​

  • LangGraph 允许显式循环（如 Agent 多次调用工具）。
  • 为防止无限循环，可设置 递归限制（recursion_limit）。
  • 内部通过计数器实现，达到上限后抛出异常。

3. 条件边的路由函数​

  • 输入：当前 State
  • 输出：字符串（下一个节点名）或列表（并行分支，高级用法）
  • 示例：
def route_after_model(state):
    if state.get("needs_tool"):
        return "tool_node"
    else:
        return END

4. Checkpoint 与中断​

  • 在每个节点执行前保存状态快照（若启用）。
  • 中断可由：
    • 节点内主动调用 interrupt() 函数。
    • 达到特定条件边时设置。
  • 恢复时从最后一个快照继续执行，无需重新运行已完成节点。

四、LangGraph vs. 传统 DAG 框架（如 LangChain Chain）​

特性LangGraph传统 LangChain Chain
执行流程可任意循环、分支、有条件跳转线性或有限的 DAG
状态管理显式、持久化、可回溯隐式传递，无持久化
多角色协同原生支持（多个节点各自承担不同角色）较困难
人机交互内置中断/恢复机制需手动实现
适用场景复杂 Agent、多步对话、长时间任务固定流程、单次 RAG

五、简单示例（助你理解）​

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class MyState(TypedDict):
    messages: List[str]
    step: int

def node_a(state: MyState):
    return {"messages": ["A running"]}

def node_b(state: MyState):
    return {"messages": ["B running"]}

def should_continue(state: MyState):
    if state["step"] < 3:
        return "node_b"
    else:
        return END

builder = StateGraph(MyState)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.set_entry_point("a")
builder.add_edge("a", "b")
builder.add_conditional_edges("b", should_continue)
graph = builder.compile()

result = graph.invoke({"messages": [], "step": 0})

总结一句话​

LangGraph = 状态机 + 图执行器 + 可持久化内存，它让你像画流程图一样编写具有循环、分支和记忆能力的复杂 Agent 工作流。

如果你还想了解：

  • 如何实现一个 ReAct Agent
  • Human-in-the-loop 的具体代码
  • 如何与 LangGraph Platform 或 LangSmith 集成

我可以继续为你展开讲解。