用 FastAPI + WebRTC + RAG 搭一个简化版 Voice Agent Demo

最近我在拆智能客服和 Voice Agent 平台时，发现很多文章会把概念讲得很完整：ASR、RAG、LLM、TTS、WebRTC、转人工、工单流转……每个词都对，但读完以后还是不知道一个最小系统到底怎么串起来。

这篇文章不追求做一个生产级语音客服平台，而是搭一个能跑通链路的最小 demo：

• 浏览器通过 WebRTC 建立实时音频通道；
• 前端用 DataChannel 向后端发送一条用户问题；
• FastAPI 负责信令接口和后端服务；
• 后端用一个很轻量的 RAG 检索模块，从本地知识库里找上下文；
• 最后把带引用依据的回答返回给浏览器。

为了让 demo 足够短，本文先把 ASR 和 TTS 做成可替换接口：WebRTC 音频通道已经建立，真正的语音识别和语音合成可以在后续接入。也就是说，这不是“完整商用 Voice Agent”，而是一个适合继续扩展的骨架。

一、先看最小架构

一个最小 Voice Agent 可以拆成 6 层：

如果放到真实智能客服场景里，链路会变成：

用户语音
  -> VAD 判断是否在说话
  -> ASR 转文字
  -> RAG 检索知识库 / 订单 / 工单 / CRM
  -> LLM 生成回答
  -> 规则层判断是否转人工
  -> TTS 合成语音
  -> WebRTC / 电话线路回传给用户

本文先实现最小闭环：

WebRTC 建连
  -> DataChannel 发送问题
  -> 本地知识库检索
  -> 返回带依据的回答

这样做的好处是：先把实时通道、信令、知识库检索、回复格式跑通，再逐步替换 ASR、LLM、TTS。

二、为什么用 FastAPI + WebRTC + RAG

模块	在 demo 里的作用	真实 Voice Agent 中的作用
FastAPI	提供 /offer 信令接口、托管前端页面	承接 Web 服务、鉴权、日志、业务接口
WebRTC	建立浏览器和服务端之间的实时连接	支撑低延迟语音通话、打断、实时互动
DataChannel	传输文本问题和回答结果	可传 ASR 中间结果、状态事件、调试日志
RAG	从本地知识库找上下文	接企业知识库、FAQ、工单、产品文档
Answer Composer	拼装回答	可替换为大模型、工作流或规则引擎

一个常见误区是：Voice Agent 不只是“LLM + TTS”。在客服场景里，真正难的是实时交互、知识命中、业务边界和可追溯。用户问一句“我这个订单为什么还没发货”，系统不仅要会说话，还要知道：

• 应该查哪个订单；
• 有没有权限看订单；
• 回答依据来自哪里；
• 什么时候必须转人工；
• 有没有把这次对话记录进工单。

所以一个哪怕很小的 demo，也应该把链路拆清楚。

三、项目目录

voice-agent-demo/
  main.py
  static/
    index.html

安装依赖：

pip install fastapi "uvicorn[standard]" aiortc

这里没有强行接入某个大模型 SDK。为了让 demo 能直接跑，RAG 和回答生成先用本地函数模拟。后面可以把 compose_answer() 换成 OpenAI、通义、DeepSeek、Qwen、GLM 或企业内部模型。

四、后端代码：FastAPI + aiortc

新建 main.py：

import asyncio
import json
from typing import List, Dict

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


app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")

pcs = set()


KNOWLEDGE_BASE: List[Dict[str, str]] = [
    {
        "title": "Voice Agent 首响延迟",
        "content": "Voice Agent 的首响延迟通常由 VAD、ASR、LLM 推理、RAG 检索和 TTS 合成共同决定。企业客服场景建议把首轮响应控制在 1.5 秒以内。",
    },
    {
        "title": "RAG 在智能客服中的作用",
        "content": "RAG 可以把企业 FAQ、产品手册、订单规则、售后政策等内容检索出来，作为大模型回答的依据，降低幻觉风险。",
    },
    {
        "title": "转人工策略",
        "content": "当用户情绪升高、连续两轮未解决、涉及退款投诉、隐私权限或高风险业务动作时，Voice Agent 应主动转人工。",
    },
    {
        "title": "WebRTC 的价值",
        "content": "WebRTC 适合浏览器侧低延迟音视频通信，可用于网页客服、在线咨询、实时语音助手等场景。",
    },
]


