스트리밍

TanStack AI는 실시간 채팅 환경을 위한 스트리밍 응답을 지원합니다. 스트리밍을 사용하면 전체 응답이 완료될 때까지 기다리지 않고 응답이 생성되는 즉시 표시할 수 있습니다.

스트리밍 작동 방식

chat()은 사양에 따른 AG-UI 청크의 비동기 이터러블을 반환합니다. chunk.type에 따라 분기합니다.

import { chat } from "@tanstack/ai";
import { openaiText } from "@tanstack/ai-openai";

const stream = chat({
  adapter: openaiText("gpt-5.5"),
  messages: [{ role: "user", content: "Hello!" }],
});

for await (const chunk of stream) {
  if (chunk.type === "TEXT_MESSAGE_CONTENT") {
    console.log(chunk.delta);
  }
  if (chunk.type === "RUN_FINISHED") {
    console.log(chunk.usage);
    console.log(chunk.metadata?.tanstack?.finishReason);
  }
}

서버 측 스트리밍

toServerSentEventsResponse를 사용해 스트림을 HTTP 응답으로 변환합니다.

import { chat, toServerSentEventsResponse } from "@tanstack/ai";
import { openaiText } from "@tanstack/ai-openai";

export async function POST(request: Request) {
  const { messages } = await request.json();

  const stream = chat({
    adapter: openaiText("gpt-5.5"),
    messages,
  });

  // Convert to HTTP response with proper headers
  return toServerSentEventsResponse(stream);
}

클라이언트 측 스트리밍

useChat 훅이 스트리밍을 자동으로 처리합니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";

const { messages, sendMessage, isLoading } = useChat({
  connection: fetchServerSentEvents("/api/chat"),
});

// Messages update in real-time as chunks arrive
messages.forEach((message) => {
  // Message content updates incrementally
});

스트림 이벤트(AG-UI Protocol)

TanStack AI는 스트리밍을 위해 AG-UI Protocol을 구현합니다. 스트림 이벤트에는 다양한 유형의 데이터가 포함됩니다.

AG-UI 이벤트

공개 StreamChunk는 AG-UI 이벤트 유형을 따릅니다. TanStack의 추가 정보는 metadata.tanstack 아래에 있습니다.

`chunk.type`	읽는 값
`RUN_STARTED`	`threadId`, `runId`
`TEXT_MESSAGE_START` / `CONTENT` / `END`	`messageId`, `delta`
`TOOL_CALL_START` / `ARGS` / `END`	`toolCallId`, `toolCallName`, 인자 `delta`. 파싱된 input과 output은 `UIMessage` 파트에 있습니다.
`REASONING_*` / `REASONING_ENCRYPTED_VALUE`	사고 콘텐츠입니다. 사고와 추론을 참고하세요.
`STEP_STARTED` / `STEP_FINISHED`	`stepName`만 포함합니다.
`CUSTOM`	`name`과 `value`입니다(샌드박스 파일, Code Mode, `structured-output.`, `.session-id`, `emitCustomEvent` 호출). 사용자 지정 이벤트를 참고하세요.
`RUN_FINISHED` / `RUN_ERROR`	프로세스 내부의 `chat()`은 여전히 TanStack `TokenUsage`(`promptTokens`)를 사용합니다. SSE/HTTP wire는 사양의 `usage` 배열(`inputTokens`)을 사용합니다. `finishReason`은 `metadata.tanstack.finishReason`입니다. 사용자 지정 서버는 이벤트 메타데이터를 참고하세요.

스레드와 실행

모든 스트림에는 두 ID가 사용되며, 이는 저장소 계층이 아니라 AG-UI protocol 자체에서 제공됩니다.

스레드(threadId)는 대화입니다. 모든 교환, 새로고침, 기기에서 유지되는 안정적인 식별자입니다.
실행(runId)은 그 안에서의 한 번의 실행입니다. 하나의 RUN_STARTED와 해당 RUN_FINISHED(또는 RUN_ERROR) 사이의 모든 과정입니다. 시작할 때마다 새 실행 ID가 발급되므로 스레드에는 수명 동안 여러 실행이 누적됩니다.

실행은 단일 모델 응답으로 제한되지 않습니다. 도구 호출과 후속 응답은 같은 실행 안에서 스트리밍됩니다. 따라서 반복 횟수와 관계없이 전체 에이전트 사이클이 하나의 실행입니다.

