跳到主要内容

Non-blocking and Reactive（ノンブロッキングとリアクティブ）

备注

非同期およびリアクティブのサポートは実験的なものです。このページの非同期およびリアクティブ API には @Experimental アノテーションが付いています。 API や動作は今後のリリースで変更される可能性があります。同期 API と TokenStream API には影響しません。

デフォルトでは、AI Service の呼び出しは、やり取り全体の間、呼び出しスレッドをブロックします。LLM の呼び出し、ツールの 実行、チャットメモリへのアクセス、ガードレールのすべてが、メソッドが戻る前に行われます。この方式はシンプルで、 ほとんどのアプリケーションでうまく機能します。

リアクティブなスタック（Quarkus/Mutiny、Vert.x、Spring WebFlux）上で構築している場合や、1 つのスレッドで 多数の同時のやり取りを処理する必要がある場合、LangChain4j は同じやり取りをブロックせずに実行できます。AI Service のメソッドで 異なる戻り値の型を宣言するだけで選択できます。インターフェースの他の部分は何も変更する必要はありません。

4 つのモード​

戻り値の型性質
String、POJO、Result<T>、…同期。呼び出しスレッドをブロックする
TokenStreamコールバックによるストリーミング
CompletableFuture<T>、CompletionStage<T>単一のレスポンス。ノンブロッキング
Flow.Publisher<AiServiceStreamingEvent>、Flow.Publisher<String>イベントのストリーム。ノンブロッキング
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);
}

ブロックせずに 1 つのレスポンスを取得する​

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

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

future をキャンセルする（future.cancel(true)）と、呼び出し元が解放され、以降の LLM のラウンドが停止し、 実行中の HTTP 呼び出しの中断をベストエフォートで試みます。

备注

future は、モデル自身のスレッド（HTTP モデルの場合は、レスポンスを読み取る トランスポートの I/O ワーカー）で完了します。そのため、明示的な executor なしで追加された継続（thenApply、thenAccept など）は、 そのスレッドで実行され、そこでブロックすると、実行中のすべての呼び出しのスループットが低下します。継続はノンブロッキングに保つか、 独自の Executor を渡してください: future.thenApplyAsync(fn, executor)。

回答をストリーミングする​

Flow.Publisher<String> を返すメソッドは、回答のテキストをチャンクごとにストリーミングします。

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

publisher は**コールド（cold）**です。サブスクライブするまで何も起こらず、サブスクリプションごとに新しいやり取りが開始されます。

起きることすべてをストリーミングする​

Flow.Publisher<AiServiceStreamingEvent> は、回答のテキストだけでなく、やり取り全体を公開します。これは、コールバックベースの TokenStream が 提供するのと同じ情報です。

イベント意味
PartialResponseEvent回答のチャンク
PartialThinkingEventモデルの推論のチャンク
PartialToolCallEvent、CompleteToolCallEvent組み立て中のツール呼び出しと、その完了
BeforeToolExecutionEvent、AfterToolExecutionEvent実行される直前のツールと、その結果
IntermediateResponseEventツール呼び出しの 1 ラウンドを終えたレスポンス
RetrievedContentsEventRAG で取得されたコンテンツ
ToolCompensatedEventやり取りがキャンセルまたは失敗した後に、完了済みのツールが補償されたこと
RawEventLangChain4j がマッピングしていない、プロバイダー固有のイベント
FinalResponseEvent最終的な回答
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() { }
});

新しいイベントタイプが今後追加される可能性があるため、認識できないイベントも適切に処理してください。default 分岐のない 網羅的な型 switch は書かないでください。

备注

onNext の中でブロックしないでください。イベントは、それを生成したスレッドで配信されます。トークンレベルのイベントではモデルのトランスポートの I/O ワーカー、 ツール実行のイベントではツール呼び出しを完了したスレッドです。重い イベントごとの処理は、独自の Executor にオフロードしてください。イベントは有界バッファを介して中継されるため、大きく遅れた サブスクライバーは、無制限にバッファリングされる代わりに IllegalStateException で終了します。サイズはデフォルトで 16384 イベントで、 AiServices.builder(...).streamingBufferSize(int) で設定できます。

サードパーティのリアクティブ型​

AI Service のメソッドは JDK の型（CompletableFuture と Flow.Publisher）を返すため、この API が 特定のリアクティブプログラミングライブラリに縛られることはありません。

Reactor の型は langchain4j-reactor モジュールを通じてサポートされています。単一レスポンスのモードには Mono<T>、 リアクティブなモードには Flux<AiServiceStreamingEvent> を使います。依存関係を追加するだけで十分で、アダプターは ServiceLoader を通じて自動的に登録されます。

