BytesRouter

主题

LangChain4j 文档版本 1.22.0 源快照 e98c0e1c 更新于 2026-10-14

非阻塞与响应式

NOTE

异步与响应式支持尚属实验性。本页上的异步与响应式 API 标注了 @Experimental；API 与行为在未来版本中仍可能变化。同步与 TokenStream API 不受影响。

默认情况下，调用一个 AI 服务会阻塞调用线程整个交互过程：LLM 调用、工具执行、聊天记忆访问与护栏都在方法返回之前发生。这很简单，且对大多数应用都行之有效。

如果你构建在响应式技术栈（Quarkus/Mutiny、Vert.x、Spring WebFlux）之上，或需要一个线程驱动多个并发交互，LangChain4j 能在不阻塞的情况下运行同样的交互。你只需在你的 AI 服务方法上声明一个不同的返回类型来抉择——接口的其他一切都不变。

四种模式

返回类型性质
String、一个 POJO、Result<T>……同步，阻塞调用线程
TokenStream通过回调流式
CompletableFuture<T>、CompletionStage<T>单个响应，非阻塞
Flow.Publisher<AiServiceStreamingEvent>、Flow.Publisher<String>一个事件流，非阻塞
java
interface Assistant {

    // synchronous
    String chat(String message);

    // one response, without blocking the caller
    CompletableFuture<String> chatAsync(String message);

    // the answer, streamed token by token
    Flow.Publisher<String> chatStreaming(String message);

    // everything that happens during the interaction, as events
    Flow.Publisher<AiServiceStreamingEvent> chatEvents(String message);
}

在不阻塞的情况下获得单个响应

java
CompletableFuture<String> future = assistant.chatAsync("Tell me a joke");

future.thenAccept(System.out::println);

取消该 future（future.cancel(true)）会释放调用方、停止后续的 LLM 轮次，并尽力尝试中止正在进行的 HTTP 调用。

NOTE

该 future 在模型自身的线程上完成——对于 HTTP 模型，是读取响应的传输层 I/O 工作线程。因此未显式指定执行器而附加的续接（thenApply、thenAccept……）会在该线程上运行，在那里阻塞会拖累所有在途调用的吞吐。请保持续接非阻塞，或传入你自己的 Executor：future.thenApplyAsync(fn, executor)。

流式输出答案

一个返回 Flow.Publisher<String> 的方法会逐块流式输出答案的文本：

java
Flow.Publisher<String> publisher = assistant.chatStreaming("Tell me a joke");

该 publisher 是冷的（cold）：在你订阅之前什么都不会发生，且每次订阅都会启动一次新的交互。

流式呈现所发生的一切

Flow.Publisher<AiServiceStreamingEvent> 呈现整个交互过程，而不仅是答案文本——即基于回调的 TokenStream 给你的那些信息：

事件含义
PartialResponseEvent答案的一个块
PartialThinkingEvent模型推理的一个块
PartialToolCallEvent、CompleteToolCallEvent一个正在拼装、随后完成的工具调用
BeforeToolExecutionEvent、AfterToolExecutionEvent一个即将运行的工具，及其结果
IntermediateResponseEvent结束某轮工具调用的那个响应
RetrievedContentsEvent由 RAG 检索到的内容
ToolCompensatedEvent一个已完成的工具在交互被取消或失败后得到了补偿
RawEvent一个 LangChain4j 未映射的提供者专有事件
FinalResponseEvent最终答案
java
assistant.chatEvents("What is the weather in Munich?").subscribe(new Flow.Subscriber<>() {

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        subscription.request(Long.MAX_VALUE);
    }

    @Override
    public void onNext(AiServiceStreamingEvent event) {
        // the event types are nested in AiServiceStreamingEvent:
        // import dev.langchain4j.service.AiServiceStreamingEvent.PartialResponseEvent;
        if (event instanceof PartialResponseEvent partial) {
            System.out.print(partial.partialResponse().text());
        }
    }

    @Override
    public void onError(Throwable error) { }

    @Override
    public void onComplete() { }
});

未来可能新增事件类型，因此请优雅地处理无法识别的事件——不要写一个没有默认分支的穷举类型切换。

