AI异步任务架构设计：SSE、检查点与幂等性实现断点续传

1. 项目概述：当AI生成任务“卡在”99%

相信很多在业务中集成过AI生成能力（无论是文本、图像还是代码）的开发者，都遇到过这个让人血压飙升的场景：你提交了一个耗时较长的生成任务，比如生成一份长篇报告、一张高分辨率图片，或者一段复杂的代码。前端进度条在稳步推进，用户满怀期待地等待着。突然，在进度达到90%、95%甚至99%的时候，请求超时了，连接断开了，或者后端服务因为某个不可预知的错误崩溃了。用户看到的是一个“生成失败”的提示，而你的服务器可能已经为这个任务消耗了大量的计算资源。更糟糕的是，用户重试，一切又得从头开始，这不仅浪费资源，体验也极其糟糕。

这个问题的核心，在于我们将AI生成这类长时间运行、非确定性的异步任务，错误地用处理传统短时、同步请求的思维去架构了。传统HTTP请求-响应模型是“一锤子买卖”，连接断开就意味着事务结束。但AI生成任务的生命周期远超一次HTTP连接所能维持的时间，其内部状态（生成了多少内容、模型参数、随机种子等）是连续且宝贵的。

我最近在重构一个智能文档生成系统时，就深度踩了这个坑。我们的需求是用户输入一个主题，AI自动生成结构完整、数据翔实的万字行业分析报告。初期方案简单粗暴：前端发起一个POST请求，后端调用大模型API，同步等待所有内容生成完毕，一次性返回。结果就是，10次请求里有3次会因为网络波动或服务端超时设置而中断，用户只能看到“请求超时”，然后骂骂咧咧地离开。

为了解决这个问题，我引入了一套组合拳：SSE（Server-Sent Events）用于实时流式输出、检查点（Checkpoint）用于保存中间状态、幂等（Idempotency）设计用于安全重试。这套方案的核心思想是：将一次性的“生成”动作，拆解为一个可暂停、可恢复、可观测的“状态流”。最终，我们实现了即使连接在最后1%断开，用户重新进入页面，任务也能从断点继续，并且之前已经生成的内容能立刻呈现出来，体验无缝衔接。下面，我就把这套方案的详细设计思路、技术选型考量、实操步骤以及填坑经验分享出来。

2. 核心架构设计与思路拆解

面对“AI生成到90%断了”的难题，我们不能只治标（比如简单调大超时时间），而需要一套治本的架构。这套架构需要满足几个核心目标：状态可持久化、进度可观测、操作可安全重试。我选择的三个核心技术组件正是为了分别解决这三个问题。

2.1 为什么是SSE，而不是WebSocket或长轮询？

首先解决“进度可观测”和“实时输出”的问题。我们需要一种机制，让服务器能在任务执行过程中，主动、持续地向客户端推送状态更新和已生成的内容片段。

• 长轮询（Long Polling）：技术简单，但效率低下。客户端需要反复发起请求，在任务完成或超时前，服务器会挂起连接。对于长达数分钟的任务，这会造成大量无效的连接占用和频繁的请求-响应开销，不是优雅的解决方案。
• WebSocket：功能强大，支持全双工通信。但对于我们“服务器向客户端单向推送任务进度和结果片段”这个核心场景来说，它有点“杀鸡用牛刀”。WebSocket需要更复杂的连接管理（握手、心跳、协议升级），并且对于只需要接收信息的客户端来说，引入了不必要的复杂性。
• SSE（Server-Sent Events）：它是基于HTTP的单向通信协议，专为服务器向客户端推送事件流而设计。其优势非常契合我们的场景：
  1. 协议简单：本质上是保持一个HTTP连接，以text/event-stream格式持续发送数据流。客户端使用标准的EventSourceAPI即可连接监听，无需额外库。
  2. 自动重连：EventSource内置了断线重连机制。当连接意外断开，客户端会自动尝试重新连接，并在重连时发送上一次接收到的事件ID。这为我们实现“断点续传”提供了天然支持。
  3. 与HTTP生态无缝集成：鉴权、缓存、代理等都可以沿用现有的HTTP基础设施，部署和调试成本低。