flowchart LR
    subgraph thread ["Thread — threadId (stable)"]
        direction LR
        subgraph r1 ["Run r1 — finished"]
            direction TB
            e1["RUN_STARTED → text → tool call → tool result → final text → RUN_FINISHED"]
        end
        subgraph r2 ["Run r2 — finished"]
            direction TB
            e2["RUN_STARTED → text → RUN_FINISHED"]
        end
        subgraph r3 ["Run r3 — running"]
            direction TB
            e3["RUN_STARTED → text"]
        end
        r1 --> r2 --> r3
    end

실행 ID는 일시적이므로 오래 유지되는 항목은 스레드를 기준으로 합니다. 재개 가능한 스트림은 runId별 전달을 기록하고, 서버 영속성은 threadId별 대화 기록을 저장합니다. 미디어 생성 훅도 threadId를 받지만, 이때는 대화가 아니라 슬롯을 나타냅니다. ID 맵을 참고하세요.

도구 입력과 출력

SSE와 HTTP의 TOOL_CALL_END에는 파싱된 input이 포함되지 않습니다. 프로세스 내부의 chat()에는 여전히 input이 있습니다. 도구 입력과 출력은 UIMessage 파트에도 저장됩니다. 서버에서는 청크를 StreamProcessor에 전달합니다. 클라이언트에서는 useChat의 messages를 읽습니다.

Server:

import { chat, StreamProcessor, toolDefinition } from "@tanstack/ai";
import { openaiText } from "@tanstack/ai-openai";
import { z } from "zod";

const weatherTool = toolDefinition({
  name: "get_weather",
  description: "Get weather for a location",
  inputSchema: z.object({
    location: z.string(),
    unit: z.enum(["celsius", "fahrenheit"]).optional(),
  }),
});

const stream = chat({
  adapter: openaiText("gpt-5.5"),
  messages: [
    { role: "user", content: "What's the weather in Paris?" },
  ],
  tools: [weatherTool],
});

const processor = new StreamProcessor();
for await (const chunk of stream) {
  processor.processChunk(chunk);
}
processor.finalizeStream();

for (const message of processor.getMessages()) {
  for (const part of message.parts) {
    if (part.type === "tool-call") {
      console.log(part.name, part.input, part.output);
    }
  }
}

클라이언트에서는 .client() 도구를 useChat에 전달합니다. part.name을 확인하면 part.input과 part.output의 타입 범위가 좁혀집니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";
import { toolDefinition } from "@tanstack/ai";
import { z } from "zod";

const weatherTool = toolDefinition({
  name: "get_weather",
  description: "Get weather for a location",
  inputSchema: z.object({
    location: z.string(),
    unit: z.enum(["celsius", "fahrenheit"]).optional(),
  }),
}).client(async (input) => {
  return { location: input.location };
});

const { messages } = useChat({
  connection: fetchServerSentEvents("/api/chat"),
  tools: [weatherTool],
});

for (const message of messages) {
  for (const part of message.parts) {
    if (part.type === "tool-call" && part.name === "get_weather") {
      console.log(part.input?.location);
    }
  }
}

사고 청크

사고 콘텐츠는 REASONING_* 및 REASONING_ENCRYPTED_VALUE 이벤트에서 제공됩니다. STEP_STARTED와 STEP_FINISHED에는 stepName만 포함됩니다. message.parts에서 ThinkingPart를 읽습니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";

const { messages } = useChat({
  connection: fetchServerSentEvents("/api/chat"),
});

for (const message of messages) {
  for (const part of message.parts) {
    if (part.type === "thinking") {
      console.log("Thinking:", part.content);
    }
  }
}

사고 콘텐츠는 UIMessage 객체에서 자동으로 ThinkingPart로 변환됩니다. 서명되지 않은 사고 콘텐츠는 UI에 유지됩니다. 서명된 사고 콘텐츠는 원래 순서대로 모델에 재생됩니다. 전체 렌더링 패턴은 사고와 추론을 참고하세요.

연결 어댑터

TanStack AI는 다양한 스트리밍 protocol을 위한 연결 어댑터를 제공합니다.

서버 보낸 이벤트 (SSE)

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";

const { messages } = useChat({
  connection: fetchServerSentEvents("/api/chat"),
});

HTTP 스트림

import { useChat, fetchHttpStream } from "@tanstack/ai-react";

const { messages } = useChat({
  connection: fetchHttpStream("/api/chat"),
});

사용자 지정 스트림

