# Milestone: Real STT + TTS Integration

> **日期：** 2026-03-16
> **阶段：** Phase 2 + Phase 4（合并完成）
> **状态：** ✅ 单元测试全部通过，等待实机验证

---

## 1. 概述

本次 milestone 将 openclaw-proxy 的 STT（语音识别）和 TTS（语音合成）从 mock 替换为真实实现，打通完整的语音链路：

```
ESP32 麦克风 → Opus 音频 → [STT: Whisper] → 文字
    → [LLM: OpenClaw CLI] → 回复文字
    → [TTS: edge-tts + ffmpeg] → Opus 音频 → ESP32 扬声器
```

**之前（Phase 0.5）：** STT 固定返回 `"测试语音输入"`，TTS 固定返回 159 字节静音帧。
**现在：** STT 使用 Whisper base 模型做真实中文语音识别，TTS 使用 edge-tts 合成语音并通过 ffmpeg 编码为 Opus。

---

## 2. 技术选型与理由

### 2.1 STT — OpenAI Whisper (base)

| 项目 | 选择 |
|------|------|
| 引擎 | `openai-whisper` 20250625 |
| 模型 | `base`（74M 参数，~140MB 磁盘） |
| 运行方式 | 本地 CPU 推理（无 GPU） |
| 语言 | 固定 `zh`（中文） |

**为什么选 Whisper base：**
- 4GB 内存 + 4 核 CPU 环境下，`base` 是精度和速度的最佳平衡点
- `tiny` 太不准确，`small` 在 CPU 上太慢（>10s/句）
- 本地运行，无需外部 API，无网络延迟
- Whisper 内置 ffmpeg 调用，能直接处理 Opus 格式输入

**不选 FunASR / SenseVoice 的原因：**
- 依赖更重，安装复杂
- Whisper 已满足演示需求，后续可替换

### 2.2 TTS — edge-tts + ffmpeg

| 项目 | 选择 |
|------|------|
| 合成引擎 | `edge-tts` 7.2.7（微软 Edge 在线 TTS） |
| 声音 | `zh-CN-XiaoxiaoNeural`（女声，自然度高） |
| 输出格式 | MP3（edge-tts 原生输出） |
| 编码转换 | `ffmpeg` → Opus（16kHz, mono, 24kbps） |

**为什么选 edge-tts：**
- 零成本、无需 API key
- 异步原生（`async for` 流式获取音频）
- 中文语音质量极佳（微软 Neural TTS）
- 依赖轻量，纯 Python + aiohttp

**不选 CosyVoice / GPT-SoVITS 的原因：**
- 需要 GPU 或大量 CPU 资源
- 部署复杂度高，不适合当前 4GB 内存环境

---

## 3. 架构变更

### 3.1 变更前后对比

```
变更前（Phase 0.5）：
┌─────────────┐     ┌──────────────────┐     ┌──────────────┐
│  ESP32      │────►│  process_stt()   │     │  OpenClaw    │
│  Opus bytes │     │  return 固定文字  │────►│  CLI         │
└─────────────┘     └──────────────────┘     └──────┬───────┘
                    ┌──────────────────┐            │
                    │  process_tts()   │◄───────────┘
                    │  return 静音帧   │
                    └──────────────────┘

变更后（本次 Milestone）：
┌─────────────┐     ┌──────────────────────────────────┐     ┌──────────────┐
│  ESP32      │────►│  process_stt()                   │     │  OpenClaw    │
│  Opus bytes │     │  tmpfile(.opus) → Whisper base   │────►│  CLI         │
└─────────────┘     │  → asyncio.to_thread() → 文本    │     └──────┬───────┘
                    └──────────────────────────────────┘            │
                    ┌──────────────────────────────────┐            │
                    │  process_tts()                   │◄───────────┘
                    │  edge-tts → MP3 → ffmpeg → Opus  │
                    └──────────────────────────────────┘
```

### 3.2 数据流详细说明

