@可持久流/tanstack-ai-交通

TanStack AI 持久流的适配器.

这个软件包做什么

这个包裹给你:

• TanStack 兼容客户端 connection (单位:千美元)subscribe + 键 send(中文(简体) )
• 服务器帮助将 TanStack 块写入持久流
• 用于回声用户提示和执行 JSON 流格式的聊天会话助手
• SSR水合+还原抵消

如果您想要聊天会话在刷新/重新连接中幸存下来,并且可以在多个客户端之间共享,请使用此功能.

安装

pnpm add @durable-streams/tanstack-ai-transport @durable-streams/client

快开始

客户端: 创建持久会话连接

import { durableStreamConnection } from "@durable-streams/tanstack-ai-transport"

const connection = durableStreamConnection({
  sendUrl: "/api/chat?id=chat_123",
  readUrl: "/api/chat-stream?id=chat_123",
  initialOffset: undefined,
})

将此连接传递给 useChat 从 @tanstack/ai-react。 。 。 .

服务器: 将模型块写入可持久流

import { toDurableChatSessionResponse } from "@durable-streams/tanstack-ai-transport"

return toDurableChatSessionResponse({
  stream: {
    writeUrl,
    headers,
  },
  newMessages: [latestUserMessage],
  responseStream, // AsyncIterable<TanStackChunk>
})

此选项将新用户消息附加到耐用流, 管道 AI 响应块, 并返回空成功响应 .

整合指南(建议聊天会话流程)

(1) 客户端连接

const connection = durableStreamConnection({
  sendUrl: `/api/chat?id=${chatId}`,
  readUrl: `/api/chat-stream?id=${encodeURIComponent(chatId)}`,
  initialOffset: resumeOffsetFromSSR,
})

使用方式 useChat({ id: chatId, connection, live: true })。 。 。 .

2) POST路线(/api/chat(中文(简体) )

• 验证聊天 ID
• 构建可持久流写入 URL
• 保留 newMessages 明确( 通常是最新的用户消息)
• 启动您的模型流( E)responseStream(中文(简体) )
• 返回时 toDurableChatSessionResponse(...)

3)获取代理路线(/api/chat-stream(中文(简体) )

• 接受聊天 id 查询段落
• 在服务器上从此 id 创建上游持久读取的 URL
• 转发查询等参数 offset, (中文(简体) ), live等 (中文(简体) ).
• 在服务器上添加可持久读取的认证信头
• 向客户交回上游身体/头部

(4) SSR水合物+还原

聊天页面加载器 :

• materializeSnapshotFromDurableStream({ readUrl, headers })
• 发送 messages + 键 offset 到客户端
• 用那个 offset 作为 initialOffset 打开 durableStreamConnection

这样可以避免在第一个订阅时重放整个历史.

API 参考

客户端 APIs

durableStreamConnection(options)

创建与 TanStack 兼容的程序 connection 对象为:

• subscribe(abortSignal?) => AsyncIterable<TanStackChunk>
• send(messages, data?, abortSignal?) => Promise<void>

DurableStreamConnectionOptions数字 :

• sendUrl: string (要求) - 地点 send(...) 职位
• readUrl?: string - 在哪里,没有 subscribe(...) 读取( 默认为 sendUrl(中文(简体) )
• initialOffset?: string - 用于恢复的初步持久抵销
• emitSnapshotOnSubscribe?: boolean (默认) true- 发射合成 MESSAGES_SNAPSHOT 初次补习
• headers?: HeadersInit - 既适用于阅读请求,也适用于写作请求
• fetchClient?: typeof fetch - 自定义获取执行

行为:

• send(...) 职位 { messages, data } 改为 sendUrl
• subscribe(...) 读取耐用的 JSON 流批量并生成 TanStack 块
• 内部抵销随着批次的到来而得到更新,因此以后的订阅将继续从最近看到的抵销中进行

materializeSnapshotFromDurableStream(options)

读取非活性可持久流并实现 TanStack 消息状态 .

输入 :

• readUrl: string
• headers?: HeadersInit
• offset?: string

回返情况:

• { messages: Array<any>; offset?: string }

sanitizeChunkForStorage(chunk)

删除重复 content 打开 TEXT_MESSAGE_CONTENT 块( 维护) delta减少所储存的有效载荷尺寸.

服务器 APIs

toDurableChatSessionResponse(options)

聊天会话的高级助手写作.

ToDurableChatSessionResponseOptions数字 :

• stream: DurableChatSessionStreamTarget (单位:千美元)writeUrl, (中文(简体) ), headers, (中文(简体) ), createIfMissing(中文(简体) )
• newMessages: DurableSessionMessage[] - 明确附上迅速信息
• responseStream: AsyncIterable<TanStackChunk> - 模型块源
• mode?: "immediate" | "await" (默认) immediate(中文(简体) )
• waitUntil?: (promise: Promise<unknown>) => void

注释:

• 聊天会话流总是 application/json
• 附录 newMessages 使用 toMessageEchoChunks(...)
• 在写前先消毒块

toDurableStreamResponse(source, options)

用于任意 Aync 块源的下级通用写法 .

ToDurableStreamResponseOptions数字 :

• stream: DurableStreamTarget
• mode?: "immediate" | "await"
• waitUntil?: (promise: Promise<unknown>) => void
• exposeLocationHeader?: boolean

ensureDurableChatSessionStream(streamTarget)

使用聊天会话时确保存在可持久流,执行 application/json。 。 。 .

toMessageEchoChunks(message)

将用户/系统/助理消息转换为:

• TEXT_MESSAGE_START
• TEXT_MESSAGE_CONTENT (如果文本存在)
• TEXT_MESSAGE_END

这在提供时保留了明确的消息ID.

appendSanitizedChunksToStream(stream, chunks, contentType?)

附加块数组 sanitizeChunkForStorage 应用.

pipeSanitizedChunksToStream(source, stream, contentType?)

管道合成块源 sanitizeChunkForStorage 应用.

共享类型

• TanStackChunk
• DurableSessionConnection
• DurableStreamConnection (别名)
• DurableStreamConnectionOptions
• DurableChatSessionStreamTarget
• DurableStreamTarget
• ToDurableStreamResponseMode
• ToDurableStreamResponseOptions
• ToDurableChatSessionResponseOptions
• DurableSessionMessage
• DurableSessionMessagePart
• WaitUntil

反应合同

toDurableStreamResponse(..., { mode: "immediate" })

• 状态 : 201
• 页眉 : Location: <read-url>
• 车身 : { "streamUrl": "<read-url>" }

toDurableStreamResponse(..., { mode: "await" })

• 状态 : 200
• 页眉 : Location: <read-url>
• 车身 : { "streamUrl": "<read-url>", "finalOffset": "<offset>" }

toDurableChatSessionResponse(..., { mode: "immediate" })

• 状态 : 202
• 体 : 空

toDurableChatSessionResponse(..., { mode: "await" })

• 状态 : 200
• 体 : 空

waitUntil 使用量

return toDurableChatSessionResponse({
  stream: { writeUrl, headers },
  newMessages,
  responseStream,
  waitUntil: ctx.waitUntil.bind(ctx),
})

使用 waitUntil 在工人风格的运行时间中,背景写在返回指针响应后继续.