Response Streaming

Response Streaming (응답 스트리밍)

LLM 은 토큰을 하나씩 생성해요. 그래서 많은 프로바이더가 전체 텍스트가 생성될 때까지 기다리는 대신 토큰 단위로 응답을 스트리밍하는 방법을 제공해요. 사용자가 알 수 없는 시간 동안 기다리지 않고 거의 즉시 읽기 시작할 수 있어서 체감 UX 가 크게 좋아져요. 이 페이지는 저수준 스트리밍 API 를 다뤄요.

출처: 공식문서

StreamingChatModel

ChatModel과 LanguageModel 에는 대응하는 StreamingChatModel, StreamingLanguageModel 인터페이스가 있어요. 비슷한 API 에 응답을 스트리밍하는 점만 달라요. 이들은 StreamingChatResponseHandler 구현을 인자로 받아요.

public interface StreamingChatResponseHandler {

    default void onPartialResponse(String partialResponse) {}
    default void onPartialResponse(PartialResponse partialResponse, PartialResponseContext context) {}

    default void onPartialThinking(PartialThinking partialThinking) {}
    default void onPartialThinking(PartialThinking partialThinking, PartialThinkingContext context) {}

    default void onPartialToolCall(PartialToolCall partialToolCall) {}
    default void onPartialToolCall(PartialToolCall partialToolCall, PartialToolCallContext context) {}

    default void onCompleteToolCall(CompleteToolCall completeToolCall) {}
    default void onUnmappedRawEvent(Object rawEvent) {}

    void onCompleteResponse(ChatResponse completeResponse);

    void onError(Throwable error);
}

이벤트별로 할 일을 정의할 수 있어요:

  • 다음 부분 텍스트 응답이 생성될 때 — onPartialResponse(...). 프로바이더에 따라 부분 응답 텍스트는 한 개 이상의 토큰일 수 있어요. 토큰이 생기는 즉시 UI 로 보낼 수 있죠.
  • 다음 부분 추론 텍스트가 생성될 때 — onPartialThinking(...).
  • 다음 부분 도구 호출이 생성될 때 — onPartialToolCall(...).
  • 단일 도구 호출 스트리밍이 끝났을 때 — onCompleteToolCall(...).
  • 프로바이더가 타입 콜백으로 노출하지 않는 원시 스트리밍 이벤트를 방출할 때 — onUnmappedRawEvent(Object).
  • 생성이 끝났을 때 — onCompleteResponse(ChatResponse) (완전한 AiMessage + ChatResponseMetadata).
  • 오류 시 — onError(Throwable).

사용 예시:

StreamingChatModel model = OpenAiStreamingChatModel.builder()
    .apiKey(System.getenv("OPENAI_API_KEY"))
    .modelName(GPT_4_O_MINI)
    .build();

String userMessage = "Tell me a joke";

model.chat(userMessage, new StreamingChatResponseHandler() {

    @Override
    public void onPartialResponse(String partialResponse) {
        System.out.println("onPartialResponse: " + partialResponse);
    }

    @Override
    public void onPartialThinking(PartialThinking partialThinking) {
        System.out.println("onPartialThinking: " + partialThinking);
    }

    @Override
    public void onPartialToolCall(PartialToolCall partialToolCall) {
        System.out.println("onPartialToolCall: " + partialToolCall);
    }

    @Override
    public void onCompleteToolCall(CompleteToolCall completeToolCall) {
        System.out.println("onCompleteToolCall: " + completeToolCall);
    }

    @Override
    public void onCompleteResponse(ChatResponse completeResponse) {
        System.out.println("onCompleteResponse: " + completeResponse);
    }

    @Override
    public void onError(Throwable error) {
        error.printStackTrace();
    }
});

더 간결하게 LambdaStreamingResponseHandler 클래스로 람다 기반 핸들러를 만들 수 있어요:

import static dev.langchain4j.model.LambdaStreamingResponseHandler.onPartialResponse;

model.chat("Tell me a joke", onPartialResponse(System.out::print));

onPartialResponseAndError(...) 로 부분 응답과 오류 모두를 정의할 수도 있어요.

리액티브 API (실험적)