注意：SSE有一个重要限制，即它是文本协议，传输二进制数据（如图片流）需要先进行Base64编码。对于文生图场景，需要评估编码带来的带宽开销。在我们的文本生成场景中，这是最佳选择。

因此，选择SSE是为了以最小的成本和复杂度，实现稳定的进度推送和内容流式输出，并利用其自动重连特性为恢复任务铺路。

2.2 检查点（Checkpoint）：任务状态的“存档点”

检查点的概念来源于游戏和分布式计算，其核心是将任务的中间状态持久化。对于AI生成任务，这个“状态”远比一个简单的进度百分比复杂。

一个AI文本生成任务的检查点至少需要包含：

1. 任务元数据：任务ID、用户ID、创建时间、使用的模型、参数（如temperature, top_p）。
2. 已生成的完整内容：这是最重要的部分。不能只存个进度，必须把已经流式输出给客户端的每一个片段都完整保存下来。
3. 模型内部状态（如果可能）：对于某些可以控制生成过程的模型，可能需要保存解码器的隐藏状态、已生成的token序列等。但大多数云端API不暴露此细节，所以我们的检查点更多是“结果缓存”+“续命参数”。
4. 续命参数：为了能从断点继续，我们需要记录“最后一句完整的话是什么”、“生成到哪个章节了”，或者直接保存最后一次向模型发送的“prompt + 已生成内容”作为新的上下文，以便继续请求。

检查点的存储介质选择：

• 内存（如Redis）：存取速度快，适合高频更新的临时状态。但服务重启数据即丢失，不适合作为唯一存储。我们用它来存“实时进度”和“临时内容缓冲”。
• 持久化数据库（如PostgreSQL, MySQL）：作为最终状态的存储。当任务完成或客户端确认接收后，将完整内容归档至此。也可用于存储检查点，但频繁更新长文本字段可能对数据库有压力。
• 对象存储（如S3, MinIO）：存储大型内容（如长文本、图片）的绝佳场所。可以将每次推送的内容片段追加存储到一个文件中，或者直接存储完整的生成结果。成本低，扩展性好。

在我们的方案中，采用“Redis + 数据库”的混合模式。Redis存储活跃任务的实时状态和最新内容片段，数据库存储任务元数据和最终完成态。检查点的写入时机是关键。

2.3 幂等（Idempotency）：安全重试的保障

幂等性意味着同一个操作执行一次或多次，对系统状态产生的影响是相同的。在任务中断重试的场景下，幂等设计是防止重复生成、数据混乱的“安全锁”。

为什么需要幂等？假设一个任务在90%时连接断开，客户端自动重连（SSE特性）。如果没有幂等控制，服务器可能：

1. 错误地启动了一个全新的任务，从0%开始。
2. 或者，虽然尝试恢复旧任务，但因为重复的请求，导致旧任务被重复提交了多次，造成资源浪费和结果错乱。

实现任务级别的幂等，通常通过**唯一的幂等键（Idempotency Key）**来实现。这个键通常由客户端在首次请求时生成（如一个UUID），并随请求头（如Idempotency-Key: <uuid>）发送。

• 服务器端收到请求后，首先以这个幂等键为主键，查询是否存在未完成或已完成的任务记录。
• 如果存在未完成的记录，则直接返回该任务的当前状态和SSE连接信息，实现“重连续传”。
• 如果存在已完成的记录，则直接返回最终结果。
• 如果不存在记录，则创建新任务，并将幂等键与任务绑定。

这样，无论客户端因为网络问题发送了多少次重连请求，系统都只会有一个对应的任务在执行或已执行完毕，完美避免了重复劳动。

3. 系统流程与核心环节实现

下面，我将结合代码片段（以Node.js + Express为例）和流程图，详细拆解这套系统是如何协同工作的。

3.1 任务生命周期与状态流转

一个具备断点续传能力的AI生成任务，其状态机比简单的“进行中/完成/失败”要复杂。

