前面几篇文章分别拆了 Voice Agent 的几个关键模块：WebRTC 音频输入、VAD 语音检测、流式 ASR、RAG 检索、流式 TTS、全链路延迟、安全转人工和工单回流。

这一篇做第一阶段收官。

本文要解决的问题很直接：

怎么把前面分散的代码，合成一个可以本地跑起来的 Voice Agent v1？

最终方案是：

• 用 FastAPI 提供 HTTP 和 WebSocket 接口；
• 用统一配置管理 ASR、LLM、TTS Provider；
• 用会话状态机管理 LISTENING / THINKING / SPEAKING / TRANSFERRING；
• 用 t0-t8 埋点记录端到端延迟；
• 用 RAG 做知识库检索和引用校验；
• 用音频播放队列模拟流式 TTS 播放；
• 用 interrupt 接口模拟用户打断；
• 用 handoff ticket 模拟转人工和上下文交接。

本文代码是一个本地可运行 demo，不代表生产级电话系统。
ASR、LLM、TTS 都使用 mock provider，目的是验证链路、状态和日志。真实项目中只需要替换 Provider 层。

本文测试数据为合成数据，只用于验证工程链路，不代表任何真实模型或厂商效果。

---

一、适用范围

这篇适合几类读者：

1. 已经了解 Voice Agent 基本链路；
2. 已经写过 ASR、RAG 或 TTS 的单模块 demo；
3. 想把语音识别、知识库、LLM、TTS 和转人工串起来；
4. 需要一个能跑通的最小工程骨架；
5. 想看全链路延迟应该怎么埋点。

不适合的场景：

1. 想直接接真实电话线路；
2. 想直接做生产环境 WebRTC 网关；
3. 想横向评测不同 ASR / TTS 厂商；
4. 想直接拿这份代码上线。

这篇的重点不是“模型多强”，而是“工程链路是否能闭环”。

---

二、运行环境

本文代码在以下环境验证：

Python 版本：Python 3.13
操作系统：Linux
主要依赖：
  fastapi==0.115.6
  uvicorn[standard]==0.34.0
  pydantic==2.10.4
  pytest==8.3.4
  httpx==0.28.1

是否需要 API Key：不需要
音频格式：demo 中用 text: 前缀模拟音频输入
采样率：配置项中默认 16 kHz
外部工具：curl，可选

说明一下这里为什么不用真实 ASR、LLM、TTS。

如果第一版就把三类外部服务全部接进去，调试时会很难判断问题来自哪里：

• 是 WebSocket 没收到音频？
• 是 VAD 没检测出语音？
• 是 ASR 超时？
• 是 RAG 没召回？
• 是 LLM 没返回？
• 是 TTS 没生成音频？
• 是播放队列没消费？

所以 v1 先用 mock provider 把链路跑通。
等状态、日志、打断和转人工都稳定后，再替换真实服务。

这也是我现在做 Voice Agent demo 比较推荐的顺序：先做骨架，再换模型，不要一上来就让三个外部服务一起开盲盒。

---

三、完整链路图

本文实现的链路如下：

WebRTC / WebSocket 音频输入
        ↓
VAD 检测是否有人说话
        ↓
ASR 转写用户语音
        ↓
会话状态管理
        ↓
RAG 检索知识库
        ↓
引用校验 / 低置信拒答
        ↓
LLM 生成回复
        ↓
流式 TTS 生成音频 chunk
        ↓
音频播放队列
        ↓
用户打断时取消队列
        ↓
安全转人工
        ↓
工单和上下文交接

本文 demo 里有两个输入入口：

POST /demo/turn
  用文本模拟一轮语音输入，方便本地调试。

WS /ws/audio/{session_id}
  接收 bytes，模拟后端已经拿到音频帧。

生产环境中，浏览器侧可以通过 WebRTC 采集麦克风音频。服务端可以用 aiortc、媒体网关或通信平台拿到 PCM 帧。本文不展开 WebRTC 信令和媒体协商，只保留 Voice Agent 后端处理链路。

---

四、项目目录

项目目录如下：

voice_agent_v1/
├── app.py
├── audio_queue.py
├── config.py
├── metrics.py
├── pipeline.py
├── providers.py
├── rag.py
├── run_demo.py
├── schemas.py
├── state.py
├── requirements.txt
├── data/
│   └── kb.jsonl
└── tests/
    ├── conftest.py
    └── test_pipeline.py

各文件作用：

文件	作用
app.py	FastAPI 入口，提供 HTTP 和 WebSocket 接口
config.py	统一配置
schemas.py	请求和响应结构
state.py	会话状态机和历史上下文
metrics.py	t0-t8 延迟埋点
rag.py	本地 RAG 检索
providers.py	ASR、LLM、TTS Provider 抽象和 mock 实现
audio_queue.py	播放队列和取消逻辑
pipeline.py	Voice Agent 主链路
run_demo.py	本地端到端测试脚本
tests/	最小测试用例

---

五、安装依赖

新建目录：

mkdir voice_agent_v1
cd voice_agent_v1

创建虚拟环境：

python -m venv .venv
source .venv/bin/activate

安装依赖：

pip install -r requirements.txt

requirements.txt：

fastapi==0.115.6
uvicorn[standard]==0.34.0
pydantic==2.10.4
pytest==8.3.4
httpx==0.28.1

---

六、统一配置文件

先写 config.py。

这里把采样率、Provider 类型、知识库路径、RAG 参数都放到一个配置对象里。

后面要从 mock ASR 换成真实 ASR，只需要改环境变量，不需要改 pipeline 主流程。

from dataclasses import dataclass
import os


@dataclass(frozen=True)
class Settings:
    app_name: str = "voice-agent-v1"
    sample_rate: int = int(os.getenv("VA_SAMPLE_RATE", "16000"))
    frame_ms: int = int(os.getenv("VA_FRAME_MS", "20"))
    rag_top_k: int = int(os.getenv("VA_RAG_TOP_K", "3"))
    rag_min_score: float = float(os.getenv("VA_RAG_MIN_SCORE", "0.08"))
    asr_provider: str = os.getenv("VA_ASR_PROVIDER", "mock")
    llm_provider: str = os.getenv("VA_LLM_PROVIDER", "mock")
    tts_provider: str = os.getenv("VA_TTS_PROVIDER", "mock")
    kb_path: str = os.getenv("VA_KB_PATH", "data/kb.jsonl")
    max_history: int = int(os.getenv("VA_MAX_HISTORY", "12"))


settings = Settings()

这一步的关键不是配置项多少，而是要避免后面在各个文件里到处写死参数。

Voice Agent 一旦接入真实业务，很多参数都会变成调优项，例如：

• VAD 灵敏度；
• ASR 热词；
• RAG 召回阈值；
• TTS 分句长度；
• 音频队列最大长度；
• 转人工触发次数；
• 单轮超时时间。

