Streaming

개요

Contexa의 Streaming은 LLM 출력 내에서 마커 기반 프로토콜을 사용하여 서술 텍스트와 구조화된 JSON 응답을 분리합니다. DefaultStandardStreamingService는 원시 LLM 청크를 처리하고, 완전한 문장으로 버퍼링하며, 프로토콜 마커를 감지하고, JSON 페이로드를 추출한 후, 클라이언트 전달을 위해 모든 것을 SSE 이벤트로 래핑합니다.

Streaming 아키텍처

LLM 또는 Pipeline에서 나온 청크가 브라우저로 전달되는 흐름은 다음과 같습니다.

DefaultStandardStreamingService는 SSE 경로에서 다음 순서로 동작합니다.

1. AICoreOperations.processStream() 또는 PipelineOrchestrator.executeStream()를 호출해 원시 청크와 프로토콜 마커가 포함된 Flux<String>를 받습니다.
2. 각 청크를 StreamingContext에 누적하여 최종 응답 마커와 문장 버퍼링 상태를 추적합니다.
3. 서술형 텍스트는 SentenceBuffer를 통해 점진적으로 방출됩니다.
4. 업스트림에서 ###GENERATING_RESULT###를 내보내면 즉시 SSE 클라이언트로 전달합니다.
5. 누적 데이터에 ###FINAL_RESPONSE###가 포함되면 구조화된 꼬리 부분을 추출해 최종 SSE 페이로드로 전송합니다.
6. 각 항목은 ServerSentEvent로 래핑되어 브라우저에 전달됩니다.

StandardStreamingService

모든 Streaming 작업을 위한 주요 인터페이스입니다. 두 가지 실행 경로와 Streaming 및 비Streaming 모드를 모두 지원합니다.

StreamingProtocol

텍스트, JSON, 제어 신호를 분리하기 위해 Streaming Pipeline에서 사용되는 마커 상수를 정의합니다.

StreamingContext

Streaming 세션을 추적하는 상태 유지 컨텍스트 객체입니다. 청크를 누적하고, 마커를 감지하며, 원시 LLM 청크를 완전한 문장으로 변환하기 위한 SentenceBuffer를 관리합니다.

주요 기능:

• 청크 누적 — 마커 감지를 위한 appendChunk().
• 문장 버퍼링 — 전송 전 청크를 완전한 문장으로 그룹화하는 getSentenceBuffer().
• JSON 추출 — 누적된 스트림에서 JSON 페이로드를 감지하고 추출하는 extractJsonPart().
• 상태 추적 — 중복 발송을 방지하기 위한 isFinalResponseStarted() 및 isJsonSent().

ChunkProcessor

Streaming 처리 체인에 연결할 수 있는 커스텀 청크 처리 로직을 위한 인터페이스입니다. SSE 서비스 경로 자체는 StreamingContext와 SentenceBuffer를 사용하고, 직접 UnifiedLLMOrchestrator.stream()를 호출하는 경로에서는 별도의 청크 프로세서가 사용될 수 있습니다.

JsonStreamingProcessor

JsonStreamingProcessor는 ChunkProcessor 구현체로, 스트림에서 JSON_START와 JSON_END 마커를 감지하고 그 사이의 JSON 페이로드를 누적합니다. 이 프로세서는 DefaultStreamingHandler가 UnifiedLLMOrchestrator.stream()를 직접 사용할 때 활용되며, 이 페이지에서 설명하는 SSE 서비스 경로는 StreamingContext와 SentenceBuffer를 사용해 이미 스트리밍 중인 청크를 브라우저 친화적인 이벤트로 패키징합니다.

처리 흐름

1. JSON_START 전 텍스트는 STREAMING_MARKER와 함께 발행됩니다.
2. JSON_START가 감지되면 GENERATING_RESULT_MARKER가 발행됩니다.
3. 마커 사이의 콘텐츠는 JSON 버퍼에 누적됩니다.
4. JSON_END가 감지되면 JSON이 FINAL_RESPONSE_MARKER와 함께 발행됩니다.
5. JSON_END 없이 스트림이 완료되면 JSON이 복구되어 발행됩니다.

코드 예제

컨트롤러에서 StandardStreamingService 사용

@GetMapping(value = "/analyze/stream",
        produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> analyzeStream(
        @RequestParam String query) {

    AIRequest<SecurityContext> request = buildRequest(query);

    return streamingService.stream(request, aiNativeProcessor);
}

@PostMapping("/analyze")
public Mono<AnalysisResponse> analyze(
        @RequestBody AnalysisRequest body) {

    AIRequest<SecurityContext> request = buildRequest(body);

    return streamingService.process(
            request, aiNativeProcessor, AnalysisResponse.class);
}

직접 Pipeline Streaming

@GetMapping(value = "/chat/stream",
        produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> chatStream(
        @RequestParam String message) {

    AIRequest<ChatContext> request = buildChatRequest(message);

    return streamingService.stream(request, pipelineOrchestrator);
}

관련 문서

설정