流式输出（Streaming）完全指南：几行代码，让你的 AI 应用从"能用"变"好用"

你兴冲冲地把 AI 接进了自己的应用，结果用户反馈：

"你这个 AI 是不是挂了？我点了按钮什么都没有……"

等了 20 秒，突然"啪"——一大段文字糊脸上。

这个翻车场景，几乎每个第一次接 AI API 的开发者都经历过。更扎心的是：你的模型选得很好，Prompt 也调得很精，但用户体验就是差那么一口气——而那口气，叫流式输出（Streaming）。

ChatGPT 让全世界觉得 AI 很"聪明"的秘密，有一半藏在那个打字机效果里。今天我们把它彻底拆解：从底层原理到生产级代码，从基础实现到进阶技巧，一篇全搞定。

---

一、为什么"等待"会杀死你的 AI 应用

先看一组数据：

Nielsen Norman Group 的研究早就给出了结论——响应时间超过 1 秒，用户开始失去注意力；超过 10 秒，用户会直接离开。而一个普通的 GPT-4o 请求，生成 500 字的回答需要多久？15 到 30 秒。

非流式 vs 流式，感知差距到底有多大？

同一个 Prompt，非流式需要等待 15 秒，流式首字 0.3 秒开始显示——用户感知时间缩短超过 80%。这不是锦上添花，这是生死线。

人类天生对"有反应"的系统更有耐心。哪怕速度一样慢，只要屏幕上有东西在动，用户就会等。这是认知心理学的基本规律，也是流式输出的核心价值所在。

---

二、流式输出的底层原理，3 分钟讲透

用一个生活比喻来理解：

技术层面，流式输出依赖的是 SSE（Server-Sent Events）协议，配合 HTTP 的 chunked transfer encoding（分块传输编码） 实现。

整个流程是这样的：

Client                          Server (LLM API)
|                                    |
|--- POST /chat/completions -------->|
|    { stream: true }               |
|                                    |  开始生成 token
|<-- data: {"choices":[{"delta":{"content":"流"}}]} --|
|<-- data: {"choices":[{"delta":{"content":"式"}}]} --|
|<-- data: {"choices":[{"delta":{"content":"输"}}]} --|
|<-- data: {"choices":[{"delta":{"content":"出"}}]} --|
|<-- data: [DONE] ------------------|
|                                    |

每一个 data: 块就是一个 SSE 事件，携带一小段 JSON，里面的 delta.content 就是当前这个 token 的文字内容。当服务器发送 data: [DONE] 时，代表生成结束。

你在浏览器 DevTools 的 Network 面板里，把请求类型筛选为 EventStream，就能亲眼看到这些数据包一条一条流进来——那个感觉很奇妙，像是在看 AI 的"思维流"。

---

三、实战代码：三种场景手把手实现

场景一：Python 后端（最简版）

import requests
import json

response = requests.post(
"https://api.884819.xyz/v1/chat/completions",  # 兼容 OpenAI 标准的 API 服务
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"model": "gpt-4o",
"messages": [{"role": "user", "content": "用100字介绍流式输出"}],
"stream": True  # 关键参数，不加这个就是非流式
},
stream=True  # requests 库也要设置，否则会等全部返回再处理
)

for line in response.iter_lines():
if line:
decoded = line.decode('utf-8')
if decoded.startswith("data: "):
data_str = decoded[6:]  # 去掉 "data: " 前缀
if data_str == "[DONE]":
break
chunk = json.loads(data_str)
delta = chunk["choices"][0]["delta"]
if "content" in delta:
print(delta["content"], end="", flush=True)

场景二：Node.js 后端（官方 SDK）

import OpenAI from "openai";

const client = new OpenAI({
apiKey: "YOUR_API_KEY",
baseURL: "https://api.884819.xyz/v1", // 使用兼容 OpenAI 标准的 API 服务，无需修改任何 SDK 代码
});

async function streamChat() {
const stream = await client.chat.completions.create({
model: "gpt-4o",
messages: [{ role: "user", content: "用100字介绍流式输出" }],
stream: true,
});

for await (const chunk of stream) {
const content = chunk.choices[0]?.delta?.content ?? "";
process.stdout.write(content); // 实时输出，不换行
}

console.log("\n[生成完毕]");
}

streamChat();

场景三：前端浏览器（fetch + ReadableStream）

这是最常见的场景，也是踩坑最多的地方：

async function streamToDOM(prompt, targetElement) {
const response = await fetch("https://api.884819.xyz/v1/chat/completions", {
method: "POST",
headers: {
"Content-Type": "application/json",
Authorization: "Bearer YOUR_API_KEY",
},
body: JSON.stringify({
model: "gpt-4o",
messages: [{ role: "user", content: prompt }],
stream: true,
}),
});

const reader = response.body.getReader();
const decoder = new TextDecoder("utf-8");
let buffer = "";

while (true) {
const { done, value } = await reader.read();
if (done) break;

buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\n");
buffer = lines.pop(); // 保留最后一个可能不完整的行

for (const line of lines) {
if (line.startsWith("data: ")) {
const data = line.slice(6);
if (data === "[DONE]") return;
try {
const chunk = JSON.parse(data);
const content = chunk.choices[0]?.delta?.content ?? "";
targetElement.textContent += content; // 实时更新 DOM
} catch (e) {
// JSON 解析失败，跳过（网络抖动可能导致数据截断）
}
}
}
}
}