这些都不适合散落在业务代码里。

---

七、请求和响应结构

创建 schemas.py。

from pydantic import BaseModel, Field
from typing import Any, Optional


class DemoTurnRequest(BaseModel):
    session_id: str = Field(default="demo-session")
    text: str


class SourceDoc(BaseModel):
    doc_id: str
    title: str
    content: str
    score: float = 0.0


class TurnResult(BaseModel):
    session_id: str
    user_text: str
    answer_text: str
    state: str
    sources: list[SourceDoc] = []
    latency_ms: dict[str, float] = {}
    handoff_ticket: Optional[dict[str, Any]] = None
    audio_chunks: int = 0


class HealthResult(BaseModel):
    ok: bool
    app: str

TurnResult 里有几个字段比较重要：

字段	说明
user_text	ASR 后的用户文本
answer_text	LLM 生成的回复
state	当前会话状态
sources	RAG 命中的知识库引用
latency_ms	全链路延迟
handoff_ticket	转人工工单
audio_chunks	TTS 生成的音频分片数量

真实系统里，audio_chunks 不会直接返回给前端，而是进入播放器或音频轨道。
本文为了方便验证，把它作为调试指标返回。

---

八、t0-t8 延迟埋点

创建 metrics.py。

Voice Agent 的延迟不能只看“总耗时”。

如果只知道一轮花了 1800ms，是没法定位问题的。
需要知道慢在哪里：

• VAD 慢？
• ASR 慢？
• RAG 慢？
• LLM 首 token 慢？
• TTS 首包慢？
• 音频队列积压？

所以这里做 t0-t8。

from time import perf_counter


class LatencyTracker:
    def __init__(self) -> None:
        self._marks: dict[str, float] = {}

    def mark(self, name: str) -> None:
        self._marks[name] = perf_counter()

    def has(self, name: str) -> bool:
        return name in self._marks

    def duration_ms(self, start: str, end: str) -> float:
        if start not in self._marks or end not in self._marks:
            return -1.0
        return round((self._marks[end] - self._marks[start]) * 1000, 2)

    def summary(self) -> dict[str, float]:
        pairs = {
            "vad_ms": ("t0_audio_in", "t1_vad_done"),
            "asr_ms": ("t1_vad_done", "t2_asr_final"),
            "rag_ms": ("t2_asr_final", "t3_rag_done"),
            "llm_first_token_ms": ("t3_rag_done", "t4_llm_first_token"),
            "llm_total_ms": ("t3_rag_done", "t5_llm_done"),
            "tts_first_chunk_ms": ("t5_llm_done", "t6_tts_first_chunk"),
            "queue_ms": ("t6_tts_first_chunk", "t7_playable_queued"),
            "turn_total_ms": ("t0_audio_in", "t8_turn_done"),
        }
        return {name: self.duration_ms(a, b) for name, (a, b) in pairs.items()}

本文采用的 t0-t8 定义：

点位	含义
t0_audio_in	收到音频输入
t1_vad_done	VAD 判断完成
t2_asr_final	ASR 最终文本完成
t3_rag_done	RAG 检索完成
t4_llm_first_token	LLM 第一个 token 返回
t5_llm_done	LLM 完整回复完成
t6_tts_first_chunk	TTS 首个音频 chunk 生成
t7_playable_queued	首个可播放 chunk 进入队列
t8_turn_done	本轮处理完成

这里有个容易踩坑的点：

TTS 首包延迟不要从“用户开始说话”算，而要从“LLM 可用于合成的文本产生后”算。

否则 ASR、RAG、LLM 的耗时都会混进 TTS 指标里，最后会误判 TTS 慢。

---

九、会话状态机

创建 state.py。

Voice Agent 不是一问一答脚本。
它必须知道自己现在处在哪个阶段。

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class ConversationState(str, Enum):
    IDLE = "IDLE"
    LISTENING = "LISTENING"
    THINKING = "THINKING"
    SPEAKING = "SPEAKING"
    TRANSFERRING = "TRANSFERRING"
    CLOSED = "CLOSED"


@dataclass
class Message:
    role: str
    content: str


@dataclass
class SessionContext:
    session_id: str
    state: ConversationState = ConversationState.IDLE
    history: list[Message] = field(default_factory=list)
    last_sources: list[dict] = field(default_factory=list)
    handoff_ticket: Optional[dict] = None
    interrupted: bool = False

    def add_message(self, role: str, content: str, max_history: int = 12) -> None:
        self.history.append(Message(role=role, content=content))
        if len(self.history) > max_history:
            self.history = self.history[-max_history:]

    def brief_history(self) -> str:
        lines = [f"{m.role}: {m.content}" for m in self.history[-6:]]
        return "\n".join(lines)


class SessionStore:
    def __init__(self) -> None:
        self._sessions: dict[str, SessionContext] = {}

    def get(self, session_id: str) -> SessionContext:
        if session_id not in self._sessions:
            self._sessions[session_id] = SessionContext(session_id=session_id)
        return self._sessions[session_id]

    def reset(self, session_id: str) -> None:
        self._sessions.pop(session_id, None)

状态流转大致是：

IDLE
  ↓
LISTENING
  ↓
THINKING
  ↓
SPEAKING
  ↓
IDLE

如果触发转人工：

SPEAKING
  ↓
TRANSFERRING

如果用户打断：

SPEAKING
  ↓
LISTENING

这里最重要的是：用户打断后，不能把未完整播放的内容当成用户已经听懂或确认的事实。

所以后面 interrupt 逻辑会写入一条 system message：

用户打断了上一轮播报，上一轮未完整播放内容不进入用户已确认事实。

这条记录对后续多轮对话很有用。

---

十、本地知识库和 RAG 检索

创建知识库文件 data/kb.jsonl。

{"doc_id":"kb-001","title":"Voice Agent WebRTC 接入说明","content":"浏览器侧可以通过 WebRTC 采集麦克风音频，服务端拿到 PCM 帧后进入 VAD、ASR、RAG、LLM 和 TTS 链路。本文 demo 用 WebSocket 接收音频帧，生产环境可替换为 aiortc 或媒体网关。"}
{"doc_id":"kb-002","title":"产品演示预约流程","content":"用户咨询产品演示时，需要确认公司名称、业务场景、预计通话量和期望接入渠道。信息明确后可以创建预约工单并转给售前同事。"}
{"doc_id":"kb-003","title":"安全转人工规则","content":"当用户要求转人工、表达投诉、涉及合同退款、连续两轮未命中知识库或情绪明显升级时，Voice Agent 应停止继续猜测，生成摘要和上下文后转人工。"}
{"doc_id":"kb-004","title":"RAG 引用校验规则","content":"RAG 命中结果需要保留 doc_id、title、score 和原文片段。低于阈值时不应强答，应拒答或转人工确认。"}

