5장: 스트리밍 LLM 응답 처리

data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":null}]}
 
data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"안녕"},"finish_reason":null}]}
 
data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"하세요"},"finish_reason":null}]}
 
data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}
 
data: [DONE]

event: message_start
data: {"type":"message_start","message":{"id":"msg_abc","type":"message","role":"assistant","model":"claude-sonnet-4-20250514","usage":{"input_tokens":25,"output_tokens":1}}}
 
event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
 
event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"안녕하세요"}}
 
event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":", 무엇을"}}
 
event: content_block_stop
data: {"type":"content_block_stop","index":0}
 
event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":15}}
 
event: message_stop
data: {"type":"message_stop"}

data: {"candidates":[{"content":{"parts":[{"text":"안녕"}],"role":"model"},"finishReason":null,"index":0}]}
 
data: {"candidates":[{"content":{"parts":[{"text":"하세요"}],"role":"model"},"finishReason":null,"index":0}]}
 
data: {"candidates":[{"content":{"parts":[{"text":"!"}],"role":"model"},"finishReason":"STOP","index":0}],"usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":5}}

type Provider = "openai" | "anthropic" | "google";
 
interface StreamToken {
  content: string;
  isFinished: boolean;
  usage?: { inputTokens: number; outputTokens: number };
}
 
function parseStreamEvent(
  provider: Provider,
  eventData: string
): StreamToken | null {
  if (eventData === "[DONE]") {
    return { content: "", isFinished: true };
  }
 
  const data = JSON.parse(eventData);
 
  switch (provider) {
    case "openai": {
      const delta = data.choices?.[0]?.delta;
      const finished = data.choices?.[0]?.finish_reason === "stop";
      return {
        content: delta?.content ?? "",
        isFinished: finished,
        usage: data.usage
          ? {
              inputTokens: data.usage.prompt_tokens,
              outputTokens: data.usage.completion_tokens,
            }
          : undefined,
      };
    }
 
    case "anthropic": {
      if (data.type === "content_block_delta") {
        return {
          content: data.delta?.text ?? "",
          isFinished: false,
        };
      }
      if (data.type === "message_stop") {
        return { content: "", isFinished: true };
      }
      if (data.type === "message_delta") {
        return {
          content: "",
          isFinished: data.delta?.stop_reason != null,
          usage: data.usage
            ? {
                inputTokens: 0,
                outputTokens: data.usage.output_tokens,
              }
            : undefined,
        };
      }
      return null;
    }
 
    case "google": {
      const candidate = data.candidates?.[0];
      const text = candidate?.content?.parts?.[0]?.text ?? "";
      const finished = candidate?.finishReason === "STOP";
      return { content: text, isFinished: finished };
    }
  }
}

토큰 1: {"na
토큰 2: me": "
토큰 3: 홍길동",
토큰 4:  "age":
토큰 5:  30, "
토큰 6: skills
토큰 7: ": ["Py
토큰 8: thon"]}

/**
 * 불완전한 JSON 문자열을 파싱 가능한 형태로 복구합니다.
 * 열린 괄호/중괄호를 추적하여 닫아줍니다.
 */
function parsePartialJSON(incomplete: string): unknown | null {
  // 빈 문자열이면 null 반환
  if (!incomplete.trim()) return null;
 
  // 완전한 JSON이면 그대로 파싱
  try {
    return JSON.parse(incomplete);
  } catch {
    // 계속 진행
  }
 
  // 열린 구조를 닫아서 파싱 시도
  let fixed = incomplete;
  const stack: string[] = [];
  let inString = false;
  let escaped = false;
 
  for (const char of incomplete) {
    if (escaped) {
      escaped = false;
      continue;
    }
    if (char === "\\") {
      escaped = true;
      continue;
    }
    if (char === '"') {
      inString = !inString;
      continue;
    }
    if (inString) continue;
 
    if (char === "{") stack.push("}");
    else if (char === "[") stack.push("]");
    else if (char === "}" || char === "]") stack.pop();
  }
 
  // 문자열이 열려있으면 닫기
  if (inString) fixed += '"';
 
  // 미완성 값 처리 (trailing comma 등)
  fixed = fixed.replace(/,\s*$/, "");
 
  // 열린 구조 닫기
  while (stack.length > 0) {
    fixed += stack.pop();
  }
 
  try {
    return JSON.parse(fixed);
  } catch {
    return null;
  }
}

import { useState, useEffect, useRef, useCallback } from "react";
 
interface StreamingMessageProps {
  endpoint: string;
  message: string;
  onComplete?: (fullText: string) => void;
}
 
function StreamingMessage({
  endpoint,
  message,
  onComplete,
}: StreamingMessageProps) {
  const [tokens, setTokens] = useState<string[]>([]);
  const [isStreaming, setIsStreaming] = useState(false);
  const abortRef = useRef<AbortController | null>(null);
 
  const startStreaming = useCallback(async () => {
    setIsStreaming(true);
    setTokens([]);
 
    const controller = new AbortController();
    abortRef.current = controller;
 
    try {
      const response = await fetch(endpoint, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ message }),
        signal: controller.signal,
      });
 
      const reader = response.body?.getReader();
      const decoder = new TextDecoder();
 
      if (!reader) return;
 
      let buffer = "";
      const accumulated: string[] = [];
 
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
 
        buffer += decoder.decode(value, { stream: true });
        const events = buffer.split("\n\n");
        buffer = events.pop() ?? "";
 
        for (const event of events) {
          const dataLine = event
            .split("\n")
            .find((l) => l.startsWith("data: "));
          if (!dataLine) continue;
 
          const data = dataLine.slice(6);
          if (data === "[DONE]") break;
 
          accumulated.push(data);
          setTokens([...accumulated]);
        }
      }
 
      onComplete?.(accumulated.join(""));
    } catch (error) {
      if (error instanceof DOMException && error.name === "AbortError") {
        // 사용자가 취소함
      } else {
        console.error("스트리밍 오류:", error);
      }
    } finally {
      setIsStreaming(false);
    }
  }, [endpoint, message, onComplete]);
 
  useEffect(() => {
    startStreaming();
    return () => abortRef.current?.abort();
  }, [startStreaming]);
 
  const handleStop = () => {
    abortRef.current?.abort();
  };
 
  return (
    <div className="streaming-message">
      <div className="content">
        {tokens.join("")}
        {isStreaming && <span className="cursor-blink">|</span>}
      </div>
      {isStreaming && (
        <button onClick={handleStop} className="stop-button">
          생성 중단
        </button>
      )}
    </div>
  );
}