#### STT 链路：`process_stt(audio_bytes: bytes) -> str`

```
1. ESP32 发送原始 Opus 音频字节
2. 写入临时文件（NamedTemporaryFile, suffix=".opus"）
3. Whisper load_audio() 内部调用 ffmpeg 将 Opus 解码为 16kHz float32 PCM
4. Whisper base 模型推理（通过 asyncio.to_thread 在线程池执行，不阻塞事件循环）
5. 提取 result["text"]，strip() 去除首尾空白
6. 返回识别文本
```

**关键设计决策：**
- **Lazy loading：** Whisper 模型不在 `import` 时加载，而是首次调用 `process_stt()` 时通过 `_get_whisper_model()` 延迟加载。这样做的好处：
  - 测试不需要加载真实模型（mock `_get_whisper_model` 即可）
  - 服务启动快，模型加载延迟到第一个请求
  - 模型全局复用，只加载一次
- **asyncio.to_thread：** Whisper 推理是 CPU 密集型操作，直接在事件循环中运行会阻塞所有 WebSocket 连接。用 `to_thread` 将其放到线程池执行。
- **临时文件：** Whisper 的 `transcribe()` 接受文件路径（内部用 ffmpeg 解码），不接受 bytes。因此需要写临时文件。Linux 上 `NamedTemporaryFile(delete=True)` 在 `with` 块内文件可通过路径访问，退出后自动删除。

#### TTS 链路：`process_tts(text: str) -> bytes`

```
1. 创建 edge_tts.Communicate(text=text, voice="zh-CN-XiaoxiaoNeural")
2. async for 遍历 communicate.stream()，收集所有 type="audio" 的 chunk
3. 将 MP3 chunks 拼接为完整 MP3 字节
4. 启动 ffmpeg 子进程进行格式转换：
   ffmpeg -i pipe:0 -c:a libopus -b:a 24k -ar 16000 -ac 1 -f opus pipe:1
   - 输入：stdin 管道接收 MP3 字节
   - 输出：stdout 管道输出 Opus 字节
5. 检查 ffmpeg 返回码，非零则抛出 RuntimeError
6. 返回 Opus 字节
```

**关键设计决策：**
- **管道传输：** MP3 → Opus 转换通过 stdin/stdout 管道完成，无需写中间文件。
- **ffmpeg 参数：** `libopus` 编码器，24kbps 码率（语音足够），16kHz 采样率，单声道。与 ESP32 硬件解码能力匹配。
- **错误处理：** ffmpeg 失败时读取 stderr 日志并抛出 RuntimeError，上层可捕获处理。

---

## 4. 代码变更详情

### 4.1 修改的文件

| 文件 | 变更类型 | 说明 |
|------|----------|------|
| `main.py` | 修改 | 替换 mock STT/TTS 为真实实现 |
| `tests/test_main.py` | 修改 | 更新测试用例匹配新实现 |
| `requirements.txt` | 修改 | 新增 whisper 和 edge-tts 依赖 |

### 4.2 `main.py` 变更明细

#### 新增导入（第 1-14 行）

```python
import tempfile          # STT 临时文件
import edge_tts          # TTS 引擎
import whisper           # STT 引擎
```

#### 新增配置常量（第 29-31 行）

```python
WHISPER_MODEL_SIZE = "base"                    # Whisper 模型大小
EDGE_TTS_VOICE = "zh-CN-XiaoxiaoNeural"       # TTS 声音
FFMPEG_BIN = which("ffmpeg") or "ffmpeg"       # ffmpeg 可执行文件路径
```

#### 新增 Whisper 模型懒加载（第 66-77 行）

```python
_whisper_model: whisper.Whisper | None = None

def _get_whisper_model() -> whisper.Whisper:
    """Return the Whisper model, loading it on first call."""
    global _whisper_model
    if _whisper_model is None:
        logger.info("Loading Whisper '%s' model…", WHISPER_MODEL_SIZE)
        _whisper_model = whisper.load_model(WHISPER_MODEL_SIZE)
    return _whisper_model
```

