Streaming-Antworten von LLM APIs mit Java 21 und Spring Boot

1. Übersicht

Eine lange Antwort von einem LLM kann 10–30 Sekunden dauern, bis sie vollständig generiert ist. Wenn du auf die vollständige Antwort wartest, bevor du sie zurückgibst, sieht der Benutzer einen leeren Bildschirm die ganze Zeit über. Streaming ermöglicht es dir, jeden Token zu empfangen, sobald das Modell ihn generiert, daher erscheint das erste Zeichen normalerweise in weniger als einer Sekunde.

Dieser Artikel zeigt dir, wie du einen Stream von einer OpenAI-kompatiblen Chat Completions API mit java.net.http.HttpClient und dann mit Spring WebClient liest. Danach erfährst du, wie du den Stream über Server-Sent Events (SSE) an den Browser pushst und Tests für den gesamten Ablauf schreibst.

2. Streaming-Format von LLM APIs

Wenn du eine Anfrage mit "stream": true sendest, gibt die API nicht ein einziges JSON-Objekt zurück. Stattdessen gibt sie Content-Type: text/event-stream zurück, und jedes Event ist eine Zeile mit 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]

Hier sind drei Punkte zu beachten:

• Der Text liegt in choices[0].delta.content. Einige Chunks haben dieses Feld nicht, zum Beispiel hat der erste Chunk nur role, oder der letzte hat nur finish_reason.
• Events werden durch eine leere Zeile voneinander getrennt.
• data: [DONE] signalisiert das Ende des Streams. Dieser Wert ist kein JSON, daher darfst du ihn nicht in Jackson verarbeiten.

3. Parse-Logik in eine separate Klasse auslagern