class TokenBatcher {
  private buffer: string[] = [];
  private rafId: number | null = null;
  private onFlush: (tokens: string[]) => void;
 
  constructor(onFlush: (tokens: string[]) => void) {
    this.onFlush = onFlush;
  }
 
  add(token: string) {
    this.buffer.push(token);
 
    if (this.rafId === null) {
      this.rafId = requestAnimationFrame(() => {
        this.onFlush([...this.buffer]);
        this.buffer = [];
        this.rafId = null;
      });
    }
  }
 
  dispose() {
    if (this.rafId !== null) {
      cancelAnimationFrame(this.rafId);
    }
  }
}

import { useChat } from "ai/react";
 
function ChatInterface() {
  const {
    messages,
    input,
    handleInputChange,
    handleSubmit,
    isLoading,
    stop,
    error,
    reload,
  } = useChat({
    api: "/api/chat",
    onFinish: (message) => {
      console.log("완료:", message.content);
    },
    onError: (error) => {
      console.error("오류:", error);
    },
  });
 
  return (
    <div className="chat-container">
      <div className="messages">
        {messages.map((msg) => (
          <div key={msg.id} className={`message ${msg.role}`}>
            {msg.content}
          </div>
        ))}
      </div>
 
      <form onSubmit={handleSubmit}>
        <input
          value={input}
          onChange={handleInputChange}
          placeholder="메시지를 입력하세요"
          disabled={isLoading}
        />
        {isLoading ? (
          <button type="button" onClick={stop}>
            중단
          </button>
        ) : (
          <button type="submit">전송</button>
        )}
      </form>
 
      {error && (
        <div className="error">
          오류가 발생했습니다.
          <button onClick={() => reload()}>재시도</button>
        </div>
      )}
    </div>
  );
}

import { streamText } from "ai";
import { anthropic } from "@ai-sdk/anthropic";
 
export async function POST(request: Request) {
  const { messages } = await request.json();
 
  const result = streamText({
    model: anthropic("claude-sonnet-4-20250514"),
    messages,
  });
 
  return result.toDataStreamResponse();
}

async function resilientStream(
  endpoint: string,
  payload: Record<string, unknown>,
  options: {
    maxRetries?: number;
    timeoutMs?: number;
    onToken: (token: string) => void;
    onError: (error: Error) => void;
  }
) {
  const { maxRetries = 3, timeoutMs = 30000, onToken, onError } = options;
  let retries = 0;
  let accumulatedText = "";
 
  while (retries <= maxRetries) {
    try {
      const controller = new AbortController();
      const timeout = setTimeout(
        () => controller.abort(),
        timeoutMs
      );
 
      const response = await fetch(endpoint, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({
          ...payload,
          // 재시도 시 이미 받은 텍스트 이후부터 요청
          resumeAfter: accumulatedText || undefined,
        }),
        signal: controller.signal,
      });
 
      clearTimeout(timeout);
 
      if (response.status === 429) {
        const retryAfter = parseInt(
          response.headers.get("Retry-After") ?? "5"
        );
        await delay(retryAfter * 1000);
        retries++;
        continue;
      }
 
      if (!response.ok) {
        throw new Error(`HTTP ${response.status}`);
      }
 
      // 정상 스트리밍 처리
      await processStream(response, (token) => {
        accumulatedText += token;
        onToken(token);
      });
 
      return; // 성공적으로 완료
    } catch (error) {
      retries++;
      if (retries > maxRetries) {
        onError(
          error instanceof Error
            ? error
            : new Error("Unknown error")
        );
        return;
      }
      // 지수 백오프
      await delay(Math.pow(2, retries) * 1000);
    }
  }
}
 
function delay(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}