创建 rag.py。

为了让代码少依赖外部包，这里没有用向量数据库，而是写了一个本地简化版 TF-IDF 检索。

它不适合生产，但适合验证 RAG 链路：

• 能读知识库；
• 能根据 query 召回；
• 能返回 score；
• 能低分拒答；
• 能把引用传给 LLM。

import json
import math
from collections import Counter
from pathlib import Path
from schemas import SourceDoc


def tokenize(text: str) -> list[str]:
    # 中文 demo 里用字符 bigram + 英文空格词，避免额外分词依赖。
    text = text.lower().strip()
    words = [w for w in text.replace("，", " ").replace("。", " ").split() if w]
    chars = [c for c in text if "\u4e00" <= c <= "\u9fff"]
    bigrams = ["".join(chars[i:i + 2]) for i in range(max(0, len(chars) - 1))]
    return words + bigrams + chars


class SimpleRAG:
    def __init__(self, kb_path: str, top_k: int = 3, min_score: float = 0.08) -> None:
        self.kb_path = Path(kb_path)
        self.top_k = top_k
        self.min_score = min_score
        self.docs: list[SourceDoc] = []
        self.doc_vectors: list[Counter] = []
        self.idf: dict[str, float] = {}
        self._load()

    def _load(self) -> None:
        if not self.kb_path.exists():
            raise FileNotFoundError(f"KB file not found: {self.kb_path}")
        raw_docs = []
        with self.kb_path.open("r", encoding="utf-8") as f:
            for line in f:
                if line.strip():
                    raw_docs.append(json.loads(line))
        self.docs = [SourceDoc(**d) for d in raw_docs]
        tokenized = [tokenize(d.title + " " + d.content) for d in self.docs]
        df = Counter()
        for tokens in tokenized:
            df.update(set(tokens))
        total = len(tokenized)
        self.idf = {term: math.log((total + 1) / (freq + 1)) + 1 for term, freq in df.items()}
        self.doc_vectors = [self._vectorize(tokens) for tokens in tokenized]

    def _vectorize(self, tokens: list[str]) -> Counter:
        tf = Counter(tokens)
        return Counter({term: count * self.idf.get(term, 0.0) for term, count in tf.items()})

    @staticmethod
    def _cosine(a: Counter, b: Counter) -> float:
        common = set(a) & set(b)
        numerator = sum(a[t] * b[t] for t in common)
        norm_a = math.sqrt(sum(v * v for v in a.values()))
        norm_b = math.sqrt(sum(v * v for v in b.values()))
        if norm_a == 0 or norm_b == 0:
            return 0.0
        return numerator / (norm_a * norm_b)

    def retrieve(self, query: str) -> list[SourceDoc]:
        q_vec = self._vectorize(tokenize(query))
        scored: list[SourceDoc] = []
        for doc, vec in zip(self.docs, self.doc_vectors):
            score = self._cosine(q_vec, vec)
            if score >= self.min_score:
                scored.append(doc.model_copy(update={"score": round(score, 4)}))
        scored.sort(key=lambda d: d.score, reverse=True)
        return scored[:self.top_k]

这里有一个关键原则：

RAG 没召回，不要让 LLM 继续强答。

所以后面 mock LLM 里会判断：

elif not sources:
    answer = "这个问题我暂时没有在知识库里找到可靠依据。为了避免乱答，我建议转人工确认后再处理。"

这比“编一个听起来像答案的答案”更适合客服场景。

---

十一、可替换的 ASR / LLM / TTS Provider

创建 providers.py。

Provider 层是整个 v1 里最值得保留的设计。

因为真实项目里，ASR、LLM、TTS 很可能不是同一个厂商，也不一定长期固定。

import asyncio
import re
from abc import ABC, abstractmethod
from typing import AsyncIterator
from schemas import SourceDoc


class ASRProvider(ABC):
    @abstractmethod
    async def transcribe(self, audio_bytes: bytes) -> str:
        raise NotImplementedError


class LLMProvider(ABC):
    @abstractmethod
    async def stream_answer(self, user_text: str, sources: list[SourceDoc], history: str) -> AsyncIterator[str]:
        raise NotImplementedError


class TTSProvider(ABC):
    @abstractmethod
    async def stream_speech(self, text: str) -> AsyncIterator[bytes]:
        raise NotImplementedError


class MockASRProvider(ASRProvider):
    async def transcribe(self, audio_bytes: bytes) -> str:
        await asyncio.sleep(0.03)
        if audio_bytes.startswith(b"text:"):
            return audio_bytes[5:].decode("utf-8", errors="ignore").strip()
        # 没有接真实 ASR 时，用固定文本验证链路。
        return "我想预约 Voice Agent 产品演示，想了解怎么转人工"


class MockLLMProvider(LLMProvider):
    async def stream_answer(self, user_text: str, sources: list[SourceDoc], history: str) -> AsyncIterator[str]:
        await asyncio.sleep(0.02)
        if self._needs_handoff(user_text):
            answer = "我先为您转接人工同事。已保留当前问题、对话摘要和相关上下文，人工接起后不用重复说明。"
        elif not sources:
            answer = "这个问题我暂时没有在知识库里找到可靠依据。为了避免乱答，我建议转人工确认后再处理。"
        else:
            top = sources[0]
            answer = f"根据知识库《{top.title}》，{top.content}"
        for token in re.split(r"([，。！？])", answer):
            if not token:
                continue
            await asyncio.sleep(0.01)
            yield token

    @staticmethod
    def _needs_handoff(text: str) -> bool:
        keywords = ["转人工", "人工", "投诉", "听不懂", "不满意", "合同", "退款"]
        return any(k in text for k in keywords)


class MockTTSProvider(TTSProvider):
    async def stream_speech(self, text: str) -> AsyncIterator[bytes]:
        # demo 用 bytes 模拟音频 chunk。真实项目里这里替换为厂商 TTS 的二进制音频流。
        parts = [p for p in re.split(r"(?<=[。！？])", text) if p.strip()]
        for idx, part in enumerate(parts, start=1):
            await asyncio.sleep(0.02)
            yield f"AUDIO_CHUNK_{idx}:{part}".encode("utf-8")


def build_asr_provider(name: str) -> ASRProvider:
    if name == "mock":
        return MockASRProvider()
    raise ValueError(f"Unsupported ASR provider: {name}")


def build_llm_provider(name: str) -> LLMProvider:
    if name == "mock":
        return MockLLMProvider()
    raise ValueError(f"Unsupported LLM provider: {name}")


def build_tts_provider(name: str) -> TTSProvider:
    if name == "mock":
        return MockTTSProvider()
    raise ValueError(f"Unsupported TTS provider: {name}")

后续如果要接真实服务，可以这样改：

