Go 接入 OpenAI Responses API 流式输出：SSE 事件拼接、断线重试与取消

Go 服务把 Responses API 当成普通 JSON 接口调用时，短文本场景基本不会出问题；一旦切换到流式输出模式，真正难处理的是事件边界问题。一段完整的回复可能被拆成多次增量文本分片，连接中断后又不能把已经展示给用户的片段无脑重复拼接，用户主动点取消时也必须第一时间释放背后的长连接资源。

先把 Responses API 流式调用拆成四个状态

流式接口不是读到一段字符串就直接返回给前端。在实际的客服回复类场景里，Go 服务至少要区分 idle、streaming、completed 和 partial 四个状态：只有收到完成信号，才把缓冲区里的完整内容提交给下游服务；网络断开、超时或用户主动取消，都应该停留在未完成状态。

官方 openai-go 仓库把 Responses API 和流式读取逻辑封装在同一套 SDK 里。后续版本迭代后事件类型和模型常量可能发生变化，下面的代码重点是生命周期处理逻辑，实际落地的时候大家要以本地锁定版本的类型定义为准。

最小可用写法：增量事件只负责拼接

把「读取事件」和「提交业务结果」两个逻辑拆分开，代码后续会更容易测试。事件循环里只做三件事：识别增量文本、记录完成状态、把异常直接抛给上层处理，不要在每个分片到达的时候就直接写数据库。

type StreamResult struct {
    Text      string
    Completed bool
}

func readAnswer(ctx context.Context, stream *responses.ResponseStream) (StreamResult, error) {
    var result StreamResult
    var buf strings.Builder

    for stream.Next() {
        event := stream.Current()
        switch event.Type {
        case "response.output_text.delta":
            buf.WriteString(event.Delta)
        case "response.completed":
            result.Completed = true
        }
    }
    if err := stream.Err(); err != nil {
        return StreamResult{Text: buf.String()}, fmt.Errorf("stream interrupted: %w", err)
    }
    if !result.Completed {
        return StreamResult{Text: buf.String()}, errors.New("stream ended without completed event")
    }
    result.Text = buf.String()
    return result, nil
}

不同 SDK 版本的流对象和事件访问器可能略有差异，不要把示例里用到的类型名当成永久不变的API定义。核心处理逻辑是通用的：循环结束后要检查流返回的错误，完成事件缺失的情况下不提交结果，已经收到的分片内容只作为排查日志或者前端临时展示使用。

SSE 事件为什么不能按“每行一个答案”处理

SSE 在传输层是以事件帧为单位发送，事件数据可能跨越底层的网络读取边界，单次网络读取操作不一定刚好拿到一个完整事件。SDK 负责把字节流整理成独立事件后，业务代码还要按事件的具体类型做分流处理。遇到没见过的未知事件不要直接当成正文追加，否则后续官方新增的元数据字段可能混到最终回答里。

这条规则也能避免重复拼接的问题：重连后发起的新请求要当成一次全新的调用，不能把新请求返回的完整回答拼到旧请求的半成品后面。如果产品层需要做断点续传，应该由服务层设计明确的幂等键和版本号来实现，不能依赖大模型两次输出的内容刚好一致。

断线重试的边界：没有展示内容时重试会更安全

请求刚建立就失败，或者还没有向用户展示任何正文内容时，可以按指数退避策略做一次有限次数的重试。已经显示了「正在生成」之外的正文后再做静默重放，很可能造成前端展示重复回答，甚至误触发重复的工具调用和重复计费请求。

func runWithOneRetry(parent context.Context, ask func(context.Context) (StreamResult, error)) (StreamResult, error) {
    ctx, cancel := context.WithTimeout(parent, 45*time.Second)
    defer cancel()

    first, err := ask(ctx)
    if err == nil {
        return first, nil
    }
    if first.Text != "" || errors.Is(ctx.Err(), context.Canceled) {
        return first, err
    }

    retryCtx, retryCancel := context.WithTimeout(parent, 45*time.Second)
    defer retryCancel()
    return ask(retryCtx)
}

示例只演示策略逻辑，不包含完整的网络重放代码。生产环境里还要额外记录尝试次数、请求唯一标识、首个分片到达时间和最终状态，方便后续排查问题属于服务端错误、客户端超时还是用户主动取消。

context 取消要贯穿到流和上层请求

浏览器断开连接、用户点击停止生成、网关到达超时阈值，最终都应该触发同一个 context 的取消逻辑。只在外层设置定时器却不把 context 传给底层SDK，常见后果就是前端页面已经结束请求，后台和大模型服务的连接还在默默消耗资源读数据。

处理取消逻辑时不要把它误记成服务故障。监控里可以单独统计 context canceled、超时、远端错误和正常完成这几类指标，几类指标对应的优化方向完全不同。完成事件正常到达后再关闭流，异常分支里要保证资源清理逻辑一定会执行。

常见问题：Go 流式接入 Responses API 的几个误区

每收到一段 delta 就写一次数据库，行不行？

不建议。增量写入会放大数据库的压力，也让断线后的半成品内容很难回滚。更稳妥的做法是前端先临时展示，等收到完成事件后一次性提交写入数据库。

连接断开后能不能总是自动重试？

不能。没有展示正文且错误属于短暂网络问题时可做有限重试；已经展示过内容、用户主动取消或鉴权失败时，要直接停止重放。

收到 EOF 就代表回答完成了吗？

不一定。需要同时拿到 SDK 识别的完成事件，并确认流本身没有返回错误，否则只能把当前内容标记为 partial 状态。

上线前的流式检查表

• 锁定 openai-go 版本，编译验证当前事件类型和流对象所有可用方法。
• 覆盖正常完成、空输出、远端错误、网络中断、超时和用户取消等所有分支场景。
• 记录尝试次数与完成状态，禁止 partial 结果直接覆盖数据库里的最终答案。
• 重试最多设置有限次数，并且依据「是否已展示正文」判断当前请求能不能重放。