완전히 사용자 지정된 요청에는 fetcher transport를 사용합니다. fetcher는 요청 입력과 AbortSignal을 받고, Response(클라이언트가 SSE 본문을 파싱함) 또는 AsyncIterable<StreamChunk>를 반환합니다. 이 값은 동기적으로, Promise로, 또는 async function*으로 반환할 수 있습니다.

import { useChat } from "@tanstack/ai-react";

const { messages } = useChat({
  fetcher: ({ messages, data }, { signal }) =>
    fetch("/api/chat", {
      method: "POST",
      body: JSON.stringify({ messages, ...data }),
      signal,
    }),
});

참고: 하위 수준의 stream() 연결 어댑터는 AsyncIterable<StreamChunk>를 동기적으로 반환해야 하는 팩토리(예: generator)를 받습니다. Promise를 반환하는 async (...) => {...} 함수는 허용하지 않습니다. 연결 어댑터가 특별히 필요한 경우가 아니라면 위의 fetcher transport를 사용하는 것이 좋습니다.

스트림 진행 상황 모니터링

콜백을 사용해 스트림 진행 상황을 모니터링할 수 있습니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";

const { messages } = useChat({
  connection: fetchServerSentEvents("/api/chat"),
  onChunk: (chunk) => {
    console.log("Received chunk:", chunk);
  },
  onFinish: (message) => {
    console.log("Stream finished:", message);
  },
});

스트림 취소

진행 중인 스트림을 취소합니다.

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";

const { stop } = useChat({
  connection: fetchServerSentEvents("/api/chat"),
});

// Cancel the current stream
stop();

stop()을 호출하면 기반 fetch가 중단됩니다. 그 결과 발생하는 AbortError는 예상된 정상 동작입니다. 해당 턴의 보류 중인 클라이언트 도구 작업은 재개되지 않습니다. 이후 해당 턴에 대한 addToolResult()는 무시됩니다. 이는 스트림 중간에서 연결이 끊기는 경우와 다릅니다. 잘린 스트림은 StreamTruncatedError를 발생시키고 클라이언트를 error 상태로 전환합니다. 기반 동작은 연결 어댑터를 참고하세요.

서버에서는 클라이언트 연결이 끊길 때 채팅 실행이 취소되도록 toServerSentEventsResponse(stream, { abortController })에 AbortController를 전달합니다.

import { chat, toServerSentEventsResponse } from "@tanstack/ai";
import { openaiText } from "@tanstack/ai-openai";

export async function POST(request: Request) {
  const { messages } = await request.json();
  const stream = chat({ adapter: openaiText("gpt-5.5"), messages });

  const abortController = new AbortController();
  return toServerSentEventsResponse(stream, { abortController });
}

메시지 대기열 처리

기본적으로 스트림이 이미 진행 중일 때 sendMessage를 호출하면 메시지를 삭제하지 않고 대기열에 추가합니다. 현재 실행이 성공적으로 완료되면 자동으로 전송됩니다. queue 옵션으로 설정할 수 있으며, 다음 세 가지 형식을 지원합니다.

QueueConfig 객체
일반적인 축약 문자열
전략 함수

import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";

const { messages, queue, sendMessage, cancelQueued, isLoading } = useChat({
  connection: fetchServerSentEvents("/api/chat"),
  queue: { whenBusy: "queue", drain: "fifo", maxSize: 5 },
});

whenBusy — 클라이언트가 사용 중일 때(스트리밍 중, 전송을 처리 중, 또는 대기열을 비우는 중) 도착한 전송을 처리하는 방식입니다.
- "queue"(기본값) — 메시지를 보류하고 실행이 성공적으로 완료되면 전송합니다. 메시지가 queue 또는 messages에 나타나면 composer를 비웁니다.
- "drop" — 전송을 무시합니다(프로미스는 여전히 resolve되며 예외를 발생시키지 않음). 메시지는 queue나 messages에 나타나지 않으므로, 사용자가 재시도할 수 있도록 composer 텍스트를 유지하고 피드백을 표시합니다.
- "interrupt" — 현재 스트림을 중단하고 새 메시지를 즉시 전송합니다. stop()과 달리 이미 대기열에 추가된 메시지는 삭제하지 않으며, 중단 전송이 성공하면 계속 처리됩니다.
drain — 대기열 항목을 꺼내는 방식입니다. "fifo"(기본값)는 순서대로 하나씩 전송하고, "batch"는 실행이 성공적으로 완료되면 현재 대기열의 모든 항목을 하나의 전송으로 병합합니다(문자열 콘텐츠는 \n으로 결합하고 멀티모달 콘텐츠는 순서대로 연결하며, 메시지별 body로 전송할 때는 마지막 항목의 body가 우선함).
maxSize — 대기열에 추가할 수 있는 메시지 수를 제한합니다(0은 대기열에 추가하지 않음을 의미함).
onOverflow — "reject"(기본값)는 maxSize에 도달하면 전송을 조용히 무시하고(예외를 발생시키지 않음), "drop-oldest"는 공간을 만들기 위해 가장 오래된 대기열 항목을 제거합니다.

