流式输出：打字机效应背后的秘密与 Go 实战

大家好，我是极客老墨。

你有没有注意到，ChatGPT 的回答从来不是"啪"地一下全部出现，而是一个字、一个词地慢慢冒出来？

第一次见到这个效果，我以为不过是个前端动画——等后端完整结果回来再放个打字机特效。后来我亲手对接了 API 才发现：根本不是动画，模型确实是一边生成一边发送的。

这个"一字一字吐"的机制，就是今天要聊的流式输出（Streaming）。它是提升 AI 应用用户体验（UX）最核心的技术之一。

一、为什么非要流式输出不可？

做个简单的算术题：你让大模型写一篇 500 字的技术分析，DeepSeek V4 Flash 平均要 3~8 秒才能生成完整响应。

非流式方案：用户提交问题，盯着转圈的 loading 等 8 秒，然后文字突然全部出现。 流式方案：用户提交问题，约 0.2 秒后（首字延迟，TTFT）就开始出字，用户边读 AI 边写。

总耗时其实差不多，但流式输出让用户的感知速度有明显提升。在用户眼里，AI 是在"实时思考"，而不是在"黑盒等待"。我自己做产品的经验是：只要涉及文本生成，默认开启流式，哪怕是内部工具。

二、SSE：大模型流式输出的标准协议

要把数据一小块一小块地推给客户端，目前行业通用的标准是 SSE（Server-Sent Events，服务端推送事件）。

1. SSE 是什么？

SSE 是一个基于 HTTP 的单向推送标准。

  • 类比一下：WebSocket 是电话（双向通话）；SSE 是广播（服务器一直播，你只管听）。
  • 优势：轻量级、原生支持 HTTP、自带自动重连、不需要像 WebSocket 那样维护复杂的长连接握手。

2. SSE 消息长什么样？

SSE 的数据格式极其简单，就是纯文本。每个事件由 data: 开头，以两个换行符 \n\n 结尾。

1data: {"content": "你"}
2
3data: {"content": "好"}
4
5data: {"content": "世"}
6
7data: [DONE]

浏览器接收到这种特定格式的 HTTP 响应，就会实时触发回调，把内容传给 JS 处理。

三、Go 实战：三层架构实现打字机效果

在生产环境里，我们通常不会让客户端直连 DeepSeek（为了保护 API Key）。标准的架构是：前端 → 你的 Go 后端 → DeepSeek API。

你的 Go 后端需要扮演一个"中转站"的角色：一边读取 DeepSeek 的流，一边把流转发给前端。

1. 核心工具准备

Go 社区里用得最多的 OpenAI 兼容客户端是 go-openai，DeepSeek 的 API 完全兼容，直接改 BaseURL 就能用。

1go get github.com/sashabaranov/go-openai

2. 后端实现：SSE 转发器

关键点在于使用 http.Flusher 接口，强制将缓冲区的数据实时推出去。

 1func handleStream(w http.ResponseWriter, r *http.Request) {
 2    // 1. 设置 SSE 必需的 Header
 3    w.Header().Set("Content-Type", "text/event-stream")
 4    w.Header().Set("Cache-Control", "no-cache")
 5    w.Header().Set("Connection", "keep-alive")
 6
 7    flusher, ok := w.(http.Flusher)
 8    if !ok {
 9        http.Error(w, "Streaming unsupported!", 500)
10        return
11    }
12
13    // 2. 初始化 DeepSeek 客户端（API Key 从环境变量读取）
14    config := openai.DefaultConfig(os.Getenv("DEEPSEEK_API_KEY"))
15    config.BaseURL = "https://api.deepseek.com"
16    client := openai.NewClientWithConfig(config)
17
18    req := openai.ChatCompletionRequest{
19        Model: "deepseek-v4-flash",
20        Messages: []openai.ChatCompletionMessage{
21            {Role: openai.ChatMessageRoleUser, Content: question},
22        },
23        Stream: true, // api调用也需要按照流式输出
24    }
25
26    stream, err := client.CreateChatCompletionStream(r.Context(), req)
27    if err != nil {
28        fmt.Fprintf(w, "event: error\ndata: %s\n\n", err.Error())
29        return
30    }
31    defer stream.Close()
32
33    // 3. 循环读取并转发
34    for {
35        chunk, err := stream.Recv()
36        if err == io.EOF {
37            fmt.Fprintf(w, "data: [DONE]\n\n")
38            flusher.Flush()
39            return
40        }
41        if err != nil {
42            fmt.Fprintf(w, "event: error\ndata: %s\n\n", err.Error())
43            flusher.Flush()
44            return
45        }
46
47        content := chunk.Choices[0].Delta.Content
48        payload, _ := json.Marshal(map[string]string{"content": content})
49        fmt.Fprintf(w, "data: %s\n\n", payload)
50        // 关键：每次写完必须 Flush，否则数据积在缓冲区
51        flusher.Flush()
52    }
53}

