EventSource로 LLM 스트림을 처리할 수 없는 이유와 해결책

채팅 완료(chat completion)를 스트리밍으로 받고 싶습니다. 응답은 text/event-stream 형식이므로, 브라우저에 내장된 EventSource가 가장 당연한 도구처럼 보입니다:

const es = new EventSource("https://api.openai.com/v1/chat/completions");

하지만 이 코드는 작동하지 않습니다. 그리고 그 이유가 사소해서 문제가 됩니다. EventSource는 GET 요청만 보낼 수 있으며, 헤더를 단 하나도 설정할 수 없습니다. 그런데 채팅 완료 엔드포인트는 정반대를 요구합니다: POST 요청, Authorization: Bearer … 헤더, 그리고 메시지를 담은 JSON 본문이 필요합니다. 플랫폼이 SSE(Server-Sent Events)를 위해 제공하는 유일한 도구는 구조적으로 LLM 요청을 처리할 수 없습니다.

그래서 우리는 fetch로 전환하게 되고 — 이제 파서(parser)는 우리가 직접 소유하게 됩니다:

const res = await fetch(url, { method: "POST", headers, body ");
const reader = res.body!.getReader();
const decoder = new TextDecoder();
...

이 모든 주석들은 사람들이 실제로 버그로 배포하는 부분입니다. 이것은 모든 스트리밍 통합에서 아무도 블로그 게시물을 작성하지 않는 부분입니다. 제대로 고쳐봅시다.

일반적인 옵션들 (그리고 그 문제점)

1. 네이티브 EventSource. GET만 가능하고 헤더 설정이 불가능합니다 — LLM API에는 부적합합니다. 그리고 작동하더라도, 깨끗하게 종료되는 스트림을 포함하여 모든 스트림 종료 시 재연결됩니다. 만약 LLM 응답에 오류가 있다면: 깨끗한 종료는 모델이 _완료했다_는 의미입니다.

2. @microsoft/fetch-event-source. 잘 알려진

sse-wire

sse-wire는 의존성 없이 fetch 기반의 SSE 클라이언트이며, for await로 소비할 수 있습니다:

import { sse } from "sse-wire";

const controller = new AbortController();
...

이것이 일반적인 경우에 필요한 모든 기능입니다. 이 라이브러리가 강력한 세 가지 이유가 있습니다:

• 헤더를 포함하여 POST 요청을 보냅니다. 내부적으로 fetch를 사용하기 때문에, 메서드(method), 본문(body), 헤더(headers)가 1급 시민(first-class)으로 다뤄집니다. 만약 지정하지 않았다면 자동으로 accept: text/event-stream을 설정하며, 전달된 signal도 그대로 통과시켜서 이미 가지고 있는 AbortController로 요청과 스트림을 모두 취소할 수 있습니다.
• 비동기 제너레이터(async generator)입니다. for await (const event of sse(...)). [DONE]에 도달하면 break하고, try/finally로 감싸서 다른 모든 반복 가능한 객체처럼 구성할 수 있습니다. 리스너를 연결하거나 .close()를 호출할 필요가 없습니다.
• 깔끔한 종료는 완료를 의미합니다. 스트림이 정상적으로 닫히면(모델이 끝났다는 의미) 루프가 종료됩니다. 완료된 생성에 대해 조용히 재연결하여 다시 요청하는 일은 없습니다.

누수 없는 취소 (Cancellation that doesn't leak)

Abort를 호출하면 sse-wire는 표준 AbortError를 발생시키고, 진행 중인(in-flight) fetch와 스트림 리더(stream reader)를 모두 취소합니다. 루프에서 일찍 break하더라도 마찬가지입니다. 이때도 finally 블록에서 리더가 취소되므로 연결이 끊긴 채로 남아있는 일이 없습니다. 이는 테스트 케이스에 의해 검증되며, 해당 테스트는 abort와 조기 break 모두에서 리더의 cancel() 메서드가 실제로 실행되는지 확인합니다.

const controller = new AbortController();
setTimeout(() => controller.abort(), 5_000); // 또는 AbortSignal.timeout(5_000)

실제 상황에서만 재연결 (Reconnection, only when it's real)

재연결은 선택적이며 전송 계층 오류(transport errors)에 대해서만 발생합니다. 즉, 답변 중간에 연결이 끊기는 경우에만 작동하며, 깔끔한 종료 시에는 절대 재요청하지 않습니다:

for await (const event of sse(url, { method: "POST", headers, body, reconnect: true })) {
  handle(event);
}

실제로 연결이 끊어지면, 동일한 지터(jitter)의 지수 백오프(exponential backoff)를 기다리고 (서버의 retry: 지시어가 기본 지연 시간이 됩니다), 마지막으로 본 이벤트에 Last-Event-ID를 설정하여 요청을 재전송하며, 같은 루프로 계속 이벤트를 생성합니다. 따라서 소비자(consumer)는 그 경계(seam)를 전혀 알아채지 못합니다.

파서가 가장 어려운 부분입니다

와이어 형식(wire format)이 까다롭기 때문에 sse-wire가 이를 처리해 줍니다: CR / LF / CRLF 종료 문자 (심지어 두 청크에 걸쳐 분리된 CRLF도), \n으로 연결된 다중 라인 data: 필드, 주석, 선행 BOM(Byte Order Mark), id 내의 NUL 바이트, 그리고 줄바꿈이 없는 마지막 라인까지 모두 처리합니다. 청크 및 UTF-8 경계를 넘어 버퍼링하기 때문에, 멀티바이트 문자가 중간에 분리되어도 올바르게 디코딩됩니다. 이 라이브러리는 동일한 페이로드(payload)를 모든 바이트 오프셋마다 분할하여 테스트하고, 이벤트가 일회성 파싱과 동일하게 나오는지를 검증합니다.

따라서 fetch 본체뿐만 아니라 모든 바이트/텍스트 스트림에 이 파서를 직접 사용할 수 있습니다:

import { parseSSE } from "sse-wire";

for await (const event of parseSSE(someByteStream)) {
...

이는 파이프라인의 앞단입니다

sse-wire는 스트리밍 구조화된 출력 흐름(structured-output flow)에서 전송(transport) 단계를 담당하며, 가장 먼저 위치합니다. 이 라이브러리의 두 형제들은 sse-wire가 생성하는 이벤트부터 처리합니다:

fetch → SSE (sse-wire) → 부분 JSON 파싱 (trickle-json) → 스키마로 복구/강제 변환 (coerce-json) → 검증

• sse-wire — 요청을 POST하고 이벤트를 스트리밍합니다. (이 라이브러리)
• trickle-json — 스트리밍되는 토큰 델타(token deltas)를 각 청크에서 사용 가능한 최적의 유효 부분 값으로 조립하며, 오류를 발생시키지 않습니다.
• coerce-json — 해당 값을 Zod / JSON Schema에 맞게 복구하고 강제 변환(coerce)하며, 모든 수정 사항을 기록합니다.

이 세 가지를 모두 사용하여 완료 스트림을 전송하고, 도착하는 대로 JSON을 조립하며, 결과를 스키마로 강제 변환하는 예시입니다:

sse-wire가 전송 계층(transport)을 담당하고; trickle-json은 청크마다 에러를 발생시키지 않으면서 최고의 값을 제공하며; coerce-json은 이를 사용자의 스키마에 맞게 조정하여 결과를 전달합니다. 이 세 가지 라이브러리 모두 의존성이 없고 독립적으로 작동하므로, 필요한 부분만 채택할 수 있습니다.

"하지만 제공업체(provider)가 구조화된 출력/공식 SDK를 이미 처리하지 않나요?" SDK들이 도움이 되기는 하지만, 스트림을 해당 요청 생명주기 및 제공업체의 형태에 종속시킵니다. sse-wire는 순수하고 프레임워크에 구애받지 않는(framework-agnostic) fetch SSE 클라이언트입니다. 어떤 파이프라인에도 넣고, 어떤 text/event-stream(LLM 여부 무관)을 가리키게 하며, 테스트에서 모킹할 수 있고, 깨끗한 이벤트를 다음 단계로 전달합니다.

사용해보기

npm install sse-wire

• npm: https://www.npmjs.com/package/sse-wire
• GitHub: https://github.com/H1manshu01/sse-wire
• 런타임 의존성 없음, ESM + CJS 지원, 전체 타입 정의, 최소 크기 약 1.4 kB+brotli 압축, 출처(provenance)와 함께 게시됨. Node ≥ 18, 브라우저, Bun에서 실행 가능.

만약 이 라이브러리가 처리해서는 안 될 스트림을 잘못 처리한다면 — 종료 조건(terminator edge case), 청크 경계(chunk boundary), 누수되는 중단(abort that leaks) 등 — 재현 가능한 예시와 함께 이슈를 제기해 주세요. 파서의 동등성(parser-parity)과 리더 누수 방지(no-leaked-reader) 보장이 핵심이므로, 알려주시면 감사하겠습니다. ⭐ 수동으로 작성한 fetch 루프를 아껴준다면 더욱 감사합니다.