# VAD 检测与话轮切换完整实现

## TEN VAD 检测原理 (2026-09-17)

### 核心参数
```yaml
# config.yaml
vad:
  threshold: 0.3        # VAD 敏感度 (0-1)
  min_speech_ms: 300    # 最短语音 300ms (防止误触发)
  min_silence_ms: 500   # 最短静音 500ms (判断"说完")
```

### 检测流程
```
音频输入 (16kHz) -> 每16ms一帧 -> TEN VAD判断 -> is_speech?
    ├─ True -> 开始说话 -> 收集音频到 buffer
    └─ False -> 之前说话？
            ├─ 是 -> 记录静音时间 -> 静音>500ms? -> ✅用户说完
            │                              └─ 否 -> 继续监听
            └─ 否 -> 继续监听
```

### 性能数据
| 指标 | TEN VAD | Silero VAD |
|------|---------|------------|
| 单帧延迟 | **0.234ms** | 0.337ms |
| 帧粒度 | 256 samples (16ms) | 512 samples (32ms) |
| 库大小 | 306KB | 2.2MB |
| RTF | 0.0146 | ~0.02 |

### 代码实现
```python
class TenVAD:
    def __init__(self):
        self.threshold = 0.3
        self.min_speech_ms = 300
        self.min_silence_ms = 500
        self._is_speaking = False
        self._speech_start_time = 0.0
        self._silence_start_time = 0.0
        self._buffer = bytearray()
        self._vad = TenVad(hop_size=256, threshold=self.threshold)
    
    def process_frame(self, audio_data: bytes) -> Optional[bool]:
        audio_np = np.frombuffer(audio_data, dtype=np.int16)[:256]
        score, is_speech = self._vad.process(audio_np)
        
        now = time.time() * 1000
        
        if is_speech:
            if not self._is_speaking:
                # 刚开始说话
                self._is_speaking = True
                self._speech_start_time = now
                self._buffer = bytearray()
                return True  # 通知上层：用户开始说话
            else:
                # 正在说话，收集音频
                self._buffer.extend(audio_data)
                self._silence_start_time = 0
        else:
            # 检测到静音
            if self._is_speaking:
                # 之前一直在说话，现在停了
                if now - self._speech_start_time < self.min_speech_ms:
                    # 说话太短，不算
                    self._is_speaking = False
                    self._buffer = bytearray()
                    return False
                if self._silence_start_time == 0:
                    self._silence_start_time = now
                if now - self._silence_start_time >= self.min_silence_ms:
                    # 静音超过阈值，用户说完了
                    self._is_speaking = False
                    return False  # 通知上层：用户说完话了
        return None  # 继续监听
```

## ReplyOnPause 简化接口

### 核心思想
**"你只管写'收到语音后怎么处理'，它帮你搞定'什么时候该响应'"**

### 使用示例
```python
from fastrtc_style import ReplyOnPause

# 只需定义处理器
async def voice_handler(audio: bytes):
    text = await stt.recognize(audio)      # STT
    response = await llm.chat(text)        # LLM
    async for chunk in tts.synthesize_stream(response):
        yield chunk                        # TTS

# 一行启动
handler = ReplyOnPause(voice_handler)
await handler.start_streaming(audio_source)
```

### 与传统实现对比
| 特性 | 手动实现 (v5) | ReplyOnPause |
|------|--------------|-------------|
| 代码量 | 100+ 行 | ~20 行 |
| VAD 管理 | 手动 | ✅ 自动 |
| 状态机 | 手动 | ✅ 自动 |
| 静音判断 | 手动 | ✅ 自动 |
| 可控性 | 高（可调参数） | 低（黑盒） |
| 适用场景 | 生产/IoT | 快速原型 |

### 完整对比
```
手动实现 (v5):
  ┌─────────────────────────────────────┐
  │ 音频输入                            │
  │   ↓                                 │
  │ TEN VAD.process_frame()             │
  │   ↓                                 │
  │ if vad_result is True:              │
  │     state = LISTENING               │
  │ elif vad_result is False:           │
  │     buffer = vad.get_buffer()       │
  │     handle_speech(buffer)           │
  │       ↓                             │
  │     STT → LLM → TTS                 │
  └─────────────────────────────────────┘

ReplyOnPause:
  ┌─────────────────────────────────────┐
  │ 音频输入                            │
  │   ↓ (自动检测 VAD + 静音)           │
  │ 自动触发 voice_handler(audio)       │
  │   ↓                                 │
  │ STT → LLM → TTS                     │
  └─────────────────────────────────────┘
```

## Barge-in 三层架构

### 1. 回声抑制
```python
class EchoSuppressor:
    def is_echo(self, asr_text: str) -> bool:
        # 比对播放文本与 ASR 识别结果
        clean_play = self._current_speech.replace("。", "").strip()
        clean_asr = asr_text.replace("。", "").strip()
        return len(clean_asr) > 10 and clean_asr in clean_play
```

### 2. 语义判停
```python
class BargeInClassifier:
    async def classify(self, transcript: str) -> str:
        # 小模型快速分类：filler (嗯/对) vs real (真打断)
        # 延迟 < 200ms
        text = transcript.strip().lower()
        if text in {"嗯", "对", "好的", "明白"}:
            return "filler"
        # 调用轻量级分类器
        ...
```

### 3. 状态机兜底
```python
async def _cancel_routine(self):
    # 统一取消：清空播放队列、中止 TTS、重置 VAD
    await self.play_queue.clear()
    if self._tts_task:
        self._tts_task.cancel()
    self.vad.reset()
```

## 文件位置
- `agent.py`: 主 Agent 实现 (v5, 745行)
- `fastrtc_style.py`: ReplyOnPause 封装
- `config.yaml`: 配置文件