핸들러 기반 API 외에도, StreamingChatModel 은 호출 스레드를 블로킹하지 않고 같은 응답을 Flow.Publisher 로 넘겨줄 수 있어요:

Flow.Publisher<ChatModelStreamingEvent> publisher = model.chat(chatRequest);

스트림은 PartialThinking, PartialResponse, PartialToolCall, CompleteToolCall 이벤트를 도착하는 대로 방출하고, 그 사이사이에 RawStreamingEvent 를 끼우고, 마지막에 조립된 ChatResponse 를 담은 단일 종단 CompleteResponse 를 방출해요. 텍스트만 방출하는 편의 오버로드도 있어요: Flow.Publisher<String> textOnly = model.chat("Tell me a joke");

publisher 는 cold 라서 구독 전에는 아무 일도 일어나지 않고, 각 구독이 새 요청을 보내요. 구독 취소는 진행 중인 HTTP 호출을 중단하려 시도해요.

이벤트는 모델 자신의 스레드(HTTP 모델은 transport I/O 워커)에서 전달돼요. onNext 에서 블로킹하거나 무거운 작업을 하면 스트림이 멈추고 동시성 상황에서 모든 in-flight 호출의 처리량이 떨어져요. 무거운 작업은 자신의 executor 로 offload 하세요. LLM 응답은 의미 있게 스로틀할 수 없어서 구현은 이를 즉시 소비하고 경계(buffer)로 이벤트를 중계해요. 넉넉히 요청하세요(예: Long.MAX_VALUE) — 적게 요청하는 구독자는 버퍼를 소진하고 오류로 종료될 수 있어요.

AI Service 레벨의 대응책(도구 실행·RAG 콘텐츠도 표면화)은 Non-blocking and Reactive 를 참고하세요.

Unmapped Raw Events (실험적)

일부 LLM 프로바이더는 LangChain4j 가 아직 전용 콜백으로 매핑하지 않는 추가 스트리밍 이벤트를 방출해요 — 예를 들어 OpenAI 서버 측 도구의 라이프사이클 이벤트(web_search) 같은 것(response.web_search_call.in_progress, response.web_search_call.searching, response.web_search_call.completed)이요.

onUnmappedRawEvent(Object rawEvent) 콜백은 이 이벤트에 접근하게 해줘요. 이미 타입 콜백으로 노출된 이벤트(부분 응답·추론·도구 호출)는 여기서 반복되지 않아요 — 그래서 중복 없이 둘 다 소비할 수 있어요. 구체적 타입은 프로바이더별로 달라요: OpenAI·Anthropic·Google AI Gemini·Mistral·Ollama 는 dev.langchain4j.http.client.sse.ServerSentEvent, OpenAI(official) Responses API 는 com.openai.models.responses.ResponseStreamEvent, Chat Completions API 는 ChatCompletionChunk, Bedrock 은 ConverseStreamOutput, Google GenAI 는 GenerateContentResponse 등. 그래서 보통 instanceof 로 검사해 캐스팅해요.

AI Services 를 쓸 때 같은 이벤트는 TokenStream.onUnmappedRawEvent(Consumer<Object>) 콜백으로도 받을 수 있어요.

스트리밍 취소

onPartialResponse(PartialResponse, PartialResponseContext), onPartialThinking(PartialThinking, PartialThinkingContext), onPartialToolCall(PartialToolCall, PartialToolCallContext) 콜백에서 컨텍스트의 StreamingHandle 로 스트리밍을 취소할 수 있어요:

model.chat(userMessage, new StreamingChatResponseHandler() {

    @Override
    public void onPartialResponse(PartialResponse partialResponse, PartialResponseContext context) {
        process(partialResponse);
        if (shouldCancel()) {
            context.streamingHandle().cancel();
        }
    }

    @Override
    public void onCompleteResponse(ChatResponse completeResponse) {
        System.out.println("onCompleteResponse: " + completeResponse);
    }

    @Override
    public void onError(Throwable error) {
        error.printStackTrace();
    }
});

StreamingHandle.cancel() 이 호출되면 LangChain4j 가 연결을 닫고 스트리밍을 멈춰요. 취소 후에는 StreamingChatResponseHandler 가 더 이상 콜백을 받지 않아요.

더 알아보기