上一篇我用 FastAPI + WebRTC + RAG 跑通了一个简化版 Voice Agent Demo：浏览器和后端能建连，用户问题可以通过 DataChannel 发到后端，后端再用一个本地知识库返回回答。

但那个 demo 还缺最关键的一步：浏览器麦克风里的声音，怎么实时变成文字？

如果要用 FastAPI + WebRTC 接入流式 ASR，可以把链路拆成四步：

1. 浏览器用 getUserMedia() 获取麦克风音频；
2. WebRTC 把音频轨道传给后端；
3. 后端用 aiortc 接收音频帧，并转换成 ASR 可用的 PCM 数据；
4. 流式 ASR 返回识别文本，再通过 DataChannel 或 WebSocket 推回前端。

也就是说，Voice Agent 从“文本问答 Demo”走向“语音助手 Demo”，中间最关键的一层就是流式 ASR。 没有 ASR，WebRTC 只是在传音频；接入 ASR 后，系统才真正能理解用户说了什么。

一、整体链路：从麦克风到实时转写

一个最小的实时语音识别链路可以这样理解：

浏览器麦克风
  -> WebRTC audio track
  -> FastAPI / aiortc 接收音频帧
  -> 转成 PCM / wav chunk
  -> 发送给流式 ASR
  -> 得到 transcript
  -> DataChannel / WebSocket 返回前端
  -> 后续接 RAG / LLM / TTS

这里有几个点很容易混在一起：

• FastAPI 不是直接处理音频流，它主要负责 HTTP 接口、信令、页面托管和业务接口；
• WebRTC 负责低延迟传输浏览器音频；
• aiortc 负责在 Python 后端接住 WebRTC audio track；
• ASR 负责把音频转成文字；
• DataChannel 或 WebSocket 负责把识别结果推回浏览器。

如果是 AI 语音客服、网页语音助手、实时语音对话系统，这条链路基本都绕不开。差别只在于：ASR 用本地模型还是云服务，后面接不接 RAG，最后是否用 TTS 把回答读出来。

二、为什么 Voice Agent 要优先接流式 ASR

做 Voice Agent 时，我不建议一开始就把 ASR、RAG、LLM、TTS、转人工全部堆到一个 demo 里。更稳的顺序是：

1. 先跑通 WebRTC 建连；
2. 再接流式 ASR，让语音能变成文字；
3. 然后接 RAG，把回答拉回知识库；
4. 再接 LLM 生成回答；
5. 最后接 TTS、打断、转人工和日志。

原因很简单：语音链路出问题时，排查成本很高。

如果用户说话后系统没反应，可能是麦克风没采到音，也可能是 WebRTC 没建连，可能是音频格式不对，也可能是 ASR 没返回，还可能是后面的模型卡住了。

所以第二步先接 ASR，是为了把“实时音频输入”这件事单独验证清楚。

三、前端：浏览器获取麦克风音频

前端先创建 RTCPeerConnection，再通过 getUserMedia() 获取麦克风音频轨道。

const pc = new RTCPeerConnection();
const dc = pc.createDataChannel("events");

dc.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log("ASR result:", data);
};

const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
for (const track of stream.getTracks()) {
  pc.addTrack(track, stream);
}

这里的 audio track 会通过 WebRTC 传到后端。前端不需要自己把音频拆成 PCM，也不需要手动上传 wav 文件。

如果只是做最小 demo，浏览器端只要负责三件事：

• 采集麦克风；
• 建立 WebRTC 连接；
• 接收后端返回的识别结果。

四、后端：FastAPI 负责信令，aiortc 负责接音频

后端先提供一个 /offer 接口，接收浏览器传来的 SDP offer，并返回 answer。

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from aiortc import RTCPeerConnection, RTCSessionDescription

app = FastAPI()
pcs = set()


@app.post("/offer")
async def offer(request: Request):
    params = await request.json()
    offer = RTCSessionDescription(sdp=params["sdp"], type=params["type"])

    pc = RTCPeerConnection()
    pcs.add(pc)

    @pc.on("track")
    def on_track(track):
        if track.kind == "audio":
            # 这里把音频轨道交给 ASR 处理
            start_asr_task(track)

    await pc.setRemoteDescription(offer)
    answer = await pc.createAnswer()
    await pc.setLocalDescription(answer)

    return JSONResponse({
        "sdp": pc.localDescription.sdp,
        "type": pc.localDescription.type,
    })

