[Vue3] 프론트엔드 문자 스트리밍 처리에 관한 것

이 글은 2023년 12월 12일에 작성된 Medium 글을 옮겨왔습니다.

ChatGPT가 답변을 실시간으로 타이핑하듯 보여주는 것을 보고, 프론트엔드에서 스트리밍을 어떻게 처리해야 할지 고민했던 적이 있습니다. 처음에는 WebSocket을 써야 하나 싶었는데, 알고 보니 브라우저에 이미 Streams API라는 좋은 도구가 있더라고요.

Streams API는 네트워크로 수신된 스트리밍 데이터를 자바스크립트로 다룰 수 있게 해주는 Web API입니다. 거의 모든 브라우저에서 지원하고 웹 워커에서도 사용 가능합니다.

스트리밍이란 네트워크를 통해 수신하려는 리소스를 작은 덩어리로 쪼개서 조금씩 처리하는 것을 말합니다. 브라우저는 영상과 같은 미디어 자산을 수신할 때 이미 이 작업을 수행하고 있죠.

Streams API를 사용하면 원시 데이터를 버퍼, 문자열, 블롭으로 한 번에 생성할 필요 없이, 가능한 만큼 조금씩 자바스크립트로 처리할 수 있습니다.

더 많은 것들도 가능합니다. 스트리밍이 시작하거나 끝났음을 감지할 수 있고, 스트림을 서로 연결하거나, 오류를 처리하거나, 스트림을 취소하거나, 읽는 속도에 반응할 수도 있습니다.

스트리밍 데이터를 fetch로 가져왔을 경우 응답 본문은 ReadableStream입니다. ReadableStream.getReader()로 만든 리더를 통해 읽을 수 있습니다.

참고로 WritableStream을 사용하면 스트리밍 데이터를 작성할 수도 있습니다.

스트리밍 데이터 처리 실습

그럼 스트림을 받아와 보겠습니다. fetch 메소드로 간단하게 받을 수 있습니다.

const { body } = await fetch(url);
// body is ReadableStream
const reader = body.getReader();

while (true) {
  const readable = await reader.read();
  console.table(readable);
  if (readable.done) break;
}

위에서 언급한 것처럼 응답 바디는 ReadableStream 형태이므로, getReader() 메소드를 통해 읽을 수 있습니다.

콘솔에는 아래와 같이 출력됩니다.

{ done: false, value: Uint8Array(27)[100, 97, 116, ...] }
{ done: false, value: Uint8Array(27)[100, 97, 116, ...] }
(...)
{ done: true, value: undefined }

스트리밍이 끝나는 시점에 done 값이 true로 들어옵니다. true가 나올 때까지 while 문을 이용해 청크를 처리하면 됩니다.

value에는 Uint8Array 타입(8비트 부호 없는 정수 배열)의 배열이 세팅됩니다. 그래서 디코딩하는 작업이 필요합니다. TextDecoder를 통해 아래와 같이 디코딩할 수 있습니다.

const decoder = new TextDecoder();

while (true) {
  const readable = await reader.read();
  if (readable.done) break;
  const chunks = decoder.decode(readable.value);
  console.log(chunks);
}

디코딩된 chunks를 출력해보면 아래와 같습니다.

data: {"data": "\uac00"}
data: {"data": "\uc5d0"}
data: {"data": "\uac8c"}
data: {"data": " "}
data: {"data": "\uc0c1"}
...

제가 사용한 API에서는 data로 두 번 감싸져 있는 형태였는데, 청크 데이터 형태는 스트리밍을 제공하는 서버의 구현에 따라 다릅니다. SSE(Server-Sent Events) 형식을 사용하는 경우가 많으니 참고하시기 바랍니다.

Vue3 Composable 구현

GPT-4처럼 실시간으로 답변을 보여주고 싶어서 컴포저블을 만들어 보았습니다. 처음에는 단순하게 시작했다가, 에러 처리나 상태 관리 문제를 만나면서 조금씩 개선하게 되었습니다.

import { ref, readonly, type Ref } from 'vue';

