Spring WebFlux와 SSE로 구현하는 AI 스트리밍 API: 실시간 대화 경험 (5편)

@Service
@RequiredArgsConstructor
@Slf4j
public class FortuneAIService {

    private final ChatClient chatClient;
    private final PromptOrchestrator promptOrchestrator;

    public Flux<String> streamFortune(String userQuery) {
        // 4편에서 구축한 오케스트레이터로 프롬프트 빌드 (Persona + RAG)
        Prompt prompt = promptOrchestrator.buildFortunePrompt(userQuery);

        // Ollama로부터 토큰 단위로 수신되는 Flux<String> 반환
        return chatClient.prompt(prompt)
                .stream()
                .content()
                .doOnSubscribe(sub -> log.info("AI 스트리밍 시작"))
                .doOnComplete(() -> log.info("AI 스트리밍 정상 종료"))
                .doOnError(e -> log.error("스트리밍 에러 발생: {}", e.getMessage()));
    }
}

@RestController
@RequestMapping("/api/fortune")
@RequiredArgsConstructor
@Slf4j
public class FortuneStreamController {

    private final FortuneAIService fortuneAIService;

    @PostMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> getFortuneStream(@RequestBody FortuneRequest request) {
        log.info("사용자 고민 접수: {}", request.content());

        return fortuneAIService.streamFortune(request.content())
                .map(token -> ServerSentEvent.<String>builder()
                        .event("token")
                        .data(token)
                        .build())
                // 스트림 종료 알림 이벤트 부착
                .concatWith(Mono.just(ServerSentEvent.<String>builder()
                        .event("done")
                        .data("[DONE]")
                        .build()))
                // 클라이언트 연결 끊김 시 백엔드 파이프라인 정리
                .doOnCancel(() -> log.info("클라이언트 브라우저 닫힘 감지. 스트리밍 강제 중단"));
    }
}

async function requestFortuneStreaming(userQuery) {
    const chatContainer = document.getElementById("chat-box");
    
    // 1. POST 요청 발송
    const response = await fetch("/api/fortune/stream", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ content: userQuery })
    });

    // 2. 바이트 스트림 리더 획득
    const reader = response.body.getReader();
    const decoder = new TextDecoder("utf-8");
    let accumulatedText = "";

    // 3. 청크 스트림 비동기 읽기
    while (true) {
        const { value, done } = await reader.read();
        if (done) break;

        const chunk = decoder.decode(value, { stream: true });
        
        // SSE 포맷 ('data: 토큰\n\n') 파싱
        const lines = chunk.split("\n");
        for (const line of lines) {
            if (line.startsWith("data:")) {
                const token = line.replace("data:", "").trim();
                if (token === "[DONE]") {
                    console.log("스트리밍 완료!");
                    return;
                }
                accumulatedText += token;
                // DOM에 실시간 타이핑 렌더링
                chatContainer.innerText = accumulatedText;
            }
        }
    }
}