上个月帮朋友收拾一个 uni-app 做的 AI 助手。接口早就接通了，功能测试全绿，可用户一上手就皱眉——点完发送，界面空白干等七八秒，然后整段回答”啪”地一下全砸在屏幕上。
问题既不在模型，也不在网络。出在请求这一层。项目里用的是 uni.request，而它的设计模型是”拿到完整响应体后触发 success”，压根没给增量回调留口子。服务端明明是一个 token 一个 token 往外推的，到了客户端被整个吞下去，等流结束才一起吐出来。
下面把这套方案从前协议层写到页面层。微信小程序是重点，H5 顺手一起做掉，App 端我会在最后说清楚为什么暂时不建议硬上。
一、先搞清楚 SSE 到底长什么样
很多大模型厂商（OpenAI 系接口、DeepSeek、通义、豆包等）的 stream: true 走的都是 SSE，也就是 Server-Sent Events。它的格式比想象中朴素得多：
data: {"choices":[{"delta":{"content":"你"}}]}
data: {"choices":[{"delta":{"content":"好"}}]}
data: {"choices":[{"delta":{"content":"，"}}]}
data: [DONE]
规则就三条：每个字段写成 field: value；冒号后面的那一个空格可有可无，剥掉；一串字段之后遇到一个空行，就表示”一个完整事件到此结束”。一个事件里可以有多行 data:，按换行拼起来才是最终 payload。
协议本身没什么难度。真正让人翻车的是分块边界。
TCP 层面的 chunk 是按字节切的，它完全不关心你在哪儿换行，更不关心一个汉字占几个字节。你打开 onChunkReceived 收到的 ArrayBuffer，长这样是家常便饭：
第 1 块： data: {"choices":[{"delta":{"content":"<半个汉
第 2 块： 字的后两个字节"}}]} nn
如果每收到一块就直接 JSON.parse，第一块必崩，第二块也是废的。所以我们需要两层缓冲：字节层负责把半个字符补齐，行层负责把半个事件补齐。少一层都不行。
二、字节层：一个能处理跨块字符的 UTF-8 解码器
小程序环境没有 TextDecoder（H5 有，后面会分端处理），所以这里得自己写。核心思路很简单：把上一次没解完的尾巴存下来，跟新到的字节拼在一起再解，解到最后一个完整字符为止，剩下的继续留到下一轮。
// utils/utf8-stream.js
/**
* 分块到达的 ArrayBuffer 不保证落在字符边界上，
* 一个汉字（3 字节）很可能被切成 1+2 两次送达。
* 这里把不完整的尾巴留下来，等下一块拼接。
*/
export class Utf8ChunkDecoder {
constructor() {
this.tail = new Uint8Array(0)
}
decode(chunk) {
const incoming = new Uint8Array(chunk)
const buf = new Uint8Array(this.tail.length + incoming.length)
buf.set(this.tail, 0)
buf.set(incoming, this.tail.length)
let out = ''
let i = 0
while (i < buf.length) {
const b0 = buf[i]
if (b0 < 0x80) {
// 单字节，ASCII
out += String.fromCharCode(b0)
i += 1
} else if (b0 < 0xe0) {
// 两字节，后面还需要 1 个字节
if (i + 1 >= buf.length) break
out += String.fromCharCode(((b0 & 0x1f) << 6) | (buf[i + 1] & 0x3f))
i += 2
} else if (b0 < 0xf0) {
// 三字节，汉字大多在这里
if (i + 2 >= buf.length) break
out += String.fromCharCode(
((b0 & 0x0f) << 12) |
((buf[i + 1] & 0x3f) << 6) |
(buf[i + 2] & 0x3f)
)
i += 3
} else {
// 四字节，emoji 之类，输出成代理对
if (i + 3 >= buf.length) break
let cp =
((b0 & 0x07) << 18) |
((buf[i + 1] & 0x3f) << 12) |
((buf[i + 2] & 0x3f) << 6) |
(buf[i + 3] & 0x3f)
cp -= 0x10000
out += String.fromCharCode(0xd800 + (cp >> 10), 0xdc00 + (cp & 0x3ff))
i += 4
}
}
this.tail = buf.slice(i)
return out
}
reset() {
this.tail = new Uint8Array(0)
}
}
注意 break 那几处。只要发现当前字节不足以构成一个完整字符，立刻跳出循环，把 i 之后的内容整个存进 tail。buf.slice(i) 返回的是新数组，不会因为 buf 被回收而出问题。
三、行层：SSE 事件解析器
字节层解决了字符完整性问题，接下来要把文本流切成一个个事件。同样的道理——一个 chunk 可能只包含半行，也可能一口气包含三四个完整事件，甚至末尾跟着半行。所以解析器只做一件事：按 n 逐行消费，遇到空行就吐出一个事件。
// utils/sse-parser.js
/**
* 逐段喂进去即可，内部自己处理跨块的半行。
* 一次 push 可能吐出 0 个、1 个或多个完整事件。
*/
export class SseParser {
constructor(onEvent) {
this.onEvent = onEvent
this.raw = ''
this.dataLines = []
}
push(text) {
this.raw += text
let idx
while ((idx = this.raw.indexOf('n')) !== -1) {
let line = this.raw.slice(0, idx)
this.raw = this.raw.slice(idx + 1)
// 兼容 CRLF
if (line.endsWith('r')) line = line.slice(0, -1)
if (line === '') {
// 空行 = 一个事件结束
if (this.dataLines.length) {
const payload = this.dataLines.join('n')
this.dataLines = []
this.onEvent(payload)
}
continue
}
if (line.startsWith('data:')) {
// 冒号后那个空格按规范要剥掉，可有可无
this.dataLines.push(line.slice(5).replace(/^ /, ''))
}
// event: / id: / retry: 以及 ": ping" 心跳注释本场景用不到，忽略
}
}
reset() {
this.raw = ''
this.dataLines = []
}
}
整个文件没有任何正则回溯，全是 indexOf 加 slice。原因很实在：流式场景下这个函数每秒会被调用几十次，正则写爽了，低端安卓机就卡给你看。
四、请求层：按端拆分，别想着用一套写法包打天下
这是最关键的一步。为什么会卡住很多人，是因为大家习惯性地想用一个 uni.request 把三端都覆盖掉，而增量回调这件事上，三端根本没有统一的 API。
- 微信小程序：
wx.request传enableChunked: true，然后监听返回的requestTask.onChunkReceived。 - H5：用原生
fetch，读response.body.getReader()。 - App（nvue/vue 原生渲染）：没有现成的增量回调，需要另辟蹊径，后面单独说。
所以这里用条件编译拆成两条路径，对外暴露同一个函数签名：
// utils/stream-chat.js
import { Utf8ChunkDecoder } from './utf8-stream'
import { SseParser } from './sse-parser'
/**
* @param {Object} opts
* @param {string} opts.url
* @param {Object} opts.body
* @param {Object} opts.headers
* @param {Function} opts.onDelta 收到一段增量文本
* @param {Function} opts.onDone
* @param {Function} opts.onError
* @returns {{ abort: Function }}
*/
export function streamChat(opts) {
const { url, body, headers = {}, onDelta, onDone, onError } = opts
const extract = (payload) => {
if (payload === '[DONE]') return
try {
const json = JSON.parse(payload)
const delta = json.choices && json.choices[0] && json.choices[0].delta
if (delta && delta.content) onDelta(delta.content)
} catch (e) {
// 有些网关会插空 data 或心跳，JSON 解析失败忽略即可
}
}
const parser = new SseParser(extract)
// #ifdef MP-WEIXIN
const decoder = new Utf8ChunkDecoder()
const task = wx.request({
url,
method: 'POST',
header: {
'Content-Type': 'application/json',
Accept: 'text/event-stream',
...headers
},
data: body,
enableChunked: true,
responseType: 'arraybuffer',
success: () => { onDone && onDone() },
fail: (err) => { onError && onError(err) }
})
task.onChunkReceived((res) => {
parser.push(decoder.decode(res.data))
})
return { abort: () => task.abort() }
// #endif
// #ifdef H5
const controller = new AbortController()
fetch(url, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Accept: 'text/event-stream',
...headers
},
body: JSON.stringify(body),
signal: controller.signal
})
.then(async (resp) => {
if (!resp.ok) throw new Error('HTTP ' + resp.status)
const reader = resp.body.getReader()
const td = new TextDecoder('utf-8')
for (;;) {
const { done, value } = await reader.read()
if (done) break
// stream: true 让解码器自己留住半个字符，H5 不用手写上面的解码器
parser.push(td.decode(value, { stream: true }))
}
onDone && onDone()
})
.catch((err) => {
if (err && err.name === 'AbortError') return
onError && onError(err)
})
return { abort: () => controller.abort() }
// #endif
}
两个细节值得说一下。
第一，Accept: text/event-stream 这个头别省。部分云厂商的网关会根据它来决定是否启用流式转发，漏了就被降级成一次性响应。
第二，onChunkReceived 里绝对不要做任何耗时操作，更不要直接往 setData 里塞。它跑在逻辑层，是串行的，回调一堵，后面的 chunk 全排队等着，体感直接从”逐字”退化成”一段一段跳”。
五、渲染层：把 40ms 内的字符攒起来再刷
假设模型吐字速度是每秒 40 个 token，如果你每来一个字就改一次响应式数据，等于每秒触发 40 次视图更新。在中低端安卓机上，这个频率足够让滚动条一顿一顿的。
做法是加一个时间窗合并：增量先往一个非响应式的普通变量里堆，每 40 毫秒才把它同步到页面上一次。40ms 差不多是两帧多一点，人眼已经分辨不出差别了，但更新次数直接砍掉八成以上。
<template>
<view>
<scroll-view scroll-y :scroll-into-view="anchor">
<view v-for="(m, i) in messages" :key="i" :id="'msg-' + i">
<text>{{ m.role === 'user' ? '我：' : 'AI：' }}</text>
<text>{{ m.content }}</text>
</view>
<view id="anchor"></view>
</scroll-view>
<input v-model="draft" placeholder="说点什么" />
<button v-if="!loading" @click="send">发送</button>
<button v-else @click="stop">停止</button>
</view>
</template>
<script setup>
import { ref, onUnmounted } from 'vue'
import { streamChat } from '@/utils/stream-chat'
const messages = ref([])
const draft = ref('')
const loading = ref(false)
const anchor = ref('anchor')
let controller = null
let raw = '' // 累积的完整文本，故意不放进 ref
let flushTimer = null
function flush() {
flushTimer = null
const last = messages.value[messages.value.length - 1]
if (last) last.content = raw
}
function onDelta(delta) {
raw += delta
if (!flushTimer) flushTimer = setTimeout(flush, 40)
}
function send() {
const text = draft.value.trim()
if (!text || loading.value) return
draft.value = ''
messages.value.push({ role: 'user', content: text })
messages.value.push({ role: 'assistant', content: '' })
raw = ''
loading.value = true
controller = streamChat({
url: 'https://your-api.example.com/v1/chat/completions',
body: {
model: 'your-model',
stream: true,
messages: messages.value
.filter((m) => m.content)
.map((m) => ({ role: m.role, content: m.content }))
},
onDelta,
onDone() {
flush()
loading.value = false
controller = null
},
onError(err) {
flush()
loading.value = false
controller = null
const last = messages.value[messages.value.length - 1]
if (last && !last.content) {
last.content = '请求失败：' + (err.errMsg || err.message || '未知错误')
}
}
})
}
function stop() {
if (!controller) return
controller.abort()
flush()
loading.value = false
controller = null
}
onUnmounted(() => {
if (controller) controller.abort()
if (flushTimer) clearTimeout(flushTimer)
})
</script>
raw 这个变量放在 ref 外面是有意的。它是纯数据缓冲区，参与响应式只会平白增加依赖收集和 diff 的开销。真正需要在视图里体现的，只有 flush 那一刻的最终值。
另外 onUnmounted 里那两行别偷懒。用户聊到一半退出页面，流还在后台推进，定时器还在往已经销毁的组件上写数据，轻则报一堆警告，重则在小程序里留下悬挂的请求句柄。
六、几个不看文档就会踩的坑
开发者工具里看不出流式效果。这是最高频的误判。微信开发者工具的模拟器在网络层做了聚合，你经常能看到回复一次性全出来，然后怀疑代码写错了。切成”真机调试”，问题当场消失。以后凡是和分块传输相关的调试，一律上真机。
Nginx 缓冲没关。服务端接口明明写好的是流式，客户端却收不到增量，八成是中间这一层把响应体缓存住了。需要在对应的 location 里关掉：
location /v1/ {
proxy_pass http://backend;
proxy_buffering off;
proxy_cache off;
proxy_set_header Connection '';
proxy_http_version 1.1;
chunked_transfer_encoding on;
}
其中 proxy_http_version 1.1 和清空 Connection 头是配套的，缺了任何一个，Nginx 会退回 HTTP/1.0 去和后端通信，流式直接失效。
基础库版本。enableChunked 需要微信基础库 2.20.2 及以上。低版本不会报错，它会安安静静地按普通请求处理，你会收到一个完整的响应体，然后界面上什么都没有。可以在启动时做一次版本判断，不满足就干脆降级成非流式，至少功能是完整的。
别假设一个 chunk 对应一个 token。网络好的时候，iOS 微信经常把两三个 SSE 事件合并成一个 chunk 发过来。如果你的解析逻辑写成”收到一块就 parse 一次”，在真机上会出现随机丢字。前面那个基于缓冲区的解析器已经规避了这个问题，但自己改代码时要注意别退化回去。
推理模型的思考链字段。如果用的是带推理过程的模型，增量可能出现在 delta.reasoning_content 或者 delta.reasoning 里，和正文的 delta.content 是分开的。做折叠展示时记得两个都要接。
七、App 端怎么办
坦白讲，App 端目前没有一条让人舒服的路。
可选方案大致三种：一是用 renderjs，在视图层里跑原生 XMLHttpRequest，靠 onprogress 读 responseText 的增量——能跑通，但视图层和逻辑层之间的通信本身就有损耗，token 密集时反而不如节流后的一次性渲染流畅；二是写一个 UTS 插件，直接调用原生网络库；三是干脆不做流式，用 loading 骨架屏扛过等待时间。
我的建议是先评估收益。如果 App 端只是顺带，优先选第三条，把精力放在小程序和 H5 上；如果 App 是主战场，再考虑第二条，把字节解码的部分下沉到原生层，跨端通信只传已经切好的字符串。
整套方案最麻烦的地方，其实不是 SSE 协议，而是那两层缓冲。把字节边界和行边界都处理干净之后，剩下的就是普通的业务代码了。