class RealASRProvider(ASRProvider):
    async def transcribe(self, audio_bytes: bytes) -> str:
        # 这里调用真实 ASR SDK 或 HTTP API
        ...

然后在工厂函数里加：

if name == "real_asr":
    return RealASRProvider()

这样 pipeline 不需要知道你用的是哪家 ASR。

---

十二、音频播放队列和取消逻辑

创建 audio_queue.py。

用户打断不是简单地“下一轮开始”。

真实语音交互里，用户打断时至少要做三件事：

1. 停止继续生成 TTS；
2. 清空还没播放的音频队列；
3. 修改会话状态，回到监听用户说话。

本文先实现第 2 和第 3 件事。

import asyncio


class PlaybackQueue:
    def __init__(self) -> None:
        self.queue: asyncio.Queue[bytes] = asyncio.Queue()
        self.cancel_event = asyncio.Event()

    async def put(self, chunk: bytes) -> None:
        if not self.cancel_event.is_set():
            await self.queue.put(chunk)

    async def cancel(self) -> int:
        self.cancel_event.set()
        cleared = 0
        while not self.queue.empty():
            self.queue.get_nowait()
            self.queue.task_done()
            cleared += 1
        return cleared

    def reset(self) -> None:
        self.cancel_event.clear()

这里的 cancel_event 很关键。

TTS 流式生成时，不应该一边用户已经打断了，一边继续往播放队列里塞音频。
否则就会出现一个很奇怪的现象：用户已经开始说下一句话了，AI 还在播放上一句的尾巴。

这就是语音 Agent 里常见的“残留播放”。

---

十三、主链路 Pipeline

创建 pipeline.py。

这个文件是 v1 的主菜，所有模块会在这里串起来。

from config import Settings
from metrics import LatencyTracker
from rag import SimpleRAG
from providers import build_asr_provider, build_llm_provider, build_tts_provider
from state import ConversationState, SessionStore, SessionContext
from audio_queue import PlaybackQueue
from schemas import TurnResult


class EnergyVAD:
    def __init__(self, min_non_zero_ratio: float = 0.01) -> None:
        self.min_non_zero_ratio = min_non_zero_ratio

    def is_speech(self, audio_bytes: bytes) -> bool:
        if audio_bytes.startswith(b"text:"):
            return len(audio_bytes) > len(b"text:")
        if not audio_bytes:
            return False
        non_zero = sum(1 for b in audio_bytes if b != 0)
        return (non_zero / len(audio_bytes)) >= self.min_non_zero_ratio


class VoiceAgentPipeline:
    def __init__(self, settings: Settings) -> None:
        self.settings = settings
        self.sessions = SessionStore()
        self.vad = EnergyVAD()
        self.asr = build_asr_provider(settings.asr_provider)
        self.llm = build_llm_provider(settings.llm_provider)
        self.tts = build_tts_provider(settings.tts_provider)
        self.rag = SimpleRAG(settings.kb_path, top_k=settings.rag_top_k, min_score=settings.rag_min_score)
        self.playback: dict[str, PlaybackQueue] = {}

    def _queue(self, session_id: str) -> PlaybackQueue:
        if session_id not in self.playback:
            self.playback[session_id] = PlaybackQueue()
        return self.playback[session_id]

    async def handle_audio_turn(self, session_id: str, audio_bytes: bytes) -> TurnResult:
        session = self.sessions.get(session_id)
        queue = self._queue(session_id)
        queue.reset()
        tracker = LatencyTracker()

        tracker.mark("t0_audio_in")
        session.state = ConversationState.LISTENING
        is_speech = self.vad.is_speech(audio_bytes)
        tracker.mark("t1_vad_done")
        if not is_speech:
            tracker.mark("t8_turn_done")
            return TurnResult(
                session_id=session_id,
                user_text="",
                answer_text="未检测到有效语音",
                state=session.state.value,
                latency_ms=tracker.summary(),
            )

        user_text = await self.asr.transcribe(audio_bytes)
        tracker.mark("t2_asr_final")
        session.add_message("user", user_text, max_history=self.settings.max_history)

        session.state = ConversationState.THINKING
        sources = self.rag.retrieve(user_text)
        session.last_sources = [s.model_dump() for s in sources]
        tracker.mark("t3_rag_done")

        answer_parts: list[str] = []
        first_token = True
        async for token in self.llm.stream_answer(user_text, sources, session.brief_history()):
            if first_token:
                tracker.mark("t4_llm_first_token")
                first_token = False
            answer_parts.append(token)
        answer_text = "".join(answer_parts)
        if first_token:
            tracker.mark("t4_llm_first_token")
        tracker.mark("t5_llm_done")

        session.add_message("assistant", answer_text, max_history=self.settings.max_history)
        session.state = ConversationState.SPEAKING
        audio_chunks = 0
        first_chunk = True
        async for chunk in self.tts.stream_speech(answer_text):
            if queue.cancel_event.is_set():
                break
            if first_chunk:
                tracker.mark("t6_tts_first_chunk")
                first_chunk = False
            await queue.put(chunk)
            audio_chunks += 1
            if not tracker.has("t7_playable_queued"):
                tracker.mark("t7_playable_queued")
        if first_chunk:
            tracker.mark("t6_tts_first_chunk")
            tracker.mark("t7_playable_queued")

        if self._needs_handoff(user_text, answer_text, sources):
            session.state = ConversationState.TRANSFERRING
            session.handoff_ticket = self._build_handoff_ticket(session, user_text, answer_text)
        else:
            session.state = ConversationState.IDLE
            session.handoff_ticket = None

        tracker.mark("t8_turn_done")
        return TurnResult(
            session_id=session_id,
            user_text=user_text,
            answer_text=answer_text,
            state=session.state.value,
            sources=sources,
            latency_ms=tracker.summary(),
            handoff_ticket=session.handoff_ticket,
            audio_chunks=audio_chunks,
        )

    async def interrupt(self, session_id: str) -> dict:
        session = self.sessions.get(session_id)
        queue = self._queue(session_id)
        cleared = await queue.cancel()
        session.interrupted = True
        session.state = ConversationState.LISTENING
        session.add_message("system", "用户打断了上一轮播报，上一轮未完整播放内容不进入用户已确认事实。")
        return {"session_id": session_id, "cleared_audio_chunks": cleared, "state": session.state.value}

    @staticmethod
    def _needs_handoff(user_text: str, answer_text: str, sources: list) -> bool:
        trigger_words = ["转人工", "投诉", "不满意", "退款", "合同"]
        return any(w in user_text for w in trigger_words) or not sources or "建议转人工" in answer_text

    @staticmethod
    def _build_handoff_ticket(session: SessionContext, user_text: str, answer_text: str) -> dict:
        return {
            "type": "human_handoff",
            "session_id": session.session_id,
            "last_user_text": user_text,
            "assistant_summary": answer_text[:120],
            "history": session.brief_history(),
            "sources": session.last_sources,
        }