interface StreamOptions {
  body: ReadableStream<Uint8Array>;
  onChunk?: (data: string) => void;
  onReady?: (result: { data: string }) => void;
  onError?: (error: Error) => void;
}

const resolveStream = async (
  data: Ref<string>,
  { body, onChunk = () => {}, onReady = () => {} }: Omit<StreamOptions, 'onError'>
) => {
  const reader = body.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const readable = await reader.read();
    if (readable.done) break;

    const decodedValue = decoder.decode(readable.value);
    const chunks = decodedValue
      .replaceAll(/^data: /gm, '')
      .split('\n')
      .map((c) => {
        if (c && c.length > 1) {
          try {
            return JSON.parse(c.replaceAll(/\n/g, ''));
          } catch {
            return null;
          }
        }
        return null;
      })
      .filter(Boolean);

    for (const chunk of chunks) {
      if (!chunk?.data) continue;
      data.value += chunk.data;
      onChunk(chunk.data);
    }
  }
  onReady({ data: data.value });
};

const chatStream = ({
  body,
  onChunk = () => {},
  onReady = () => {},
  onError = () => {},
}: StreamOptions) => {
  const data = ref('');

  resolveStream(data, { body, onChunk, onReady }).catch((error) => {
    onError(error);
  });

  return {
    data: readonly(data),
  };
};

export const useChatStream = () => {
  return {
    chatStream,
  };
};

이 컴포저블의 동작 방식은 다음과 같습니다.

• body(ReadableStream)를 받아서
• TextDecoder로 디코딩하고
• while 문으로 청크를 순차 처리합니다
• 각 청크가 파싱되면 onChunk 콜백을 호출하고
• 모든 청크 처리가 완료되면 onReady를 호출합니다
• 에러가 발생하면 onError를 호출합니다

외부에서 데이터를 직접 변경할 수 없도록 readonly로 감싸서 반환했습니다.

사용 예시

이 컴포저블 함수는 아래와 같이 사용할 수 있습니다.

<script setup lang="ts">
import { reactive } from 'vue';
import { useChatStream } from '@/composables/useChatStream';

const { chatStream } = useChatStream();
const answer = reactive({ status: null as string | null, message: '' });

const sendQuestion = async () => {
  const { body } = await fetch('/api/chat');

  if (!body) return;

  chatStream({
    body,
    onChunk: (data) => {
      answer.status = 'streaming';
      answer.message += data;
    },
    onReady: () => {
      answer.status = 'done';
    },
    onError: (error) => {
      answer.status = 'error';
      console.error(error);
    },
  });
};
</script>

<template>
  <div>
    <p>Answer: {{ answer.message }}</p>
    <span v-if="answer.status === 'streaming'">...</span>
  </div>
</template>

여기까지가 문자열 스트리밍을 다루는 기초적인 내용입니다. Vue3 기준으로 작성했지만, 대부분의 코드가 일반 자바스크립트이기 때문에 다른 프레임워크에서도 참고하기 어렵지 않을 것입니다.

보너스: pipeThrough

글을 작성하면서 스트리밍 문자열을 디코딩하는 다른 방법도 알게 되었습니다.

위에서 사용한 방법:

const reader = body.getReader();
const decoder = new TextDecoder();

while (true) {
  const readable = await reader.read();
  if (readable.done) break;
  const decodedValue = decoder.decode(readable.value);
}

pipeThrough()를 활용하는 방법:

const reader = body.pipeThrough(new TextDecoderStream()).getReader();

while (true) {
  const readable = await reader.read();
  if (readable.done) break;
  // readable.value는 이미 디코딩된 문자열
}

이 방법에는 두 가지 장점이 있습니다.

• TextDecoder로 매번 디코딩하는 로직을 줄일 수 있습니다.
• 파이핑을 하면 그 동안 스트림이 잠기기(lock) 때문에 다른 곳에서 동시에 읽는 것을 방지할 수 있습니다.

처음 구현할 때는 첫 번째 방법을 사용했는데, 나중에 pipeThrough를 알게 되어서 리팩토링할 기회가 있다면 적용해볼 생각입니다.

MDN 문서에서 더 자세한 내용을 확인할 수 있습니다.