这段代码只做两件事：完成 WebRTC 信令协商，并在收到音频轨道时启动 ASR 任务。

真正关键的地方在 track.recv()。

五、aiortc 接收到的音频帧怎么转 PCM

在 aiortc 里，音频轨道是一个不断产生 frame 的对象。后端可以在循环里持续接收音频帧。

async def consume_audio(track, asr_client):
    while True:
        frame = await track.recv()
        pcm = frame.to_ndarray()
        await asr_client.send_audio(pcm)

这个例子只是说明思路。真实接 ASR 时，还要处理三个问题：

• 采样率是否符合 ASR 要求，例如 16k 或 48k；
• 声道数是否需要转成 mono；
• frame 数据是否需要重新编码成 PCM bytes。

很多 ASR 服务并不直接接受 numpy.ndarray，而是要求 bytes。这时可以把音频帧转成指定格式后再发送。

async def consume_audio(track, asr_client):
    while True:
        frame = await track.recv()
        pcm_array = frame.to_ndarray()

        # 根据 ASR 服务要求做重采样、声道转换和编码
        pcm_bytes = convert_to_pcm16(pcm_array)

        await asr_client.send_audio(pcm_bytes)

这里的 convert_to_pcm16() 可以根据实际 ASR 服务来实现。比如本地模型、云厂商 ASR、FunASR、Whisper streaming、SenseVoice、讯飞、阿里云、火山等，对输入格式的要求都不完全一样。

六、把 ASR 封装成一个可替换接口

为了避免 demo 和某一个 ASR 厂商绑死，建议先抽象一个 StreamingASRClient。

class StreamingASRClient:
    async def start(self):
        pass

    async def send_audio(self, pcm_bytes: bytes):
        pass

    async def receive_text(self) -> dict:
        pass

    async def close(self):
        pass

后面无论接本地模型还是云服务，都尽量保持这几个方法不变。

这样做的好处是：Voice Agent 的主链路不用关心 ASR 厂商是谁，只关心“音频发出去，文字拿回来”。

一个识别结果可以统一成这样的结构：

{
  "type": "asr_partial",
  "text": "我想查一下订单",
  "is_final": false,
  "latency_ms": 320
}

最终句可以这样返回：

{
  "type": "asr_final",
  "text": "我想查一下订单为什么还没有发货",
  "is_final": true,
  "latency_ms": 860
}

这两个字段很重要：

• is_final=false 表示中间识别结果，前端可以实时显示；
• is_final=true 表示一句话基本结束，可以交给 RAG 或 LLM。

七、识别结果怎么回传前端

识别结果可以通过 DataChannel 回传，也可以通过 WebSocket 回传。

如果这只是一个 WebRTC 语音助手 Demo，DataChannel 更简单。因为音频轨道和事件消息都挂在同一个 PeerConnection 上，状态更集中。

@pc.on("datachannel")
def on_datachannel(channel):
    pc.asr_channel = channel

当 ASR 返回文字时：

channel.send(json.dumps({
    "type": "asr_final",
    "text": "我想查一下订单为什么还没有发货",
    "is_final": True
}, ensure_ascii=False))

如果你希望 ASR 和 WebRTC 解耦，比如后面要接多个页面、多个会话、客服工作台、日志系统，那么 WebSocket 会更清晰。

我的建议是：

• demo 阶段用 DataChannel；
• 生产环境根据会话管理复杂度选择 WebSocket 或消息队列；
• 无论用哪种方式，都要带上 session_id、trace_id 和时间戳。

八、实时 ASR 接入后，还要看哪些指标

接入 ASR 不等于语音助手就能用了。至少要看这几个指标：

1. 首字延迟：用户开始说话后，多久能看到第一段识别文本；
2. 最终句延迟：用户说完后，多久返回稳定结果；
3. 识别准确率：业务名词、数字、地址、订单号是否识别正确；
4. 噪声鲁棒性：办公室、门店、电话外放环境下是否能用；
5. 中断处理：用户说到一半停顿，系统是否误判结束；
6. 热词能力：产品名、品牌名、业务术语能不能识别；
7. 日志追踪：每次识别是否记录音频时长、延迟、模型版本和错误原因。

