为 TanStack AI 与 Vercel AI SDK 接入 Durable Transport用 Durable Streams 构建可恢复、可协作的 AI 会话【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric本文介绍 Electric 项目发布的基于 Durable Streams 的 AI SDK 传输层适配器面向 TanStack AI 的 Connection Adapter 与面向 Vercel AI SDK 的 Durable Transport。围绕韧性resilience、可恢复性resumability与多人/多 Agent 协作三大主题你会掌握如何在几乎不改动现有 AI 工程代码的前提下把客户端与服务端之间的通信替换为持久化、可寻址的流式会话并通过 Electric Cloud 实现零运维托管。背景为什么 AI 应用的通信层需要持久化大多数 AI 应用只要在连接上出任何问题就会断裂网络时好时坏、用户刷新页面、组件 re-render、浏览器标签被挂起这些都会打断一次长时间运行的流式生成active generation。传统做法是让 agent 直接向 UI 流式输出 token——这是单用户 单 Agent 的请求/响应范式脆弱且无法协作详见 Building AI apps? You need sync 中对 resumability、multi-device、multi-user 的系统性分析。Durable Streams 正是为这类问题设计的底层原语它是持久化、可寻址的二进制流带有可靠投递协议天然提供韧性和可恢复性——连接断开后客户端保存上次处理到的 offset重连时从该位置继续读取数据不丢失。在此基础上叠加 Durable Sessions 模式就能进一步支持多标签页、多设备、多用户与多 Agent 的实时与异步协作见 Durable Sessions——the key pattern for collaborative AI。两个主流 AI SDK——TanStack AI 与 Vercel AI SDK——都意识到了这些挑战并分别提供了 Transport / Connection Adapter 扩展点允许开发者接入自定义传输层。本次发布的durable-streams/tanstack-ai-transport与durable-streams/aisdk-transport正是利用这些扩展点把客户端 服务端的通信整体替换为基于 Durable Streams 的同步范式。Durable Streams 与 Durable Sessions两个前置概念在深入适配器之前先明确底层两个概念Durable Streams是持久化、追加式append-only、可寻址的 HTTP 流。每个流拥有独立 URL客户端可以从任意 offset 读取历史数据catch-up也可以切换到实时订阅模式long-poll 或 SSE。offset 是不透明字符串服务端通过Stream-Next-Offset响应头告知下次读取位置进度完全由客户端保存。协议的六个基本操作是PUT /streams/my-stream # Create创建 POST /streams/my-stream # Append追加 GET /streams/my-stream?offset… # Read按 offset 读取 HEAD /streams/my-stream # Metadata元信息 POST /streams/my-stream # Close携带 Stream-Closed: true 关闭 DELETE /streams/my-stream # Delete删除完整的协议语义偏移量、消息边界、幂等生产者、live 模式、生命周期与 CDN 缓存可参考 Electric Streams 协议总览五分钟跑通一个可恢复流的最小示例见 Quickstart。Durable Sessions 模式则是把整个会话做成持久化状态把用户消息、工具调用、流式 LLM 响应、多模态数据帧全部写入同一条 Durable Stream任何客户端加入会话时从自己的 offset 订阅同一条流即可自然保持跨端实时同步。会话数据持久化后客户端可以随时读取、加入、甚至 fork 会话这为异步协作、审计日志与决策追踪提供了基础。本次发布的 transport 适配器正是把这一模式以可插拔传输层的形式落地到两个主流 AI SDK。TanStack AI 集成Durable Connection AdapterTanStack AI 是一个类型安全、框架无关的 AI SDK通过统一接口对接多个 LLM 提供方无厂商锁定、无私有格式纯 TypeScript 实现。扩展点Connection AdapterTanStack AI 提供了 Connection Adapter 扩展点用于自定义客户端与服务端之间的通信方式。durable-streams/tanstack-ai-transport包里的 durable connection adapter 正是挂在这个扩展点上为 TanStack AI 应用注入韧性、可恢复性与协作能力。其核心思路与接入方式在 TanStack AI 集成文档 中有完整说明。工作原理当ChatClient初始化时拉取既有消息历史并建立对某条 Durable Stream 的订阅所有写入用户消息、工具调用、流式 LLM 响应等都通过这条流写入并被消费客户端 服务端通信因此变得可恢复并且解锁了对多条并发消息与多个 token 流的支持任何连接到同一会话的客户端都订阅并写入同一条流因此无论数据来自哪个用户或哪个 Agent所有标签页、设备、用户都能实时保持一致由于会话数据是持久化的客户端可以随时读取、加入、fork 会话实现异步协作、可审计性与决策追踪。这本质上是一种基于同步sync-based的架构把复杂的持久化、消息投递与分布式数据问题交由底层处理正如原博客所言这是构建 AI 应用唯一合理的做法。服务端把/api/chat的响应换成一个 durable 会话在 TanStack AI 应用的服务端/api/chat处理器或任何你调用 LLM 的地方中把原来的toServerSentEventsResponse(stream)替换为toDurableChatSessionResponse(...)import { toDurableChatSessionResponse } from durable-streams/tanstack-ai-transport return toDurableChatSessionResponse({ stream: { writeUrl, headers, }, newMessages: [latestUserMessage], responseStream })这个函数做了三件事把用户消息持久化进 Durable Stream使其成为会话历史的一部分把流式 LLM 响应管道写入Durable Stream返回一个签名的流 URL供客户端订阅和消费。客户端把ChatClient换成 durable 连接客户端通过useChathook 使用 durable connection adapterimport { useMemo } from react import { useChat } from tanstack/ai-react import { durableStreamConnection } from durable-streams/tanstack-ai-transport export function Chat({ id, initialMessages, initialOffset }) { const connection useMemo( () durableStreamConnection({ sendUrl: /api/chat?id${id}, readUrl: /api/stream?id${id}, initialOffset }), [id, initialOffset] ) const { messages, sendMessage } useChat({ connection, id, initialMessages, live: true }) // ... }其中initialMessages已发生的聊天历史与initialOffset订阅恢复位置可以来自任何来源。例如如果你把消息存在数据库里就从数据库取消息历史——但务必把 offset 与数据在同一个事务里一起存储避免出现竞态条件。消息历史与可恢复性反转控制原则可恢复性要求客户端在恢复活动订阅之前先加载会话的初始消息历史。而消息历史的存储方式可以有多种可以直接从会话流中读取并物化materialize也可以先把消息物化后存进 Postgres 之类的数据库。因此这些 durable transport 集成的设计遵循**反转控制inversion of control**原则集成并不规定你如何处理消息持久化与检索而是把选择权交给你这一原则在 TanStack AI 集成文档 的 Initial message history 一节有专门阐述。最简单的方式是使用集成提供的materializeSnapshotFromDurableStream辅助函数直接从会话流中物化出历史。如果把它放进 TanStackRouteloader即可同时支持客户端渲染与 SSRimport { materializeSnapshotFromDurableStream } from durable-streams/tanstack-ai-transport const { messages, offset } await materializeSnapshotFromDurableStream({ readUrl, headers, })推荐的会话流程集成文档给出了一个完整的四步参考流程可与上面的代码片段组合成生产可用的会话闭环客户端连接durableStreamConnection({ sendUrl:/api/chat?id${chatId}, readUrl:/api/chat-stream?id${encodeURIComponent(chatId)}, initialOffset: resumeOffsetFromSSR })配合useChat({ id: chatId, connection, live: true })使用POST 路由POST /api/chat校验 chat id、构造 durable stream 写 URL、显式保留newMessages通常只放最新 prompt、启动模型的responseStream最后返回toDurableChatSessionResponse(...)GET 代理路由GET /api/chat-stream接收 chatid在 durable stream 服务端构造上游读 URL透传offset、live等读参数加上服务端侧读鉴权请求头返回上游的 body 与 headersSSR 水合与恢复页面 loader 中调用materializeSnapshotFromDurableStream拿到{ messages, offset }把offset作为initialOffset传给客户端连接避免首次订阅时重放整个会话。端到端数据流把这些片段组合起来一次完整的会话流程是客户端渲染时取得初始消息历史并从正确的 offset 订阅会话流调用sendMessage或append时用户消息照常发送到后端由你现有的 AI 工程逻辑与 LLM provider 指令处理LLM provider 流式返回时toDurableChatSessionResponse把活动生成内容流式写入 Durable Stream所有客户端订阅同一条流任何数据同步进来都会响应式更新——无论它来自哪个用户或 Agent并支持多条交错并发的生成流。由于底层是持久化、可寻址的流客户端可以在任意时刻加入并从当前 offset 追赶进度既支持实时协作也支持事后异步访问详见 Durable Sessions 参考实现 中对完整会话同步的描述。Vercel AI SDK 集成Durable TransportVercel AI SDK 是一个面向 TypeScript 的 AI 工具包。它提供了 Transport 接口作为扩展点durable-streams/aisdk-transport包里的 Durable Transport 正是挂在这个接口上。客户端替换默认 transport把默认 transport 换成createDurableChatTransportimport { useChat } from ai-sdk/react import { createDurableChatTransport } from durable-streams/aisdk-transport const transport createDurableChatTransport({ api: /api/chat }) const chat useChat({ transport, resume: true })服务端包装 AI SDK 消息流在服务端用toDurableStreamResponse包装 AI SDK UI 消息流import { toDurableStreamResponse } from durable-streams/aisdk-transport return toDurableStreamResponse({ source: result.toUIMessageStream(), stream: { writeUrl: buildWriteStreamUrl(streamPath), readUrl: buildReadProxyUrl(request, streamPath), headers: DURABLE_STREAMS_WRITE_HEADERS, }, })服务端把 AI SDK 的 chunk 写入 Durable Streams并通过响应头Location与响应体{ streamUrl }把读 URL 返回给客户端。集成文档Vercel AI SDK 集成指出这是想让一次生成在刷新后存活、并能干净地重连到同一条流时应使用的集成方式——你保持正常的useChat流程只需替换 transport。Resume 流程刷新安全的一次生成为了让一次生成在页面刷新后无缝恢复集成文档给出了四个步骤生成进行中时把当前活动的 stream id持久化到该聊天的会话状态里增加一个重连端点例如GET /api/chat/:id/stream该端点在没有活动生成时返回204有活动生成时返回200并携带Location与{ streamUrl }在useChat中开启resume: true。值得一提的是Vercel AI SDK 的数据流协议本身是围绕单用户 单 Agent 设计的例如data: {type:finish}没有携带 message id无法区分多条并发消息的结束而 Durable State 层正好可以在 Electric 投递协议之上、以端到端类型安全的方式韧性传输这类结构化协议——这一对比在原博客及 Durable Sessions 文章 的 Durable transport 一节中有详细剖析。托管在 Electric Cloud 上零运维部署两个集成都支持开箱即用的托管方案。Electric Cloud 是 Electric 的托管同步平台提供完全托管的 Durable Streams 实现即文档中的Electric Streams参见 streams 总览。在 Hosted Durable Streams 发布公告 中披露了托管版本的关键能力读请求不命中源站Electric Cloud 的 Sync CDN 承担所有读流量单个流经测试可支撑大量并发连接高吞吐写入小消息写入吞吐可观可持续吞吐达 MB 级每秒简单定价读免费每月赠送一定的写入额度之后按量付费协议正确性保障数百项服务端与客户端一致性测试多语言客户端TypeScript、Python、Go、Rust、Java、Swift 等语言的客户端均通过完整一致性测试。在仓库中也可以看到协议与实现的落地证据packages/durable-streams-rust/提供了 Rust 实现的 Durable Streams 服务端含src/handlers.rs的协议处理与bench-latency/延迟基准而packages/agents-runtime/、packages/agents-server/等包则直接依赖durable-streams/client构建其 Agent 运行时与实体流见packages/agents-server/src/stream-client.ts。接入托管服务的大致路径是注册 Electric Cloud、创建 service然后直接对流的 URL 执行PUT/POST/GET——全部是普通 HTTP无需自建基础设施。如果你希望把 durable、可恢复的流式能力放在已有的 AI 流式 API前面不改上游协议还可以使用durable-streams/proxy它把上游响应持久化进 Durable Streams客户端通过requestId即可在刷新与重连后恢复到同一条流见 Durable Proxy。下一步注册 Electric Cloud按 TanStack AI 集成文档 与 Vercel AI SDK 集成文档 完成接入从 Quickstart 开始用 curl 五分钟跑通创建流 → 追加 → 读取 → 实时 tail想理解底层协议细节offset 语义、幂等生产者、SSE/long-poll 两种 live 模式、CDN 缓存阅读 Electric Streams 协议总览想构建完整的协作型会话复用多条消息与多个活跃生成、presence、Agent 注册参考 Durable Sessions 模式详解 及其参考实现。总之Durable Streams transport 集成让你以最小代码改动、零基础设施投入为 TanStack AI 与 Vercel AI SDK 应用补上韧性、可恢复性与多用户多 Agent 协作能力——这正是 AI 应用与 Agent 系统走向生产环境与团队协作所必需的基础设施层。【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考