🚨 小白最容易踩的 3 个坑

Python 的 requests 库默认会把响应全部下载完再返回，必须两处都设置。

网络传输不保证每个 data: 块完整到达，可能一次 read() 里包含半条数据。上面代码里的 buffer 逻辑就是为了解决这个问题——这是最常见的 bug 来源。

---

四、进阶技巧：让流式输出更丝滑的 5 个方法

① 打字机光标动画

只需几行 CSS，就能让输出过程更有"AI 思考中"的感觉：

.streaming-cursor::after {
content: "▋";
animation: blink 0.7s infinite;
}

@keyframes blink {
0%, 100% { opacity: 1; }
50% { opacity: 0; }
}

生成完成后，移除 streaming-cursor 类即可。

② 流式输出中途取消（AbortController）

用户点击"停止生成"按钮，怎么实现？

const controller = new AbortController();

// 停止按钮
stopBtn.onclick = () => controller.abort();

const response = await fetch(url, {
method: "POST",
signal: controller.signal, // 挂载 signal
body: JSON.stringify({ stream: true, ...params }),
});

③ Markdown 实时渲染

AI 输出的内容经常包含 Markdown 格式（代码块、加粗、列表），如果等全部生成完再渲染，体验大打折扣。

推荐方案：使用 marked.js + 防抖，每收到 N 个 token 渲染一次，而不是每个 token 都触发渲染（避免频繁 DOM 操作导致性能问题）。

let renderTimer = null;
let accumulatedText = "";

function onChunkReceived(content) {
accumulatedText += content;
clearTimeout(renderTimer);
renderTimer = setTimeout(() => {
targetElement.innerHTML = marked.parse(accumulatedText);
}, 50); // 50ms 防抖
}

④ 多轮对话的流式上下文管理

流式输出时，你需要在本地把 AI 的回答"攒"完整，再加入对话历史：

full_response = ""
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
full_response += delta.content
print(delta.content, end="", flush=True)

流式结束后，把完整回复加入历史
messages.append({"role": "assistant", "content": full_response})

⑤ 流式 + Function Calling 的处理

Function Calling 在流式模式下，tool_calls 的参数也是分块返回的，需要拼接：

tool_call_args = ""
for chunk in stream:
delta = chunk.choices[0].delta
if delta.tool_calls:
for tc in delta.tool_calls:
if tc.function.arguments:
tool_call_args += tc.function.arguments

流结束后，解析完整的函数参数
import json
args = json.loads(tool_call_args)

---

五、选对平台，流式体验才有保障

写到这里要说一个经常被忽略的问题：代码写对了，体验还是差——锅可能在 API 服务本身。

流式输出的体验，很大程度取决于两个指标：

• TTFT（Time to First Token）：首 token 延迟，决定用户"感知响应速度"
• 吞吐速度（tokens/s）：决定文字流出的顺畅程度

主流模型的流式输出速度参考（实测数据，网络环境影响较大）：

Q：我用的 API 流式输出断断续续，首字特别慢，怎么办？

>

A：流式体验很大程度取决于 API 服务的基础设施，不完全是代码问题。本文所有示例使用的 [api.884819.xyz](https://api.884819.xyz) 针对流式场景做了专项优化，首 token 延迟稳定在 300ms 以内，且完全兼容 OpenAI SDK——切换成本为零，只需改一行 base_url。

---

六、今日收获 Checklist + 立刻行动

读完这篇，你应该已经掌握：

• [x] 流式输出的核心价值：首字 0.3 秒 vs 等待 15 秒，体验天壤之别
• [x] SSE 协议的工作原理：token 逐个返回，[DONE] 信号结束
• [x] Python / Node.js / 前端浏览器三种场景的完整实现代码
• [x] 最容易踩的 3 个坑：stream 参数、buffer 处理、[DONE] 判断
• [x] 5 个进阶技巧：光标动画、中途取消、Markdown 渲染、上下文管理、Function Calling

1. 访问 [api.884819.xyz](https://api.884819.xyz) 获取 API Key

2. 复制本文任意代码示例，填入 Key

3. 运行，看着文字一个一个蹦出来的那一刻——你会上瘾的 😄

---

### 📌 下一篇预告

>

流式输出解决了"看起来快"的问题——但如果你想让 AI 真的快呢？

>

下一篇，我们聊一个更硬核的话题：《AI 应用性能优化全攻略：从 Prompt 压缩到并发调度，把响应速度压到极限》。

>

我们会揭秘：为什么同样的模型，有人的应用快 3 倍？Prompt 长度和速度到底是什么关系？多轮对话越聊越慢怎么破？并发 100 个请求不被限流的正确姿势是什么？

>

关注/收藏，下周见。

---