interface Assistant {

    Mono<String> answer(String userMessage);

    Flux<AiServiceStreamingEvent> events(String userMessage);
}
备注

Mutiny の Uni/Multi は、まだサポートされていません。接続点（ServiceLoader で検出される CompletableFutureAdapter と PublisherAdapter の SPI）は用意されていますが、それらのアダプターは同梱されていないため、そのような 戻り値の型を宣言すると、現状では失敗します。CompletableFuture または Flow.Publisher を使い、呼び出し側で変換してください。

Flux<String> は、同じモジュール内の、より古い TokenStream ベースのアダプターが処理します。 AI Servicesを参照してください。これはこの機能より前から存在し、ここで説明している ノンブロッキングのパスには含まれず、すべてのプロバイダーで動作します。

ノンブロッキングであるべきもの​

ノンブロッキングは、すべてのレイヤーで成り立つ必要があります。どこか 1 か所でもブロックするステップがあると、呼び出し全体が再びブロッキングになります。各レイヤーには、 既存のブロッキングメソッドに加えて、対応する非同期のメソッドがあります。

レイヤーブロッキングノンブロッキングの対応メソッド
チャットモデル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(...)
Web 検索search(...)searchAsync(...)
MCPMcpClient.executeTool(...)executeToolAsync(...)
ガードレールの実行ChatExecutor.execute(...)ChatExecutor.executeAsync(...)
HTTP クライアントexecute(...)executeAsync(...)、stream(...)

対応するメソッドを実装していないコンポーネントは、黙ってブロックするのではなく、明示的に失敗します。返される future または publisher は、コンポーネント名と不足しているメソッドを示す AsyncNotSupportedException で失敗します。 AsyncNotSupportedException は UnsupportedFeatureException の一種であるため、1 つの catch 節で、どちらの 「サポートされない」ケースも捕捉でき、再試行されることはありません。

避けられないブロッキングコード​

ツールは典型的なケースです。データベースやブロッキングする HTTP API を呼び出すツールを、ノンブロッキングにすることはできません。そのような ツールはオフロードされるため、モデル自身のスレッドをブロックすることはありません。

备注

Java 21 以降では、オフロード用の executor は仮想スレッドを作成するため、ブロッキングするツールは仮想スレッドをパークさせます。 仮想スレッドはキャリアスレッドを占有せず、そこからアンマウントされます。LangChain4j のターゲットは Java 17 であり、Java 17 ～ 20 では同じ executor は上限のないプラットフォームスレッドのプールになります。これは、負荷時のリソースの使われ方が大きく異なります。これが問題になる場合は、 独自の有界な Executor を指定してください（後述）。

これは @Tool アノテーションが付いたメソッドに適用されます。executeAsync をオーバーライドしていない手書きの ToolExecutor は、 オプトインしていない他のコンポーネントと同様に、代わりに AsyncNotSupportedException で明示的に失敗します。

ツールは、非同期およびリアクティブのモードでは、デフォルトで並行して実行されます。1 つずつ実行するには、 シングルスレッドの executor を渡します。

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 Service の作成時に拒否される

ツールのエラー​

2 つのツールエラーのデフォルトは、意図的に逆になっています。実行の失敗を LLM に送ると、ツールのバグがあなたから隠され、 モデルがそれを回避して回答をでっち上げることを促してしまいます。そのため、非同期モードでは、例外自体が LLM に伝えてよい内容を示している場合を除き、 代わりに呼び出しを失敗させます （ToolErrorVisibleToLlmを参照）。一方、不正な形式の引数文字列は、モデルが生成したものであり、 通常は指摘されれば修正できるため、 呼び出しを失敗させるのではなく、返送されます。

どちらも引き続き設定可能で、明示的に設定したハンドラーはすべてのモードで使用されます。

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

これらのハンドラーでできることについては、エラー処理を参照してください。

ツールの並行性​

これらのモードでは、ツールは常に Executor 上で実行されるため、1 つの LLM レスポンスに含まれる複数のツール呼び出しは、同時に 実行されます。ツールが並行して実行しても安全でない場合は、シングルスレッドの executor を渡してください。 executeToolsConcurrently(Executors.newSingleThreadExecutor())。

イベントの順序​