#### 替换 `process_stt`（第 83-102 行）

**之前：** 固定返回 `"测试语音输入"`（2 行逻辑）
**之后：** 完整 Whisper 推理流程（12 行逻辑）

```python
async def process_stt(audio_bytes: bytes) -> str:
    logger.info("STT: received %d bytes of Opus audio", len(audio_bytes))
    model = _get_whisper_model()
    with tempfile.NamedTemporaryFile(suffix=".opus", delete=True) as f:
        f.write(audio_bytes)
        f.flush()
        result = await asyncio.to_thread(
            model.transcribe, f.name, language="zh"
        )
    text = result["text"].strip()
    logger.info("STT: transcribed → %s", text)
    return text
```

#### 替换 `process_tts`（第 147-187 行）

**之前：** 固定返回 `b"\xfc\xff\xfe" * 53` 静音帧（3 行逻辑）
**之后：** edge-tts 合成 + ffmpeg Opus 编码（25 行逻辑）

```python
async def process_tts(text: str) -> bytes:
    logger.info("TTS: synthesising → %s", text)

    # 1. edge-tts → MP3
    communicate = edge_tts.Communicate(text=text, voice=EDGE_TTS_VOICE)
    mp3_chunks: list[bytes] = []
    async for chunk in communicate.stream():
        if chunk["type"] == "audio":
            mp3_chunks.append(chunk["data"])
    mp3_bytes = b"".join(mp3_chunks)

    # 2. MP3 → Opus via ffmpeg
    proc = await asyncio.create_subprocess_exec(
        FFMPEG_BIN, "-i", "pipe:0",
        "-c:a", "libopus", "-b:a", "24k",
        "-ar", "16000", "-ac", "1",
        "-f", "opus", "pipe:1",
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    opus_bytes, stderr = await proc.communicate(input=mp3_bytes)

    if proc.returncode != 0:
        err_msg = stderr.decode(errors="replace").strip()
        logger.error("ffmpeg failed (rc=%d): %s", proc.returncode, err_msg)
        raise RuntimeError(f"ffmpeg TTS conversion error: {err_msg}")

    logger.info("TTS: generated %d bytes of Opus audio", len(opus_bytes))
    return opus_bytes
```

#### 未变更的部分

以下代码**完全未修改**：
- Pydantic 模型（`HardwareAction`, `HardwareEvent`, `StateNotification`, `TextReply`）
- `call_openclaw()` — 仍通过 CLI 调用 OpenClaw
- `handle_text_frame()` — 文本帧路由
- `handle_binary_frame()` — 二进制帧管线编排（STT → LLM → TTS 时序不变）
- `websocket_chat()` — WebSocket 端点
- `/health` — 健康检查
- 入口点 — uvicorn 启动

### 4.3 `tests/test_main.py` 变更明细

#### 新增导入

```python
from unittest.mock import MagicMock  # 用于 mock Whisper 模型（同步对象）
```

#### 替换的测试

| 旧测试 | 新测试 | 变更说明 |
|--------|--------|----------|
| `test_process_stt_returns_mock_text` | `test_process_stt_calls_whisper` | mock `_get_whisper_model`，验证临时文件、语言参数、文本 strip |
| `test_process_tts_returns_bytes` | `test_process_tts_returns_opus_bytes` | mock edge-tts stream + ffmpeg subprocess，验证 MP3 拼接和 Opus 输出 |

#### 新增的测试

| 测试 | 说明 |
|------|------|
| `test_process_tts_raises_on_ffmpeg_failure` | ffmpeg 返回非零退出码时应抛出 `RuntimeError` |

#### 修改的测试

| 测试 | 变更说明 |
|------|----------|
| `test_websocket_binary_frame_full_pipeline` | 从 mock subprocess 改为 mock 三个管线函数（`process_stt`, `call_openclaw`, `process_tts`），更好地隔离 WebSocket 集成测试 |

