Streaming de resposta da LLM API com Java 21 e Spring Boot

1. Visão geral

Uma resposta longa de um LLM pode levar 10–30 segundos para ser gerada. Se você esperar a frase inteira pronta antes de devolver, o usuário verá uma tela em branco o tempo todo. Streaming permite receber cada token assim que o modelo o gera, então o primeiro caractere geralmente aparece em menos de um segundo.

Este artigo guia você a ler stream de uma API compatível com OpenAI Chat Completions usando java.net.http.HttpClient, depois usando Spring WebClient. A seguir, você vai encaminhar o stream para o navegador via Server-Sent Events (SSE) e escrever testes para todo o fluxo.

2. Formato do stream da LLM API

Quando você envia uma requisição com "stream": true, a API não retorna um único JSON. Em vez disso, ela retorna Content-Type: text/event-stream, e cada evento é uma linha 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]

Há três pontos que você precisa lembrar:

• O texto está em choices[0].delta.content. Alguns chunks não têm este campo, por exemplo o primeiro chunk só tem role, ou o último chunk só tem finish_reason.
• Os eventos são separados por uma linha em branco.
• data: [DONE] sinalizando que o stream terminou. Este valor não é JSON, então você não deve passá-lo para Jackson.

3. Separar a lógica de parse em uma classe