NOTE

不要在 onNext 中阻塞。事件在产生它们的那个线程上投递：词元级事件在模型的传输 I/O 工作线程上，工具执行事件在完成某个工具调用的那个线程上。请把繁重的逐事件工作卸载到你自己的 Executor。事件通过一个有界缓冲区中转，因此落后太多的订阅者会以 IllegalStateException 终止，而不是无限制地缓冲；该大小默认为 16384 个事件，可用 AiServices.builder(...).streamingBufferSize(int) 配置。

第三方响应式类型

AI 服务方法返回 JDK 类型——CompletableFuture 与 Flow.Publisher——因此该 API 不会把你绑到某个特定的响应式编程库。

Reactor 类型通过 langchain4j-reactor 模块支持——单响应模式用 Mono<T>，响应式模式用 Flux<AiServiceStreamingEvent>。添加依赖即可；适配器通过 ServiceLoader 自行注册。

java
interface Assistant {

    Mono<String> answer(String userMessage);

    Flux<AiServiceStreamingEvent> events(String userMessage);
}

NOTE

Mutiny 的 Uni/Multi 尚不支持。接入点是现成的——通过 ServiceLoader 发现的 CompletableFutureAdapter 与 PublisherAdapter SPI——但没有为它们提供适配器，因此声明这样的返回类型目前会失败。请使用 CompletableFuture 或 Flow.Publisher，并在调用处做转换。

Flux<String> 由同一模块中较早的、基于 TokenStream 的适配器提供——参见 AI 服务。它早于本特性，不属于此处所述的非阻塞路径，且对所有提供者都有效。

哪些必须是阻塞的

非阻塞必须在每一层都成立：任何一处有阻塞步骤都会让整个调用重新变为阻塞。每一层在其现有阻塞方法旁边都有一个异步对应物：

层阻塞非阻塞对应物
聊天模型chat(ChatRequest)chatAsync(ChatRequest)，以及 StreamingChatModel 上返回 Publisher 的 chat(ChatRequest)
嵌入模型embed(...)embedAsync(...)
打分模型scoreAll(...)scoreAsync(...)
聊天记忆add、messages、setaddAsync、messagesAsync、setAsync
聊天记忆存储getMessages、updateMessages、deleteMessagesgetMessagesAsync、updateMessagesAsync、deleteMessagesAsync
护栏validate(...)validateAsync(...)
工具ToolExecutor.execute(...)ToolExecutor.executeAsync(...)
RAGRetrievalAugmentor.augment(...)、检索器、路由器、聚合器、查询转换器augmentAsync(...)、retrieveAsync(...)、routeAsync(...)、aggregateAsync(...)、transformAsync(...)
嵌入存储search(...)searchAsync(...)
网页搜索search(...)searchAsync(...)
MCPMcpClient.executeTool(...)executeToolAsync(...)
护栏执行ChatExecutor.execute(...)ChatExecutor.executeAsync(...)
HTTP 客户端execute(...)executeAsync(...)、stream(...)

一个尚未实现其对应物的组件会响亮地失败，而不是悄悄阻塞：返回的 future 或 publisher 会以一个点名该组件与缺失方法的 AsyncNotSupportedException 失败。AsyncNotSupportedException 是一个 UnsupportedFeatureException，因此一个 catch 子句就能覆盖"不支持"的两种情形，且它绝不会被重试。

无法避免的阻塞代码

工具是常见情形：一个调用数据库或阻塞 HTTP API 的工具无法变为非阻塞。这类工具会被卸载（offload），以免阻塞模型自身的线程。

NOTE

在 Java 21 及更新版本上，卸载执行器创建虚拟线程，因此一个阻塞的工具会挂起一个虚拟线程，它从其载体线程卸载而非占用它。LangChain4j 以 Java 17 为目标，而在 Java 17-20 上，同一个执行器是一个无界平台线程池——在负载下是截然不同的资源特征。如果这对你重要，请提供你自己的有界 Executor（见下文）。

这适用于 @Tool 注解的方法。一个手写的 ToolExecutor 若不覆写 executeAsync，则会像任何未选择接入的组件那样，以 AsyncNotSupportedException 响亮失败。