#### 未变更的测试（6 个）

- `test_call_openclaw_parses_cli_output`
- `test_call_openclaw_raises_on_cli_failure`
- `test_call_openclaw_raises_on_empty_payloads`
- `test_health_endpoint`
- `test_websocket_connect_and_disconnect`
- `test_websocket_text_frame_wake_word`
- `test_websocket_text_frame_malformed_json`

### 4.4 `requirements.txt` 变更

```diff
 fastapi>=0.115.0
 uvicorn[standard]>=0.34.0
 websockets>=14.0
 pydantic>=2.0
+openai-whisper>=20250625
+edge-tts>=7.0
 pytest>=8.0
 httpx>=0.28.0
 pytest-asyncio>=0.25.0
```

---

## 5. 测试结果

```
$ python -m pytest tests/ -v

tests/test_main.py::test_process_stt_calls_whisper PASSED            [  9%]
tests/test_main.py::test_call_openclaw_parses_cli_output PASSED      [ 18%]
tests/test_main.py::test_call_openclaw_raises_on_cli_failure PASSED  [ 27%]
tests/test_main.py::test_call_openclaw_raises_on_empty_payloads PASSED [ 36%]
tests/test_main.py::test_process_tts_returns_opus_bytes PASSED       [ 45%]
tests/test_main.py::test_process_tts_raises_on_ffmpeg_failure PASSED [ 54%]
tests/test_main.py::test_health_endpoint PASSED                      [ 63%]
tests/test_main.py::test_websocket_connect_and_disconnect PASSED     [ 72%]
tests/test_main.py::test_websocket_text_frame_wake_word PASSED       [ 81%]
tests/test_main.py::test_websocket_text_frame_malformed_json PASSED  [ 90%]
tests/test_main.py::test_websocket_binary_frame_full_pipeline PASSED [100%]

============================== 11 passed in 2.11s ==============================
```

**全部 11 个测试通过**（原 10 个 + 新增 1 个 ffmpeg 错误处理测试）。

---

## 6. 依赖清单

### 运行时依赖

| 包 | 版本 | 用途 | 安装大小 |
|----|------|------|----------|
| `openai-whisper` | 20250625 | 语音识别引擎 | ~10MB（代码） |
| `torch` | 2.10.0 | Whisper 推理后端 | ~2GB（含 CUDA stubs） |
| `edge-tts` | 7.2.7 | 微软 TTS 合成 | ~50KB |
| `aiohttp` | 3.13+ | edge-tts 的 HTTP 传输层 | ~1MB |

### 系统依赖

| 工具 | 版本 | 用途 |
|------|------|------|
| `ffmpeg` | 7.0.2 | Opus ↔ WAV/MP3 音频转码 |
| `libopus` | （含于 ffmpeg） | Opus 编解码器 |

### Whisper 模型文件

| 模型 | 参数量 | 磁盘 | 内存 | 首次加载 |
|------|--------|------|------|----------|
| `base` | 74M | ~140MB | ~300MB | 需联网下载到 `~/.cache/whisper/` |

---

## 7. 配置参数

所有配置集中在 `main.py` 顶部 Configuration 区域：

```python
# ── Configuration ────────────────────────────────────────────────────
OPENCLAW_CLI = which("openclaw") or "openclaw"    # OpenClaw CLI 路径
OPENCLAW_AGENT_ID = "main"                         # 使用的 agent ID
OPENCLAW_TIMEOUT = 120                             # CLI 超时（秒）
WHISPER_MODEL_SIZE = "base"                        # Whisper 模型：tiny/base/small/medium/large
EDGE_TTS_VOICE = "zh-CN-XiaoxiaoNeural"           # TTS 声音选择
FFMPEG_BIN = which("ffmpeg") or "ffmpeg"           # ffmpeg 路径
```

**可调参数说明：**