[客户端] --(1. 创建请求，携带 Idempotency-Key)--> [服务器] [服务器] --(2. 检查幂等键)--> [Redis/DB] |-> 键存在且任务未完成 -> 跳转至步骤6 (重连) |-> 键存在且任务已完成 -> 直接返回最终结果 |-> 键不存在 -> 继续步骤3 [服务器] --(3. 创建任务记录，状态=PENDING)--> [DB] [服务器] --(4. 异步执行生成任务)--> [Worker/Async Process] [客户端] --(5. 建立SSE连接 /tasks/:id/events)--> [服务器] [服务器] --(6. 将SSE连接与任务ID绑定)--> [Redis] [Worker] --(7. 生成内容片段，发布事件)--> [Redis Pub/Sub] [SSE Handler] --(8. 监听事件，推送至客户端)--> [客户端] [Worker] --(9. 每生成一段，更新检查点)--> [Redis] [客户端] --(10. 接收并显示片段)--> [UI] |-> 连接断开 -> 客户端EventSource自动重连，携带Last-Event-ID |-> 重连成功 -> 服务器根据Last-Event-ID从检查点获取历史事件并重放，然后继续推送新事件 [Worker] --(11. 生成完成，更新任务状态=COMPLETED，持久化结果)--> [DB] [服务器] --(12. 发送`[DONE]`事件，关闭SSE流)--> [客户端]

3.2 关键代码实现解析

1. 创建幂等任务端点

// 使用 Express 框架 app.post('/api/generate/report', async (req, res) => { const { topic, parameters } = req.body; const idempotencyKey = req.headers['idempotency-key']; // 客户端提供 if (!idempotencyKey) { return res.status(400).json({ error: 'Idempotency-Key header is required' }); } // 检查幂等键 const existingTask = await db.task.findUnique({ where: { idempotencyKey } }); // 情况1：任务已存在且完成 if (existingTask && existingTask.status === 'COMPLETED') { return res.json({ taskId: existingTask.id, status: 'COMPLETED', resultUrl: existingTask.resultUrl // 直接返回结果地址 }); } // 情况2：任务已存在且未完成（可能是PENDING或PROCESSING） if (existingTask && ['PENDING', 'PROCESSING'].includes(existingTask.status)) { // 返回任务信息，让客户端去连接对应的SSE流 return res.json({ taskId: existingTask.id, status: existingTask.status, streamUrl: `/api/tasks/${existingTask.id}/events` // SSE流地址 }); } // 情况3：全新任务 const newTask = await db.task.create({ data: { idempotencyKey, topic, parameters, status: 'PENDING', userId: req.user.id // 假设有用户认证 } }); // 异步触发任务执行（放入消息队列，如Bull、RabbitMQ） await taskQueue.add('generate-report', { taskId: newTask.id, topic, parameters }); // 更新任务状态为处理中（可选，也可由Worker自己更新） await db.task.update({ where: { id: newTask.id }, data: { status: 'PROCESSING' } }); res.json({ taskId: newTask.id, status: 'PROCESSING', streamUrl: `/api/tasks/${newTask.id}/events` }); });

2. SSE事件流端点实现

这是连接客户端，实现实时推送和断点续传的核心。