def retrieve(query: str, top_k: int = 2) -> List[Dict[str, str]]:
    """一个极简 RAG 检索器：用关键词重合度模拟召回。

    生产环境可以替换为向量数据库，例如 Milvus、Qdrant、pgvector、Elasticsearch
    或企业知识库自己的召回接口。
    """
    query_chars = set(query.lower())
    scored = []

    for doc in KNOWLEDGE_BASE:
        text = (doc["title"] + doc["content"]).lower()
        score = len(query_chars & set(text))
        scored.append((score, doc))

    scored.sort(key=lambda item: item[0], reverse=True)
    return [doc for score, doc in scored[:top_k] if score > 0]


def compose_answer(query: str, docs: List[Dict[str, str]]) -> Dict[str, object]:
    if not docs:
        return {
            "answer": "知识库里没有找到足够依据，建议转人工或补充知识库内容。",
            "citations": [],
            "route": "human_fallback",
        }

    context = "\n".join([f"- {doc['title']}：{doc['content']}" for doc in docs])

    # 这里先用模板模拟 LLM。真实项目中可以把 query + context 交给大模型。
    answer = (
        f"根据当前知识库，我会这样判断：{docs[0]['content']} "
        f"如果这是线上客服场景，还需要记录本轮问题、命中的知识条目和是否触发转人工策略。"
    )

    return {
        "answer": answer,
        "citations": [doc["title"] for doc in docs],
        "route": "rag_answer",
        "debug_context": context,
    }


async def consume_audio(track):
    """接收浏览器传来的音频轨道。

    这个 demo 只消费音频帧，避免连接阻塞。
    后续可以在这里接 VAD / ASR，把音频转成文字后交给 RAG。
    """
    while True:
        try:
            frame = await track.recv()
            # 为了避免刷屏，这里不打印每一帧。
            # 可以在这里加入音量检测、VAD、ASR streaming。
        except Exception:
            break


@app.get("/")
async def index():
    with open("static/index.html", "r", encoding="utf-8") as f:
        return HTMLResponse(f.read())


@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("datachannel")
    def on_datachannel(channel):
        @channel.on("message")
        def on_message(message):
            if not isinstance(message, str):
                return

            try:
                payload = json.loads(message)
                query = payload.get("query", "").strip()
            except json.JSONDecodeError:
                query = message.strip()

            docs = retrieve(query)
            result = compose_answer(query, docs)
            channel.send(json.dumps(result, ensure_ascii=False))

    @pc.on("track")
    def on_track(track):
        if track.kind == "audio":
            asyncio.create_task(consume_audio(track))

    @pc.on("connectionstatechange")
    async def on_connectionstatechange():
        if pc.connectionState in ["failed", "closed", "disconnected"]:
            await pc.close()
            pcs.discard(pc)

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

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


@app.on_event("shutdown")
async def on_shutdown():
    await asyncio.gather(*[pc.close() for pc in pcs])
    pcs.clear()

这段代码做了几件事：

1. /offer 接收浏览器发来的 WebRTC offer；
2. 服务端创建 RTCPeerConnection 并返回 answer；
3. 浏览器音频轨道会进入 consume_audio()；
4. 文本问题通过 DataChannel 进入后端；
5. 后端先检索知识库，再返回答案、引用来源和路由类型。

五、前端代码：浏览器建立 WebRTC 连接

新建 static/index.html：