工具在异步与响应式模式下默认并发运行。要让它们一次只运行一个，请传入一个单线程执行器：

java
AiServices.builder(Assistant.class)
        .chatModel(model)
        .tools(new MyTools())
        .executeToolsConcurrently(Executors.newSingleThreadExecutor())
        .build();

对于 RAG，一个未实现 retrieveAsync 的检索器默认会失败，而不是悄悄阻塞。请改在 DefaultRetrievalAugmentor 或 EmbeddingStoreContentRetriever 上用 offloadBlocking(true) 选择卸载。

与同步模式不同的默认值

异步与响应式模式不只是把同一交互换到另一个线程上跑——有几个默认值是有意不同的。如果你在迁移一个已有方法，这些就是需要检查的地方。

同步 / TokenStreamCompletableFuture / Flow.Publisher
多个工具调用顺序执行并发执行
工具执行错误送回给 LLM让调用失败，除非该异常实现了 ToolErrorVisibleToLlm
工具参数解析错误让调用失败送回给 LLM
@Moderate支持在 AI 服务创建时被拒绝

工具错误

这两种工具错误的默认值是有意相反的。把一个执行失败送给 LLM 会对你隐瞒一个你工具里的 bug，并诱导模型编造一个答案来绕过它，所以异步模式会让调用失败，除非该异常本身表明可以让 LLM 知道什么（见 ToolErrorVisibleToLlm）。另一方面，一个畸形的参数字符串是模型产生的东西，被告知后它通常能修正，所以会被送回，而不是让调用失败。

二者都可配置，且一个显式配置的处理器会被每种模式使用：

java
AiServices.builder(Assistant.class)
        .chatModel(model)
        .tools(new MyTools())
        .toolExecutionErrorHandler(myExecutionErrorHandler)
        .toolArgumentsErrorHandler(myArgumentsErrorHandler)
        .build();

关于这些处理器能做什么，参见错误处理。

工具并发

因为在这些模式下工具总是在一个 Executor 上运行，一次 LLM 响应中的多个工具调用会同时运行。如果你的工具不适合并发运行，请传入一个单线程执行器：executeToolsConcurrently(Executors.newSingleThreadExecutor())。

事件顺序

在响应式路径上，一个工具在其 CompleteToolCallEvent 被发出后就立即开始，因此它会与该轮的其余部分重叠。所以某轮的 BeforeToolExecutionEvent / AfterToolExecutionEvent 可能在该轮的 IntermediateResponseEvent 之前到达。请把 IntermediateResponseEvent 当作一个逐轮标记，而非一道所有该轮工具事件都必先于它的屏障。基于回调的 TokenStream 总是先报告中间响应。

控制执行器与传播上下文

凡 LangChain4j 在调用线程之外运行工作之处——并发工具调用、卸载的检索、重试退避——它都从一个可插拔的接入点获取执行器，即 ExecutorProvider SPI：

java
public interface ExecutorProvider {

    Executor executor();

    default CapturedContext captureContext() {
        return CapturedContext.NONE;
    }
}

通过 ServiceLoader 注册一个，或为测试与非-DI 应用以编程方式注册：

java
ExecutorProvider.set(() -> myExecutor);

若没有它，LangChain4j 在 Java 21 及更新版本上使用一个每任务一虚拟线程的执行器，在 Java 17-20 上使用一个无界平台线程池。

NOTE

单次调用会跨越多个线程，因此环境性的 ThreadLocal 状态——MDC 日志上下文、追踪 span、安全上下文——不会像完全同步模式那样被自动传播。要让它们随工作走，请实现 captureContext()。LangChain4j 每次调用会在调用 AI 服务方法的那个线程上调用它一次，并让该调用卸载给执行器的每个任务（工具、阻塞的 RAG 阶段、审核）都带着被捕获的上下文运行。用 ExecutorProvider.set(() -> myExecutor) 注册的 lambda 做不到这一点，因此请改注册一个类，例如配合 Micrometer 上下文传播：

java
import dev.langchain4j.spi.CapturedContext;
import dev.langchain4j.spi.ExecutorProvider;
import io.micrometer.context.ContextSnapshotFactory;
import java.util.concurrent.Executor;