这一段代码里，最核心的是 handle_audio_turn()。

它把一轮语音交互拆成了这些阶段：

收到音频
→ VAD
→ ASR
→ 写入用户消息
→ RAG
→ LLM
→ 写入助手消息
→ TTS
→ 音频入队
→ 判断是否转人工
→ 返回结果

如果你后续想把 mock ASR 换成真实 ASR，不要改这里。
只换 Provider。

如果你后续想把本地 RAG 换成向量数据库，也尽量不要改主链路。
只换 SimpleRAG.retrieve() 的实现。

这个边界一旦守住，Voice Agent 工程就不容易长成一团耳机线。

---

十四、FastAPI 接口

创建 app.py。

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from config import settings
from pipeline import VoiceAgentPipeline
from schemas import DemoTurnRequest, HealthResult, TurnResult

app = FastAPI(title=settings.app_name)
pipeline = VoiceAgentPipeline(settings)


@app.get("/health", response_model=HealthResult)
async def health() -> HealthResult:
    return HealthResult(ok=True, app=settings.app_name)


@app.post("/demo/turn", response_model=TurnResult)
async def demo_turn(req: DemoTurnRequest) -> TurnResult:
    audio_bytes = ("text:" + req.text).encode("utf-8")
    return await pipeline.handle_audio_turn(req.session_id, audio_bytes)


@app.post("/demo/interrupt/{session_id}")
async def interrupt(session_id: str) -> dict:
    return await pipeline.interrupt(session_id)


@app.websocket("/ws/audio/{session_id}")
async def audio_ws(websocket: WebSocket, session_id: str) -> None:
    await websocket.accept()
    try:
        while True:
            audio_bytes = await websocket.receive_bytes()
            result = await pipeline.handle_audio_turn(session_id, audio_bytes)
            await websocket.send_json(result.model_dump())
    except WebSocketDisconnect:
        return

这里提供三个接口：

接口	作用
GET /health	健康检查
POST /demo/turn	用文本模拟一轮语音输入
POST /demo/interrupt/{session_id}	模拟用户打断
WS /ws/audio/{session_id}	接收音频 bytes

为什么保留 /demo/turn？

因为它非常适合调试主链路。

真实音频链路没接好时，你可以先用文本把后半段跑通：

RAG → LLM → TTS → 队列 → 状态 → 转人工

等后半段稳定，再接真实音频。

---

十五、本地运行脚本

创建 run_demo.py。

import asyncio
from config import settings
from pipeline import VoiceAgentPipeline


async def main() -> None:
    pipeline = VoiceAgentPipeline(settings)
    result = await pipeline.handle_audio_turn(
        session_id="local-demo",
        audio_bytes="text:我想预约 Voice Agent 产品演示，需要了解 WebRTC 接入方式".encode("utf-8"),
    )
    print(result.model_dump_json(indent=2))

    interrupt_result = await pipeline.interrupt("local-demo")
    print(interrupt_result)

    result2 = await pipeline.handle_audio_turn(
        session_id="local-demo",
        audio_bytes="text:那还是帮我转人工吧".encode("utf-8"),
    )
    print(result2.model_dump_json(indent=2))


if __name__ == "__main__":
    asyncio.run(main())

启动：

python run_demo.py

我本地运行后，第一轮输出如下：

{
  "session_id": "local-demo",
  "user_text": "我想预约 Voice Agent 产品演示，需要了解 WebRTC 接入方式",
  "answer_text": "根据知识库《产品演示预约流程》，用户咨询产品演示时，需要确认公司名称、业务场景、预计通话量和期望接入渠道。信息明确后可以创建预约工单并转给售前同事。",
  "state": "IDLE",
  "sources": [
    {
      "doc_id": "kb-002",
      "title": "产品演示预约流程",
      "content": "用户咨询产品演示时，需要确认公司名称、业务场景、预计通话量和期望接入渠道。信息明确后可以创建预约工单并转给售前同事。",
      "score": 0.517
    },
    {
      "doc_id": "kb-001",
      "title": "Voice Agent WebRTC 接入说明",
      "content": "浏览器侧可以通过 WebRTC 采集麦克风音频，服务端拿到 PCM 帧后进入 VAD、ASR、RAG、LLM 和 TTS 链路。本文 demo 用 WebSocket 接收音频帧，生产环境可替换为 aiortc 或媒体网关。",
      "score": 0.1542
    }
  ],
  "latency_ms": {
    "vad_ms": 0.0,
    "asr_ms": 30.8,
    "rag_ms": 0.19,
    "llm_first_token_ms": 33.03,
    "llm_total_ms": 108.26,
    "tts_first_chunk_ms": 21.37,
    "queue_ms": 0.02,
    "turn_total_ms": 180.92
  },
  "handoff_ticket": null,
  "audio_chunks": 2
}

然后模拟用户打断：

{
  "session_id": "local-demo",
  "cleared_audio_chunks": 2,
  "state": "LISTENING"
}

再说“那还是帮我转人工吧”，输出如下：

{
  "session_id": "local-demo",
  "user_text": "那还是帮我转人工吧",
  "answer_text": "我先为您转接人工同事。已保留当前问题、对话摘要和相关上下文，人工接起后不用重复说明。",
  "state": "TRANSFERRING",
  "sources": [
    {
      "doc_id": "kb-003",
      "title": "安全转人工规则",
      "content": "当用户要求转人工、表达投诉、涉及合同退款、连续两轮未命中知识库或情绪明显升级时，Voice Agent 应停止继续猜测，生成摘要和上下文后转人工。",
      "score": 0.43
    },
    {
      "doc_id": "kb-004",
      "title": "RAG 引用校验规则",
      "content": "RAG 命中结果需要保留 doc_id、title、score 和原文片段。低于阈值时不应强答，应拒答或转人工确认。",
      "score": 0.1888
    }
  ],
  "latency_ms": {
    "vad_ms": 0.0,
    "asr_ms": 30.61,
    "rag_ms": 0.18,
    "llm_first_token_ms": 30.92,
    "llm_total_ms": 84.72,
    "tts_first_chunk_ms": 20.46,
    "queue_ms": 0.01,
    "turn_total_ms": 156.52
  },
  "handoff_ticket": {
    "type": "human_handoff",
    "session_id": "local-demo",
    "last_user_text": "那还是帮我转人工吧",
    "assistant_summary": "我先为您转接人工同事。已保留当前问题、对话摘要和相关上下文，人工接起后不用重复说明。",
    "history": "user: 我想预约 Voice Agent 产品演示，需要了解 WebRTC 接入方式\nassistant: 根据知识库《产品演示预约流程》，用户咨询产品演示时，需要确认公司名称、业务场景、预计通话量和期望接入渠道。信息明确后可以创建预约工单并转给售前同事。\nsystem: 用户打断了上一轮播报，上一轮未完整播放内容不进入用户已确认事实。\nuser: 那还是帮我转人工吧\nassistant: 我先为您转接人工同事。已保留当前问题、对话摘要和相关上下文，人工接起后不用重复说明。",
    "sources": [
      {
        "doc_id": "kb-003",
        "title": "安全转人工规则",
        "content": "当用户要求转人工、表达投诉、涉及合同退款、连续两轮未命中知识库或情绪明显升级时，Voice Agent 应停止继续猜测，生成摘要和上下文后转人工。",
        "score": 0.43
      }
    ]
  },
  "audio_chunks": 1
}

