Respuesta de streaming desde LLM API con Java 21 y Spring Boot

1. Resumen

Una respuesta larga desde un LLM puede tardar 10–30 segundos en generarse completamente. Si esperas a tener la respuesta completa antes de devolverla, el usuario verá una pantalla en blanco durante todo ese tiempo. El streaming te permite recibir cada token tan pronto como el modelo lo genera, por lo que la primera letra suele aparecer en menos de un segundo.

Este artículo te guía a través de leer un stream de una API compatible con OpenAI Chat Completions usando java.net.http.HttpClient, luego usando Spring WebClient. Después, transmitirás el stream al navegador a través de Server-Sent Events (SSE) y escribirás pruebas para todo el flujo.

2. Formato del stream de LLM API

Cuando envías una solicitud con "stream": true, la API no devuelve un JSON único. En cambio, devuelve Content-Type: text/event-stream, y cada evento es una línea 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]

Hay tres puntos que debes recordar:

• El texto está en choices[0].delta.content. Algunos chunks no tienen este campo, por ejemplo, el primer chunk solo tiene role, o el chunk final solo tiene finish_reason.
• Los eventos están separados por una línea en blanco.
• data: [DONE] señala el final del stream. Este valor no es JSON, por lo que no debes pasarlo a Jackson.

3. Separa la lógica de parsing en una clase independiente