3. 前端实现：EventSource 消费

浏览器原生的 EventSource 对象天然支持 SSE，调用很简单。

1const es = new EventSource('/stream?q=你好');
2es.onmessage = (event) => {
3    if (event.data === '[DONE]') {
4        es.close();
5        return;
6    }
7    const { content } = JSON.parse(event.data);
8    document.getElementById('content').innerText += content;
9};

EventSource 有一个限制：只支持 GET 请求。如果你需要 POST（比如传一个完整的 messages 数组），要么改用 fetch + ReadableStream 手动解析 SSE，要么用社区库 @microsoft/fetch-event-source。

四、V4 Thinking 模式下的流式处理

上一篇我们提到 DeepSeek V4 支持 Thinking 模式（通过 thinking: {"type": "enabled"} 开启）。开启后，模型会先输出推理过程（reasoning_content），再输出最终回答（content）。

在流式模式下，这两个字段是分阶段出现在 SSE chunk 里的：

1// 阶段 1：推理过程（reasoning_content 有值，content 为空）
2data: {"choices":[{"delta":{"reasoning_content":"让我先分析"}}]}
3data: {"choices":[{"delta":{"reasoning_content":"这个问题..."}}]}
4
5// 阶段 2：最终回答（reasoning_content 为空，content 有值）
6data: {"choices":[{"delta":{"content":"Go 语言的"}}]}
7data: {"choices":[{"delta":{"content":"goroutine..."}}]}
8
9data: [DONE]

处理逻辑是在循环里分别拼接两个字段：

 1var reasoningBuf, contentBuf strings.Builder
 2for {
 3    chunk, err := stream.Recv()
 4    if err == io.EOF { break }
 5
 6    delta := chunk.Choices[0].Delta
 7    if delta.ReasoningContent != "" {
 8        reasoningBuf.WriteString(delta.ReasoningContent)
 9        // 可以实时推给前端，展示在折叠区域
10    }
11    if delta.Content != "" {
12        contentBuf.WriteString(delta.Content)
13        // 推给前端主区域
14    }
15}

这里有一个和上一篇相呼应的点：reasoning_content 不能回灌到多轮对话的 messages 数组里。流式拼接完以后，只把 contentBuf 的内容存进历史，reasoningBuf 的内容记日志或展示后丢弃。

五、上线避坑：为什么你的流式"卡住了"？

本地跑得好好的流式，一上线发现变成了"等 8 秒一起出"。我遇到过的原因按频率排：

Nginx 的 proxy_buffering。Nginx 默认会缓存后端的响应，攒够了再一次性发给客户端。解法是在对应 location 里加上 proxy_buffering off;，或者让 Go 后端在响应头里带上 X-Accel-Buffering: no——后者更灵活，只影响流式接口。

1location /stream {
2    proxy_pass http://backend;
3    proxy_buffering off;      # 方案一：Nginx 配置
4    # 方案二：后端响应头 X-Accel-Buffering: no
5}

忘记 Flush()。Go 的 http.ResponseWriter 默认有缓冲。不手动调用 flusher.Flush()，数据会一直攒在内存里，直到连接关闭才一股脑发出去。每次 Fprintf 之后都要紧跟一个 Flush()。

CDN 或云负载均衡器的响应缓冲。AWS ALB、Cloudflare 等中间层也可能缓冲 SSE 响应。通常需要设置 Content-Type: text/event-stream 并确认中间层对此类型不做缓冲。

TTFT（Time To First Token）偏高。如果首字延迟超过 2 秒，问题可能不在流式转发，而在模型本身——比如 prompt 太长导致预填充时间拉长，或者 Thinking 模式下推理阶段本身就需要几秒。可以从 API 响应的第一个 SSE chunk 到达时间来判断。

老墨总结

流式输出的核心是 SSE 协议 + http.Flusher 手动推流。代码量不大，但上线后要关注的细节不少。

老墨说： 我现在判断一个 AI 应用是否"能用"，第一个看的就是 TTFT——首字出来要多久。流式的意义不是让总耗时变短，而是把用户等待的感知从"黑盒"变成"实时"。如果 TTFT 超过 3 秒，用户会以为系统卡了，不管后面生成得多流畅。

V4 Thinking 模式给流式加了一层复杂度（reasoning_content 的分阶段处理），但也给产品带来了新的展示可能——把推理过程实时展示出来，让用户看到 AI “在想什么”。

下一篇聊大模型的最后一个基础课题——它的局限性与开发者的应对策略。

完整示例代码见 Github

关注公众号：极客老墨

更多 AI 应用开发、工程实践和效率工具分享，欢迎扫码关注。

极客老墨微信公众号二维码

相关阅读