<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8" />
  <title>Mini Voice Agent Demo</title>
  <style>
    body {
      font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
      max-width: 860px;
      margin: 40px auto;
      line-height: 1.7;
    }
    textarea {
      width: 100%;
      min-height: 90px;
      font-size: 15px;
    }
    button {
      margin: 8px 8px 8px 0;
      padding: 8px 14px;
      cursor: pointer;
    }
    pre {
      background: #f6f8fa;
      padding: 14px;
      overflow: auto;
    }
  </style>
</head>
<body>
  <h1>Mini Voice Agent Demo</h1>
  <p>这个页面会建立 WebRTC 音频通道，并通过 DataChannel 向后端发送问题。</p>

  <button id="connect">连接 Voice Agent</button>
  <span id="status">未连接</span>

  <h2>提问</h2>
  <textarea id="query">智能客服什么时候应该转人工？</textarea>
  <br />
  <button id="send">发送问题</button>

  <h2>返回结果</h2>
  <pre id="result">等待回答...</pre>

  <script>
    let pc;
    let channel;

    const statusEl = document.getElementById("status");
    const resultEl = document.getElementById("result");

    document.getElementById("connect").onclick = async () => {
      pc = new RTCPeerConnection();
      channel = pc.createDataChannel("agent-text");

      channel.onopen = () => {
        statusEl.textContent = "DataChannel 已连接";
      };

      channel.onmessage = (event) => {
        const data = JSON.parse(event.data);
        resultEl.textContent = JSON.stringify(data, null, 2);
      };

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

      const offer = await pc.createOffer();
      await pc.setLocalDescription(offer);

      const response = await fetch("/offer", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify(pc.localDescription),
      });

      const answer = await response.json();
      await pc.setRemoteDescription(answer);
      statusEl.textContent = "WebRTC 已连接，等待 DataChannel 打开";
    };

    document.getElementById("send").onclick = () => {
      if (!channel || channel.readyState !== "open") {
        resultEl.textContent = "请先点击“连接 Voice Agent”。";
        return;
      }

      const query = document.getElementById("query").value;
      channel.send(JSON.stringify({ query }));
    };
  </script>
</body>
</html>

这里有一个小设计：浏览器会申请麦克风权限，把音频 track 加到 WebRTC 连接里；但问题文本通过 DataChannel 发送。这样可以把“实时通道”和“RAG 回答”先跑通。

如果后续要变成真正语音版，只需要把链路改成：

audio track -> ASR streaming -> transcript -> retrieve() -> compose_answer() -> TTS -> audio track

六、运行 demo

启动服务：

uvicorn main:app --reload

打开浏览器：

http://127.0.0.1:8000

操作顺序：

1. 点击“连接 Voice Agent”；
2. 允许浏览器使用麦克风；
3. 在输入框里输入问题；
4. 点击“发送问题”；
5. 查看返回的 answer、citations 和 route。

示例问题：

智能客服什么时候应该转人工？

可能得到的结果：

{
  "answer": "根据当前知识库，我会这样判断：当用户情绪升高、连续两轮未解决、涉及退款投诉、隐私权限或高风险业务动作时，Voice Agent 应主动转人工。如果这是线上客服场景，还需要记录本轮问题、命中的知识条目和是否触发转人工策略。",
  "citations": [
    "转人工策略",
    "RAG 在智能客服中的作用"
  ],
  "route": "rag_answer",
  "debug_context": "- 转人工策略：当用户情绪升高、连续两轮未解决、涉及退款投诉、隐私权限或高风险业务动作时，Voice Agent 应主动转人工。\n- RAG 在智能客服中的作用：RAG 可以把企业 FAQ、产品手册、订单规则、售后政策等内容检索出来，作为大模型回答的依据，降低幻觉风险。"
}

七、这个 demo 离生产级 Voice Agent 还差什么

上面的 demo 只解决了最小闭环，离真正可用的智能客服平台还有几层关键能力。