La lógica de parsing debe estar en una clase independiente para facilitar las pruebas. Esta clase solo necesita Jackson, que ya está disponible en cualquier 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();

    /** Recibe una línea SSE completa, por ejemplo: data: {...} */
    public Optional<String> parseLine(String line) {
        if (!line.startsWith("data:")) {
            return Optional.empty();
        }
        return parsePayload(line.substring(5).trim());
    }

    /** Recibe la parte del payload después de "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);
        }
    }
}

Cómo funciona esta clase:

• parseLine ignora líneas vacías y líneas como event: o : keep-alive.
• El método path(...) de Jackson no lanza una excepción cuando falta un nodo. Si el chunk no tiene content, el resultado es Optional.empty().
• [DONE] se bloquea antes de llamar a readTree, por lo que no causa un error de parsing.

4. Streaming con java.net.http.HttpClient

Si solo necesitas una CLI o un trabajo simple, el HttpClient disponible en el JDK es suficiente. BodyHandlers.ofLines() devuelve un Stream<String> que se lee gradualmente a medida que los datos llegan de la red, sin esperar a que se descargue el cuerpo completo:

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": "Explica la relación entre Order y Customer en 3 oraciones"}]}
            """;

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

Explicación de las líneas principales:

• El código de estado debe verificarse primero. Cuando la API devuelve un error (401, 429…), el body es JSON de error, no SSE.
• takeWhile detiene el stream cuando encuentra [DONE]. El bloque try-with-resources cierra el stream, y la conexión se libera en consecuencia.
• System.out.flush() ayuda a que el texto aparezca inmediatamente. Sin esta línea, la terminal puede agrupar la salida e imprimirla más tarde.

El resultado es que el texto aparece gradualmente en la terminal:

Un Customer puede tener muchas Orders, mientras que cada Order pertenece exactamente a un Customer...

Nota: timeout(...) del HttpRequest solo se aplica hasta que se reciben los headers de respuesta. No limita el tiempo de lectura del body. Si el modelo se “congela” en el medio, el bucle esperará indefinidamente. Esta es una razón para cambiar a WebClient en aplicaciones reales.

5. Streaming con Spring WebClient

En un servicio Spring Boot, WebClient te proporciona un Flux<String> con soporte para backpressure, timeout y cancelación. Necesitas agregar la dependencia:

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

La configuración de conexión se coloca en application.properties:

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

Luego, declara el 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();
    }
}

El servicio llama a la API y devuelve un Flux que contiene los 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("Cliente canceló el stream para la pregunta: {}", question));
    }
}

Cada paso en la cadena de procesamiento:

• bodyToFlux(SSE_TYPE) para que Spring separe automáticamente los eventos por líneas en blanco y elimine el prefijo data:. No tienes que manejar el buffer manualmente.
• takeWhile completa el Flux cuando encuentra [DONE].
• timeout(Duration.ofSeconds(30)) se aplica al intervalo entre dos elementos consecutivos, incluido el primer elemento. Si no hay un token nuevo en 30 segundos, Flux reporta un TimeoutException. Esto es lo que le faltaba al HttpClient en la sección 4.
• Cuando la API devuelve 4xx/5xx, retrieve() lanza WebClientResponseException antes de que haya tokens.

5.1. Guardar la respuesta completa después de que stream finalice

Generalmente necesitas guardar la respuesta completa para registrar el historial o calcular el costo. StringBuilder debe crearse por separado para cada suscripción. Por lo tanto, envuelves la cadena de procesamiento en Flux.defer:

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

Si StringBuilder se declara fuera de defer, dos solicitudes simultáneas escribirían en el mismo buffer. Este error es difícil de detectar en pruebas manuales porque generalmente solo envías una solicitud a la vez.

6. Transmitir el stream al navegador

El controller devuelve Flux<ServerSentEvent<...>> con 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("El modelo está ocupado, por favor intenta de nuevo."))
                        .event("error").build()));
    }
}

Puedes probar el endpoint con curl -N. La bandera -N desactiva el buffer de curl:

curl -N "http://localhost:8080/api/chat/stream?q=Order%20es%20un%20pedido"

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

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

event:token
data:{"text":" un pedido"}

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

Hay tres decisiones de diseño notables:

• Los tokens se envuelven en JSON (ChatChunk), no se envía una cadena sin procesar. Según la especificación SSE, el lado receptor elimina un espacio inmediatamente después de data:. Un token " es" enviado como texto sin procesar se convertiría en "es" en el navegador, y las palabras se pegarían. Un token que contiene \n también se dividiría en múltiples líneas data:. Envolverlo en JSON ayuda a evitar ambos errores.
• Los errores se envían como un evento error. Una vez que se ha enviado el primer byte, el estado 200 ya está en los headers. No puedes cambiar a 500. El cliente debe escuchar los eventos error.
• El evento done se envía explícitamente. El EventSource del navegador se reconecta automáticamente cuando se cierra el stream. El cliente necesita el evento done para saber que la respuesta está completa y llamar a close() por sí solo.

Cuando el usuario cierra la pestaña, WebFlux cancela la suscripción. La señal de cancelación se propaga al WebClient, Reactor Netty cierra la conexión con la API del LLM, y el proveedor detiene la generación de tokens. Gracias a esto, no pagas por la parte de la respuesta que nadie lee.

7. Escribir pruebas

7.1. Prueba del parser

class ChatChunkParserTest {

    private final ChatChunkParser parser = new ChatChunkParser();

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

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

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

La primera prueba confirma que el espacio al comienzo del token se mantiene. Este es precisamente el detalle más fácil de perder en todo el flujo.

7.2. Prueba del controlador con WebTestClient

Simulas ChatStreamService para que las pruebas no llamen a la API real. @MockitoBean está disponible desde Spring Boot 3.4. Con versiones anteriores, usas @MockBean.

@WebFluxTest(ChatController.class)
class ChatControllerTest {

    @Autowired
    WebTestClient webTestClient;

    @MockitoBean
    ChatStreamService chatStreamService;

    @Test
    void whenStreaming_thenSendsTokensThenDone() {
        when(chatStreamService.streamAndRecord("hi"))
                .thenReturn(Flux.just("Hola", " mun", "do"));

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

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

La segunda prueba simula un error que ocurre en el medio. El cliente recibe los tokens ya generados, luego el evento error, y la respuesta sigue teniendo el estado 200.

8. Errores comunes y cuándo no debes hacer streaming

• Usar bodyToMono(String.class). Este enfoque acumula todo el body antes de devolverlo, por lo que pierdes todos los beneficios del streaming.
• Reverse proxy buffer response. Nginx almacena en buffer la respuesta de forma predeterminada, y el cliente recibirá todos los tokens a la vez al final. Necesitas desactivar proxy_buffering para esta ruta, o agregar el header X-Accel-Buffering: no. La compresión gzip en la capa proxy también puede causar buffering.
• Usar .block() en un controlador Spring MVC. Si el proyecto usa spring-boot-starter-web (Servlet), aún puedes devolver un Flux o usar SseEmitter. No hagas block() y luego escribas cada parte, porque cada solicitud ocupará un hilo durante 30 segundos.
• Mostrar output estructurado antes de recibirlo completamente. Si le pides al modelo que devuelva JSON para analizar o llamar a una herramienta, los chunks incompletos no se pueden analizar. En este caso, aún puedes hacer streaming, pero debes acumular suficiente antes de validar. Si no necesitas mostrar gradualmente, un request normal (sin streaming) es más simple.
• Calcular el uso de tokens. De forma predeterminada, los chunks del stream no tienen el campo usage. Muchas APIs soportan una opción como "stream_options": {"include_usage": true} para enviar un chunk adicional al final que contiene usage y choices vacío. El parser de la sección 3 ignora este chunk de manera segura porque path(0) no existe.

Nota: No debes reintentar automáticamente después de haber enviado algunos tokens al usuario. Un reintento generará una respuesta diferente, que se concatenará con la parte anterior, resultando en contenido confuso. Solo reintenta cuando el error ocurre antes del primer token, por ejemplo, cuando recibes 429 o 503 de retrieve().

9. Conclusión

Hacer streaming de respuestas desde una API LLM es esencialmente leer Server-Sent Events. Lees cada línea data:, extraes choices[0].delta.content, y te detienes cuando encuentras [DONE]. El HttpClient del JDK es suficiente para herramientas pequeñas. En un servicio Spring Boot, WebClient combinado con Flux te proporciona timeout adicional entre tokens y la capacidad de cancelación que se propaga al upstream.

Al transmitir el stream al navegador, debes envolver los tokens en JSON para preservar espacios y saltos de línea. También debes enviar los eventos done y error explícitamente, porque el código de estado se fija desde el primer byte. Finalmente, verifica tu reverse proxy para asegurarte de que no almacena silenciosamente en buffer toda la respuesta.