从「录音」到「流式语音回答」：一次吃透 ASR + 流式 TTS 全链路

语音助手听起来很高端，但拆开看，就是「听 → 想 → 说」三段：ASR 把你的话变成文字，LLM 把文字变成回答，TTS 再把回答读出来。 难点不在任何一段本身，而在每一段选什么传输协议——有的用普通 HTTP，有的用 SSE，有的必须用 WebSocket。本文用一个真实项目拆透这三段的选型逻辑和落地实现。

为什么要学这个

豆包、Siri、小爱同学这些语音助手，核心都跑在下面这条链路上：

麦克风 → 识别成文字 → 大模型思考 → 流式吐文字 → 合成语音 → 流式播放

作为 Agent 开发者，你迟早会遇到「让 AI 开口说话」的需求。而这条链路里藏着三个协议选型问题，是普通文本聊天不会遇到的：

1. 录音怎么传？ 录完一次性上传就行，还是边录边传？
2. AI 回答怎么推？ 文字要「打字机式」地蹦出来，SSE 就够了；但文字变成语音后是二进制，SSE 传不了。
3. 语音怎么播？ 不能等整段音频生成完再播，要「边合成边播放」，这又需要一个播放侧机制。

本文按「从简单到复杂」的顺序，把这三段逐一拆开。你会看到：协议不是越多越好，而是每一段数据形态决定一段协议。

---

一、语音识别 ASR：最朴素的一次请求

一句话： ASR（自动语音识别）把音频变成文字。这里用非流式方案——用户说完一句话，录音整体上传，识别一次，返回结果。对「按住说话、说完出字」的场景，这是最简单也够用的选择。

前端：MediaRecorder 录音

用浏览器原生 MediaRecorder 把麦克风声音攒成 Blob：

const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
const mediaRecorder = new MediaRecorder(stream, {
  mimeType: 'audio/ogg;codecs=opus',
});
const chunks = [];
mediaRecorder.ondataavailable = (e) => chunks.push(e.data);

mediaRecorder.onstop = async () => {
  const blob = new Blob(chunks, { type: mediaRecorder.mimeType });
  // 关键：Blob 塞进 FormData 的 audio 字段，一次性 POST 给后端
  const formData = new FormData();
  formData.append('audio', blob, 'record.ogg');
  const res = await fetch('/speech/asr', { method: 'POST', body: formData });
  const { text } = await res.json(); // ← 识别出的文字
};
mediaRecorder.start(250); // 每 250ms 收集一次数据，避免大块内存

后端：Nest 接文件，调腾讯 ASR SDK

Nest 用 FileInterceptor 接收上传的文件，再调腾讯云 SDK：

@Post('asr')
@UseInterceptors(FileInterceptor('audio'))
async recognize(@UploadedFile() file) {
  const text = await this.speechService.recognizeBySentence(file);
  return { text };
}

async recognizeBySentence(file) {
  const result = await this.asrClient.SentenceRecognition({
    EngSerViceType: '16k_zh',                 // 16k 中文普通话
    SourceType: 1,                             // 1 = 语音数据以 base64 传入
    Data: file.buffer.toString('base64'),
    DataLen: file.buffer.length,
    VoiceFormat: 'ogg-opus',                   // 和前端录音 mimeType 对齐
  });
  return result.Result ?? '';
}

要点：

• 为什么非流式就够：一句话识别，等用户说完再整体识别，等待成本可接受；流式识别（边说边出字）是为长语音、低延迟设计的，复杂度高不少。
• Data 传的是 base64，所以内存里要留一份完整音频——录制时间过长时这是个隐患，但对一句话绰绰有余。
• 前端录音的 mimeType（ogg-opus）必须和后端 VoiceFormat 保持一致，否则腾讯识别会失败。

---

二、流式文本：SSE 让回答「打字机式」蹦出来

一句话： SSE（Server-Sent Events）是一条只读的 HTTP 长连接，服务端能持续往同一条连接推文本。AI 回答是一句句生成的，用 SSE 推给前端，就能边生成边显示。

为什么不能等全文