对 AI 语音客服来说，ASR 错了，后面的 RAG 和 LLM 很可能都会错。所以 ASR 是 Voice Agent 链路里最应该先测清楚的一层。

九、接上 RAG 后，链路会变成什么样

当 ASR 返回 asr_final 后，就可以把文字交给 RAG。

async def on_asr_final(text: str):
    docs = retrieve(text)
    answer = compose_answer(text, docs)
    return answer

这一步的重点不是“让大模型自由发挥”，而是让回答有依据。

比如用户问：“我的订单为什么还没发货？”

系统应该先根据用户身份和订单号查业务系统，再检索售后政策、物流规则、发货说明。如果知识库和业务系统都没有依据，就应该返回“当前资料不足，需要转人工”，而不是编一个看起来很顺的答案。

这也是 Voice Agent 和普通语音聊天机器人的区别：它不只是把话听懂，还要知道哪些话能回答，哪些话必须交给人工。

十、常见问题

FastAPI + WebRTC 怎么实现实时语音识别？

核心做法是：浏览器用 WebRTC 发送麦克风音频，FastAPI 提供 /offer 信令接口，后端用 aiortc 接收 audio track，再把音频帧转换成 ASR 可用的 PCM 数据，最后把识别结果通过 DataChannel 或 WebSocket 返回前端。

WebRTC 音频怎么传给后端 ASR？

浏览器通过 getUserMedia() 获取麦克风音频，再用 pc.addTrack() 加入 PeerConnection。后端 aiortc 会在 on_track 事件里拿到 audio track，然后通过 await track.recv() 持续接收音频帧。

aiortc 接收到的音频帧怎么转 PCM？

可以先用 frame.to_ndarray() 取出音频数据，再根据 ASR 服务要求做重采样、声道转换和 PCM16 编码。不同 ASR 服务对采样率、声道数、编码格式要求不同，不能直接假设所有服务都接受同一种格式。

Voice Agent 为什么需要流式 ASR？

因为 Voice Agent 需要低延迟理解用户说话内容。如果只等用户整段说完再上传音频文件，响应会很慢，也很难支持打断、实时字幕和多轮对话。流式 ASR 可以边听边识别，是实时语音助手和 AI 语音客服的基础能力。

DataChannel 和 WebSocket 哪个更适合返回识别结果？

demo 阶段 DataChannel 更简单，因为音频和文本事件都在同一个 WebRTC 连接里。生产环境如果需要更复杂的会话管理、日志系统、客服工作台或多端同步，WebSocket 或消息队列会更清晰。

实时 ASR 延迟应该怎么优化？

可以从几个方向入手：减少音频 buffer，选择更低延迟的 ASR 服务，避免不必要的格式转换，使用 VAD 判断说话边界，把 ASR 中间结果及时返回前端，并记录首字延迟和最终句延迟。

十一、总结

用 FastAPI + WebRTC 接入流式 ASR，本质上是在解决一个问题：浏览器里的实时语音，如何稳定地变成后端可处理的文字。

最小链路可以概括为：

getUserMedia 采集麦克风
  -> WebRTC 传输音频
  -> aiortc 接收音频帧
  -> 转成 PCM
  -> 流式 ASR 返回 transcript
  -> DataChannel / WebSocket 推回前端
  -> 后续接 RAG、LLM、TTS 和转人工

如果第一篇文章解决的是“Voice Agent 的最小闭环怎么搭”，那么这一篇解决的是“语音输入层怎么补上”。

下一步再往下做，就可以把 asr_final 接到 RAG，要求回答必须带引用依据；再把回答接到 TTS，形成真正的实时语音问答闭环。

但在继续加能力之前，我建议先把 ASR 这一层测清楚：首字延迟、最终句延迟、识别准确率、业务热词、噪声鲁棒性和日志追踪。因为在 AI 语音客服里，听错一句话，后面所有智能都可能变成“认真地答错”。