AI 流式输出中途断开时怎么保存已生成内容

AI 流式输出中途断开时，最重要的原则是：不要把保存动作放在“完整响应返回”之后。服务端应该以请求唯一标识维护一条可更新的流记录，每收到一组文本增量就保存一个带序号的快照；连接关闭时保留已确认内容，并把状态标记为 interrupted 或 partial，而不是误标成完成。

先把一条流看成可恢复的记录

流式接口通常通过 SSE 持续发送事件，文本只是其中一种增量事件。以 OpenAI Responses API 为例，请求打开 stream=true 后，客户端可以处理 response.output_text.delta，并在 response.completed 到达时确认完整响应；事件流里还可能出现错误事件。不要把每个事件直接当成最终答案，先建立一条有版本的记录。

最小字段可以是：request_id（业务请求幂等键）、sequence（已确认的片段序号）、content（当前正文快照）、status、provider_response_id、error_message 和 updated_at。request_id 建唯一索引，更新条件同时带上旧序号，才能防止较早的重试覆盖较新的内容。

收到增量就保存，但不要每个字符都写一次

实现时把“展示”和“持久化”分开：前端可以即时渲染，服务端则把增量先放进内存缓冲，在达到固定片段数、时间间隔或换行边界时写入快照。遇到完成、错误或连接异常时无条件刷最后一批数据。下面的 Node.js 结构展示了关键边界，saveCheckpoint 应在事务中按 requestId 做条件更新。

import OpenAI from "openai";

const client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

async function consume(requestId, input) {
  let sequence = 0;
  let content = "";
  let lastFlush = Date.now();
  const stream = await client.responses.create({
    model: process.env.OPENAI_MODEL,
    input,
    stream: true,
  });

  // 保存最近一次完整快照，重连时从它继续，而不是重放所有事件。
  const flush = async (status, errorMessage = "") => {
    await saveCheckpoint({ requestId, sequence, content, status, errorMessage });
    lastFlush = Date.now();
  };

  try {
    for await (const event of stream) {
      if (event.type === "response.output_text.delta") {
        content += event.delta;
        sequence += 1;
        // 用频率阈值控制写入量，但不影响前端即时输出。
        if (sequence % 8 === 0 || Date.now() - lastFlush >= 1000) {
          await flush("partial");
        }
      } else if (event.type === "response.completed") {
        await flush("completed");
      } else if (event.type === "error") {
        await flush("error", event.message || "stream error");
      }
    }
  } catch (error) {
    // 连接异常也要刷盘，已生成内容不能随异常一起丢失。
    await flush("interrupted", String(error.message || error));
    throw error;
  }
  return { requestId, sequence, content };
}

这里的关键不是某个 SDK 的字段名，而是状态转换：增量事件只推进序号和正文，完成事件才进入 completed；异常路径只保存已经收到的内容。若生产环境按字符量很大，可以把每次快照改成“追加片段表 + 定期正文合并”，但恢复逻辑仍要有一个明确的最后确认序号。

断线恢复要靠幂等键和最后确认序号

重连时不要无条件把新响应拼到旧正文后面。客户端先读取服务端保存的 sequence，再使用同一个 request_id 发起恢复请求；服务端只接受大于已确认序号的片段，或者用版本条件覆盖同一快照。若上游不能从指定序号继续，宁可新建一次带 parent_request_id 的重试记录，也不要把两次生成结果静默混在一起。

还要区分三种“断开”：用户主动取消、网络断开、上游返回错误。它们都可以保留正文，但恢复策略不同：主动取消通常直接展示已生成内容；网络断开可以允许一次重连；上游错误则保留错误原因并等待人工或业务重试。状态字段不能只用一个布尔值，否则排查时无法知道内容为什么停在半截。

回滚、告警和常见问题

如果发现重复文本，先暂停自动重试，保留原始 request_id、最后序号和上游响应标识，再从最近一次快照恢复。数据库写入失败时不能继续把内存缓冲当作“已保存”，应降低并发或进入降级队列。对持续超过业务阈值仍为 partial 的记录报警，并把快照保留时间与敏感内容清理策略一起定义。

为什么只在流结束时保存会丢内容？

因为连接断开、进程重启或上游错误都可能发生在完成事件之前，结束回调根本不会执行。增量快照才是可恢复数据。

保存每个片段是不是最安全？

不一定。每个片段落库会放大写入压力；更实用的做法是按数量或时间批量刷盘，并在异常与完成路径强制保存。

重连后如何判断能不能继续拼接？

比较同一 request_id 的最后确认序号和新事件的序号。无法证明连续时就新建重试记录，不要直接追加到旧正文。

官方流式说明见 OpenAI Streaming API responses。将事件消费、快照保存、状态机和重试边界拆开后，即使 AI 输出只完成了一半，系统也能给用户一个可继续处理的结果，而不是只剩一条空记录。