大模型生成几百字要好几秒。如果等全文生成完一次性返回，用户看到的是「转圈 8 秒 + 瞬间全出」；用 SSE 就是「第 1 秒开始出字，一路蹦完」。体验差异，就是流式存在的意义。

后端：Nest 的 @Sse 装饰器

@Sse 是 Nest 对 SSE 的一等支持，只需要返回一个 Observable，框架负责把它序列化成 SSE 帧：

@Controller('ai')
export class AiController {
  @Sse('chat/stream')
  chatStream(@Query('query') query: string): Observable<{ data: string }> {
    return from(this.aiService.streamChain(query)).pipe(
      map((chunk) => ({ data: chunk })), // 每一帧都包成 { data }
    );
  }
}

LangChain 侧，模型是原生支持流式输出的（Runnable 链调 .stream() 就是个异步生成器）：

async *streamChain(query: string): AsyncGenerator<string> {
  const stream = await this.chain.stream({ query });
  for await (const chunk of stream) yield chunk;
}

前端：EventSource 三个回调就能接

const es = new EventSource('/ai/chat/stream?query=' + encodeURIComponent(query));
es.onmessage = (event) => {
  answer += event.data;   // 每帧拼到已有文本后面，实时更新 DOM
  render(answer);
};
es.onerror = () => es.close(); // 连接结束时触发，即流结束

要点：

• SSE 是纯文本协议（text/event-stream），每一帧以 data: xxx\n\n 形式传输——它只适合传 UTF-8 文本。
• SSE 是单向的（服务端→客户端），且只能走 GET；客户端没法往连接里发数据。
• 这里埋下了下一个问题的种子：如果 AI 输出的不是文字而是语音二进制，SSE 就无能为力了。

---

三、语音合成 TTS：为什么必须「流式」

一句话： TTS（语音合成）把文字变成音频。问题是：文字是流式到达的，你不可能等整段文字生成完再合成语音——那样第一个字出声要等好几秒。

非流式 vs 流式 TTS

如果用非流式，延迟 = LLM 吐完整段文字的时间 + TTS 合成整段音频的时间，语音对话体验直接崩塌。所以只要是想「边回答边说话」的对话式场景，就必须流式 TTS。

腾讯流式 TTS 的协议形态

腾讯的流式合成接口 TextToStreamAudioWSv2，本身就是一条 WebSocket 连接（wss://tts.cloud.tencent.com/stream_wsv2），而不是 HTTP：

• 你通过这条 WS 把文本片段发过去（ACTION_SYNTHESIS），文本发完再发一个结束信号（ACTION_COMPLETE）。
• 腾讯把合成好的音频帧从同一条 WS 推回来——而且推回来的是二进制帧，不是文本。

先用 ws 库写的最小验证脚本把协议跑通（真实项目里独立验证过，逻辑后面被原样搬进了服务）：

ws.on('message', (data, isBinary) => {
  if (isBinary) {
    // 音频数据！直接写进文件 / 转给前端
    writeStream.write(data);
    return;
  }
  const msg = JSON.parse(data.toString());
  if (msg.ready === 1) {
    // 服务端就绪后，开始逐段发文本
    ws.send(JSON.stringify({
      session_id: sid, action: 'ACTION_SYNTHESIS', data: '第一句文本',
    }));
  }
  if (msg.final === 1) ws.close(); // 合成结束
});

要点：

• 关键洞察：TTS 和 ASR 一样，也有「流式/非流式」之分，语音对话必须选流式。
• 腾讯流式 TTS 的传输通道是 WebSocket + 二进制帧——这决定了我们服务端也得用 WS 去对接，而不是普通 HTTP 调用。
• 每一次 ACTION_SYNTHESIS 发一段文本，ACTION_COMPLETE 标记文本发完；收到的二进制帧就是可直接播放的音频（如 mp3 帧）。
• 连接 URL 需要按腾讯协议做 HMAC-SHA1 签名（把 Action/AppId/SecretId/时间戳等参数排序拼接再签名）。这个独立脚本只验证了协议；真正要放进服务，还得解决「怎么把 LLM 的流式文本喂给这条 WS」——就是下一节的核心。

---

四、用 WebSocket 打通「文本 → 语音」的流式管道

