LLM APIからのストリーミング応答をJava 21とSpring Bootで実装

1. 概要

LLMからの長い回答は生成に10～30秒かかることがあります。回答全体が完成するのを待ってから返すと、その間ユーザーは空の画面を見つめることになります。ストリーミングを使うと、モデルが生成したトークンをリアルタイムで受け取ることができるため、最初の文字は通常1秒以内に表示されます。

このガイドでは、OpenAI Chat Completions互換のAPIからストリームをJDKのjava.net.http.HttpClientで読み込む方法、その後Spring WebClientを使う方法を説明します。次に、ストリームをServer-Sent Events (SSE)経由でブラウザに送信し、フロー全体をテストする方法を取り上げます。

2. LLM APIのストリーム形式

"stream": trueでリクエストを送信すると、APIは単一のJSONを返しません。代わりに、Content-Type: text/event-streamを返し、各イベントはdata:行です：

data: {"id":"c1","choices":[{"index":0,"delta":{"role":"assistant","content":""}}]}

data: {"id":"c1","choices":[{"index":0,"delta":{"content":"Xin"}}]}

data: {"id":"c1","choices":[{"index":0,"delta":{"content":" chào"}}]}

data: {"id":"c1","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: [DONE]

3つのポイントを覚えておく必要があります：

• テキストはchoices[0].delta.contentにあります。チャンクによってはこのフィールドがない場合があります。例えば、最初のチャンクはroleのみ、最後のチャンクはfinish_reasonのみです。
• イベントは空行で区切られています。
• data: [DONE]はストリームが完了したことを示します。この値はJSONではないため、Jacksonに渡してはいけません。

3. パース処理をクラスに分離

パース処理は独立したクラスに置いてテスト可能にするべきです。このクラスに必要なのはJacksonだけで、これはすべてのSpring Boot Webスターターに含まれています：

package dev.nonla.chat;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.util.Optional;

public class ChatChunkParser {

    public static final String DONE = "[DONE]";
    private final ObjectMapper mapper = new ObjectMapper();

    /** SSEの1行全体を受け取る、例：data: {...} */
    public Optional<String> parseLine(String line) {
        if (!line.startsWith("data:")) {
            return Optional.empty();
        }
        return parsePayload(line.substring(5).trim());
    }

    /** "data:"の後のペイロード部分を受け取る */
    public Optional<String> parsePayload(String payload) {
        if (payload.isEmpty() || DONE.equals(payload)) {
            return Optional.empty();
        }
        try {
            JsonNode content = mapper.readTree(payload)
                    .path("choices").path(0).path("delta").path("content");
            return content.isTextual() && !content.asText().isEmpty()
                    ? Optional.of(content.asText())
                    : Optional.empty();
        } catch (JsonProcessingException e) {
            throw new IllegalStateException("無効なチャンク: " + payload, e);
        }
    }
}

このクラスの動作方法：

• parseLineは空行やevent:または: keep-aliveのような行をスキップします。
• Jacksonのpath(...)はノードが存在しない場合、例外をスローしません。チャンクにcontentがない場合、結果はOptional.empty()です。
• [DONE]はreadTree呼び出しの前に処理されるため、パースエラーが発生しません。

4. java.net.http.HttpClientでのストリーミング

CLIまたは簡単なジョブだけが必要な場合、JDKに付属のHttpClientで十分です。BodyHandlers.ofLines()は、ネットワークからデータが到着したら段階的に読み込まれるStream<String>を返します。全体のボディが読み込まれるまで待つ必要はありません：

package dev.nonla.chat;

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodyHandlers;
import java.time.Duration;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class StreamingChatDemo {

    public static void main(String[] args) throws Exception {
        String baseUrl = System.getenv("LLM_BASE_URL");
        String apiKey = System.getenv("LLM_API_KEY");

        String body = """
            {"model": "gpt-4o-mini", "stream": true,
             "messages": [{"role": "user",
                           "content": "3文でOrderとCustomerの関係を説明してください"}]}
            """;

        HttpRequest request = HttpRequest.newBuilder(URI.create(baseUrl + "/v1/chat/completions"))
                .header("Authorization", "Bearer " + apiKey)
                .header("Content-Type", "application/json")
                .header("Accept", "text/event-stream")
                .timeout(Duration.ofSeconds(60))
                .POST(HttpRequest.BodyPublishers.ofString(body))
                .build();

        HttpClient client = HttpClient.newHttpClient();
        HttpResponse<Stream<String>> response = client.send(request, BodyHandlers.ofLines());

        if (response.statusCode() != 200) {
            String error = response.body().collect(Collectors.joining("\n"));
            throw new IllegalStateException("HTTP " + response.statusCode() + ": " + error);
        }

        ChatChunkParser parser = new ChatChunkParser();
        try (Stream<String> lines = response.body()) {
            lines.takeWhile(line -> !line.equals("data: [DONE]"))
                 .map(parser::parseLine)
                 .flatMap(Optional::stream)
                 .forEach(token -> {
                     System.out.print(token);
                     System.out.flush();
                 });
        }
        System.out.println();
    }
}

主要な行の説明：

• ステータスコードを最初に確認する必要があります。APIがエラーを返す場合（401, 429…）、ボディはSSEではなくJSONエラーです。
• takeWhileは[DONE]に達するとストリームを停止します。try-with-resourcesブロックはストリームをクローズし、接続も解放されます。
• System.out.flush()はテキストをすぐに表示するのに役立ちます。この行がないと、ターミナルは出力をバッファして後で印刷するかもしれません。

結果はテキストがターミナルに段階的に表示されます：

各Customerは複数のOrderを配置できますが、各Orderは正確に1つのCustomerに属しています...

注意: HttpRequestのtimeout(...)はレスポンスヘッダーを受け取るまでのみ適用されます。ボディ読み込み時間の制限ではありません。モデルが途中で「ハング」した場合、ループは永遠に待機します。これが実際のアプリケーションでWebClientに移行する理由の1つです。

5. Spring WebClientでのストリーミング

Spring Bootサービスでは、WebClientはバックプレッシャー、タイムアウト、キャンセルをサポートするFlux<String>を提供します。dependencyを追加する必要があります：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

接続設定はapplication.propertiesに配置されます：

llm.base-url=${LLM_BASE_URL}
llm.api-key=${LLM_API_KEY}
llm.model=gpt-4o-mini

次に、WebClientビーンを宣言します：

@Configuration
class LlmClientConfig {

    @Bean
    WebClient llmWebClient(@Value("${llm.base-url}") String baseUrl,
                           @Value("${llm.api-key}") String apiKey) {
        return WebClient.builder()
                .baseUrl(baseUrl)
                .defaultHeader(HttpHeaders.AUTHORIZATION, "Bearer " + apiKey)
                .build();
    }
}

ServiceはAPIを呼び出し、トークンを含むFluxを返します：

@Service
public class ChatStreamService {

    private static final Logger log = LoggerFactory.getLogger(ChatStreamService.class);
    private static final ParameterizedTypeReference<ServerSentEvent<String>> SSE_TYPE =
            new ParameterizedTypeReference<>() {};

    private final WebClient llmWebClient;
    private final String model;
    private final ChatChunkParser parser = new ChatChunkParser();

    public ChatStreamService(WebClient llmWebClient, @Value("${llm.model}") String model) {
        this.llmWebClient = llmWebClient;
        this.model = model;
    }

    public Flux<String> stream(String question) {
        Map<String, Object> body = Map.of(
                "model", model,
                "stream", true,
                "messages", List.of(Map.of("role", "user", "content", question)));

        return llmWebClient.post()
                .uri("/v1/chat/completions")
                .contentType(MediaType.APPLICATION_JSON)
                .accept(MediaType.TEXT_EVENT_STREAM)
                .bodyValue(body)
                .retrieve()
                .bodyToFlux(SSE_TYPE)
                .mapNotNull(ServerSentEvent::data)
                .takeWhile(data -> !ChatChunkParser.DONE.equals(data))
                .mapNotNull(data -> parser.parsePayload(data).orElse(null))
                .timeout(Duration.ofSeconds(30))
                .doOnCancel(() -> log.info("クライアントが質問のストリームをキャンセルしました: {}", question));
    }
}

処理チェーンの各ステップ：

• bodyToFlux(SSE_TYPE)により、Springは空行でイベントを自動的に分割し、data:プレフィックスを削除します。バッファを手動で処理する必要はありません。
• takeWhileは[DONE]に達するとFluxを完了します。
• timeout(Duration.ofSeconds(30))は連続する要素間の間隔に適用されます（最初の要素を含む）。30秒以内に新しいトークンがない場合、FluxはTimeoutExceptionを報告します。これはセクション4のHttpClientに欠けていたものです。
• APIが4xx/5xxを返す場合、retrieve()はトークンが生成される前にWebClientResponseExceptionをスローします。

5.1. ストリーム完了後に回答全体を保存

履歴を記録したり、コストを計算したりするために、完全な回答を保存する必要があることがよくあります。StringBuilderは各subscriptionごとに個別に作成する必要があります。そのため、処理チェーンをFlux.deferでラップします：

public Flux<String> streamAndRecord(String question) {
    return Flux.defer(() -> {
        StringBuilder full = new StringBuilder();
        return stream(question)
                .doOnNext(full::append)
                .doOnComplete(() -> log.info("完全な回答（{} 文字）: {}",
                        full.length(), full));
    });
}

StringBuilderがdeferの外で宣言されている場合、2つの同時リクエストが同じバッファに書き込みます。手動テストでは通常1回に1つのリクエストだけを送信するため、このバグは検出が難しいです。

6. ブラウザへのストリーム送信

Controllerはproduces = text/event-streamでFlux<ServerSentEvent<...>>を返します：

public record ChatChunk(String text) {}

@RestController
class ChatController {

    private final ChatStreamService chatStreamService;

    ChatController(ChatStreamService chatStreamService) {
        this.chatStreamService = chatStreamService;
    }

    @GetMapping(path = "/api/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    Flux<ServerSentEvent<ChatChunk>> stream(@RequestParam("q") String question) {
        return chatStreamService.streamAndRecord(question)
                .map(token -> ServerSentEvent.builder(new ChatChunk(token)).event("token").build())
                .concatWith(Mono.just(ServerSentEvent.builder(new ChatChunk("")).event("done").build()))
                .onErrorResume(ex -> Mono.just(ServerSentEvent
                        .builder(new ChatChunk("モデルが忙しいため、再度お試しください。"))
                        .event("error").build()));
    }
}

curl -Nでエンドポイントをテストできます。フラグ-Nはcurlのバッファリングを無効にします：

curl -N "http://localhost:8080/api/chat/stream?q=Order%20la%20gi"

event:token
data:{"text":"Order"}

event:token
data:{"text":" は"}

event:token
data:{"text":" 注文です"}

event:done
data:{"text":""}

注目すべき設計決定が3つあります：

• トークンはJSON（ChatChunk）でラップされ、生文字列は送信されません。 SSE仕様に従うと、受信側はdata:直後の1つのスペースを削除します。生の" là"トークンはブラウザでは"là"になり、単語がくっついてしまいます。\nを含むトークンも複数のdata:行に分割されます。JSONでのラップは両方のエラーを防ぎます。
• エラーはerrorイベントとして送信されます。 最初のバイトが送信されると、ステータス200はヘッダーに含まれます。500に変更することはできません。クライアント側がerrorイベントをリッスンする必要があります。
• doneイベントが明示的に送信されます。 ブラウザのEventSourceはストリームがクローズされると自動的に再接続しようとします。クライアントは回答が完了したことを知り、close()を呼び出すためにdoneイベントが必要です。

ユーザーがタブを閉じると、WebFluxはサブスクリプションをキャンセルします。キャンセル信号はWebClientまで遡り、Reactor Nettyはアップストリームの接続をクローズし、APIプロバイダーはトークン生成を停止します。これにより、誰も読んでいない回答部分の料金を払う必要がなくなります。

7. テストの記述

7.1. パーサーのテスト

class ChatChunkParserTest {

    private final ChatChunkParser parser = new ChatChunkParser();

    @Test
    void givenContentChunk_whenParseLine_thenReturnsToken() {
        String line = "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\" こんにちは\"}}]}";

        assertThat(parser.parseLine(line)).contains(" こんにちは");
    }

    @Test
    void givenDoneOrRoleOnlyChunk_whenParseLine_thenReturnsEmpty() {
        assertThat(parser.parseLine("data: [DONE]")).isEmpty();
        assertThat(parser.parseLine(
                "data: {\"choices\":[{\"delta\":{\"role\":\"assistant\"}}]}")).isEmpty();
        assertThat(parser.parseLine(": keep-alive")).isEmpty();
    }

    @Test
    void givenBrokenJson_whenParseLine_thenThrows() {
        assertThatThrownBy(() -> parser.parseLine("data: {oops"))
                .isInstanceOf(IllegalStateException.class)
                .hasMessageContaining("無効なチャンク");
    }
}

最初のテストはトークンの先頭のスペースが保持されることを確認します。これは全体的なフローの中で最も失いやすい詳細です。

7.2. WebTestClientを使ったControllerのテスト

ChatStreamServiceをモックして、テストが実際のAPIを呼び出さないようにします。@MockitoBeanはSpring Boot 3.4から利用可能です。より古いバージョンでは、@MockBeanを使用します。

@WebFluxTest(ChatController.class)
class ChatControllerTest {

    @Autowired
    WebTestClient webTestClient;

    @MockitoBean
    ChatStreamService chatStreamService;

    @Test
    void whenStreaming_thenSendsTokensThenDone() {
        when(chatStreamService.streamAndRecord("hi"))
                .thenReturn(Flux.just("Xin", " chào", "!"));

        List<ServerSentEvent<String>> events = webTestClient.get()
                .uri("/api/chat/stream?q=hi")
                .accept(MediaType.TEXT_EVENT_STREAM)
                .exchange()
                .expectStatus().isOk()
                .returnResult(new ParameterizedTypeReference<ServerSentEvent<String>>() {})
                .getResponseBody()
                .collectList()
                .block(Duration.ofSeconds(5));

        assertThat(events).extracting(ServerSentEvent::event)
                .containsExactly("token", "token", "token", "done");
        assertThat(events.get(1).data()).isEqualTo("{\"text\":\" chào\"}");
    }

    @Test
    void whenUpstreamFails_thenSendsErrorEvent() {
        when(chatStreamService.streamAndRecord("hi"))
                .thenReturn(Flux.concat(Flux.just("Xin"),
                        Flux.error(new TimeoutException())));

        List<ServerSentEvent<String>> events = webTestClient.get()
                .uri("/api/chat/stream?q=hi")
                .accept(MediaType.TEXT_EVENT_STREAM)
                .exchange()
                .returnResult(new ParameterizedTypeReference<ServerSentEvent<String>>() {})
                .getResponseBody()
                .collectList()
                .block(Duration.ofSeconds(5));

        assertThat(events).extracting(ServerSentEvent::event)
                .containsExactly("token", "error");
    }
}

2番目のテストは、途中で発生するエラーをシミュレートします。クライアントは既に生成されたトークンを受け取り、その後errorイベント、およびレスポンスはまだステータス200を持ちます。

8. よくあるエラーとストリーミングすべきでない場合

• bodyToMono(String.class)を使用する。 このアプローチはボディ全体をバッファしてから返すため、ストリーミングの利点がすべて失われます。
• リバースプロキシがレスポンスをバッファする。 Nginxはデフォルトでレスポンスをバッファし、クライアントはすべてのトークンを最後に一度に受け取ります。このルートのproxy_bufferingを無効にするか、ヘッダーX-Accel-Buffering: noを追加する必要があります。プロキシレイヤーでのgzip圧縮もバッファリングを引き起こす可能性があります。
• Spring MVCコントローラーで.block()を使用する。 プロジェクトがspring-boot-starter-web（Servlet）を使用している場合、Fluxを返すかSseEmitterを使用できます。.block()してから一部を書き込まないでください。各リクエストは30秒間スレッドを保持することになります。
• 不完全なデータから構造化された出力を表示する。 モデルがJSONを返して解析したり、ツールを呼び出したりするように要求すると、不完全なチャンクは解析できません。この場合、ストリーミング可能ですが、十分に集約してから検証する必要があります。段階的表示が不要な場合は、通常のリクエスト（ストリーミングなし）がより簡単です。
• トークン使用量の計算。 デフォルトでは、ストリーミング時のチャンクにはusageフィールドがありません。多くのAPIは"stream_options": {"include_usage": true}のようなオプションをサポートして、usageと空のchoicesを含む追加の最終チャンクを送信します。セクション3のパーサーは、path(0)が存在しないため、このチャンクを安全にスキップします。

注意: トークンの一部をユーザーに送信した後は、自動的に再試行しないでください。再試行は別の回答を生成し、古い部分に接続すると、内容がごちゃごちゃになります。最初のトークンの前にエラーが発生した場合のみ再試行してください。例えば、retrieve()から429または503を受け取った場合です。

9. 結論

LLM APIからのストリーミング応答は本質的にServer-Sent Eventsを読むことです。各data:行を読み、choices[0].delta.contentを取得し、[DONE]に達したら停止します。JDKのHttpClientは小さなツールには十分です。Spring Bootサービスでは、WebClientとFluxの組み合わせが、トークン間のタイムアウト機能と上流へのキャンセル機能を提供します。

ストリームをブラウザに転送するとき、スペースと改行を保持するためにトークンをJSONでラップする必要があります。また、ステータスコードは最初のバイトから固定されているため、doneとerrorイベントを明示的に送信する必要があります。最後に、リバースプロキシをチェックして、応答全体を黙ってバッファしていないことを確認してください。