일반적인 WhenBusy 문자열(예: queue: "interrupt")을 { whenBusy: "interrupt" }의 축약형으로 전달하거나, 전송별 작업을 제어하는 QueueStrategy 함수를 전달할 수도 있습니다. 전략 형식은 항상 FIFO로 처리되며(batch는 사용하지 않음), 작업에는 WhenBusy 타입이 사용됩니다. sendMessage의 두 번째 인자에 지정한 전송별 whenBusy는 설정과 전략을 모두 재정의합니다.

대기열을 비우는 경우와 플러시하는 경우

비우기(자동 전송) — 도구 후속 작업이 완료된 경우를 포함해 스트림이 성공적으로 완료된 후에만 수행됩니다.
플러시(전송하지 않고 폐기) — 활성 생성의 오류/중단(사용자의 stop(), 실제 스트림 오류), clear(), unsubscribe(), reload()에서 수행됩니다. 인터럽트는 이전 실행을 중단하지만 플러시하지 않으며, 남은 항목은 인터럽트를 발생시킨 턴이 성공하면 처리됩니다.
interrupt는 플러시하지 않음 — 기존 대기열 항목은 유지되고 인터럽트 턴이 성공한 후 처리됩니다.

useChat은 보류 중인 대기열을 messages와 구분해 렌더링할 수 있도록 queue로 노출하며, 전송 전에 항목을 취소하는 cancelQueued(id)도 제공합니다.

function PendingQueue() {
  return (
    <>
      {queue.map((q) => (
        <div key={q.id} className="pending">
          {typeof q.content === "string" ? q.content : "[attachment]"}
          <button onClick={() => cancelQueued(q.id)}>Cancel</button>
        </div>
      ))}
    </>
  );
}

sendMessage의 두 번째 인자를 사용해 단일 전송에 대해 설정된 정책을 재정의합니다. 같은 객체에서 해당 요청에만 사용할 추가 JSON인 body도 허용하며, forwardedProps에 병합됩니다.

sendMessage("Never mind, do this instead", {
  whenBusy: "interrupt",
  body: { source: "composer" },
});

참고: 이는 기본 동작의 변경입니다. 스트리밍 중 전송된 메시지는 이전에는 조용히 삭제되었습니다. 이제 queue: "drop"(또는 { whenBusy: "drop" })을 선택해 이전 동작을 복원하거나 queue: "interrupt"를 선택하지 않는 한 대기열에 추가됩니다.

모범 사례

로딩 상태 처리 - 로딩 표시기를 표시하려면 isLoading을 사용합니다.
오류 처리 - 스트림 오류가 있는지 error 상태를 확인합니다.
마운트 해제 시 취소 - 컴포넌트가 마운트 해제될 때 스트림을 정리합니다.
렌더링 최적화 - 성능을 위해 필요한 경우 업데이트를 일괄 처리합니다.
진행 상황 표시 - 스트리밍되는 부분 콘텐츠를 표시합니다.
대기열 메시지 구분 렌더링 - queue를 사용해 보류 중인 전송을 messages와 구분해 표시합니다.

다음 단계

연결 어댑터 - 다양한 연결 유형을 알아봅니다.
API 레퍼런스 - 스트리밍 API를 살펴봅니다.

스트리밍 작동 방식​

서버 측 스트리밍​

클라이언트 측 스트리밍​

스트림 이벤트(AG-UI Protocol)​

AG-UI 이벤트​

스레드와 실행​

도구 입력과 출력​

사고 청크​

연결 어댑터​

서버 보낸 이벤트 (SSE)​

HTTP 스트림​

사용자 지정 스트림​

스트림 진행 상황 모니터링​

스트림 취소​

메시지 대기열 처리​

대기열을 비우는 경우와 플러시하는 경우​

모범 사례​

다음 단계​