Streaming de réponses depuis LLM API avec Java 21 et Spring Boot

1. Aperçu

Une longue réponse d’une LLM peut prendre 10 à 30 secondes pour être générée. Si vous attendez la phrase complète avant de la retourner, l’utilisateur regardera un écran vide pendant tout ce temps. Le streaming vous permet de recevoir chaque token dès que le modèle les génère, donc le premier caractère apparaît généralement en moins d’une seconde.

Cet article vous guide pour lire un flux depuis une API compatible OpenAI Chat Completions en utilisant java.net.http.HttpClient, puis Spring WebClient. Ensuite, vous envoyez ce flux au navigateur via Server-Sent Events (SSE) et écrivez des tests pour tout le pipeline.

2. Format du streaming d’une LLM API

Lorsque vous envoyez une requête avec "stream": true, l’API ne retourne pas un seul JSON. Au lieu de cela, elle retourne Content-Type: text/event-stream, et chaque événement est une ligne 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]

Il y a trois points à retenir :

• Le texte se trouve à choices[0].delta.content. Certains chunks n’ont pas ce champ, par exemple le premier chunk n’a que role, ou le dernier n’a que finish_reason.
• Les événements sont séparés par une ligne vide.
• data: [DONE] signale la fin du streaming. Cette valeur n’est pas du JSON, vous ne devez donc pas la passer à Jackson.

3. Extraire la logique de parsing dans une classe séparée