Die Parse-Logik sollte in einer separaten Klasse liegen, um sie leicht zu testen. Diese Klasse braucht nur Jackson, das bereits in jedem Spring Boot Web Starter vorhanden ist:

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();

    /** Erhält eine ganze SSE-Zeile, z.B.: data: {...} */
    public Optional<String> parseLine(String line) {
        if (!line.startsWith("data:")) {
            return Optional.empty();
        }
        return parsePayload(line.substring(5).trim());
    }

    /** Erhält die Payload nach "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("Ungültiger Chunk: " + payload, e);
        }
    }
}

So funktioniert diese Klasse:

• parseLine ignoriert leere Zeilen und Zeilen wie event: oder : keep-alive.
• Jacksons path(...) wirft keine Exception, wenn ein Node fehlt. Wenn ein Chunk kein content hat, ist das Ergebnis Optional.empty().
• [DONE] wird vor dem Aufruf von readTree gefiltert, daher verursacht es keinen Parse-Fehler.

4. Streaming mit java.net.http.HttpClient

Wenn du nur eine CLI oder einen einfachen Job brauchst, reicht der in der JDK enthaltene HttpClient aus. BodyHandlers.ofLines() gibt einen Stream<String> zurück, der allmählich gelesen wird, wenn Daten aus dem Netzwerk ankommen, ohne auf den vollständigen Body zu warten:

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": "Erläutere die Beziehung zwischen Order und Customer in 3 Sätzen"}]}
            """;

        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();
    }
}

Erläuterung der wichtigsten Zeilen:

• Der Status-Code muss zuerst überprüft werden. Wenn die API einen Fehler zurückgibt (401, 429…), ist der Body ein JSON-Error statt SSE.
• takeWhile beendet den Stream, wenn [DONE] gefunden wird. Der try-with-resources-Block schließt den Stream, und die Verbindung wird freigegeben.
• System.out.flush() erzwingt, dass der Text sofort angezeigt wird. Ohne diese Zeile kann das Terminal die Ausgabe puffern und später auf einmal drucken.

Das Ergebnis ist, dass der Text allmählich im Terminal angezeigt wird:

Eine Order ist ein Kundenauftrag, und ein Kunde kann viele Orders aufgeben...

Hinweis: timeout(...) des HttpRequest gilt nur bis zum Empfang der Response-Header. Es begrenzt nicht die Zeit zum Lesen des Body. Wenn das Modell stecken bleibt, wartet die Schleife unbegrenzt. Dies ist ein Grund, in echten Anwendungen zu WebClient zu wechseln.

5. Streaming mit Spring WebClient

In einem Spring Boot Service gibt dir WebClient einen Flux<String> mit Unterstützung für Backpressure, Timeout und Cancellation. Du musst diese Abhängigkeit hinzufügen:

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

Die Verbindungskonfiguration wird in application.properties eingestellt:

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

Danach deklarierst du einen WebClient Bean:

@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();
    }
}

Der Service ruft die API auf und gibt einen Flux mit den Tokens zurück:

@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("Client hat den Stream für die Frage storniert: {}", question));
    }
}

Jeder Schritt in der Verarbeitungskette:

• bodyToFlux(SSE_TYPE) lässt Spring Events automatisch nach leeren Zeilen trennen und das data:-Präfix entfernen. Du musst den Buffer nicht selbst verwalten.
• takeWhile beendet Flux, wenn [DONE] gefunden wird.
• timeout(Duration.ofSeconds(30)) gilt für das Intervall zwischen aufeinanderfolgenden Elementen, einschließlich des ersten Elements. Wenn nach 30 Sekunden kein neuer Token kommt, meldet Flux eine TimeoutException. Das ist, was HttpClient in Abschnitt 4 nicht hat.
• Wenn die API 4xx/5xx zurückgibt, wirft retrieve() eine WebClientResponseException, bevor es Tokens gibt.

5.1. Die gesamte Antwort speichern, nachdem das Streaming beendet ist

Du musst oft die gesamte Antwort speichern, um die Historie zu erfassen oder Kosten zu berechnen. Ein StringBuilder muss für jedes Subscription separat erstellt werden. Daher wickelst du die Verarbeitungskette in Flux.defer ein:

public Flux<String> streamAndRecord(String question) {
    return Flux.defer(() -> {
        StringBuilder full = new StringBuilder();
        return stream(question)
                .doOnNext(full::append)
                .doOnComplete(() -> log.info("Vollständige Antwort ({} Zeichen): {}",
                        full.length(), full));
    });
}

Wenn StringBuilder außerhalb von defer deklariert wird, schreiben zwei gleichzeitige Anfragen in einen gemeinsamen Buffer. Dieser Fehler ist schwer zu erkennen, wenn du manuell testest, da du normalerweise nur eine Anfrage gleichzeitig sendest.

6. Stream zum Browser streamen

Der Controller gibt einen Flux<ServerSentEvent<...>> mit produces = text/event-stream zurück:

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("Das Modell ist beschäftigt, bitte versuchen Sie es später erneut."))
                        .event("error").build()));
    }
}

Du kannst den Endpoint mit curl -N testen. Das Flag -N deaktiviert den Buffer von curl:

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

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

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

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

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

Es gibt drei bemerkenswerte Designentscheidungen:

• Tokens werden in JSON (ChatChunk) verpackt, nicht als rohe Strings gesendet. Nach der SSE-Spezifikation wird der Empfänger einen Leeraum direkt nach data: entfernen. Das Token " ist", das roh gesendet wird, wird zu "ist" im Browser, und die Wörter kleben zusammen. Tokens, die \n enthalten, werden auch in mehrere data:-Zeilen aufgeteilt. Das Verpacken in JSON vermeidet beide Fehler.
• Fehler werden als error-Event gesendet. Sobald das erste Byte gesendet wurde, liegt der Status 200 in den Headers. Du kannst nicht zu 500 wechseln. Der Client muss auf das error-Event hören.
• Das done-Event wird explizit gesendet. Der EventSource des Browsers verbindet sich automatisch wieder, wenn der Stream schließt. Der Client braucht das done-Event, um zu wissen, dass die Antwort vollständig ist, und ruft close() auf.

Wenn der Benutzer den Tab schließt, bricht WebFlux das Subscription ab. Das Cancellation-Signal propagiert rückwärts zu WebClient, Reactor Netty schließt die Verbindung zur LLM API, und der Provider stoppt die Token-Generierung. Dadurch zahlst du nicht für den Teil der Antwort, den niemand liest.

7. Tests schreiben

7.1. Test parser

class ChatChunkParserTest {

    private final ChatChunkParser parser = new ChatChunkParser();

    @Test
    void givenContentChunk_whenParseLine_thenReturnsToken() {
        String line = "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\" chào\"}}]}";

        assertThat(parser.parseLine(line)).contains(" chào");
    }

    @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("Ungültiger Chunk");
    }
}

Der erste Test bestätigt, dass das Leerzeichen am Anfang des Tokens erhalten bleibt. Dies ist das leicht zu übersehende Detail im gesamten Ablauf.

7.2. Controller mit WebTestClient testen

Du mockst ChatStreamService, damit der Test die echte API nicht aufruft. @MockitoBean ist seit Spring Boot 3.4 verfügbar. Bei älteren Versionen verwendest du @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");
    }
}

Der zweite Test simuliert einen Fehler in der Mitte. Der Client erhält die bereits generierten Tokens, dann das error-Event, und die Response hat immer noch den Status 200.

8. Häufige Fehler und wann man nicht streamen sollte

• bodyToMono(String.class) verwenden. Dies puffert den gesamten Body und gibt ihn dann zurück, daher verlierst du alle Vorteile des Streaming.
• Reverse Proxy puffert die Response. Nginx puffert die Response standardmäßig, und der Client erhält alle Tokens auf einmal am Ende. Du musst proxy_buffering für diese Route deaktivieren oder den Header X-Accel-Buffering: no hinzufügen. Gzip-Kompression auf der Proxy-Ebene kann auch zu Pufferung führen.
• .block() im Spring MVC Controller verwenden. Wenn dein Projekt spring-boot-starter-web (Servlet) verwendet, kannst du immer noch Flux zurückgeben oder SseEmitter verwenden. Blockiere nicht und schreibe dann Stück für Stück, da jede Anfrage einen Thread für 30 Sekunden hält.
• Strukturierte Ausgabe anzeigen, bevor genug empfangen wurde. Wenn du das Modell auffordest, JSON zurückzugeben, um zu parsen oder Tools aufzurufen, können unvollständige Chunks nicht geparst werden. Dann kannst du immer noch streamen, aber musst genug sammeln, bevor du validierst. Wenn du keine schrittweise Anzeige brauchst, ist eine normale (non-streaming) Anfrage einfacher.
• Token-Verbrauch berechnen. Standardmäßig haben Chunks beim Streaming kein usage-Feld. Viele APIs unterstützen eine Option wie "stream_options": {"include_usage": true}, um einen zusätzlichen abschließenden Chunk mit usage und leeren choices zu senden. Der Parser aus Abschnitt 3 ignoriert diesen Chunk sicher, da path(0) nicht existiert.

Hinweis: Du solltest nicht automatisch wiederholen, wenn bereits Tokens an den Benutzer gesendet wurden. Ein Wiederversuch generiert eine andere Antwort und verbindet sie mit dem alten Teil, was zu durcheinander geratenen Inhalten führt. Wiederhole nur, wenn der Fehler vor dem ersten Token auftritt, z.B. bei 429 oder 503 von retrieve().

9. Fazit

Streaming-Responses von LLM APIs ist eigentlich das Lesen von Server-Sent Events. Du liest jede data:-Zeile, extrahierst choices[0].delta.content und stoppst, wenn du [DONE] findest. Der HttpClient der JDK reicht für kleine Tools aus. In einem Spring Boot Service bietet dir WebClient kombiniert mit Flux zusätzliche Timeouts zwischen Tokens und die Möglichkeit der Cancellation upstream.

Wenn du den Stream zum Browser pushst, solltest du Tokens in JSON verpacken, um Leerzeichen und Zeilenumbrüche zu bewahren. Du solltest auch die Events done und error explizit senden, da der Status-Code ab dem ersten Byte fest ist. Abschließend überprüfe deine Reverse Proxy, um sicherzustellen, dass sie nicht stillschweigend die gesamte Response puffert.