| 参数 | 可选值 | 影响 |
|------|--------|------|
| `WHISPER_MODEL_SIZE` | `tiny` / `base` / `small` / `medium` / `large` | 模型越大越准但越慢。`base` 在 4 核 CPU 上约 2-5 秒/句 |
| `EDGE_TTS_VOICE` | 见 `edge-tts --list-voices` | 可选其他中文声音如 `zh-CN-YunxiNeural`（男声）|
| ffmpeg `-b:a` | `16k` / `24k` / `32k` / `64k` | Opus 码率，`24k` 适合语音 |

---

## 8. 性能预估

基于 4 核 CPU / 4GB 内存 / 无 GPU 环境：

| 阶段 | 预估延迟 | 瓶颈 |
|------|----------|------|
| **首次 STT 调用** | 10-30 秒 | 下载 Whisper base 模型（~140MB，仅一次） |
| **STT 推理** | 2-5 秒/句 | CPU 推理，取决于音频长度 |
| **LLM（OpenClaw）** | 5-30 秒 | 取决于 agent 复杂度和外部 API |
| **TTS 合成** | 1-3 秒 | edge-tts 网络请求 + ffmpeg 转码 |
| **总延迟** | 8-38 秒 | 首次更长，后续稳定在 8-15 秒 |

**内存占用估算：**
- Whisper base 模型：~300MB
- FastAPI + Python 运行时：~200MB
- 峰值（推理中）：~500-600MB
- 剩余可用：~3.4GB（充裕）

---

## 9. 已知限制与风险

### 9.1 当前限制

| 限制 | 说明 | 影响 |
|------|------|------|
| **无音频缓冲** | 当前每个二进制帧独立触发完整管线 | ESP32 按住说话时，每个音频片段会分别处理而非合并。需要 Phase 1 的 `AudioBuffer` 实现 |
| **无分块推流** | TTS 音频一次性整包发送 | 长句子的音频可能有数十 KB，ESP32 需要一次性接收全部数据后才开始播放 |
| **单并发限制** | Whisper 推理会占用 CPU，多个同时请求会排队 | 当前单用户场景下不是问题 |
| **网络依赖** | edge-tts 需要访问微软服务器 | 服务器需有外网访问能力 |

### 9.2 风险项

| 风险 | 概率 | 缓解措施 |
|------|------|----------|
| Whisper 模型首次下载失败 | 中 | 可手动下载放到 `~/.cache/whisper/`，或预热 |
| edge-tts 微软服务不稳定 | 低 | 可降级为本地 TTS 或重试 |
| ESP32 发送的 Opus 格式不兼容 | 中 | ffmpeg 通常能处理，如不行需检查 ESP32 编码参数 |
| 长音频导致 Whisper OOM | 低 | base 模型在 4GB 环境处理 <30 秒音频没问题 |

---

## 10. 实机测试计划

### 10.1 前置准备

```bash
# 1. 确保服务器环境就绪
source venv/bin/activate
python -m pytest tests/ -v          # 所有测试通过
ffmpeg -version                      # 确认 ffmpeg 可用
python -c "import whisper; print(whisper.__version__)"   # 确认 whisper 可导入

# 2. 预热 Whisper 模型（避免首次调用等待下载）
python -c "import whisper; whisper.load_model('base')"

# 3. 确认 OpenClaw gateway 运行中
openclaw gateway status              # 或检查 :18789 端口

# 4. 启动 proxy
python main.py
```

### 10.2 模拟测试（无需 ESP32）

```bash
# 用 test_client.py 进行基本连通测试
python test_client.py
```

预期输出：
```
=== Connected ===

[TX] wake_word
[RX] {'type': 'state', 'status': 'listening'}

[TX] binary audio (80 bytes of mock Opus)
[RX] {'type': 'state', 'status': 'thinking'}
[RX] {'type': 'text_reply', 'text': '...（OpenClaw 回复）'}
[RX] binary audio: XXXX bytes

=== Done ===
```

> **注意：** `test_client.py` 发送的是 80 字节零填充 mock 数据，Whisper 可能识别为空或乱码。这是正常的——真实 Opus 音频才能产生有意义的识别结果。