リアクティブなパスでは、ツールは CompleteToolCallEvent が発行されるとすぐに開始されるため、ラウンドの残りの部分と 重なります。そのため、あるラウンドの BeforeToolExecutionEvent / AfterToolExecutionEvent が、そのラウンドの IntermediateResponseEvent より前に届くことがあります。IntermediateResponseEvent はラウンドごとのマーカーとして扱い、 そのラウンドのすべてのツールイベントがそれより前に来ることを保証するバリアとは考えないでください。コールバックベースの TokenStream は、常に 中間レスポンスを最初に報告します。

executor の制御とコンテキストの伝播​

LangChain4j が呼び出し元のスレッド以外で処理を実行する場所（並行したツール呼び出し、オフロードされた取得、リトライのバックオフ）では、 いずれも、ExecutorProvider SPI という 1 つの差し替え可能な接続点から executor を取得します。

public interface ExecutorProvider {

    Executor executor();

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

ServiceLoader を通じて登録するか、テストや DI を使わないアプリケーションの場合はプログラムで登録します。

ExecutorProvider.set(() -> myExecutor);

何も登録しない場合、LangChain4j は Java 21 以降ではタスクごとに仮想スレッドを使う executor を、 Java 17 ～ 20 では上限のないプラットフォームスレッドのプールを使用します。

备注

1 回の呼び出しは複数のスレッドをまたぐため、環境の ThreadLocal の状態（MDC のログコンテキスト、トレーシングのスパン、 セキュリティコンテキスト）は、完全な同期モードのようには自動的に伝播されません。これを処理に 追従させるには、captureContext() を実装してください。LangChain4j は、呼び出しごとに、AI Service のメソッドを呼び出したスレッドで一度だけこれを呼び出し、 その呼び出しが executor にオフロードするすべてのタスク（ツール、ブロッキングする RAG のステージ、 モデレーション）を、キャプチャしたコンテキストで実行します。ExecutorProvider.set(() -> myExecutor) で登録したラムダでは これを行えないため、たとえば Micrometer のコンテキスト伝播を使って、クラスを登録してください。

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;
    }
}
ExecutorProvider.set(new ContextPropagatingExecutorProvider());

OpenTelemetry では、captureContext() は return Context.current()::wrap; です。MicroProfile Context Propagation の場合は次のとおりです。

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 は、 同時に実行される複数のタスクをラップする場合があり、ラップされた各タスクは、完了時に、実行されているスレッドの以前のコンテキストを復元する必要があります。 タスクはプールされたスレッドで実行されるため、残されたコンテキストは無関係なタスクに漏れてしまいます。 上記の例は、これらのルールに従っています。

注意すべき点がいくつかあります。

  • コンテキストを伝播する executor だけ（ManagedExecutor、TaskDecorator でラップした executor、 Context.taskWrapping(executor) など）では不十分です。それはタスクを送信するスレッドのコンテキストをキャプチャしますが、 オフロードされる処理の多くは、モデルが回答した後に、回答を届けたスレッドから送信されます。これは captureContext() と組み合わせることができます。その場合、タスクはキャプチャされたコンテキストで実行されます。
  • LangChain4j がオフロードしない処理（ガードレール、チャットメモリ、リスナー、ストリーミングのコールバック）は、 現在のスレッドで実行され、キャプチャされたコンテキストは渡されません。
  • AI Service の呼び出しの外でオフロードされる処理にも渡されません。リトライのバックオフ、モデル 統合が独自に開始するスレッド（たとえばストリーミングのレスポンスを読み取るためのもの）、および langchain4j-agentic の 並列エージェントと非同期エージェントです。そこでコンテキストが必要な場合は、captureContext() を、コンテキストを伝播する executor と組み合わせてください。
  • Flow.Publisher、Mono、Flux を返すメソッドでは、コンテキストは、結果がサブスクライブされるときではなく、 メソッドが呼び出されたときにキャプチャされます。
  • キャプチャされたコンテキストは呼び出しの間ずっと保持されるため、オフロードされた処理は、その呼び出しを開始したリクエストが 完了した後でも、そのコンテキストで実行される可能性があります（たとえば、すでに終了したセッションの セキュリティコンテキストで）。
  • コンテキストをキャプチャするのは ExecutorProvider だけです。AI Service に渡された executor（ executeToolsConcurrently(executor) で指定したもの）上で実行されるツールも、キャプチャされたコンテキストで実行されます。
  • これは AiServices で作成された AI Service に適用されます。独自の AI Service 実装を構築するフレームワークは、 コンテキストをどのように伝播するかを自分で決定します。

InvocationContext は影響を受けません。これは、スレッドローカルを通じてではなく、常にパラメータとして明示的に渡されます。その parentInvocationId() も、キャプチャされたコンテキストでは引き継がれません。 ネストした呼び出しのリンクを参照してください。