public class ContextPropagatingExecutorProvider implements ExecutorProvider {

    private static final ContextSnapshotFactory CONTEXT_SNAPSHOT_FACTORY = ContextSnapshotFactory.builder().build();

    @Override
    public Executor executor() {
        return null; // keeps the built-in default
    }

    @Override
    public CapturedContext captureContext() {
        return CONTEXT_SNAPSHOT_FACTORY.captureAll()::wrap;
    }
}
java
ExecutorProvider.set(new ContextPropagatingExecutorProvider());

使用 OpenTelemetry 时，captureContext() 是 return Context.current()::wrap;。使用 MicroProfile Context Propagation 时：

java
private static final ThreadContext THREAD_CONTEXT = ThreadContext.builder().build();

@Override
public CapturedContext captureContext() {
    Executor callerContext = THREAD_CONTEXT.currentContextExecutor();
    return task -> () -> callerContext.execute(task);
}

captureContext() 每次调用都会被调用，所以它必须很廉价。它返回的 CapturedContext 可能包装多个同时运行的任务，且每个被包装的任务在完成时必须恢复它所运行线程的先前上下文：任务在池化线程上运行，因此遗留的上下文会泄漏到无关的任务中。上面的例子都遵循了这些规则。

有几点需要留意：

  • 单有一个传播上下文的执行器（一个 ManagedExecutor、一个用 TaskDecorator 包装的执行器、Context.taskWrapping(executor)……）是不够的：它捕获的是提交任务那个线程的上下文，而大多数卸载工作是在模型回答之后、由投递答案的那个线程提交的。它可以与 captureContext() 结合：这样任务就带着被捕获的上下文运行。
  • LangChain4j 不卸载的工作——护栏、聊天记忆、监听器、流式回调——在当前线程上运行，得不到被捕获的上下文。
  • 在 AI 服务调用之外卸载的工作也得不到它：重试退避、模型集成自行启动的线程（例如为读取流式响应的线程），以及 langchain4j-agentic 的并行与异步智能体。如果你在那里需要上下文，请把 captureContext() 与一个传播上下文的执行器结合。
  • 对于返回 Flow.Publisher、Mono 或 Flux 的方法，上下文是在方法被调用时捕获的，而非在结果被订阅时。
  • 被捕获的上下文会保留整个调用期间，因此卸载的工作可能在启动该调用的请求完成之后仍带着它运行（例如带着一个已经结束会话的安全上下文）。
  • 只有 ExecutorProvider 会捕获上下文。在传给 AI 服务的执行器上（用 executeToolsConcurrently(executor)）运行的工具，仍带着它捕获的上下文运行。
  • 这适用于用 AiServices 创建的 AI 服务。一个构建自己 AI 服务实现的框架，自行决定它如何传播上下文。

InvocationContext 不受影响——它是作为一个参数显式传递的，从不通过线程局部变量。

模型监听器不得阻塞

ChatModelListener 与 EmbeddingModelListener 回调在模型自身的线程上同步调用，从不被卸载。在异步与响应式 API 上，那就是传输层的 I/O 工作线程。一个在那里做阻塞 I/O 的监听器会拖住那个工作线程，拖累所有在途调用的吞吐——请在回调内部把这类工作卸载到你自己的执行器。参见可观测性。

提供者支持

WARNING

非阻塞模式只对已实现了它们的提供者有效。 对任何其他提供者声明一个 CompletableFuture 或 Flow.Publisher 返回类型，能通过编译，然后在运行时以一个点名该组件与缺失方法的 AsyncNotSupportedException 失败。不会静默回退到阻塞调用——这正是用意所在，但这意味着你能用的返回类型取决于你的提供者。

支持是按提供者逐个选择接入的，正在逐步推出。目前：

提供者CompletableFuture（chatAsync）Flow.Publisher（响应式 chat）
OpenAI — Chat Completions✅✅
OpenAI — Responses✅✅
Anthropic✅✅
Bedrock✅✅
其他每个提供者❌❌

除聊天模型外：OpenAI 实现了异步嵌入（embedAsync）、Cohere 异步打分（scoreAsync）与 Tavily 异步网页搜索（searchAsync）。