这说明几个点已经跑通：

1. 第一轮能完成 RAG 命中；
2. 能返回知识库引用；
3. 能记录全链路延迟；
4. 能生成 TTS 音频 chunk；
5. 用户打断后能清空播放队列；
6. 第二轮触发转人工；
7. 转人工时能带上历史上下文和引用来源。

---

十六、启动 FastAPI 服务

启动服务：

uvicorn app:app --reload --host 0.0.0.0 --port 8000

健康检查：

curl http://127.0.0.1:8000/health

预期输出：

{
  "ok": true,
  "app": "voice-agent-v1"
}

调用一轮对话：

curl -X POST http://127.0.0.1:8000/demo/turn \
  -H "Content-Type: application/json" \
  -d '{"session_id":"s1","text":"我想预约 Voice Agent 产品演示"}'

预期会看到：

{
  "session_id": "s1",
  "user_text": "我想预约 Voice Agent 产品演示",
  "answer_text": "根据知识库《产品演示预约流程》，用户咨询产品演示时，需要确认公司名称、业务场景、预计通话量和期望接入渠道。信息明确后可以创建预约工单并转给售前同事。",
  "state": "IDLE",
  "sources": [
    {
      "doc_id": "kb-002",
      "title": "产品演示预约流程",
      "content": "用户咨询产品演示时，需要确认公司名称、业务场景、预计通话量和期望接入渠道。信息明确后可以创建预约工单并转给售前同事。",
      "score": 0.517
    }
  ],
  "latency_ms": {
    "vad_ms": 0.0,
    "asr_ms": 30.0,
    "rag_ms": 0.2,
    "llm_first_token_ms": 30.0,
    "llm_total_ms": 100.0,
    "tts_first_chunk_ms": 20.0,
    "queue_ms": 0.0,
    "turn_total_ms": 180.0
  },
  "handoff_ticket": null,
  "audio_chunks": 2
}

注意：不同机器上延迟数值会有轻微差异。
本文的延迟数据只用于验证埋点，不代表真实模型耗时。

模拟打断：

curl -X POST http://127.0.0.1:8000/demo/interrupt/s1

预期输出：

{
  "session_id": "s1",
  "cleared_audio_chunks": 2,
  "state": "LISTENING"
}

触发转人工：

curl -X POST http://127.0.0.1:8000/demo/turn \
  -H "Content-Type: application/json" \
  -d '{"session_id":"s1","text":"我要投诉，帮我转人工"}'

预期会看到：

{
  "state": "TRANSFERRING",
  "handoff_ticket": {
    "type": "human_handoff",
    "session_id": "s1",
    "last_user_text": "我要投诉，帮我转人工",
    "assistant_summary": "我先为您转接人工同事。已保留当前问题、对话摘要和相关上下文，人工接起后不用重复说明。"
  }
}

---

十七、最小测试用例

创建 tests/conftest.py。

import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

创建 tests/test_pipeline.py。

import asyncio
from config import settings
from pipeline import VoiceAgentPipeline


def test_pipeline_basic_turn():
    async def run():
        p = VoiceAgentPipeline(settings)
        result = await p.handle_audio_turn("t1", "text:我想预约产品演示".encode("utf-8"))
        assert result.user_text == "我想预约产品演示"
        assert result.audio_chunks >= 1
        assert result.latency_ms["turn_total_ms"] >= 0
    asyncio.run(run())


def test_pipeline_handoff():
    async def run():
        p = VoiceAgentPipeline(settings)
        result = await p.handle_audio_turn("t2", "text:我要投诉，帮我转人工".encode("utf-8"))
        assert result.state == "TRANSFERRING"
        assert result.handoff_ticket is not None
    asyncio.run(run())


def test_interrupt_clears_queue():
    async def run():
        p = VoiceAgentPipeline(settings)
        await p.handle_audio_turn("t3", "text:介绍一下 WebRTC 接入".encode("utf-8"))
        result = await p.interrupt("t3")
        assert result["state"] == "LISTENING"
    asyncio.run(run())

运行语法检查：

python -m py_compile *.py

运行测试：

pytest -q

本地运行结果：

...                                                                      [100%]
3 passed in 0.70s

---

十八、这版 v1 现在具备哪些能力？

1. 有完整链路

虽然 ASR、LLM、TTS 都是 mock，但主链路已经完整：

音频输入 → VAD → ASR → RAG → LLM → TTS → 播放队列 → 打断 → 转人工

这比单独跑一个 ASR demo 更接近真实 Voice Agent。

2. 有统一状态

每轮对话都会进入明确状态：

LISTENING
THINKING
SPEAKING
TRANSFERRING
IDLE

这对后续处理打断、超时、转人工、异常恢复都很重要。

3. 有 Provider 抽象

ASR、LLM、TTS 都可以替换。

这意味着后续可以做几类实验：

• 同一个 pipeline 对比不同 ASR；
• 同一个 RAG 对比不同 LLM；
• 同一段回复对比不同 TTS 首包延迟；
• 同一组测试脚本对比不同模型组合的端到端耗时。

4. 有 t0-t8 延迟埋点

现在不只知道“慢”，还能知道慢在哪。

示例：

{
  "vad_ms": 0.0,
  "asr_ms": 30.8,
  "rag_ms": 0.19,
  "llm_first_token_ms": 33.03,
  "llm_total_ms": 108.26,
  "tts_first_chunk_ms": 21.37,
  "queue_ms": 0.02,
  "turn_total_ms": 180.92
}

真实项目中，建议把这些日志写入时序数据库或日志系统，后面可以按 session、provider、业务场景分析。

5. 有 RAG 引用和拒答

如果 RAG 没命中，系统不会继续编答案，而是建议转人工确认。

这对客服场景很重要。
宁可转人工，也不要用一个漂亮但错误的回答把用户带进沟里。

6. 有打断后的上下文处理

用户打断后会记录：

用户打断了上一轮播报，上一轮未完整播放内容不进入用户已确认事实。

这可以避免后续 LLM 错误地认为用户已经听完上一轮内容。

7. 有转人工工单

转人工时会生成：

