流式传输 | TanStack AI 文档

TanStack AI 支持流式响应，用于实时聊天体验。流式传输允许您在生成响应时显示响应，而不是等待完整的响应。

流式传输的工作原理

当您使用 chat() 时，它会返回一个异步可迭代的块流

typescript

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

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

// Stream contains chunks as they arrive
for await (const chunk of stream) {
  console.log(chunk); // Process each chunk
}

服务器端流式传输

使用 toServerSentEventsResponse 将流转换为 HTTP 响应

typescript

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.2"),
    messages,
  });

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

客户端流式传输

useChat hook 会自动处理流式传输

typescript

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 协议)

TanStack AI 实现了 AG-UI 协议进行流式传输。流事件包含不同类型的数据

AG-UI 事件

RUN_STARTED - 在运行开始时发出
TEXT_MESSAGE_START/CONTENT/END - 文本内容流式传输生命周期
TOOL_CALL_START/ARGS/END - 工具调用生命周期
STEP_STARTED/STEP_FINISHED - 思考/推理步骤
RUN_FINISHED - 运行完成，包含完成原因和使用情况
RUN_ERROR - 运行过程中发生错误

思考块

思考/推理由 AG-UI 事件 STEP_STARTED 和 STEP_FINISHED 表示。它们与最终响应文本分开流式传输

typescript

for await (const chunk of stream) {
  if (chunk.type === "STEP_FINISHED") {
    console.log("Thinking:", chunk.content); // Accumulated thinking content
    console.log("Delta:", chunk.delta); // Incremental thinking token
  }
}

思考内容会自动转换为 UIMessage 对象中的 ThinkingPart。它仅用于 UI，不包含发送回模型的消息。

连接适配器

TanStack AI 提供用于不同流式传输协议的连接适配器

服务器发送事件 (SSE)

typescript

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

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

HTTP 流

typescript

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

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

自定义流

typescript

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

const { messages } = useChat({
  connection: stream(async (messages, data, signal) => {
    // Custom streaming implementation
    const response = await fetch("/api/chat", {
      method: "POST",
      body: JSON.stringify({ messages, ...data }),
      signal,
    });
    // Return async iterable
    return processStream(response);
  }),
});

监控流进度

您可以使用回调函数监控流进度

typescript

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

取消流

取消正在进行的流

typescript

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

// Cancel the current stream
stop();

最佳实践

处理加载状态 - 使用 isLoading 显示加载指示器
处理错误 - 检查 error 状态以获取流失败情况
卸载时取消 - 在组件卸载时清理流
优化渲染 - 如有需要，批量更新以提高性能
显示进度 - 在流式传输时显示部分内容

下一步

连接适配器 - 了解不同的连接类型
API 参考 - 探索流式传输 API