Electric Sync构建 AI 应用的弹性架构——可恢复、多设备与多用户协作【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electricAI 应用天然是协作型的会话需要在断网、刷新、多标签页、多设备和多用户之间保持一致。本文围绕 Electric 官方博客《Building AI apps? You need sync》展开系统讲解如何利用同步引擎sync engine解决 AI 应用开发中的四大核心难题——可恢复性Resumability、可中断性Interruptibility、多设备Multi-device与多用户Multi-user并结合本仓库中 TypeScript 客户端与 React Hooks 的实际源码说明通过存储流式传输streaming via store模式在协议层是如何落地为自动重连与增量同步的。读完本文你将掌握用ShapeStream/useShape订阅数据库变更的完整写法、基于offset的同步协议原理以及一套可直接运行的多用户、多 Agent 演示应用的搭建步骤。为什么 AI 应用必须是协作式的绝大多数 AI 应用都会把模型输出的 token 流式地送到前端——这正是 Claude 和 ChatGPT 逐字打字给你看的原理。传统做法是Agent 直连 UI模型服务生成 token通过一次长连接直接推给浏览器。这条链路一断网络抖动、页面刷新、用户切走再回来流就丢了。Electric 团队的观点是这不只是边缘场景而是 AI 产品的核心诉求。当多个用户围绕同一个 AI 会话协作、当一个会话派生出越来越多的 Agent 时可恢复性、多设备一致性、多用户可见性都会成为决定产品成败的基础能力。而这四类问题恰好都能被同步引擎统一解决。Resumability把 token 流进存储再订阅存储核心改造只有一句话不要从 Agent 直接流到 UI而是把 token 写进一个存储store前端订阅这个存储。直连模式下连接断开或用户刷新页面流式内容直接丢失经由存储模式下token 先落库例如 Postgres前端只是订阅存储的变更。存储里的数据不会因客户端掉线而丢失前端重新连接时从上次读到的位置继续即可。这种模式的关键机制叫可恢复性resumability应用记住自己上次读到的位置position重连时请求从该位置继续的流。手动实现这套逻辑相当繁琐——消息投递本质上是分布式系统里关于一致性的深坑——而同步引擎把它做成了内建能力。以 Electric 为例其同步协议的核心就是客户端携带offset参数请求数据流。协议层offset 如何工作Electric 的 HTTP 同步协议参见 HTTP API 文档是这样设计的初始同步客户端以offset-1发起请求表示从给定 Shape 的日志起点给我全量数据curl -i http://localhost:3000/v1/shape?tablefoooffset-1日志分页Shape 日志可能过大无法一次返回。服务器先返回一批数据和一个electric-offset响应头客户端把该头值作为下一次请求的offset参数直到取完当前全部数据实时模式追平后客户端切换为livetrue带上offset和 Shape 的handle保持长连接实时接收增量curl -i http://localhost:3000/v1/shape?tablefoolivetruehandle3833821-1721812114261offset0_0另支持offsetnow跳过历史数据、直接获得最新的续传位置适合从零开始的应用场景。这套协议在 TypeScript 客户端中被封装成了开箱即用的 APIimport { ShapeStream } from electric-sql/client const tokenStream new ShapeStream({ params: { table: tokens, }, }) // tokenStream.subscribe(tokens ...)从源码结构看自动恢复能力就藏在客户端的状态机里。ShapeStream 的状态管理中共享状态维护一个offset字段每次构建请求 URL 时会执行url.searchParams.set(OFFSET_QUERY_PARAM, this.#shared.offset)见 shape-stream-state.ts#L266当 Shape 需要失效重取如被标记markMustRefetch时offset 会被重置回-1重新拉取全量见 shape-stream-state.ts#L27-L52。追平历史后客户端还会把upToDateOffset写入状态进入实时模式见 shape-stream-state.ts#L330-L337。也就是说记住位置、断点续传这些细节对应用开发者是完全透明的。重连的健壮性同样有源码可查。fetch.ts 定义了带完整抖动full jitter指数的退避重试策略BackoffOptions包含初始延迟、最大延迟达到后固定间隔重试例如每分钟一次与最大重试次数等待时间的取值遵循服务器Retry-After头作为下限、客户端退避值封顶的优先级规则见 fetch.ts#L30-L75 与 fetch.ts#L157-L183。这正是博客中Electric AI chat 演示应用能无感处理离线、弱网与页面刷新的底层保障。Multi-device多标签页与多设备的一致体验用户会做什么开两个浏览器标签页在跟 Claude 聊天和回邮件之间反复横跳。两个标签页同时打开时用户记不清上次用的是哪个更糟的是他会以为任务没在跑而重复发一次 prompt于是两个线程干着同一件事。谁背锅你的软件。世界不止有浏览器标签页。Agent 在后台干活用户随时可能掏出手机在排队买咖啡的间隙查看进度。浏览器里发起的会话如何让手机应用保持最新这正是同步引擎的扇出fan-out能力把一次写入弹射到任意多个订阅者。以 Electric 为例你只需要把变更写进 PostgresElectric 负责把数据可靠地扇出给任意数量的客户端——官方基准测试声称其开箱即可扩展到百万级客户端规模参见 基准测试文档。无论用户换到哪个设备、回到哪个标签页状态都与其预期完全一致。Multi-user多人共享同一个 AI 会话SaaS 产品本就围绕协作设计想想 Figma 的多人编辑。有了 AI 之后点按钮协作正在被与 Agent 交互协作取代——而 Agent 直连 UI 的直连流是单用户的撑不起多人场景。多用户的解法与可恢复性、多设备相同流经存储并扇出同时保证把正确的会话流给正确的用户。这正是 Electric 与 Figma LiveGraph 这类引擎做的事在可靠流式传输与扇出之上提供部分复制partial replication让正确的数据同步给正确的用户。Electric 用 Shapes 定义部分复制通过 where 子句过滤出只需要的内容const tokenStream new ShapeStream({ params: { table: tokens, // 只同步某个指定会话的 token where: session_id 1234, }, })这一行 where 子句从根本上改变了 AI 应用的交互体验多个用户可以实时协作于同一个 AI 会话。官方博客中的演示展示了这个场景第一个用户向 AI 提问第二个用户实时旁观发现 AI 缺少上下文后上传了一份文档AI 基于新上下文生成了更好的回答——两个人的操作都发生在同一条共享的数据流上。Interruptibility对所有人都生效的停止通过存储流式传输还带来一个附赠能力对全部用户一致的中断。直连模式下每个用户各自从 Agent 拉流停止生成只能作用于拉流的那一端。而经由存储后是 Agent 单向写入存储任何用户都可以发出一条中断指令终止 Agent 的 token 流、停止向存储写入所有并发用户因此被自然中断。示例代码// 从 OpenAI API 流式获取 token const stream await openai.chat.completions.create({ model, messages, stream: true, }) // 写入 Postgres for await (const event of stream) { pg.insert(INSERT INTO tokens value ($1), [event.message]) } // 直到被中断 function interrupt() { stream.controller.abort() }这直接修复了用户疯狂点击 stop、模型却无视并继续生成的经典痛点——中断作用在数据源头写入存储的循环对所有订阅者一次性生效。Agents are usersAgent 也是协作方能中断流程、更新数据的并不只有人类用户。Agent 不只是接口而是行为主体actor它能发通知、能更新应用状态。所以只要有一个用户在和一个 Agent 交互你就已经有了一个多用户应用——至少包含你和 AI 两个用户。多 Agent 蜂群Swarms未来不只有一个 Agent而是成群 Agent 在后台替你干活。它们需要共享上下文、具备情境感知能力。LangGraph、Mastra 等工具为 Agent 提供了共享数据层但没有解决最后一公里问题——把状态同步回用户侧应用把人留在回路里。状态不能只存在于云端用户同样有代理权agency。设想一个项目管理场景你让 AI 助手监视 todo 列表并执行任务同时另开一个会话让另一个 Agent 规划项目、生成任务。这些 Agent 需要通过共享状态如 todo 列表协作感知变更并作出反应——用户也想看到同一份状态。在 Electric AI chat 演示中负责监视 todo 列表的 Agent 侧代码是这样的const listItemsStream new ShapeStream({ url: ${ELECTRIC_API_URL}/v1/shape, params: { table: todo_items, where: list_id ${listId}, }, }) const listItemsShape new Shape(listItemsStream) async function processNextItem() { const item listItemsShape.currentRows.find((item) !item.done) if (item) { // 使用 Agent 执行该任务 } } let processing false async function processItems() { if (processing) return processing true while (listItemsShape.currentRows.some((item) !item.done)) { await processNextItem() } processing false } listItemsShape.subscribe(async () { await processItems() })而把同一份状态展示给用户的只是一个普通的 React 组件function TodoListItems() { const { data: todoListItems } useShape({ url: ${ELECTRIC_API_URL}/v1/shape, params: { table: todo_lists_items, }, }) return ( ul {todoListItems.map((todoListItem) ( li key{todoListItem.id} {todoListItem.task} {todoListItem.done span Done/span} /li ))} /ul ) }Agent 与 UI 消费的是同一个 Shape 的同一份数据。本仓库中的 useShape 实现印证了这一点它内部创建或复用ShapeStream通过useSyncExternalStoreWithSelector订阅shape的变更并在isLoading、lastSyncedAt、error、lastOffset、handle等任一字段变化时触发重渲染见 react-hooks.tsx#L140-L152——组件因此能精确地数据变一行、界面更新一行。结构化数据与混乱半径模型不仅擅长吐 token同样擅长返回结构化数据。把 token 流经存储的另一个重大优势就是这个存储可以是一个结构化数据库。这让不同 Agent 可以分工协作于共享状态的不同部分——例如一个 Agent 负责 Figma 项目的高层结构另一个 Agent 填充各画布的细节。同时调用 API 时你通常清楚它能改哪些数据爆炸半径明确知道该重新拉取什么而面对有自主性的 AI Agent你无法预知它会改什么。要么持续追踪并重新拉取一切要么监控数据变化、被自动通知。你真正需要的是声明性地描述应用、Agent 或 UI 所需的数据子集持续监控它、保持最新、并对变更作出响应——这正是 Shape带 where 子句的 Shape 定义所做的。这也是业界观点AI Agent 就是本地优先客户端的来源。Sync is the solution为什么同步是解法Sync 系统性地解决了 AI 交互体验中一系列实际问题从可恢复性、可中断性到多标签页、多设备、多用户。随着 Agent 变得更协作、更自主、数量更多共享状态、审查进度、响应变更、维护本地数据集都会变得更重要。从采用角度adoption看AI 创业团队的一个重要机会是用更聪明的 AI 系统替换现有一代软件尤其是 b2b 与企业软件——这类软件天然围绕团队协作为中心支持多角色多用户。这意味着单用户 AI 会话无法达标要替换既有企业系统并获得广泛采用AI 应用必须支持团队级协作即把多个用户和 Agent 保持在同步状态。同步是一个自己很难解好的问题——而这是你最不该占用核心产品开发时间去解决的事。有野心的 AI 应用应当构建在同步引擎如 Electric之上。动手运行多用户、多 Agent 的 AI 聊天演示本篇文档配套一个有弹性、多用户、多 Agent 的 AI 聊天演示应用electric-ai-chat。运行步骤如下适用前提本地已安装 Node、pnpm、Docker并拥有一个 OpenAI API key克隆 electric-ai-chat 示例仓库并进入目录# 获取 electric-ai-chat 示例仓库 cd electric-ai-chat安装依赖pnpm install用 Docker 启动 Postgres 和 Electricdocker compose up -d配置 OpenAI API key 并启动后端 APIexport OPENAI_API_KEYyour-openai-api-key pnpm dev:api启动演示应用pnpm dev:app浏览器打开localhost:5173即可看到token 流经存储的弹性会话、多设备扇出、多用户实时协作以及监视 todo 列表的 Agent 与用户共享同一份数据。延伸阅读同步总览 与 同步演示TypeScript 客户端文档、React 绑定文档Shapes 指南部分复制的定义方式协议实现参考HTTP 同步协议、ShapeStream 状态机、带退避的拉取与重连、useShape 实现【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Electric Sync：构建 AI 应用的弹性架构——可恢复、多设备与多用户协作
Electric Sync构建 AI 应用的弹性架构——可恢复、多设备与多用户协作【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electricAI 应用天然是协作型的会话需要在断网、刷新、多标签页、多设备和多用户之间保持一致。本文围绕 Electric 官方博客《Building AI apps? You need sync》展开系统讲解如何利用同步引擎sync engine解决 AI 应用开发中的四大核心难题——可恢复性Resumability、可中断性Interruptibility、多设备Multi-device与多用户Multi-user并结合本仓库中 TypeScript 客户端与 React Hooks 的实际源码说明通过存储流式传输streaming via store模式在协议层是如何落地为自动重连与增量同步的。读完本文你将掌握用ShapeStream/useShape订阅数据库变更的完整写法、基于offset的同步协议原理以及一套可直接运行的多用户、多 Agent 演示应用的搭建步骤。为什么 AI 应用必须是协作式的绝大多数 AI 应用都会把模型输出的 token 流式地送到前端——这正是 Claude 和 ChatGPT 逐字打字给你看的原理。传统做法是Agent 直连 UI模型服务生成 token通过一次长连接直接推给浏览器。这条链路一断网络抖动、页面刷新、用户切走再回来流就丢了。Electric 团队的观点是这不只是边缘场景而是 AI 产品的核心诉求。当多个用户围绕同一个 AI 会话协作、当一个会话派生出越来越多的 Agent 时可恢复性、多设备一致性、多用户可见性都会成为决定产品成败的基础能力。而这四类问题恰好都能被同步引擎统一解决。Resumability把 token 流进存储再订阅存储核心改造只有一句话不要从 Agent 直接流到 UI而是把 token 写进一个存储store前端订阅这个存储。直连模式下连接断开或用户刷新页面流式内容直接丢失经由存储模式下token 先落库例如 Postgres前端只是订阅存储的变更。存储里的数据不会因客户端掉线而丢失前端重新连接时从上次读到的位置继续即可。这种模式的关键机制叫可恢复性resumability应用记住自己上次读到的位置position重连时请求从该位置继续的流。手动实现这套逻辑相当繁琐——消息投递本质上是分布式系统里关于一致性的深坑——而同步引擎把它做成了内建能力。以 Electric 为例其同步协议的核心就是客户端携带offset参数请求数据流。协议层offset 如何工作Electric 的 HTTP 同步协议参见 HTTP API 文档是这样设计的初始同步客户端以offset-1发起请求表示从给定 Shape 的日志起点给我全量数据curl -i http://localhost:3000/v1/shape?tablefoooffset-1日志分页Shape 日志可能过大无法一次返回。服务器先返回一批数据和一个electric-offset响应头客户端把该头值作为下一次请求的offset参数直到取完当前全部数据实时模式追平后客户端切换为livetrue带上offset和 Shape 的handle保持长连接实时接收增量curl -i http://localhost:3000/v1/shape?tablefoolivetruehandle3833821-1721812114261offset0_0另支持offsetnow跳过历史数据、直接获得最新的续传位置适合从零开始的应用场景。这套协议在 TypeScript 客户端中被封装成了开箱即用的 APIimport { ShapeStream } from electric-sql/client const tokenStream new ShapeStream({ params: { table: tokens, }, }) // tokenStream.subscribe(tokens ...)从源码结构看自动恢复能力就藏在客户端的状态机里。ShapeStream 的状态管理中共享状态维护一个offset字段每次构建请求 URL 时会执行url.searchParams.set(OFFSET_QUERY_PARAM, this.#shared.offset)见 shape-stream-state.ts#L266当 Shape 需要失效重取如被标记markMustRefetch时offset 会被重置回-1重新拉取全量见 shape-stream-state.ts#L27-L52。追平历史后客户端还会把upToDateOffset写入状态进入实时模式见 shape-stream-state.ts#L330-L337。也就是说记住位置、断点续传这些细节对应用开发者是完全透明的。重连的健壮性同样有源码可查。fetch.ts 定义了带完整抖动full jitter指数的退避重试策略BackoffOptions包含初始延迟、最大延迟达到后固定间隔重试例如每分钟一次与最大重试次数等待时间的取值遵循服务器Retry-After头作为下限、客户端退避值封顶的优先级规则见 fetch.ts#L30-L75 与 fetch.ts#L157-L183。这正是博客中Electric AI chat 演示应用能无感处理离线、弱网与页面刷新的底层保障。Multi-device多标签页与多设备的一致体验用户会做什么开两个浏览器标签页在跟 Claude 聊天和回邮件之间反复横跳。两个标签页同时打开时用户记不清上次用的是哪个更糟的是他会以为任务没在跑而重复发一次 prompt于是两个线程干着同一件事。谁背锅你的软件。世界不止有浏览器标签页。Agent 在后台干活用户随时可能掏出手机在排队买咖啡的间隙查看进度。浏览器里发起的会话如何让手机应用保持最新这正是同步引擎的扇出fan-out能力把一次写入弹射到任意多个订阅者。以 Electric 为例你只需要把变更写进 PostgresElectric 负责把数据可靠地扇出给任意数量的客户端——官方基准测试声称其开箱即可扩展到百万级客户端规模参见 基准测试文档。无论用户换到哪个设备、回到哪个标签页状态都与其预期完全一致。Multi-user多人共享同一个 AI 会话SaaS 产品本就围绕协作设计想想 Figma 的多人编辑。有了 AI 之后点按钮协作正在被与 Agent 交互协作取代——而 Agent 直连 UI 的直连流是单用户的撑不起多人场景。多用户的解法与可恢复性、多设备相同流经存储并扇出同时保证把正确的会话流给正确的用户。这正是 Electric 与 Figma LiveGraph 这类引擎做的事在可靠流式传输与扇出之上提供部分复制partial replication让正确的数据同步给正确的用户。Electric 用 Shapes 定义部分复制通过 where 子句过滤出只需要的内容const tokenStream new ShapeStream({ params: { table: tokens, // 只同步某个指定会话的 token where: session_id 1234, }, })这一行 where 子句从根本上改变了 AI 应用的交互体验多个用户可以实时协作于同一个 AI 会话。官方博客中的演示展示了这个场景第一个用户向 AI 提问第二个用户实时旁观发现 AI 缺少上下文后上传了一份文档AI 基于新上下文生成了更好的回答——两个人的操作都发生在同一条共享的数据流上。Interruptibility对所有人都生效的停止通过存储流式传输还带来一个附赠能力对全部用户一致的中断。直连模式下每个用户各自从 Agent 拉流停止生成只能作用于拉流的那一端。而经由存储后是 Agent 单向写入存储任何用户都可以发出一条中断指令终止 Agent 的 token 流、停止向存储写入所有并发用户因此被自然中断。示例代码// 从 OpenAI API 流式获取 token const stream await openai.chat.completions.create({ model, messages, stream: true, }) // 写入 Postgres for await (const event of stream) { pg.insert(INSERT INTO tokens value ($1), [event.message]) } // 直到被中断 function interrupt() { stream.controller.abort() }这直接修复了用户疯狂点击 stop、模型却无视并继续生成的经典痛点——中断作用在数据源头写入存储的循环对所有订阅者一次性生效。Agents are usersAgent 也是协作方能中断流程、更新数据的并不只有人类用户。Agent 不只是接口而是行为主体actor它能发通知、能更新应用状态。所以只要有一个用户在和一个 Agent 交互你就已经有了一个多用户应用——至少包含你和 AI 两个用户。多 Agent 蜂群Swarms未来不只有一个 Agent而是成群 Agent 在后台替你干活。它们需要共享上下文、具备情境感知能力。LangGraph、Mastra 等工具为 Agent 提供了共享数据层但没有解决最后一公里问题——把状态同步回用户侧应用把人留在回路里。状态不能只存在于云端用户同样有代理权agency。设想一个项目管理场景你让 AI 助手监视 todo 列表并执行任务同时另开一个会话让另一个 Agent 规划项目、生成任务。这些 Agent 需要通过共享状态如 todo 列表协作感知变更并作出反应——用户也想看到同一份状态。在 Electric AI chat 演示中负责监视 todo 列表的 Agent 侧代码是这样的const listItemsStream new ShapeStream({ url: ${ELECTRIC_API_URL}/v1/shape, params: { table: todo_items, where: list_id ${listId}, }, }) const listItemsShape new Shape(listItemsStream) async function processNextItem() { const item listItemsShape.currentRows.find((item) !item.done) if (item) { // 使用 Agent 执行该任务 } } let processing false async function processItems() { if (processing) return processing true while (listItemsShape.currentRows.some((item) !item.done)) { await processNextItem() } processing false } listItemsShape.subscribe(async () { await processItems() })而把同一份状态展示给用户的只是一个普通的 React 组件function TodoListItems() { const { data: todoListItems } useShape({ url: ${ELECTRIC_API_URL}/v1/shape, params: { table: todo_lists_items, }, }) return ( ul {todoListItems.map((todoListItem) ( li key{todoListItem.id} {todoListItem.task} {todoListItem.done span Done/span} /li ))} /ul ) }Agent 与 UI 消费的是同一个 Shape 的同一份数据。本仓库中的 useShape 实现印证了这一点它内部创建或复用ShapeStream通过useSyncExternalStoreWithSelector订阅shape的变更并在isLoading、lastSyncedAt、error、lastOffset、handle等任一字段变化时触发重渲染见 react-hooks.tsx#L140-L152——组件因此能精确地数据变一行、界面更新一行。结构化数据与混乱半径模型不仅擅长吐 token同样擅长返回结构化数据。把 token 流经存储的另一个重大优势就是这个存储可以是一个结构化数据库。这让不同 Agent 可以分工协作于共享状态的不同部分——例如一个 Agent 负责 Figma 项目的高层结构另一个 Agent 填充各画布的细节。同时调用 API 时你通常清楚它能改哪些数据爆炸半径明确知道该重新拉取什么而面对有自主性的 AI Agent你无法预知它会改什么。要么持续追踪并重新拉取一切要么监控数据变化、被自动通知。你真正需要的是声明性地描述应用、Agent 或 UI 所需的数据子集持续监控它、保持最新、并对变更作出响应——这正是 Shape带 where 子句的 Shape 定义所做的。这也是业界观点AI Agent 就是本地优先客户端的来源。Sync is the solution为什么同步是解法Sync 系统性地解决了 AI 交互体验中一系列实际问题从可恢复性、可中断性到多标签页、多设备、多用户。随着 Agent 变得更协作、更自主、数量更多共享状态、审查进度、响应变更、维护本地数据集都会变得更重要。从采用角度adoption看AI 创业团队的一个重要机会是用更聪明的 AI 系统替换现有一代软件尤其是 b2b 与企业软件——这类软件天然围绕团队协作为中心支持多角色多用户。这意味着单用户 AI 会话无法达标要替换既有企业系统并获得广泛采用AI 应用必须支持团队级协作即把多个用户和 Agent 保持在同步状态。同步是一个自己很难解好的问题——而这是你最不该占用核心产品开发时间去解决的事。有野心的 AI 应用应当构建在同步引擎如 Electric之上。动手运行多用户、多 Agent 的 AI 聊天演示本篇文档配套一个有弹性、多用户、多 Agent 的 AI 聊天演示应用electric-ai-chat。运行步骤如下适用前提本地已安装 Node、pnpm、Docker并拥有一个 OpenAI API key克隆 electric-ai-chat 示例仓库并进入目录# 获取 electric-ai-chat 示例仓库 cd electric-ai-chat安装依赖pnpm install用 Docker 启动 Postgres 和 Electricdocker compose up -d配置 OpenAI API key 并启动后端 APIexport OPENAI_API_KEYyour-openai-api-key pnpm dev:api启动演示应用pnpm dev:app浏览器打开localhost:5173即可看到token 流经存储的弹性会话、多设备扇出、多用户实时协作以及监视 todo 列表的 Agent 与用户共享同一份数据。延伸阅读同步总览 与 同步演示TypeScript 客户端文档、React 绑定文档Shapes 指南部分复制的定义方式协议实现参考HTTP 同步协议、ShapeStream 状态机、带退避的拉取与重连、useShape 实现【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考