{
  "type": "human_handoff",
  "session_id": "local-demo",
  "last_user_text": "那还是帮我转人工吧",
  "assistant_summary": "...",
  "history": "...",
  "sources": [...]
}

真实项目里，这个结构可以继续写入：

• CRM；
• 工单系统；
• 企微待办；
• 飞书审批；
• 人工坐席工作台。

---

十九、指标与结果

本文只做本地链路验证，不做模型横评。

测试项如下：

指标	本文结果	说明
VAD 判断	通过	文本模拟输入可识别为有效语音
ASR 输出	通过	mock ASR 返回用户文本
RAG 召回	通过	可返回 doc_id/title/score/content
LLM 首 token	通过	有 llm_first_token_ms
TTS 首 chunk	通过	有 tts_first_chunk_ms
音频队列	通过	可统计 audio_chunks
用户打断	通过	可清空队列并切回 LISTENING
转人工	通过	可生成 handoff_ticket
单元测试	通过	3 passed in 0.70s

再次强调：

以下数据只用于验证评测代码，不代表任何真实模型或厂商成绩。

如果接真实模型，建议至少增加这些指标：

模块	建议指标
ASR	CER、WER、实体准确率、空转写率、热词召回率
RAG	Recall@K、引用正确率、拒答准确率
LLM	首 token 延迟、总生成耗时、拒答正确率
TTS	首包延迟、合成耗时、RTF、播放队列积压
打断	打断检测耗时、取消成功率、残留播放时长
稳定性	超时率、错误率、连续运行时长、并发会话数

---

二十、常见报错与排查

1. ModuleNotFoundError: No module named 'config'

错误现象：

ModuleNotFoundError: No module named 'config'

可能原因：

• 不在项目根目录运行；
• pytest 没找到项目路径；
• 文件目录和 import 不一致。

检查命令：

pwd
ls

修复方法：

cd voice_agent_v1
pytest -q

如果还是报错，确认 tests/conftest.py 已添加项目根目录：

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

---

2. KB file not found

错误现象：

FileNotFoundError: KB file not found: data/kb.jsonl

可能原因：

• 没创建 data/kb.jsonl；
• 从错误目录启动；
• 环境变量 VA_KB_PATH 配错。

检查命令：

ls data
echo $VA_KB_PATH

修复方法：

mkdir -p data

然后创建 data/kb.jsonl。

如果想自定义路径：

export VA_KB_PATH=/absolute/path/to/kb.jsonl

---

3. WebSocket 收不到数据

错误现象：

/ws/audio/{session_id} 已连接，但没有返回

可能原因：

• 客户端发送的是文本，不是 bytes；
• WebSocket 没有持续发送音频帧；
• 音频帧为空；
• VAD 判断为无语音。

检查方法：

先用 /demo/turn 验证后半段：

curl -X POST http://127.0.0.1:8000/demo/turn \
  -H "Content-Type: application/json" \
  -d '{"session_id":"debug","text":"我想预约产品演示"}'

如果这个接口正常，说明 RAG、LLM、TTS、状态机没问题。
问题大概率在音频输入层。

修复方向：

• 确认 WebSocket 发送 bytes；
• 确认音频不是全 0；
• 确认采样率和编码格式；
• 在生产环境替换更可靠的 VAD。

---

4. RAG 总是没召回

错误现象：

"sources": []

可能原因：

• rag_min_score 太高；
• 知识库内容太少；
• 用户问题和知识库表达差异太大；
• 中文分词策略太粗糙。

检查方式：

临时降低阈值：

export VA_RAG_MIN_SCORE=0.01

修复方法：

1. 增加知识库内容；
2. 优化 query rewrite；
3. 换成 embedding 检索；
4. 对业务热词做别名表；
5. 对低召回问题单独记录，后续补知识。

---

5. 用户打断后仍然播放旧内容

错误现象：

用户已经说下一句话，但上一轮音频还在播放。

可能原因：

• 播放队列没有清空；
• TTS 还在继续生成；
• 前端播放器没有停止；
• 服务端状态没有切回 LISTENING。

检查接口：

curl -X POST http://127.0.0.1:8000/demo/interrupt/s1

正常应返回：

{
  "state": "LISTENING"
}

修复方法：

• 服务端清空队列；
• TTS 流检查 cancel_event；
• 前端停止当前 audio source；
• 上下文记录用户打断，不把未播放完内容当成已确认事实。

---

6. 转人工没有带上下文

错误现象：

"handoff_ticket": null

或者工单里没有历史消息。

可能原因：

• 转人工关键词没命中；
• _needs_handoff() 条件太窄；
• session_id 每轮都变了；
• history 没写入。

检查请求：

{
  "session_id": "s1",
  "text": "我要投诉，帮我转人工"
}

修复方法：

• 保持同一个 session_id；
• 扩充转人工关键词；
• 增加连续未命中次数；
• 增加情绪升级判断；
• 在真实项目中把 handoff ticket 写入 CRM 或工单系统。

---

二十一、FAQ

1. 这个 demo 算不算真正的 WebRTC Voice Agent？

不算完整生产版。

本文只实现 Voice Agent 后端处理链路，并提供 WebSocket 接收音频 bytes 的入口。
真实 WebRTC 还需要处理：

• SDP 信令；
• ICE；
• 音频编解码；
• 抖动缓冲；
• 丢包；
• 回声消除；
• 浏览器权限；
• 媒体网关。

本文的重点是后端语音智能链路，不是 WebRTC 信令教程。

---

2. 为什么不用真实 ASR、LLM、TTS？

因为第一阶段要先确认工程闭环。

如果一开始就接三类真实服务，出问题时定位成本很高。
v1 用 mock provider 先跑通：

输入 → 状态 → 检索 → 生成 → 合成 → 队列 → 打断 → 转人工

等这个链路稳定后，再逐个替换 Provider。

---

3. 中文 ASR 应该看 WER 还是 CER？

中文更常看 CER，也就是字符错误率。

因为中文没有天然空格分词，WER 依赖分词方式，不同分词器会影响结果。
如果是业务场景，还应该额外看实体准确率，例如：

• 手机号；
• 地址；
• 日期；
• 时间；
• 金额；
• 订单号；
• 人名；
• 业务热词。

单纯 CER 低，不代表业务可用。

---

4. RAG 低分时为什么要拒答？

因为客服场景里，错误回答的成本通常高于“不知道”。

如果知识库没有依据，继续让 LLM 自由发挥，可能会产生：

• 错误流程；
• 错误承诺；
• 错误价格；
• 错误政策；
• 错误售后结论。

所以本文逻辑是：

有可靠引用 → 回答
没有可靠引用 → 拒答或转人工

这不是让 AI 变笨，而是让它别乱开业务支票。

---

5. 用户打断后，已经生成但没播放完的内容要进上下文吗？