app.get('/api/tasks/:taskId/events', async (req, res) => { const { taskId } = req.params; const lastEventId = req.headers['last-event-id']; // SSE标准重连头，客户端自动发送 // 设置SSE响应头 res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', // 重要：允许跨域（如果前端分离部署） 'Access-Control-Allow-Origin': '*' }); // 发送一个初始心跳或确认连接的事件 res.write(`id: ${Date.now()}\n`); res.write(`event: connected\n`); res.write(`data: ${JSON.stringify({ taskId })}\n\n`); // **关键：处理断点重连 - 重放历史事件** if (lastEventId) { // 根据lastEventId，从Redis或DB中查询该ID之后的所有事件 const historicalEvents = await redis.lrange(`task:${taskId}:events`, 0, -1); // 假设事件列表存在Redis let startReplaying = false; for (const eventData of historicalEvents) { const event = JSON.parse(eventData); // 找到最后一个已接收事件的位置，然后发送之后的事件 if (event.id === lastEventId) { startReplaying = true; continue; } if (startReplaying) { res.write(`id: ${event.id}\n`); res.write(`event: ${event.type}\n`); res.write(`data: ${JSON.stringify(event.data)}\n\n`); } } } // 订阅该任务的新事件（使用Redis Pub/Sub） const subscriber = redis.duplicate(); await subscriber.connect(); await subscriber.subscribe(`task:${taskId}:stream`, (message) => { const event = JSON.parse(message); // 按照SSE格式发送事件 res.write(`id: ${event.id}\n`); // 事件ID，用于断点重连 res.write(`event: ${event.type}\n`); // 事件类型，如 `chunk`, `progress`, `error` res.write(`data: ${JSON.stringify(event.data)}\n\n`); }); // 客户端关闭连接时，清理订阅 req.on('close', () => { subscriber.unsubscribe(`task:${taskId}:stream`); subscriber.quit(); console.log(`Client disconnected from stream for task ${taskId}`); }); });

3. 后台Worker与检查点更新

Worker是实际执行AI生成的部分。它需要与SSE流和检查点存储紧密配合。

// 伪代码，展示Worker逻辑 async function generateReportWorker(taskId, topic, parameters) { try { const taskStreamKey = `task:${taskId}:stream`; const checkpointKey = `task:${taskId}:checkpoint`; const eventsListKey = `task:${taskId}:events`; // 初始化检查点：从Redis加载，如果不存在则新建 let checkpoint = await redis.get(checkpointKey); let fullContent = ''; if (checkpoint) { console.log(`Resuming task ${taskId} from checkpoint.`); const cp = JSON.parse(checkpoint); fullContent = cp.fullContent || ''; // 可能需要根据cp中的信息调整模型调用参数，例如设置`prompt`为已生成内容 } else { console.log(`Starting new task ${taskId}.`); // 初始化检查点结构 checkpoint = { fullContent: '', lastSentEventId: null }; } // 模拟调用大模型API的流式接口（例如OpenAI的stream: true） const stream = await openai.chat.completions.create({ model: 'gpt-4', messages: [{ role: 'user', content: `基于主题“${topic}”生成报告。` }], stream: true, }); let accumulatedChunk = ''; for await (const part of stream) { const contentDelta = part.choices[0]?.delta?.content || ''; if (contentDelta) { accumulatedChunk += contentDelta; fullContent += contentDelta; // **策略：按句子或段落分割推送，而不是每个token都推** // 这里简单按句号分割作为示例，实际可按\n\n或固定长度分割 if (accumulatedChunk.endsWith('。') || accumulatedChunk.endsWith('.\n')) { const eventId = `evt_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`; const eventData = { id: eventId, type: 'chunk', data: { chunk: accumulatedChunk, progress: calculateProgress(fullContent) // 估算进度 } }; // 1. 发布事件到SSE流 await redis.publish(taskStreamKey, JSON.stringify(eventData)); // 2. 将事件存入历史列表，用于断线重连时重放 await redis.rpush(eventsListKey, JSON.stringify(eventData)); // 控制列表长度，防止内存无限增长，只保留最近N个事件或一段时间内的事件 await redis.ltrim(eventsListKey, -100, -1); // 只保留最后100个事件 // 3. 更新检查点（异步或定期进行，避免每次写入） // 这里采用节流方式，每推送5个片段或内容长度增长超过500字符时更新一次 if (eventsListKey.length % 5 === 0 || fullContent.length - checkpoint.fullContent.length > 500) { const newCheckpoint = { fullContent: fullContent, lastSentEventId: eventId, updatedAt: new Date().toISOString() }; await redis.setex(checkpointKey, 86400, JSON.stringify(newCheckpoint)); // 设置24小时过期 } accumulatedChunk = ''; // 清空当前累积片段 } } } // 生成完成，处理最后的累积内容（如果有） if (accumulatedChunk) { // ... 同样发布事件、存储、更新检查点 ... } // 最终步骤：发送完成事件，持久化最终结果，清理临时数据 const doneEvent = { id: `evt_final`, type: 'done', data: { resultUrl: `/api/results/${taskId}` } }; await redis.publish(taskStreamKey, JSON.stringify(doneEvent)); await db.task.update({ where: { id: taskId }, data: { status: 'COMPLETED', result: fullContent, completedAt: new Date() } }); // 可选：清理Redis中的临时数据（检查点、事件列表），或设置更短的过期时间 await redis.del(checkpointKey, eventsListKey); } catch (error) { console.error(`Task ${taskId} failed:`, error); // 发布错误事件 await redis.publish(taskStreamKey, JSON.stringify({ id: `evt_error_${Date.now()}`, type: 'error', data: { message: '生成任务失败', error: error.message } })); await db.task.update({ where: { id: taskId }, data: { status: 'FAILED', error: error.message } }); } }

4. 实操避坑指南与性能优化

在实际落地这套方案的过程中，我遇到了不少坑，也总结出一些优化经验。

4.1 检查点更新的频率与粒度

坑：最初，我每生成一个token（或一个很小的片段）就更新一次Redis检查点。这导致了极高的Redis IOPS，在并发任务多的时候，Redis成了性能瓶颈，甚至影响了事件推送的实时性。

解决方案：采用节流（Throttle）和防抖（Debounce）思想更新检查点。

• 时间阈值：至少每2-5秒才更新一次检查点，而不是实时更新。
• 内容阈值：累积生成的内容长度超过一定字符数（如500字）再更新。
• 事件阈值：每推送N个SSE事件后更新一次。
• 组合使用：在我们的最终方案中，采用了“内容增长超过500字符或每推送5个事件”的复合条件，在数据安全性和性能之间取得了很好的平衡。

4.2 SSE连接管理与资源释放

坑：当用户离开页面时，浏览器可能会关闭EventSource连接，但服务器端的响应流（res对象）可能不会立即被Node.js垃圾回收，订阅了Redis频道的连接也未释放，导致内存和连接泄漏。

解决方案：

1. 监听req.on('close')事件：这是最重要的。一旦客户端连接关闭，立即取消Redis订阅并清理相关资源（如上面的代码所示）。
2. 设置心跳与超时：在SSE流中定期发送注释行（:开头的行）作为心跳。可以设置一个服务器端的超时计时器，如果长时间未收到客户端心跳（需要客户端配合回送），则主动关闭连接。
3. 使用连接池或唯一标识：将SSE连接对象与任务ID关联存储在一个WeakMap或专门的连接管理器中，便于在任务完成或出错时，主动查找并关闭所有相关的SSE连接。

4.3 历史事件重放的策略与存储

坑：将所有历史SSE事件无限制地存入Redis列表，当生成长文档时，列表可能巨大，消耗大量内存，且在重连时重放全部事件会导致网络流量暴增和客户端处理延迟。

解决方案：

1. 限制列表长度：使用LRANGE和LTRIM命令，只保留最近50-100个事件。因为断线重连通常发生在很短的时间窗口内，不需要保存全部历史。
2. 基于检查点的增量重放：这是更优雅的方案。客户端重连时发送Last-Event-ID。服务器端不需要存储所有历史事件，而是从检查点中取出完整的已生成内容，将其作为“第一个事件”推送给客户端。然后，Worker从检查点记录的位置继续生成新内容。这样，重连时只需要发送一次“全量快照”，后续就是增量流。这要求客户端能处理这种“快照+增量”的模式。
3. 分片存储：对于超长任务，可以将事件按章节或时间分片存储在不同的Key中，重连时只加载最近的一个分片。

4.4 幂等键的生成与生命周期管理

坑：客户端生成的幂等键（如UUID）在用户刷新页面后丢失，导致无法关联到之前的任务，用户被迫开始一个新任务。

解决方案：

1. 前端持久化幂等键：在首次创建任务时，将服务器返回的taskId和客户端自己生成的idempotencyKey存储在localStorage或sessionStorage中。页面刷新后，优先尝试使用存储的idempotencyKey去恢复任务。
2. 服务端关联用户与会话：在创建任务时，不仅检查幂等键，也检查当前用户是否有未完成的相同主题任务（需定义“相同”的语义，如主题、参数一致），如果有，可以提示用户是否恢复。这可以作为幂等键机制的补充。
3. 设置合理的过期时间：在Redis中存储的检查点和任务状态，需要设置过期时间（如24小时）。防止未完成的“僵尸任务”永久占用资源。数据库中的任务记录可以保留更久，但状态应标记为EXPIRED或ABANDONED。

4.5 前端（客户端）的健壮性处理

前端并非只是简单监听SSE，它需要处理各种边界情况。

class AITaskClient { constructor(taskId, streamUrl) { this.taskId = taskId; this.streamUrl = streamUrl; this.eventSource = null; this.retryCount = 0; this.maxRetries = 5; this.accumulatedData = ''; // 用于累积已接收的数据 this.lastEventId = null; // 记录最后收到的事件ID } connect() { const headers = {}; if (this.lastEventId) { headers['Last-Event-ID'] = this.lastEventId; // 断线重连的关键 } this.eventSource = new EventSource(this.streamUrl, { headers }); this.eventSource.onmessage = (event) => { // 标准SSE消息，event.data是字符串 const data = JSON.parse(event.data); this.handleEvent(data); }; this.eventSource.addEventListener('chunk', (event) => { const data = JSON.parse(event.data); this.lastEventId = event.lastEventId; // 更新最后事件ID this.accumulatedData += data.chunk; this.updateUI(this.accumulatedData, data.progress); }); this.eventSource.addEventListener('done', (event) => { const data = JSON.parse(event.data); console.log('Task completed!', data.resultUrl); this.eventSource.close(); this.showFinalResult(data.resultUrl); }); this.eventSource.addEventListener('error', (event) => { console.error('SSE Error:', event); // EventSource在连接失败时会自动重试，但我们可以控制重试逻辑 this.eventSource.close(); this.retryCount++; if (this.retryCount < this.maxRetries) { setTimeout(() => this.connect(), 1000 * Math.pow(2, this.retryCount)); // 指数退避重连 } else { this.showFatalError('任务连接失败，请刷新页面重试。'); } }); this.eventSource.onopen = () => { console.log('SSE连接已建立'); this.retryCount = 0; // 连接成功后重置重试计数 }; } handleEvent(data) { // 处理通用事件或未指定类型的事件 console.log('Received event:', data); } updateUI(content, progress) { // 更新DOM，显示实时生成的内容和进度条 document.getElementById('output').innerText = content; document.getElementById('progressBar').style.width = `${progress}%`; } } // 使用示例：页面加载时，尝试从localStorage恢复任务 const savedTaskId = localStorage.getItem('lastTaskId'); const savedIdempotencyKey = localStorage.getItem('lastIdempotencyKey'); if (savedTaskId && savedIdempotencyKey) { // 尝试恢复任务 fetch(`/api/tasks/${savedTaskId}/status`, { headers: { 'Idempotency-Key': savedIdempotencyKey } }).then(/* ... */); }

5. 扩展思考与方案变体

这套“SSE+检查点+幂等”的模式具有很强的普适性，不仅限于AI文本生成。

• 文生图/图生图：检查点可以存储生成到第几步的潜在变量（Latent）、去噪进度等。SSE推送的不再是文本片段，而是生成过程的预览图（低分辨率或模糊版本），或者进度百分比。最终生成完成后再推送高清图URL。由于图片数据量大，推送预览图需考虑Base64编码的带宽问题，或改用WebSocket分片传输二进制数据。
• 长视频生成/渲染：检查点保存已渲染完成的帧序列或视频片段文件路径。SSE推送渲染进度和已完成的视频片段URL，前端可以边下边播。
• 复杂数据处理/ETL任务：检查点保存已处理的数据批次、转换状态。SSE推送处理进度和日志信息。

一个重要的权衡是状态恢复的粒度。完全无损的“热恢复”（从模型内部状态继续）通常需要AI服务提供方的深度支持，这对于使用云端API的我们来说很难实现。我们实现的更多是一种“温恢复”或“用户无感知的冷恢复”：保存所有已输出的结果，当连接恢复时，立刻将历史结果展示给用户，同时从断点开始请求AI服务生成剩余部分。对于用户而言，体验是连续的，这就已经解决了核心痛点。

最后，这套方案引入了一定的复杂度，包括需要维护Redis、消息队列、更复杂的状态管理。因此，它更适合于生成耗时较长（如>30秒）、结果价值高、用户等待耐心有限的核心场景。对于秒级完成的简单生成，传统的异步轮询或Callback或许仍是更经济的选择。架构的选择，永远是在用户体验、开发成本和系统复杂度之间寻找最佳平衡点。