一句话： 后端加一个 TTS 中继服务，把「SSE 流出的文本」转手推给腾讯 TTS，再把腾讯返回的二进制音频帧通过 WS 转发给前端。整条链路里，WS 是唯一能承载二进制的双向通道。

为什么这块非 WebSocket 不可

SSE 把「AI 回答的文本」推给前端没问题；但「AI 回答的语音」是二进制，SSE 传不了。所以在 SSE 之外并行拉一条 WS 通道，专门运语音。

4.1 事件桥：把「流式文本」广播给 TTS 中继

关键设计是解耦：SSE 那条链（AiController / AiService）不关心 TTS 的存在，它只负责在产出一个文本 chunk 时，往事件总线上发一个事件。TTS 中继服务（TtsRelayService）用 @OnEvent 订阅这个事件。两边通过 @nestjs/event-emitter 联系，互不 import。

事件类型先定义好：

// src/common/stream-events.ts
export const AI_TTS_STREAM_EVENT = 'ai.tts.stream';

export type AiTtsStreamEvent =
  | { type: 'start'; sessionId: string; query: string }
  | { type: 'chunk'; sessionId: string; chunk: string }
  | { type: 'end'; sessionId: string }
  | { type: 'error'; sessionId: string; error: string };

AiService 在流式循环里边 yield 给 SSE、边 emit 给事件总线，同一份文本一鱼两吃：

async *streamChain(query: string, ttsSessionId?: string): AsyncGenerator<string> {
  const stream = await this.chain.stream({ query });
  for await (const chunk of stream) {
    if (ttsSessionId) {
      this.eventEmitter.emit(AI_TTS_STREAM_EVENT, { type: 'chunk', sessionId: ttsSessionId, chunk });
    }
    yield chunk; // 这一路进 SSE 推给前端显示文字
  }
  if (ttsSessionId) {
    this.eventEmitter.emit(AI_TTS_STREAM_EVENT, { type: 'end', sessionId: ttsSessionId });
  }
}

4.2 两条通道靠 sessionId 关联

这里有个很反直觉的点：文字流和语音流是并行的两条通道。前端怎么让后端知道「这条 SSE 的文本，要合成语音发给哪条 WS」？答案是——前端先开 WS，拿 sessionId，再带着它去开 SSE：

前端:  new WebSocket('/speech/tts/ws')   →  后端回 { type:'session', sessionId }
前端:  EventSource('/ai/chat/stream?query=...&ttsSessionId=<sessionId>')
后端:  AiService emit 的事件带上 sessionId → TtsRelayService 按 sessionId 找到对应 WS

后端侧事件里始终带着 sessionId，TtsRelayService 就靠它把「文本事件」映射到「具体某条腾讯 TTS 连接 + 某条前端连接」。

4.3 TtsRelayService：一进一出的双向管道

中继服务维护一个 Map<sessionId, Session>，每个 Session 里握着两条 WS：一条连前端（clientWs），一条连腾讯（tencentWs）。核心逻辑就是订阅事件后按类型分发：

@OnEvent(AI_TTS_STREAM_EVENT)
handleAiStreamEvent(event: AiTtsStreamEvent): void {
  const session = this.sessions.get(event.sessionId);
  if (!session) return;
  switch (event.type) {
    case 'start': {
      this.ensureTencentConnection(session);          // 懒建连：第一次才连腾讯
      this.sendClientJson(session.clientWs, { type: 'tts_started' });
      break;
    }
    case 'chunk': {
      const chunk = event.chunk.trim();
      if (!session.ready || !session.tencentWs?.readyState === WebSocket.OPEN) {
        session.pendingChunks.push(chunk);            // 腾讯还没就绪：先攒着
        return;
      }
      this.sendTencentChunk(session, chunk);          // 就绪：转发给腾讯
      break;
    }
    case 'end': {
      this.flushPendingChunks(session);
      session.tencentWs.send(JSON.stringify({ session_id, action: 'ACTION_COMPLETE' }));
      break;
    }
  }
}

而腾讯那条 WS 的回调，把音频二进制帧原样转发给前端，把 ready / final / code 这类控制消息翻译成前端能懂的 JSON：

