Java 21과 Spring Boot로 LLM API의 스트리밍 응답 처리하기

1. 개요

LLM에서 긴 응답 하나를 완전히 생성하는 데 10~30초가 걸릴 수 있습니다. 전체 응답을 받을 때까지 기다린 후 반환하면 그 시간 동안 사용자는 빈 화면을 봐야 합니다. 스트리밍을 사용하면 모델이 생성하는 각 토큰을 즉시 받을 수 있으므로 첫 글자는 보통 1초 미만 후에 나타납니다.

이 글에서는 java.net.http.HttpClient를 사용하여 OpenAI Chat Completions 호환 API에서 스트림을 읽는 방법을 안내한 후, 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]

기억해야 할 세 가지 점:

• 텍스트는 choices[0].delta.content에 있습니다. 일부 청크에는 이 필드가 없습니다. 예를 들어 첫 청크는 role만 있고 마지막 청크는 finish_reason만 있습니다.
• 이벤트는 빈 줄로 구분됩니다.
• data: [DONE]은 스트림이 끝났음을 나타냅니다. 이 값은 JSON이 아니므로 Jackson에 넘기면 안 됩니다.

3. 파싱 로직을 별도 클래스로 분리

파싱 로직은 테스트하기 쉽도록 독립적인 클래스에 있어야 합니다. 이 클래스는 모든 Spring Boot starter web에 이미 포함된 Jackson만 필요합니다:

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 줄을 받습니다. 예: 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": "Order와 Customer의 관계를 3문장으로 설명하세요"}]}
            """;

        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는 정확히 한 고객에게만 속합니다...

주의: HttpRequest의 timeout(...)은 응답 헤더를 받을 때까지만 적용됩니다. 본문 읽기 시간을 제한하지 않습니다. 모델이 중간에 “멈춘다면” 루프는 무한정 기다립니다. 이는 실제 애플리케이션에서 WebClient로 전환하는 이유 중 하나입니다.

5. Spring WebClient를 사용한 스트리밍

Spring Boot 서비스에서 WebClient는 백프레셔(backpressure), 타임아웃, 취소를 지원하는 Flux<String>을 제공합니다. 다음 종속성을 추가해야 합니다:

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

서비스는 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을 별도로 생성해야 합니다. 따라서 처리 체인을 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 외부에서 선언되면 동시에 두 개의 요청이 동일한 버퍼에 쓰게 됩니다. 일반적으로 한 번에 하나의 요청만 보내기 때문에 수동 테스트에서는 이 오류를 감지하기 어렵습니다.

6. 브라우저로 스트림 푸시

컨트롤러는 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%20란%20무엇입니까"

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

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

event:token
data:{"text":" 주문"}

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

주의할 세 가지 설계 결정:

• 토큰은 JSON(ChatChunk)으로 래핑되며 원본 문자열로 전송되지 않습니다. SSE 사양에 따르면 수신자는 data: 직후에 하나의 공백을 제거합니다. 토큰 " là"를 원본으로 보내면 브라우저에서는 "là"가 되고 단어가 연결됩니다. \n을 포함하는 토큰도 여러 data: 줄로 분할됩니다. JSON으로 래핑하면 두 오류를 모두 방지할 수 있습니다.
• 오류는 error 이벤트로 전송됩니다. 첫 번째 바이트가 전송되면 상태 200이 이미 헤더에 있습니다. 500으로 변경할 수 없습니다. 클라이언트는 error 이벤트를 수신해야 합니다.
• done 이벤트는 명시적으로 전송됩니다. 브라우저의 EventSource는 스트림이 닫힐 때 자동으로 다시 연결됩니다. 클라이언트는 응답이 완료되었음을 알고 close()를 직접 호출하려면 done 이벤트가 필요합니다.

사용자가 탭을 닫으면 WebFlux는 구독을 취소합니다. 취소 신호는 WebClient로 역으로 전파되고 Reactor Netty는 LLM 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\":\" 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("청크가 유효하지 않습니다");
    }
}

첫 번째 테스트는 토큰의 시작 부분 공백이 유지되는지 확인합니다. 이는 전체 흐름에서 가장 놓치기 쉬운 세부 사항입니다.

7.2. WebTestClient로 컨트롤러 테스트

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

두 번째 테스트는 중간에 발생하는 오류를 시뮬레이트합니다. 클라이언트는 생성된 토큰을 받고 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 이벤트를 명시적으로 보내야 합니다. 마지막으로 리버스 프록시가 전체 응답을 자동으로 버퍼링하지 않는지 확인하세요.