langgraph4j 介绍

简介

LangGraph4j 是一个用于在 Java 中构建有状态的、多智能体（Multi-Agent）应用程序的库，与大型语言模型（LLM）深度集成。它灵感来源于 Python 的 LangGraph 库，旨在与 LangChain4j 和 Spring AI 等主流 Java LLM 框架无缝协作。

从本质上讲，LangGraph4j 允许你定义循环图（Cyclical Graphs），让不同的组件（智能体、工具或自定义逻辑）以有状态的方式进行交互。这对于构建需要记忆、上下文以及不同”智能体”之间协作或任务交接的复杂应用程序至关重要。

本文参考自官网：https://langgraph4j.github.io/langgraph4j/main/getting-started/

---

核心特性与优势

LangGraph4j 提供了以下特性和优势：

• 有状态执行（Stateful Execution）：在图的节点之间管理和更新共享状态，实现记忆和上下文感知。
• 循环图（Cyclical Graphs）：与传统的有向无环图（DAG）不同，LangGraph4j 支持循环，这对于需要控制流回退的智能体架构至关重要（例如智能体重试任务或请求澄清）。
• 显式控制流（Explicit Control Flow）：明确定义图中节点之间的路径和转换条件。
• 模块化（Modularity）：使用更小、可复用的组件（节点）构建复杂系统。
• 灵活性（Flexibility）：可与各种 LLM 提供商和自定义 Java 逻辑集成。
• 可观测性与调试（Observability & Debugging）：
  • 检查点（Checkpoints）：在任意时刻保存图的状态，并在后续进行回放或检查。这对于调试和理解复杂交互非常宝贵。
  • 图可视化（Graph Visualization）：使用 PlantUML 或 Mermaid 生成图的可视化表示，以理解其结构。
• 异步与流式支持（Asynchronous & Streaming Support）：使用非阻塞操作和 LLM 流式输出构建响应式应用程序。
• Playground & Studio：一个 Web UI，用于可视化检查、运行和调试你的图。

---

核心概念详解

有效使用 LangGraph4j 的关键在于理解以下概念：

StateGraph<S extends AgentState>

StateGraph 是你用来定义应用程序结构的主要类。你可以在其中添加节点和边来创建图。它由 AgentState 参数化。

AgentState

AgentState（或其子类）代表图的共享状态。它本质上是一个 Map<String, Object>，在节点之间传递。每个节点都可以读取此状态并返回对其的更新。

• Schema（模式）：状态的结构由一个”schema”定义，即一个 Map<String, Channel.Reducer>。映射中的每个键对应状态中的一个属性。
• Channel.Reducer（归约器）：Reducer 定义了如何处���状态属性的更新。例如，新值可能会覆盖旧值，也可能会被添加到现有值的列表中。
• Channel.Default<T>：为状态属性提供默认值（如果尚未设置）。
• Channel.Appender<T> / MessageChannel.Appender<M>：一种常见的 Reducer 类型，将新值追加到与状态属性关联的列表中。这对于累积消息、工具调用或其他数据序列非常有用。MessageChannel.Appender 专为聊天消息设计，还支持按 ID 删除消息。

节点（Nodes）

节点是执行操作的图的基本构建块。节点通常是一个函数（或实现 NodeAction<S> 或 AsyncNodeAction<S> 的类），它：

1. 接收当前的 AgentState 作为输入。
2. 执行一些计算（例如调用 LLM、执行工具、运行自定义业务逻辑）。
3. 返回一个 Map<String, Object> 表示对状态的更新。然后根据 schema 的 Reducer 将这些更新应用到 AgentState 中。

节点可以是同步的，也可以是异步的（CompletableFuture）。

边（Edges）

边定义了节点之间的控制流。

• 普通边（Normal Edges）：从一个节点到另一个节点的无条件转换。节点 A 完成后，控制流始终传递给节点 B。使用 addEdge(sourceNodeName, destinationNodeName) 定义。
• 条件边（Conditional Edges）：下一个节点根据当前 AgentState 动态确定。源节点完成后，会执行一个 EdgeAction<S>（或 AsyncEdgeAction<S>）函数。该函数接收当前状态并返回要执行的下一个节点的名称。这允许分支逻辑（例如，如果智能体决定使用工具，则转到”execute_tool”节点；否则转到”respond_to_user”节点）。使用 addConditionalEdges(...) 定义条件边。
• 入口点（Entry Points）：你还可以使用 addConditionalEntryPoint(...) 定义图的条件入口点。

编译（Compilation）

在 StateGraph 中定义完所有节点和边后，你可以通过 compile() 将其编译为 CompiledGraph<S extends AgentState>。这个编译后的图是你的逻辑的不可变、可运行的表示。编译过程会验证图的结构（例如检查是否有孤立节点）。

检查点（Checkpoints / 持久化）

LangGraph4j 允许你在任意步骤保存图的状态（Checkpoint）。这对于以下场景非常有用：