tencentWs.on('message', (data, isBinary) => {
  if (isBinary) {
    // 音频帧：原样转给前端（binary: true）
    if (session.clientWs.readyState === WebSocket.OPEN) {
      session.clientWs.send(data, { binary: true });
    }
    return;
  }
  const msg = JSON.parse(data.toString());
  if (Number(msg.ready) === 1) {        // 腾讯就绪 → 把攒的文本都发出去
    session.ready = true;
    this.flushPendingChunks(session);
  }
  if (Number(msg.final) === 1) {
    this.sendClientJson(session.clientWs, { type: 'tts_final' });  // 通知前端收尾
  }
});

4.4 挂载：main.ts 里的裸 WebSocketServer

有趣的是，这里没用 Nest 的 @WebSocketGateway 装饰器，而是在 main.ts 里直接挂一个原生 ws 的 WebSocketServer，按路径 /speech/tts/ws 路由，接到连接就交给中继服务：

const ttsWss = new WebSocketServer({ server, path: '/speech/tts/ws' });
ttsWss.on('connection', (socket, request) => {
  const sessionId = ttsRelayService.registerClient(socket);
  socket.on('close', () => ttsRelayService.unregisterClient(sessionId));
});

要点：

• 事件总线（EventEmitter）是解耦的关键：文本流（SSE）与语音中继（TTS）互不感知，只通过 ai.tts.stream 事件通信。
• sessionId 是双通道的关联凭证：先 WS 后 SSE，前端把 sessionId 一路透传，后端按它归并两条连接。
• pendingChunks 缓冲解决时序问题：SSE 的文本可能先于腾讯 WS 就绪到达，先攒进队列，收到 ready 再一次性 flush。
• 二进制帧 {binary:true} 原样转发：中间服务不 decode 音频，拿到就丢给前端，开销最小。
• 用原生 WebSocketServer 而非 @WebSocketGateway，同样能跑——Nest 的装饰器方案只是另一种等价写法。

---

五、前端流式播放：Audio + MediaSource + SourceBuffer

一句话： 浏览器播放音频通常给一个完整 URL 就行；但流式场景下音频是一帧帧来的，所以要把 MediaSource 当作「可不断往里灌数据的播放源」，把 WS 收到的帧持续 appendBuffer 进去，实现边收边播。

5.1 接 WS，把音频帧收进待播队列

前端这条 WS 的消息分两种：字符串是控制消息（session / tts_started / tts_final / tts_error），ArrayBuffer 是音频帧。收到音频帧不直接 append，而是先塞进队列再统一 flush：

const ws = new WebSocket('/speech/tts/ws');
ws.binaryType = 'arraybuffer';              // 关键：按二进制数组收

ws.onmessage = (event) => {
  if (typeof event.data === 'string') {
    const msg = JSON.parse(event.data);
    if (msg.type === 'session') sessionId = msg.sessionId;   // 拿关联凭证
    if (msg.type === 'tts_started') prepareStreamingAudio(); // 后端开始合成了，初始化播放器
    if (msg.type === 'tts_final' || msg.type === 'tts_error') {
      ttsStreamFinal = true;                // 流结束了，可以收尾
      flushTtsBufferQueue();
    }
  } else if (event.data instanceof ArrayBuffer) {
    ttsPendingBuffers.push(event.data);     // 先进队列
    flushTtsBufferQueue();                  // 再灌一块
  }
};

5.2 MediaSource + SourceBuffer：边灌边播

prepareStreamingAudio() 把 Audio 的 src 指向一个 MediaSource 对象，然后在 sourceopen 里创建 SourceBuffer；flushTtsBufferQueue() 是这个机制的心脏：

function prepareStreamingAudio() {
  ttsMediaSource = new MediaSource();
  ttsObjectUrl = URL.createObjectURL(ttsMediaSource);
  ttsAudioEl.src = ttsObjectUrl;                        // src 指向「可编程播放源」

  ttsMediaSource.addEventListener('sourceopen', () => {
    ttsSourceBuffer = ttsMediaSource.addSourceBuffer('audio/mpeg');
    ttsSourceBuffer.mode = 'sequence';                  // 按到达顺序拼接
    ttsSourceBuffer.addEventListener('updateend', flushTtsBufferQueue); // 播完一块再灌下一块
    flushTtsBufferQueue();
  });
}