モデルのリスナーはブロックしてはならない​

ChatModelListener と EmbeddingModelListener のコールバックは、モデル自身のスレッドで同期的に呼び出され、 オフロードされることはありません。非同期およびリアクティブ API では、これはトランスポートの I/O ワーカーを意味します。そこで ブロッキング I/O を行うリスナーは、そのワーカーを停止させ、実行中のすべての呼び出しのスループットを低下させます。そのような処理は、コールバックの内部から 独自の executor にオフロードしてください。オブザーバビリティを参照してください。

プロバイダーのサポート​

注意

ノンブロッキングのモードは、それを実装しているプロバイダーでのみ機能します。 他のプロバイダーに対して CompletableFuture または Flow.Publisher の戻り値の型を宣言すると、コンパイルは通りますが、 実行時にコンポーネント名と不足しているメソッドを示す AsyncNotSupportedException で失敗します。ブロッキング呼び出しへの 暗黙のフォールバックはありません。それが狙いですが、使用できる戻り値の型は、使用している プロバイダーに依存することを意味します。

サポートはプロバイダーごとのオプトインで、段階的に展開されています。現時点では次のとおりです。

プロバイダーCompletableFuture（chatAsync）Flow.Publisher（リアクティブな chat）
OpenAI — Chat Completions✅✅
OpenAI — Responses✅✅
Anthropic✅✅
Bedrock✅✅
その他のすべてのプロバイダー❌❌

チャットモデル以外では、OpenAI が非同期の埋め込み（embedAsync）を、Cohere が非同期のスコアリング （scoreAsync）を、Tavily が非同期の Web 検索（searchAsync）を実装しています。

オプトインしていないプロバイダーでも、同期 API と TokenStream API には影響がありません。これらは これまでどおり動作します。

呼び出し内のその他すべても同様​

そのモードをサポートするチャットモデルは必要条件ですが、十分条件ではありません。やり取りのどこか 1 か所でもブロッキングのコンポーネントがあれば、 呼び出しは同じように失敗します。実際には、次のようになります。

コンポーネント実装が必要なもの同梱の実装
チャットメモリaddAsync / messagesAsync / setAsyncMessageWindowChatMemory と InMemoryChatMemoryStore はすでに実装済み
ガードレールvalidateAsyncなし。自分で書くガードレールは、I/O を行わない場合でもこれをオーバーライドする必要がある
コンテンツリトリーバー、クエリルーター、アグリゲーターretrieveAsync、routeAsync、aggregateAsync代わりに offloadBlocking(true) でオフロードをオプトインする
埋め込みストアsearchAsync上記と同様に、リトリーバーの offloadBlocking(true) を使う
カスタムの ToolExecutorexecuteAsync@Tool アノテーション付きのメソッドは自動的にオフロードされる
チャットモデルルーター（RoutingChatModel）routeAsyncDecisionModelChatModelRouter は実装済み（その意思決定モデルが decideAsync をサポートしている場合）

ブロッキング処理を行わないガードレールは、1 行でこの契約を満たせます。

@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 スターターでは、ノンブロッキングのモードは Spring の WebClient で送信されるため、クラスパスにそれが必要です （Spring Boot 4 では spring-boot-starter-webclient、Spring Boot 3 では spring-webflux）。どの依存関係を追加するか、およびどのような カスタマイズが適用されるかについては、Spring Boot 統合を参照してください。

環境コンテキストを非同期の呼び出しに追従させるには、LangChain4j がデフォルトの executor ではなく、アプリケーション自身の executor にオフロードするようにします。

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

すると、LangChain4j はオフロードされた処理をアプリケーションのタスク executor で実行し、そのプールは spring.task.execution.* に従います。この設定は、1 つのアプリケーションコンテキストに限定されずプロセス全体に及ぶため、 デフォルトではオフです。

Spring Boot スターターのバージョン 1.23.0-beta33 以降では、呼び出し元のコンテキストもキャプチャされるため（ 上記を参照）、トレーシングのスパン、MDC、セキュリティコンテキストが、 LangChain4j がオフロードする処理に追従します。

  • アプリケーションが TaskDecorator Bean（たとえば ContextPropagatingTaskDecorator）を定義している場合は、それらが 何をキャプチャするかを定めます。これはアプリケーション自身のタスク executor の場合と同じです（Spring Boot 3 では、ちょうど 1 つだけある場合のみ）。
  • そうでない場合は、クラスパス上に Micrometer のコンテキスト伝播があれば、それが使われます。