一个尚未选择接入的提供者在同步与 TokenStream API 上不受影响——那些完全照旧工作。

同样的道理适用于调用中的其他一切

一个支持该模式的聊天模型是必要条件但非充分条件：交互中任何一处一个阻塞组件都会以同样的方式让调用失败。实践中这意味着：

组件它必须实现自带的实现
聊天记忆addAsync / messagesAsync / setAsyncMessageWindowChatMemory 与 InMemoryChatMemoryStore 已经实现
护栏validateAsync无——你写的护栏必须覆写它，即使它不做 I/O
内容检索器、查询路由器、聚合器retrieveAsync、routeAsync、aggregateAsync改用 offloadBlocking(true) 选择卸载
嵌入存储searchAsync同上，经由检索器的 offloadBlocking(true)
自定义 ToolExecutorexecuteAsync@Tool 注解的方法会为你被卸载
聊天模型路由器（RoutingChatModel）routeAsyncDecisionModelChatModelRouter 实现了，只要它的决策模型支持 decideAsync

一个不做任何阻塞工作的护栏，用一行就满足了契约：

java
@Override
public CompletableFuture<InputGuardrailResult> validateAsync(InputGuardrailRequest request) {
    return CompletableFuture.completedFuture(validate(request));
}

Spring Boot

Flux<String> 对每个提供者都继续工作，包括 ❌ 行里的那些：它由 langchain4j-reactor 中基于 TokenStream 的适配器提供，而非本页所述的非阻塞路径。

Mono<T> 与 Flux<AiServiceStreamingEvent> 来自 langchain4j-reactor 模块，并携带与 JDK 类型相同的提供者约束——见上文的第三方响应式类型。

使用 OpenAI starter 时，非阻塞模式会通过 Spring 的 WebClient 发送，因此它们需要它在 classpath 上（Spring Boot 4 上的 spring-boot-starter-webclient，Spring Boot 3 上的 spring-webflux）。关于该添加哪个依赖、以及哪些定制生效，参见 Spring Boot 集成。

要让环境上下文跟随一个异步调用，请让 LangChain4j 卸载到应用自己的执行器，而非它的默认执行器：

properties
langchain4j.executor.use-spring-task-executor=true

随后 LangChain4j 会在应用的任务执行器上运行卸载的工作，其线程池遵循 spring.task.execution.*。它默认关闭，因为该设置是进程范围的，而非限定到单个应用上下文。

从 Spring Boot starters 的 1.23.0-beta33 版本起，调用方的上下文也会被捕获（见上文），从而使追踪 span、MDC 与安全上下文跟随 LangChain4j 卸载的工作：

  • 如果应用定义了 TaskDecorator bean（例如一个 ContextPropagatingTaskDecorator），它们定义了被捕获的内容，就如同应用自己的任务执行器那样（在 Spring Boot 3 上，仅当恰好有一个时）；
  • 否则，若 Micrometer 上下文传播在 classpath 上，就用它。

友情链接

  • Agents.md面向 AI 编码助手的仓库说明文件规范
  • LangChainLangChain 是一个开放的Agent平台
  • LangSmithAgent 工程平台：可观测与评估、部署、Fleet、Engine 等
  • EmbabelEmbabel 是一个JVM平台Agent框架
  • SpringAISpringAI 是一个基于 Spring 的 AI 框架
  • MCPModel Context Protocol，连接 AI 应用与外部数据/工具的开放标准
  • LangChain4jJava 生态的 LLM 应用开发框架
  • LlamaIndex把数据接入 LLM 的框架与文档平台
  • LlamaParse智能体式 OCR 文档解析：把 PDF、扫描件变成 LLM 可用的文本与结构化数据
  • LiteParse本地优先、轻量的文档解析库，不需要 API key
  • CrewAI多智能体编排框架：以 Crew/Flow 组织自主协作的 AI Agent
  • MAFMicrosoft Agent Framework，企业级 .NET/Python 智能体框架

BytesRouter 是社区翻译站点，所有译文仅供参考，一切以官方英文原文为准。

© 2026 BytesRouter · 译文采用 CC BY 4.0 许可