LangGraph 调研

基础组件（SDK）

GitHub - langchain-ai/langgraph: Build resilient language agents as graphs.

LangGraph 文档

LangGraph 提供基础的编排组件，可以将 LangChain 组件或自定义函数编排成 Graph（支持有环的 Graph）。每个 Graph 有开始节点和结束节点，运行的时候会在中间节点流转。

编排 Graph 的基础元素：

Graph 节点（Node）：
- 每个节点包含一个字符串表示的名称，以及一个可执行的函数（表示该节点的执行逻辑）。
- 每个节点可以有输入和输出。
Graph 边（Edge）：
- 直接流转：一个节点直接连接另一个节点，表示两个节点顺序执行。
- 条件流转：需要定义一个函数，它可以根据之前节点的执行结果返回下一个要执行的节点，用于判断节点的流转逻辑。
状态（State）：
- 表示 Graph 运行期间的状态对象，用于保存上下文、节点执行结果等。
- Graph 节点可以读写状态。比如读取消息上下文作为历史消息，调用 LLM 之后，将消息添加到消息上下文。
- Graph 边可以读取状态。比如根据上一步 LLM 的调用结果判断是否需要调用工具。

SDK 能力包括：

Graph 的构建
- Graph 组件的创建（节点、边、状态）和有效性检查。
- 子 Graph（一个 Graph 作为另一个 Graph 的节点）模式，可用于多 Agent 场景。
- 运行配置（使用的大模型、API Key 等）。
- Graph 可视化（导出构建的 Graph 为图片）。
Graph 的执行
- 状态管理。
- 响应类型：流式响应和非流式响应。
- 执行次数控制。
- 支持类似 map-reduce 的执行模式。
- 支持人为介入执行流程。
- Graph 任务重跑（Debug 场景）。
预定义的 Graph 模板或者 Graph 节点
- ReAct 模板。
- Tool 执行器。
数据持久化
- 执行步骤快照。
- 历史记忆存储。

开发工具

LangGraph Studio

LangGraph Studio

LangGraph Studio 介绍视频

GitHub - langchain-ai/langgraph-studio: Desktop app for prototyping and debugging

流程：

[LangGraph] 通过 GitHub 仓库提供初始项目模板（脚手架），包含 Graph 开发模板、单元测试模板和应用配置模板等。
[开发者] 基于项目模板开发自己的 LLM 应用（Graph 形式）。
[LangGraph] 支持展示代码对应的 Graph，并支持在页面调试。

本地客户端

LangGraph CLI

在基于 LangGraph 模板开发自己的项目后，可以通过 langgraph client 启动一个 Server 用于本地调试。

原理：拉取 LangGraph 基础镜像，然后结合基础镜像和当前的仓库生成一个新的 Server 镜像，并创建用于调试的 Server 容器。

平台产品

Deploy to Cloud

部署流程：

开发者调试完成后，将 Git 仓库上传到 GitHub。
在 LangGraph Cloud 平台创建 Deployment，选择 GitHub 仓库进行部署。
部署完成后，可通过 API 触发开发的应用。

平台没开源，可能是用类似 LangServe 的方案把应用转为 API。

平台的其它功能：

部署版本历史。
可观测性：调用 Trace 记录，以及 Token 统计。（参考 LangSmith）
LangGraph Studio 在线调试。

附录：技术细节

Graph 节点流转

LangGraph 执行算法参考了 Pregel。

Graph 每步执行：

获取下一个 superstep 节点列表（下一步需要执行的任务）。
并发执行下一步的所有节点任务，等待所有节点任务完成，或者其中一个失败。
每个节点任务执行完成后，进行 commit，调用 checkpointer 写入当前任务状态（作为 Trace）。
在这一步的所有任务都执行完成后，调用 checkpointer 写入当前 superstep 的快照。

Channel 概念：一个节点的入边（节点函数的输入）。

每个 Node 会监听它对应的 Channel，如果 Channel 被写入了值，则这个 Node 会在下一步执行。

Trace

Persistence

通过实现 BaseCheckpointSaver 接口，可以实现自定义 Trace 逻辑。

class BaseCheckpointSaver(Generic[V]):
    """Base class for creating a graph checkpointer.

    Checkpointers allow LangGraph agents to persist their state
    within and across multiple interactions.

    Note:
        When creating a custom checkpoint saver, consider implementing async
        versions to avoid blocking the main thread.
    """

    def put(self, ...) -> RunnableConfig:
        """写入一个 superstep 状态快照"""
        raise NotImplementedError

    def put_writes(self, ...) -> None:
        """写入一个 superstep 中每个节点的状态"""
        raise NotImplementedError

LangGraph 官方提供了 Postgres 版本的实现：Use Postgres checkpointer for persistence。

以基于 ReAct 模式的 Graph 为例：

      +-----------+
      | __start__ |
      +-----------+
            |
            v
      +------------+
      | call_model |
      +------------+
        |        |
        v        v
    +-------+  +---------+
    | tools |  | __end__ |
    +-------+  +---------+
        |
        v
      +------------+
      | call_model |
      +------------+

不调用工具时的 Trace 示例：

[start]
call_model
[end]

调用一个工具时的 Trace 示例：

[start]
call_model
tools
call_model
[end]

@whichxjy