不要直接当作用户已听到内容。

比较稳妥的处理方式是：

已生成内容：可以记录为 assistant draft
已播放内容：可以记录为 assistant spoken
未播放内容：不要作为用户已确认事实

本文简化处理为：

用户打断了上一轮播报，上一轮未完整播放内容不进入用户已确认事实。

生产环境可以进一步记录“播放到第几个 chunk”。

---

6. TTS 首包延迟应该从哪里开始算？

建议从“LLM 产生可用于合成的文本”开始算。

如果从用户开始说话算，会把 ASR、RAG、LLM 的耗时也算进 TTS。
这样排查问题时会很混乱。

本文点位是：

t5_llm_done → t6_tts_first_chunk

如果后续做真正流式 LLM + 流式 TTS，可以改成：

t4_llm_first_sentence → t6_tts_first_chunk

---

7. 转人工应该什么时候触发？

本文只做了几类简单触发：

• 用户明确说转人工；
• 用户投诉；
• 涉及退款、合同等敏感问题；
• RAG 没有命中；
• 回复中建议转人工。

真实项目还可以增加：

• 连续两轮没有解决；
• 用户情绪升级；
• ASR 置信度过低；
• 业务规则要求人工确认；
• 涉及高风险操作；
• 用户身份验证失败。

转人工不是失败。
对 Voice Agent 来说，知道什么时候不要继续答，也是一种能力。

---

二十二、已知限制

这版 v1 还只是工程骨架，有几个明显限制。

1. 没有真实音频编解码

本文没有处理：

• PCM 转码；
• Opus 解码；
• 8 kHz / 16 kHz 重采样；
• 立体声转单声道；
• 音量归一化；
• 噪声抑制。

下一阶段如果接电话线路，这些会马上变成必修课。

---

2. VAD 很简化

本文 EnergyVAD 只是用非零字节比例判断是否有语音。

生产环境不建议这么做。
真实项目中可以替换为：

• WebRTC VAD；
• Silero VAD；
• 云厂商端点检测；
• ASR 自带端点检测；
• 多策略融合。

---

3. RAG 是本地简化版

本文 RAG 只适合验证流程，不适合生产。

生产环境建议至少支持：

• embedding 检索；
• hybrid search；
• rerank；
• 引用片段定位；
• 权限过滤；
• 多知识库路由；
• 拒答评测集。

---

4. 没有真实流式播放

本文用 bytes 模拟 TTS 音频 chunk，只统计 chunk 数量。
真实系统还需要处理：

• 音频格式；
• 播放器缓冲；
• 音频时长；
• 首包可播放；
• 队列积压；
• 用户打断时前端停止播放。

---

5. 没有并发压测

本文只做了单 session 的端到端验证。
真实服务还需要压测：

• 并发会话；
• 长连接稳定性；
• Provider 超时；
• 队列堆积；
• 内存泄漏；
• 服务重启恢复。

---

二十三、下一阶段优化方向

第一阶段到这里，Voice Agent 的骨架已经跑通。

下一阶段可以沿着四条线继续做。

1. 替换真实 ASR

重点不是“接上接口”这么简单，而是要评估：

• 电话 8 kHz 下的识别效果；
• 数字、地址、业务热词准确率；
• 噪声环境；
• 打断时 ASR 是否会误识别；
• 流式 partial result 是否稳定。

2. 替换真实 TTS

重点关注：

• 首包延迟；
• 分句策略；
• 音频队列积压；
• 打断取消；
• 播放残留；
• 语气是否适合客服场景。

3. 引入真实 WebRTC 或电话媒体流

可以选择两条路线：

浏览器 Demo：WebRTC + aiortc
电话 Demo：运营商线路 / 云通信媒体流

两条路线都要解决音频格式和实时性问题。

4. 做评测脚本

前面我们已经写过 ASR、RAG、TTS、延迟等单项评测。
现在可以开始做端到端评测：

• 一组标准用户问题；
• 一组打断测试；
• 一组转人工测试；
• 一组无答案拒答测试；
• 一组延迟统计；
• 一组连续运行稳定性测试。

只有评测脚本稳定，后面换模型才知道到底有没有变好。

---

二十四、完整测试流程回顾

从零开始跑这版 v1：

mkdir voice_agent_v1
cd voice_agent_v1

python -m venv .venv
source .venv/bin/activate

pip install -r requirements.txt

python -m py_compile *.py
python run_demo.py
pytest -q

uvicorn app:app --reload --host 0.0.0.0 --port 8000

健康检查：

curl http://127.0.0.1:8000/health

跑一轮对话：

curl -X POST http://127.0.0.1:8000/demo/turn \
  -H "Content-Type: application/json" \
  -d '{"session_id":"s1","text":"我想预约 Voice Agent 产品演示"}'

模拟打断：

curl -X POST http://127.0.0.1:8000/demo/interrupt/s1

触发转人工：

curl -X POST http://127.0.0.1:8000/demo/turn \
  -H "Content-Type: application/json" \
  -d '{"session_id":"s1","text":"我要投诉，帮我转人工"}'

如果能看到：

3 passed
state: IDLE
state: LISTENING
state: TRANSFERRING
handoff_ticket 不为空
latency_ms 有 t0-t8 汇总

说明这个 Voice Agent v1 的主链路已经跑通。

---

二十五、专栏串联

上一篇我们解决的是：

真实电话环境下怎么测 ASR，尤其是数字、地址、噪声和业务热词这类容易出错的内容。

本篇解决的是：

怎么把前面拆过的 WebRTC、VAD、ASR、RAG、TTS、打断、状态管理和转人工，整合成一个本地可运行的 Voice Agent v1。

下一篇开始，我会进入第二阶段：

在这个 v1 骨架上替换真实 ASR、TTS 和 WebRTC 音频流，并开始做端到端评测。

参考资料

• FastAPI 官方文档：WebSockets
  https://fastapi.tiangolo.com/advanced/websockets/

• FastAPI 官方文档：WebSocket Reference
  https://fastapi.tiangolo.com/reference/websockets/

• Python 官方文档：asyncio Queue
  https://docs.python.org/3/library/asyncio-queue.html

• Python 官方文档：Coroutines and Tasks
  https://docs.python.org/3/library/asyncio-task.html

• Python 官方文档：asyncio
  https://docs.python.org/3/library/asyncio.html

• MDN Web Docs：WebRTC API
  https://developer.mozilla.org/en-US/docs/Web/API/WebRTC_API

• MDN Web Docs：MediaDevices.getUserMedia()
  https://developer.mozilla.org/en-US/docs/Web/API/MediaDevices/getUserMedia

• WebRTC 官方文档：Getting started with media devices
  https://webrtc.org/getting-started/media-devices

• py-webrtcvad GitHub 仓库
  https://github.com/wiseman/py-webrtcvad