连接适配器 | TanStack AI 文档

连接适配器处理客户端和服务器之间的通信，用于流式传输聊天响应。TanStack AI 提供了内置适配器并支持自定义实现。

内置适配器

服务器发送事件 (SSE)

SSE 是大多数用例的推荐适配器。它提供可靠的流式传输和自动重连。在服务器端，使用 toServerSentEventsStream() 或 toServerSentEventsResponse() 将您的聊天流转换为 SSE 格式。

typescript

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

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

选项

typescript

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

动态值

您可以使用函数来获取在每次请求时进行评估的动态 URL 或选项

typescript

const { messages } = useChat({
  connection: fetchServerSentEvents(
    () => `/api/chat?user=${currentUserId}`,
    () => ({
      headers: { Authorization: `Bearer ${getToken()}` },
    })
  ),
});

自定义 fetch

您还可以传入自定义 fetch 客户端（用于代理、重试或自定义传输）

typescript

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

HTTP 流

对于不支持 SSE 的环境，请使用 HTTP 流适配器

typescript

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

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

自定义适配器

对于特殊用例，您可以创建自定义适配器以满足特定的协议或要求

typescript

import { stream, type ConnectionAdapter } from "@tanstack/ai-react";
import type { StreamChunk, ModelMessage } from "@tanstack/ai";

const customAdapter: ConnectionAdapter = stream(
  async (messages: ModelMessage[], data?: Record<string, any>, signal?: AbortSignal) => {
    // Custom implementation
    const response = await fetch("/api/chat", {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
      },
      body: JSON.stringify({ messages, ...data }),
      signal,
    });

    if (!response.ok) {
      throw new Error(`HTTP error! status: ${response.status}`);
    }

    // Return async iterable of StreamChunk
    return processStream(response);
  }
);

const { messages } = useChat({
  connection: customAdapter,
});

WebSocket 适配器示例

要创建基于 WebSocket 的适配器

typescript

import { stream, type ConnectionAdapter } from "@tanstack/ai-react";
import type { StreamChunk, ModelMessage } from "@tanstack/ai";

function createWebSocketAdapter(url: string): ConnectionAdapter {
  return stream(async (messages: ModelMessage[], data?: Record<string, any>) => {
    return new ReadableStream<StreamChunk>({
      async start(controller) {
        const ws = new WebSocket(url);

        ws.onopen = () => {
          ws.send(JSON.stringify({ messages, ...data }));
        };

        ws.onmessage = (event) => {
          const chunk = JSON.parse(event.data);
          controller.enqueue(chunk);
        };

        ws.onerror = (error) => {
          controller.error(error);
        };

        ws.onclose = () => {
          controller.close();
        };
      },
    });
  });
}

const { messages } = useChat({
  connection: createWebSocketAdapter("ws://:8080/chat"),
});

适配器接口

所有适配器都实现了 ConnectionAdapter 接口

typescript

interface ConnectionAdapter {
  connect(
    messages: UIMessage[] | ModelMessage[],
    data?: Record<string, any>,
    abortSignal?: AbortSignal
  ): AsyncIterable<StreamChunk>;
}

错误处理

适配器应优雅地处理错误

typescript

const adapter = stream(async (messages, data, signal) => {
  try {
    const response = await fetch("/api/chat", {
      method: "POST",
      body: JSON.stringify({ messages, ...data }),
      signal,
    });

    if (!response.ok) {
      throw new Error(`HTTP ${response.status}: ${response.statusText}`);
    }

    return processStream(response);
  } catch (error) {
    if (error.name === "AbortError") {
      // Request was cancelled
      return;
    }
    throw error;
  }
});

认证

将身份验证标头添加到适配器

typescript

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

const { messages } = useChat({
  connection: fetchServerSentEvents("/api/chat", {
    headers: {
      Authorization: `Bearer ${token}`,
    },
  }),
});

对于动态令牌，请使用函数

typescript

const { messages } = useChat({
  connection: fetchServerSentEvents("/api/chat", () => ({
    headers: {
      Authorization: `Bearer ${getToken()}`,
    },
  })),
});

最佳实践

在大多数情况下使用 SSE - 在可能的情况下，优先使用 SSE 适配器，因为它具有可靠且受支持的功能
处理重连 - 适配器应管理瞬态网络问题的重连
取消卸载时 - 当组件卸载时，清理连接以避免内存泄漏
处理错误 - 当请求失败时，向用户提供有意义的错误消息
支持中止信号 - 适配器应允许通过中止信号取消请求

下一步

流式传输 - 了解有关流式传输响应的信息
API 参考 - 探索连接适配器 API