能力	demo 现状	生产环境建议
ASR	只接收音频帧，未识别	接入流式 ASR，输出中间识别结果
VAD	未实现	用 VAD 判断用户是否说完，支持打断
RAG	关键词召回	换成向量召回 + 关键词召回 + rerank
LLM	模板生成	接大模型，并增加系统提示词和安全边界
TTS	未实现	接流式 TTS，边生成边播放
转人工	返回 route 字段	接客服工作台、工单系统、坐席分配
日志	只返回 debug_context	记录 trace_id、命中文档、耗时、模型版本
权限	未实现	按用户身份控制可查知识和可执行业务动作

我比较建议按这个顺序扩展：

1. 先接 ASR，让音频真的转成文本；
2. 再接向量知识库，提高召回效果；
3. 然后接 LLM，并要求回答必须带引用；
4. 最后接 TTS 和转人工。

这样每一步都有可观察的指标，不会一上来就堆成一个黑盒。

八、测评一个 Voice Agent Demo，要看哪些指标

如果只是能回答问题，不算真正的 Voice Agent。做智能客服平台测评时，我会优先看这些指标：

指标	为什么重要	简单测试方法
首响延迟	用户是否觉得“它在卡”	从用户停止说话到首字出现/首音播放
识别准确率	ASR 错了，后面全错	噪声、口音、业务名词混合测试
知识库命中率	决定回答是否有依据	用 30 条 FAQ 看命中文档是否正确
幻觉率	客服场景不能乱编	问知识库没有的问题，看是否承认不知道
打断能力	语音客服必须能被用户打断	TTS 播放中插话，看系统是否停播
转人工策略	高风险问题不能硬答	投诉、退款、隐私、重复失败场景
日志可追溯	方便复盘和审计	检查是否记录 trace_id、召回、模型版本

这也是我认为 Voice Agent 和普通聊天机器人最大的区别：它不只是“能不能回答”，而是“能不能在真实业务里可控地回答”。

九、下一步可以怎么改

如果要把本文 demo 往真实项目推进，可以做 4 个小版本。

v0.2：接入流式 ASR

把 consume_audio() 改造成音频缓冲器，把音频帧送给 ASR 服务：

async def consume_audio(track):
    while True:
        frame = await track.recv()
        pcm = frame.to_ndarray()
        # send pcm to streaming ASR
        # transcript = await asr_client.send(pcm)

v0.3：接入向量知识库

把 retrieve() 替换为向量检索：

def retrieve(query: str, top_k: int = 3):
    embedding = embed(query)
    docs = vector_store.search(embedding, top_k=top_k)
    return docs

v0.4：回答必须带依据

给 LLM 的提示词里加约束：

你是企业智能客服助手。
只能根据给定知识库回答。
如果知识库没有依据，请回答“当前资料不足，需要转人工”。
每次回答必须返回引用的知识条目标题。

v0.5：加入转人工

可以先用规则实现：

def should_handoff(query: str, answer: str) -> bool:
    risk_words = ["投诉", "退款", "报警", "隐私", "泄露", "人工", "主管"]
    return any(word in query for word in risk_words)

真实项目里，转人工策略还要结合用户等级、会话轮次、情绪、业务类型和坐席状态。

十、总结

这篇 demo 的重点不是“我写了一个多聪明的 AI 客服”，而是把 Voice Agent 的工程链路拆开：

• FastAPI 负责服务入口和信令；
• WebRTC 负责实时连接；
• DataChannel 负责状态和文本消息；
• RAG 负责把回答拉回知识库；
• route 字段为后续转人工和业务编排留出口；
• debug_context 为日志、审计和测评留依据。

如果你要评估一个 Voice Agent 平台，不建议只看它“声音像不像真人”。更值得看的，是它在真实业务里有没有这些能力：

1. 能不能低延迟响应；
2. 能不能基于知识库回答；
3. 能不能承认不知道；
4. 能不能在高风险场景转人工；
5. 能不能把每次回答的依据留下来。

语音只是入口，RAG 和业务控制才是智能客服能不能落地的关键。