A lógica de parse deve ficar em uma classe independente para facilitar testes. Esta classe só precisa de Jackson, que já vem incluído em qualquer Spring Boot starter 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();

    /** Recebe uma linha SSE completa, por exemplo: data: {...} */
    public Optional<String> parseLine(String line) {
        if (!line.startsWith("data:")) {
            return Optional.empty();
        }
        return parsePayload(line.substring(5).trim());
    }

    /** Recebe a parte payload apó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 inválido: " + payload, e);
        }
    }
}

Como esta classe funciona:

• parseLine ignora linhas em branco e linhas como event: ou : keep-alive.
• path(...) do Jackson não lança exception quando o nó está ausente. Se o chunk não tem content, o resultado é Optional.empty().
• [DONE] é bloqueado antes de chamar readTree, então não causa erro de parse.

4. Streaming com java.net.http.HttpClient

Se você só precisa de um CLI ou um job simples, o HttpClient nativo do JDK é suficiente. BodyHandlers.ofLines() retorna um Stream<String> que é lido gradualmente conforme os dados chegam da rede, em vez de esperar que o corpo todo seja carregado:

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": "Giải thích quan hệ Order và Customer trong 3 câu"}]}
            """;

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

Explicação das linhas principais:

• O status code deve ser verificado primeiro. Quando a API retorna um erro (401, 429…), o corpo é JSON de erro, não SSE.
• takeWhile interrompe o stream quando encontra [DONE]. O bloco try-with-resources fecha o stream e a conexão é liberada.
• System.out.flush() faz o texto aparecer imediatamente. Sem essa linha, o terminal pode agrupar o output e imprimir depois.

O resultado é o texto aparecendo gradualmente no terminal:

Một Customer có thể đặt nhiều Order, còn mỗi Order chỉ thuộc về đúng một Customer...

Nota: timeout(...) de HttpRequest se aplica apenas até receber os headers de resposta. Ele não limita o tempo de leitura do corpo. Se o modelo ‘travar’ no meio, o loop vai esperar indefinidamente. Esta é uma razão para mudar para WebClient em aplicações reais.

5. Streaming com Spring WebClient

Em um serviço Spring Boot, WebClient te dá um Flux<String> com suporte a backpressure, timeout e cancelamento. Você precisa adicionar a dependência:

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

A configuração da conexão é colocada em application.properties:

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

A seguir, declare o 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();
    }
}

O service chama a API e retorna um Flux contendo os 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("Client cancelou o stream para a pergunta: {}", question));
    }
}

Cada passo na cadeia de processamento:

• bodyToFlux(SSE_TYPE) para Spring separar automaticamente os eventos por linhas em branco e remover o prefixo data:. Você não precisa lidar com o buffer manualmente.
• takeWhile completa o Flux quando encontra [DONE].
• timeout(Duration.ofSeconds(30)) se aplica ao intervalo entre dois elementos consecutivos, incluindo o primeiro. Se não houver um novo token em 30 segundos, Flux reporta um TimeoutException. Isso é o que falta em HttpClient na seção 4.
• Quando a API retorna 4xx/5xx, retrieve() lança WebClientResponseException antes de qualquer token.

5.1. Salvar a resposta completa após o stream terminar

Você geralmente precisa salvar a resposta completa para registrar histórico ou calcular custos. StringBuilder deve ser criado separadamente para cada assinatura. Portanto, você envolve a cadeia de processamento em Flux.defer:

public Flux<String> streamAndRecord(String question) {
    return Flux.defer(() -> {
        StringBuilder full = new StringBuilder();
        return stream(question)
                .doOnNext(full::append)
                .doOnComplete(() -> log.info("Resposta completa ({} caracteres): {}",
                        full.length(), full));
    });
}

Se StringBuilder for declarado fora de defer, duas requisições simultâneas vão escrever no mesmo buffer. Este erro é difícil de detectar em testes manuais porque você geralmente envia apenas uma requisição por vez.

6. Encaminhar o stream para o navegador

O controller retorna Flux<ServerSentEvent<...>> com 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("Modelo está ocupado, tente novamente."))
                        .event("error").build()));
    }
}

Você pode testar o endpoint com curl -N. A flag -N desativa o buffer do curl:

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

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

event:token
data:{"text":" là"}

event:token
data:{"text":" đơn hàng"}

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

Há três decisões de design dignas de nota:

• Os tokens são encapsulados em JSON (ChatChunk), não são enviados como strings brutas. De acordo com a especificação SSE, o receptor descarta um espaço logo após data:. Um token " là" enviado como string bruta se tornaria "là" no navegador, e as palavras ficariam juntas. Um token contendo \n também seria dividido em várias linhas data:. Encapsular em JSON evita ambos os erros.
• Os erros são enviados como um evento error. Quando o primeiro byte foi enviado, o status 200 já está nos headers. Você não pode mudar para 500 mais. O cliente deve ouvir o evento error.
• O evento done é enviado explicitamente. EventSource do navegador se reconecta automaticamente quando o stream fecha. O cliente precisa do evento done para saber que a resposta está completa e chamar close() por si mesmo.

Quando o usuário fecha a aba, WebFlux cancela a assinatura. O sinal de cancelamento se propaga para WebClient, Reactor Netty fecha a conexão com a LLM API, e o provedor para de gerar tokens. Dessa forma, você não paga pela parte da resposta que ninguém lê.

7. Escrever testes

7.1. Testar o 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("Chunk inválido");
    }
}

O primeiro teste confirma que o espaço no início do token é preservado. Este é exatamente o detalhe mais fácil de perder em todo o fluxo.

7.2. Testar o controller com WebTestClient

Você faz mock de ChatStreamService para que o teste não chame a API real. @MockitoBean está disponível desde Spring Boot 3.4. Em versões mais antigas, você usa @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");
    }
}

O segundo teste simula um erro ocorrendo no meio. O cliente recebe tokens já gerados, depois o evento error, e a resposta ainda tem status 200.

8. Erros comuns e quando não fazer streaming

• Usar bodyToMono(String.class). Essa abordagem agrega o corpo inteiro antes de retornar, então você perde todos os benefícios do streaming.
• Reverse proxy bufferiza a resposta. Nginx bufferiza a resposta por padrão, e o cliente receberá todos os tokens de uma vez no final. Você precisa desativar proxy_buffering para esta rota ou adicionar o header X-Accel-Buffering: no. A compressão gzip no nível de proxy também pode causar bufferização.
• Usar .block() em controller Spring MVC. Se o projeto usar spring-boot-starter-web (Servlet), você ainda pode retornar Flux ou usar SseEmitter. Não faça block() e depois escreva parte por parte, pois cada requisição ocupará uma thread por 30 segundos.
• Exibir saída estruturada quando ainda não recebeu o suficiente. Se você pedir ao modelo para retornar JSON para análise ou chamar uma ferramenta, chunks incompletos não podem ser analisados. Neste caso, você ainda pode fazer stream, mas precisa reunir o suficiente antes de validar. Se não precisar exibir gradualmente, uma requisição comum (sem stream) é mais simples.
• Calcular uso de tokens. Por padrão, chunks durante streaming não têm o campo usage. Muitas APIs suportam uma opção como "stream_options": {"include_usage": true} para enviar um chunk final adicional com usage e choices vazio. O parser da seção 3 ignora este chunk com segurança porque path(0) não existe.

Nota: Você não deve fazer retry automático quando já enviou parte dos tokens para o usuário. Retry gerará uma resposta diferente, que se juntará à parte antiga e criará conteúdo embaralhado. Só faça retry quando o erro ocorrer antes do primeiro token, por exemplo ao encontrar 429 ou 503 de retrieve().

9. Conclusão

Fazer streaming de resposta de um LLM API é essencialmente ler Server-Sent Events. Você lê cada linha data:, extrai choices[0].delta.content e para quando encontra [DONE]. HttpClient do JDK é suficiente para ferramentas pequenas. Em um serviço Spring Boot, WebClient combinado com Flux lhe dá timeout entre tokens e a capacidade de cancelamento que se propaga para upstream.

Ao encaminhar o stream para o navegador, você deve encapsular tokens em JSON para preservar espaços e quebras de linha. Você também deve enviar explicitamente os eventos done e error, pois o código de status é fixado desde o primeiro byte. Finalmente, verifique o reverse proxy para garantir que ele não bufferiza silenciosamente a resposta inteira.