function flushTtsBufferQueue() {
  if (!ttsSourceBuffer || ttsSourceBuffer.updating) return; // 关键：一次只能 append 一块
  if (ttsPendingBuffers.length > 0) {
    const next = ttsPendingBuffers.shift();
    ttsSourceBuffer.appendBuffer(next);                 // 灌一块，播一块
    ttsAudioEl.play().catch(() => {});                  // 自动播放策略兜底
    return;
  }
  if (ttsStreamFinal && ttsMediaSource.readyState === 'open') {
    ttsMediaSource.endOfStream();                       // 数据灌完，通知播放器收尾
  }
}

要点：

• 三层关系：Audio 标签负责出声，MediaSource 是它背后的「可编程播放源」，SourceBuffer 是往这个源里灌数据的水龙头。
• updating 标志决定了队列存在：SourceBuffer 同一时间只能 appendBuffer 一块，必须等上一块的 updateend 事件后才能灌下一块——所以音频帧先进 ttsPendingBuffers 队列，updateend 再触发 flush，保证串行。
• mode = 'sequence'：按到达顺序把音频帧拼起来，不按时间戳重排（流式场景帧本身有序）。
• endOfStream() 是收尾信号：必须等后端明确发来 tts_final / tts_error 才知道「数据不会再来了」，否则会一直等 updateend。
• play().catch(() => {}) 兜底浏览器的自动播放策略（用户未交互前不允许出声）。

---

六、完整时序：一图看懂流式语音的协作过程

把第四、五节拼起来，整条「文字 → 语音 → 播放」的协作过程可以用一张时序图看清：四个参与者（前端、Nest 的 SSE/AiService、TtsRelayService 中继、腾讯流式 TTS），四个阶段（握手、懒建连、逐 chunk 中继、收尾）。

阶段A · 双通道握手（步 1-3）

先 WS、后 SSE：前端先开 WS 拿到 sessionId，再带着它去连 SSE。这样后端才能把「这条 SSE 的文本」对到「那条 WS 的连接」——sessionId 是双通道的唯一关联凭证。

阶段B · 懒建连腾讯（步 4-7）

start 事件触发时才连腾讯（ensureTencentConnection），不是启动就建一堆连接。腾讯回 ready 后，前端收到 tts_started 才开始初始化播放器——播放器就绪发生在音频帧到来之前，避免第一帧到了还没东西接。

阶段C · 逐 chunk 中继（步 9-15，核心）

真正的「流式」就在这一步。注意 AiService 是一鱼两吃：同一个 chunk，一边 yield 进 SSE 给前端显示文字，一边 emit 给中继去合成语音。腾讯每回一帧二进制音频，中继不 decode、原样 {binary:true} 转发，前端 appendBuffer 就播。

阶段D · 收尾（步 16-20）

LLM 流结束 → 中继发 ACTION_COMPLETE → 腾讯回 final → 前端收到 tts_final 才确认「数据不会再来了」，于是 endOfStream() 通知播放器收尾。

图上看不到的竞态保护

阶段C 有个隐患：LLM 的文本 chunk 可能先于腾讯 WS 就绪到达。pendingChunks 队列就是兜底——!ready 时先 push 进队列，腾讯 ready 后一次性 flush 补发，保证文本发送顺序和 LLM 生成顺序一致。

---

总结：一张图 + 一张表，看懂整条链路

全链路架构图

图例：🟩 = 已实现（全链路），🟦 = 腾讯云外部服务。

怎么选协议：一张决策表

收尾一句话： 整条链路没有魔法，只有一条原则——数据长什么样，就选什么协议。文本走 SSE，二进制走 WS，一次性数据走 HTTP；而三段之间要协调，靠的是一条事件总线和一把 sessionId 的钥匙。把这套选型记牢，语音助手对你来说就只是三段普通管道拼接而已。现在，你的 Agent 已经能真正开口说话了。