• 调试：检查各个时间点的状态以了解发生了什么。
• 恢复：将图恢复到先前的状态并继续执行。
• 长时间运行的进程：持久化长时间运行的智能体交互的状态。

通常你会使用 CheckpointSaver 的实现（例如用于内存存储的 MemorySaver，或自己实现持久化存储）。

---

安装

要在项目中使用 LangGraph4j，你需要将其添加为依赖。

Maven

确保你使用的是 Java 17 或更高版本。

最新稳定版本（推荐）：

<properties>
    <langgraph4j.version>1.8.20</langgraph4j.version> <!-- 请检查实际最新版本 -->
</properties>

<!-- 可选：添加 Bill of Materials (BOM) 来管理 langgraph4j 模块版本 -->
<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.bsc.langgraph4j</groupId>
            <artifactId>langgraph4j-bom</artifactId>
            <version>${langgraph4j.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<dependencies>
    <dependency>
        <groupId>org.bsc.langgraph4j</groupId>
        <artifactId>langgraph4j-core</artifactId>
    </dependency>
    <!-- 根据需要添加其他 langgraph4j 模块，例如 langgraph4j-langchain4j -->
</dependencies>

💡 请始终在 Maven Central 仓库中检查最新版本号。

开发快照版本：

如果你想使用最新的未发布功能，可以使用快照版本：

<dependency>
    <groupId>org.bsc.langgraph4j</groupId>
    <artifactId>langgraph4j-core</artifactId>
    <version>1.8.20</version> <!-- 或当前的快照版本 -->
</dependency>

你可能需要配置 settings.xml 或 pom.xml 以包含 Sonatype OSS 快照仓库：

<repositories>
    <repository>
        <id>sonatype-oss-snapshots</id>
        <url>https://central.sonatype.com/repository/maven-snapshots</url>
        <snapshots>
            <enabled>true</enabled>
        </snapshots>
    </repository>
</repositories>

---

你的第一个图 — 简单示例

让我们创建一个非常简单的图，它包含两个节点：greeter 和 responder。

• greeter 节点将向状态中添加一个问候消息。
• responder 节点将根据问候语添加一条响应消息。

1. 定义状态

我们的状态将保存一个消息列表。

import org.bsc.langgraph4j.state.AgentState;
import org.bsc.langgraph4j.state.Channels;
import org.bsc.langgraph4j.state.Channel;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;

// 为我们的图定义状态
class SimpleState extends AgentState {
    public static final String MESSAGES_KEY = "messages";

    // 定义状态的 schema
    // MESSAGES_KEY 将保存一个字符串列表，新消息会被追加到列表中
    public static final Map<String, Channel<?>> SCHEMA = Map.of(
        MESSAGES_KEY, Channels.appender(ArrayList::new)
    );

    public SimpleState(Map<String, Object> initData) {
        super(initData);
    }

    public List<String> messages() {
        return this.<List<String>>value("messages")
            .orElse(List.of());
    }
}

2. 定义节点

import org.bsc.langgraph4j.action.NodeAction;
import java.util.Map;

// 添加问候语的节点
class GreeterNode implements NodeAction<SimpleState> {
    @Override
    public Map<String, Object> apply(SimpleState state) {
        System.out.println("GreeterNode 正在执行。当前消息: " + state.messages());
        return Map.of(SimpleState.MESSAGES_KEY, "Hello from GreeterNode!");
    }
}

// 添加响应的节点
class ResponderNode implements NodeAction<SimpleState> {
    @Override
    public Map<String, Object> apply(SimpleState state) {
        System.out.println("ResponderNode 正在执行。当前消息: " + state.messages());
        List<String> currentMessages = state.messages();
        if (currentMessages.contains("Hello from GreeterNode!")) {
            return Map.of(SimpleState.MESSAGES_KEY, "Acknowledged greeting!");
        }
        return Map.of(SimpleState.MESSAGES_KEY, "No greeting found.");
    }
}

3. 定义并编译图

import org.bsc.langgraph4j.StateGraph;
import org.bsc.langgraph4j.GraphStateException;
import static org.bsc.langgraph4j.action.AsyncNodeAction.node_async;
import static org.bsc.langgraph4j.StateGraph.START;
import static org.bsc.langgraph4j.StateGraph.END;

public class SimpleGraphApp {

    public static void main(String[] args) throws GraphStateException {
        // 初始化节点
        GreeterNode greeterNode = new GreeterNode();
        ResponderNode responderNode = new ResponderNode();

        // 定义图结构
        var stateGraph = new StateGraph<>(SimpleState.SCHEMA, initData -> new SimpleState(initData))
            .addNode("greeter", node_async(greeterNode))
            .addNode("responder", node_async(responderNode))
            // 定义边
            .addEdge(START, "greeter")       // 从 greeter 节点开始
            .addEdge("greeter", "responder")  // greeter → responder
            .addEdge("responder", END)        // responder 之后结束
        ;

        // 编译图
        var compiledGraph = stateGraph.compile();

        // 运行图
        for (var item : compiledGraph.stream(
                Map.of(SimpleState.MESSAGES_KEY, "Let's begin!"))) {
            System.out.println(item);
        }
    }
}

说明

• 我们定义了 SimpleState，其中 MESSAGES_KEY 使用 AppenderChannel 来累积字符串。
• GreeterNode 添加一条”Hello”消息。
• ResponderNode 检查问候语并添加确认消息。
• StateGraph 定义了节点，边指定了流程：START → greeter → responder → END。
• stateGraph.compile() 创建可运行的 CompiledGraph。
• compiledGraph.stream(initialState) 执行图。我们遍历流以获取最终状态。流中的每一项代表节点执行后的状态。

这个示例演示了基本工作流程：定义状态 → 定义节点 → 用边连接 → 编译 → 运行。

---

运行你的图

如示例所示，通常使用以下执行方法之一来运行编译后的图：

• stream(S initialState, RunnableConfig config)：执行图并返回一个 AsyncGenerator<S>。每个产出项是节点完成后的状态 S。这对于观察每个步骤的状态或流式传输部分结果非常有用。
• invoke(S initialState, RunnableConfig config)：执行图并返回一个 CompletableFuture<S>，在图到达 END 节点后以最终状态完成。

RunnableConfig 可用于传递运行时配置。

---

Studio 🤩 — 可视化运行你的图

LangGraph4j Studio 是一个可嵌入的 Web 应用程序，用于可视化和实验你的图：

• 实时监控：查看节点执行状态和状态变更过程
• 断点调试：在任意节点设置断点，检查中间状态
• 流程回放：通过时间旅行功能回溯任意执行步骤

前往 Studio 探索 LangGraph4j Studio。

---

内置集成

LangChain4j

作为 LangChain4j 集成的默认用例，我们使用 LangGraph4j 实现了 AgentExecutor（也称为 ReACT Agent）。在项目的模块中，你可以找到完整的工作代码和测试。欢迎参考使用。

定义工具

public class TestTool {

    @Tool("tool for test AI agent executor")
    String execTest(@P("test message") String message) {
        return String.format("test tool ('%s') executed with result 'OK'", message);
    }

    @Tool("return current number of system thread allocated by application")
    int threadCount() {
        return Thread.getAllStackTraces().size();
    }
}

运行智能体

var model = OllamaChatModel.builder()
    .modelName("qwen2.5:7b")
    .baseUrl("http://localhost:11434")
    .supportedCapabilities(Capability.RESPONSE_FORMAT_JSON_SCHEMA)
    .logRequests(true)
    .logResponses(true)
    .maxRetries(2)
    .temperature(0.0)
    .build();

var agent = AgentExecutor.builder()
    .chatModel(model)
    .toolsFromObject(new TestTool())
    .build()
    .compile();

for (var item : agent.stream(Map.of("messages",
        "perform test twice and return number of current active threads"))) {
    System.out.println(item);
}

Spring AI

作为 Spring AI 集成的默认用例，我们同样使用 LangGraph4j 实现了 AgentExecutor（ReACT Agent）。

定义工具

public class TestTool {

    @Tool(description = "tool for test AI agent executor")
    String execTest(@ToolParam(description = "test message") String message) {
        return String.format("test tool ('%s') executed with result 'OK'", message);
    }

    @Tool(description = "return current number of system thread allocated by application")
    int threadCount() {
        return Thread.getAllStackTraces().size();
    }
}

运行智能体

var model = OllamaChatModel.builder()
    .ollamaApi(OllamaApi.builder().baseUrl("http://localhost:11434").build())
    .defaultOptions(OllamaOptions.builder()
        .model("qwen2.5:7b")
        .temperature(0.1)
        .build())
    .build();

var agent = AgentExecutor.builder()
    .chatModel(model)
    .toolsFromObject(new TestTool())
    .build()
    .compile();

for (var item : agent.stream(Map.of("messages",
        "perform test twice and return number of current active threads"))) {
    System.out.println(item);
}

---

核心能力概览

LangGraph4j 配备了构建复杂智能体应用程序的丰富功能：

能力	说明
异步操作	节点和边可以是异步的（返回 CompletableFuture），支持非阻塞 I/O 操作，尤其适用于 LLM 调用场景
流式传输	原生支持通过节点从 LLM 流式传输响应，实现实时输出
检查点（持久化与时间旅行）	保存和加载图的状态，支持恢复长时间运行的任务、通过检查中间状态进行调试，甚至”时间旅行”到先前的状态
图可视化	生成图的 PlantUML 或 Mermaid 图表，可视化其结构，有助于理解和调试
Playground & Studio	LangGraph4j 附带一个可嵌入的 Web UI（Studio），允许你实时可视化、运行和与图进行交互
子图（Subgraphs）	通过在父图节点中嵌套更小、可复用的图来组合复杂图，促进模块化和可复用性
并行执行	配置图的部分以并行执行多个节点，提升可并发运行任务的性能
线程（多轮对话）	在单个图实例中管理不同的并行执行线程，每个线程都有自己的检查点历史。这对于同时处理多个用户会话或对话至关重要