### 10.3 ESP32 实机测试检查清单

- [ ] ESP32 小智固件已配置 WebSocket 地址为 `ws://<服务器IP>:8000/chat`
- [ ] 服务器防火墙已开放 8000 端口
- [ ] 按下按钮 → 服务器日志显示 `Hardware event: key_down`
- [ ] 对着麦克风说中文 → 服务器日志显示 `STT: transcribed → <识别的文字>`
- [ ] 松开按钮 → 服务器日志显示 `STT: received N bytes` 和 `LLM: sending to OpenClaw`
- [ ] ESP32 扬声器播放 TTS 合成的语音回复
- [ ] 多轮对话：连续提问，agent 能记住上下文（同一 session_id）

### 10.4 日志关键字速查

正常流程的日志序列：
```
STT: received XXXX bytes of Opus audio
Loading Whisper 'base' model…              ← 仅首次出现
STT: transcribed → 你好                    ← 识别结果
LLM: sending to OpenClaw → 你好
LLM: reply ← 你好！有什么我可以帮忙的？
TTS: synthesising → 你好！有什么我可以帮忙的？
TTS: generated XXXX bytes of Opus audio
Pushing XXXX bytes of audio to client
```

异常情况日志：
```
ffmpeg failed (rc=1): ...                  ← TTS ffmpeg 转码失败
OpenClaw CLI failed (rc=N): ...            ← LLM CLI 调用失败
```

---

## 11. 文件最终状态

### `main.py`（309 行）

```
第 1-14 行    — 导入（新增 tempfile, whisper, edge_tts）
第 16-22 行   — 日志配置
第 24-31 行   — 配置常量（新增 WHISPER_MODEL_SIZE, EDGE_TTS_VOICE, FFMPEG_BIN）
第 34-63 行   — Pydantic 模型（未变）
第 66-77 行   — Whisper 模型懒加载（新增）
第 80-102 行  — process_stt()（真实 Whisper 实现）
第 105-144 行 — call_openclaw()（未变）
第 147-187 行 — process_tts()（真实 edge-tts + ffmpeg 实现）
第 190-261 行 — 帧路由 + 管线编排（未变）
第 264-309 行 — FastAPI 应用 + 入口点（未变）
```

### `tests/test_main.py`（212 行）

```
11 个测试用例：
  - 1 个 STT 单元测试（重写）
  - 3 个 LLM 单元测试（未变）
  - 2 个 TTS 单元测试（1 个重写 + 1 个新增）
  - 5 个集成测试（1 个修改 mock 策略，4 个未变）
```

### `requirements.txt`（9 行）

```
新增 2 个依赖：openai-whisper, edge-tts
```

---

## 12. 与 PLAN.md 阶段映射

| PLAN.md 阶段 | 状态 | 本次完成内容 |
|-------------|------|-------------|
| Phase 0 — Mock 骨架 | ✅ 已完成 | — |
| Phase 0.5 — OpenClaw 对接 | ✅ 已完成 | — |
| Phase 1 — 音频缓冲 | ⏳ 待做 | — |
| **Phase 2 — 真实 STT** | **✅ 本次完成** | **Whisper base 模型接入** |
| Phase 3 — OpenClaw 对接 | ✅ 已在 0.5 完成 | — |
| **Phase 4 — 真实 TTS** | **✅ 本次完成** | **edge-tts + ffmpeg Opus 编码** |
| Phase 5 — 生产加固 | ⏳ 待做 | — |

**下一步：** Phase 1（音频缓冲），支持 ESP32 的 press-to-talk 交互模式——用户按住按键持续说话，松开后将完整音频合并送入管线。

---

## 13. 快速验证命令

```bash
# 进入虚拟环境
source venv/bin/activate

# 跑单元测试（应全部通过）
python -m pytest tests/ -v

# 启动服务（另一个终端）
python main.py

# 模拟客户端测试
python test_client.py
```