La logique de parsing doit se trouver dans une classe indépendante pour être facilement testable. Cette classe n’a besoin que de Jackson, qui est déjà inclus dans tout starter web de Spring Boot :

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

    /** Reçoit une ligne SSE complète, par exemple : data: {...} */
    public Optional<String> parseLine(String line) {
        if (!line.startsWith("data:")) {
            return Optional.empty();
        }
        return parsePayload(line.substring(5).trim());
    }

    /** Reçoit la partie payload après "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("Chunk invalide : " + payload, e);
        }
    }
}

Voici comment fonctionne cette classe :

• parseLine ignore les lignes vides et les lignes comme event: ou : keep-alive.
• La méthode path(...) de Jackson ne lève pas d’exception si un nœud est manquant. Si le chunk n’a pas de content, le résultat est Optional.empty().
• [DONE] est bloqué avant d’appeler readTree, donc il ne cause pas d’erreur de parsing.

4. Streaming avec java.net.http.HttpClient

Si vous avez besoin d’une simple CLI ou d’un job unique, le HttpClient inclus dans le JDK suffit. BodyHandlers.ofLines() retourne un Stream<String> qui est lu progressivement au fur et à mesure que les données arrivent du réseau, au lieu d’attendre le chargement complet du corps :

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": "Expliquer la relation entre Order et Customer en 3 phrases"}]}
            """;

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

Explication des lignes principales :

• Le code de statut doit être vérifié en premier. Lorsque l’API retourne une erreur (401, 429…), le corps est un JSON d’erreur, pas du SSE.
• takeWhile arrête le flux lorsqu’il rencontre [DONE]. Le bloc try-with-resources ferme le flux, et la connexion est libérée en conséquence.
• System.out.flush() assure que le texte s’affiche immédiatement. Sans cette ligne, le terminal peut mettre en buffer la sortie et l’afficher plus tard.

Le résultat est un texte qui s’affiche progressivement sur le terminal :

Un Client peut passer plusieurs Commandes, tandis que chaque Commande n'appartient qu'à exactement un Client...

Remarque : Le timeout(...) d’HttpRequest ne s’applique que jusqu’à la réception des headers de réponse. Il ne limite pas le temps de lecture du corps. Si le modèle « se fige » à mi-chemin, la boucle attendra indéfiniment. C’est une raison de migrer vers WebClient dans une véritable application.

5. Streaming avec Spring WebClient

Dans un service Spring Boot, WebClient vous donne un Flux<String> avec support de la pression arrière (backpressure), timeouts et annulation. Vous devez ajouter la dépendance :

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

La configuration de la connexion est placée dans application.properties :

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

Ensuite, déclarez le bean 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();
    }
}

Le service appelle l’API et retourne un Flux contenant les tokens :

@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("Le client a annulé le flux pour la question : {}", question));
    }
}

Chaque étape du pipeline de traitement :

• bodyToFlux(SSE_TYPE) laisse Spring séparer automatiquement les événements par des lignes vides et éliminer le préfixe data:. Vous n’avez pas à gérer le buffer vous-même.
• takeWhile complète le Flux quand il rencontre [DONE].
• timeout(Duration.ofSeconds(30)) s’applique à l’intervalle entre deux éléments consécutifs, y compris le premier. Si aucun nouveau token n’arrive dans les 30 secondes, le Flux signale une TimeoutException. C’est ce qui manquait à HttpClient à la section 4.
• Lorsque l’API retourne 4xx/5xx, retrieve() lève une WebClientResponseException avant tout token.

5.1. Enregistrer la réponse complète après la fin du streaming

Vous avez souvent besoin d’enregistrer la réponse complète pour l’historique ou calculer les frais. StringBuilder doit être créé séparément pour chaque souscription. Vous enveloppez donc le pipeline de traitement dans Flux.defer :

public Flux<String> streamAndRecord(String question) {
    return Flux.defer(() -> {
        StringBuilder full = new StringBuilder();
        return stream(question)
                .doOnNext(full::append)
                .doOnComplete(() -> log.info("Réponse complète ({} caractères) : {}",
                        full.length(), full));
    });
}

Si StringBuilder était déclaré en dehors de defer, deux requêtes simultanées écriraient dans le même buffer. Cette erreur est difficile à détecter lors d’un test manuel car vous n’envoyez généralement qu’une requête à la fois.

6. Envoyer le flux au navigateur

Le contrôleur retourne un Flux<ServerSentEvent<...>> avec produces = text/event-stream :

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("Le modèle est occupé, veuillez réessayer."))
                        .event("error").build()));
    }
}

Vous pouvez tester l’endpoint avec curl -N. Le drapeau -N désactive la mise en buffer de curl :

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

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

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

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

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

Il y a trois décisions de conception à noter :

• Les tokens sont enveloppés dans du JSON (ChatChunk), pas envoyés sous forme de chaîne brute. Selon la spécification SSE, le destinataire supprimera un espace immédiatement après data:. Le token " est" envoyé brut deviendrait "est" dans le navigateur, et les mots se collent. Les tokens contenant \n se diviseraient aussi en plusieurs lignes data:. L’enveloppe JSON évite les deux erreurs.
• Les erreurs sont envoyées sous forme d’événement error. Une fois que le premier byte est envoyé, le statut 200 est déjà dans les headers. Vous ne pouvez pas le changer en 500. Le client doit écouter l’événement error.
• L’événement done est envoyé explicitement. L’EventSource du navigateur se reconnecte automatiquement lorsque le flux se ferme. Le client a besoin de l’événement done pour savoir que la réponse est terminée et appeler close() lui-même.

Lorsque l’utilisateur ferme l’onglet, WebFlux annule l’abonnement. Le signal d’annulation remonte à WebClient, Reactor Netty ferme la connexion à l’API LLM, et le fournisseur arrête de générer des tokens. De cette façon, vous ne payez pas pour la partie de la réponse que personne ne lit.

7. Écrire des tests

7.1. Test du parser

class ChatChunkParserTest {

    private final ChatChunkParser parser = new ChatChunkParser();

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

        assertThat(parser.parseLine(line)).contains(" bonjour");
    }

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

Le premier test confirme que l’espace au début du token est conservé. C’est le détail le plus facile à perdre dans tout le pipeline.

7.2. Test du contrôleur avec WebTestClient

Vous mocquez ChatStreamService pour que le test n’appelle pas l’API réelle. @MockitoBean est disponible depuis Spring Boot 3.4. Avec les versions antérieures, vous utilisez @MockBean.

@WebFluxTest(ChatController.class)
class ChatControllerTest {

    @Autowired
    WebTestClient webTestClient;

    @MockitoBean
    ChatStreamService chatStreamService;

    @Test
    void whenStreaming_thenSendsTokensThenDone() {
        when(chatStreamService.streamAndRecord("hi"))
                .thenReturn(Flux.just("Bonjour", " le", " monde"));

        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\":\" le\"}");
    }

    @Test
    void whenUpstreamFails_thenSendsErrorEvent() {
        when(chatStreamService.streamAndRecord("hi"))
                .thenReturn(Flux.concat(Flux.just("Bonjour"),
                        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");
    }
}

Le second test simule une erreur survenant à mi-chemin. Le client reçoit le token généré, puis l’événement error, et la réponse conserve le statut 200.

8. Erreurs courantes et quand ne pas utiliser le streaming

• Utiliser bodyToMono(String.class). Cette approche récupère l’intégralité du corps avant de le retourner, vous perdez donc tous les avantages du streaming.
• Le reverse proxy met en buffer la réponse. Nginx met en buffer la réponse par défaut, et le client recevra tous les tokens à la fois à la fin. Vous devez désactiver proxy_buffering pour cette route, ou ajouter l’en-tête X-Accel-Buffering: no. La compression gzip au niveau du proxy peut aussi causer une mise en buffer.
• Utiliser .block() dans un contrôleur Spring MVC. Si votre projet utilise spring-boot-starter-web (Servlet), vous pouvez toujours retourner un Flux ou utiliser SseEmitter. Ne faites pas .block() puis écrivez par parties, car chaque requête conserverait un thread pendant 30 secondes.
• Afficher une sortie structurée avant de la recevoir complètement. Si vous demandez au modèle de retourner du JSON pour le parser ou appeler un outil, les chunks incomplets ne peuvent pas être parsés. Vous pouvez toujours faire du streaming, mais vous devez accumuler assez avant de valider. Si vous n’avez pas besoin d’afficher progressivement, une requête normale (sans streaming) est plus simple.
• Compter l’utilisation des tokens. Par défaut, les chunks en streaming n’ont pas de champ usage. De nombreuses API supportent une option comme "stream_options": {"include_usage": true} pour envoyer un chunk final supplémentaire avec usage et choices vide. Le parser de la section 3 ignore ce chunk sans danger car path(0) n’existe pas.

Remarque : Ne vous relancez pas automatiquement après avoir envoyé une partie des tokens à l’utilisateur. Une nouvelle tentative génèrera une réponse différente, qui se concaténera à l’ancienne, créant un contenu brouillé. Relancez-vous seulement si l’erreur se produit avant le premier token, par exemple en cas de 429 ou 503 de la part de retrieve().

9. Conclusion

Le streaming de réponses depuis une LLM API est essentiellement la lecture d’événements Server-Sent Events. Vous lisez chaque ligne data:, extrayez choices[0].delta.content, et arrêtez quand vous rencontrez [DONE]. Le HttpClient du JDK suffit pour les petits outils. Dans un service Spring Boot, WebClient combiné avec Flux vous donne des timeouts entre les tokens et la possibilité d’annuler jusqu’aux services en amont.

Lorsque vous renvoyez le flux au navigateur, vous devez envelopper les tokens dans du JSON pour préserver les espaces et les sauts de ligne. Vous devez aussi envoyer explicitement les événements done et error, car le code de statut est fixé dès le premier byte. Enfin, vérifiez votre reverse proxy pour vous assurer qu’il ne met pas silencieusement en buffer toute la réponse.