大模型流式 JSON 怎么稳定落库：增量缓冲、完成事件与幂等写入

聊天窗口里看着顺滑的逐字输出，到了服务端往往是另一回事：网络断在第 37 个分片、网关重试了一次、模型把一个对象拆成十几段发来。要是每收到一段就调用 json.Unmarshal，或者先把片段写进业务表，最终很容易留下半截 JSON、重复记录和无法追踪的状态。

先把流式输出看成一段未提交的传输

不少模型服务会用 SSE 推送增量事件。以 Responses API 一类接口为例，调用方能收到创建、内容增量和结束相关的事件；但不同供应商的事件名、字段位置并不完全一致。服务端真正应依赖的不是某个固定名称，而是两件事：内容是否已聚合完整，以及上游是否明确给出可接受的终态。

这里别急着把 delta 反序列化。分片边界由网络和服务端缓冲决定，{"title":"春 与 天"} 分开抵达很正常。把每段先追加到内存或短期缓存，才能保证解析器面对的是一个完整文档。

接口契约里至少要有四个字段

生产环境里我们一般会把模型调用封装成内部任务，不让业务层直接接触零散事件。下面这组字段足够覆盖大多数“生成后落库”的场景，命名可以按团队现有规范调整。

如果上游支持 JSON Schema 或结构化输出，可以把字段约束提前，这能减少格式出错的概率，但不会自动解决“只收到半段”的问题。流式模式下仍要把完成事件作为提交闸门。另一个容易漏掉的边界是拒答和截断，它们有时也带着文本，但不等于一条可交付的业务数据。

Go 示例：聚合、确认、解析三步分开

下面的例子没有绑定任何一家厂商的特定 SDK。Chunk 可以来自任意 SSE 解码器，关键是把终态判断与正文解析的逻辑完全隔开。

type Chunk struct {
    RequestID string
    Delta     string
    Status    string // in_progress, completed, failed, incomplete
}

type Result struct {
    Title string `json:"title"`
    Score int    `json:"score"`
}

func collect(chunks 

这段代码有一个刻意的取舍：连接关闭时不猜测结果是否完整。宁可把任务标为待核对，也不要从残缺文本里“尽力解析”出一条看似正常的数据。对摘要、标签、风控结论这类会进入后续流程的数据，这条边界尤其重要。

完成事件之后，才开始幂等写入

解析通过后也别立刻认为任务结束。超时重试、消息重复投递和客户端刷新，都可能让同一个 request_id 再次抵达。推荐让业务表的 request_id 带唯一约束，并把状态限制在 pending、completed、failed 三种可解释的值。

func saveOnce(ctx context.Context, db *sql.DB, requestID string, r Result) error {
    tx, err := db.BeginTx(ctx, nil)
    if err != nil { return err }
    defer tx.Rollback()

    // runSQL 是项目内对事务写入的轻量封装；这里省略驱动调用细节。
    _, err = runSQL(ctx, tx, `
        INSERT INTO ai_result (request_id, title, score, status)
        VALUES (?, ?, ?, 'completed')
        ON DUPLICATE KEY UPDATE request_id = request_id`,
        requestID, r.Title, r.Score)
    if err != nil { return err }
    return tx.Commit()
}

MySQL 的唯一索引把“首次提交”和“重复到达”压缩成一次确定的判断。业务层随后查询 request_id 对应记录并返回即可。如果还要记录模型原文，建议单独放在审计表或对象存储，主业务表只保留经过 JSON 解析和字段校验后的数据。

异常时保留什么，删除什么

排查时最有价值的通常不是一长串日志，而是能把一次调用串起来的最小证据：request_id、上游终态、分片数量、完整原文的哈希、JSON 解析错误和入库结果。原始内容如果包含用户输入，应按业务的数据保留规则脱敏并设置过期时间。

• 收到失败或不完整终态：停止追加，记录原因，标记为 failed。
• 完成态但 JSON 解析失败：保留受控的原文证据，不创建业务结果。
• 事务提交失败：保留缓冲区，允许同一请求 ID 在恢复后再次提交。
• 重复请求：优先读取已有 completed 记录，避免再次请求模型。

相关问题

流式输出能不能边收边写数据库？

可以写到短期缓冲表或缓存，但不建议直接写入最终业务表。最终表应由完成态与完整 JSON 共同触发。

用了 JSON Schema 还需要自己校验吗？

需要。Schema 能收紧生成格式，服务端仍应校验必填字段、业务范围和数据库约束；它们解决的问题不一样。

没有收到完成事件时能否按超时自动提交？

不建议。超时只能说明连接异常，不能证明文本完整。可以改为待核对或重新发起同一请求，而不是提交猜测结果。

幂等键应该由谁生成？

最好由业务入口生成并贯穿网关、模型调用和数据库写入。这样用户重试、队列补偿和人工排查都能对齐同一条记录。

把“看起来完成”变成可验证的完成

流式接口的优势是反馈快，代价是调用方必须承担状态收敛。把分片累积、终态确认、JSON 解析和幂等提交拆开后，每一步都有明确的失败去向：半截文本不会污染业务表，重复请求也不会制造第二份结果。先把这条链路跑通，再去优化模型提示词和响应速度，通常更划算。