# LiveKit Voice AI - 分段测试计划

## 📋 测试总览

| 阶段 | 内容 | 预计时间 | 状态 |
|------|------|----------|------|
| **Phase 1** | 单组件测试 | 30min | ✅ 已完成 (7/7) |
| **Phase 2** | 集成测试 | 1h | ⏳ 待开始 |
| **Phase 3** | 打断逻辑测试 | 1.5h | ⏳ 待开始 |
| **Phase 4** | 真实音频流测试 | 2h | ⏳ 待开始 |
| **Phase 5** | LiveKit Room 集成 | 2h | ⏳ 待开始 |
| **Phase 6** | 移动端测试 | 3h | ⏳ 待开始 |

---

## Phase 1: 单组件测试 ✅

### 已完成
- [x] EventBus 发布/订阅
- [x] STT 组件 (FunASR HTTP)
- [x] LLM 组件 (Agnes AI)
- [x] TTS 组件 (Edge TTS)
- [x] VAD 组件 (TEN VAD 框架)
- [x] Memory 组件 (SQLite + 事实提取)
- [x] Agent Worker 集成

### 测试结果
```
📊 测试结果: 7/7 通过
```

### 已知问题
- [ ] TEN VAD 未安装：`pip install ten-vad`
- [ ] Agnes AI API Key 失效：需更新 `.env.dev`

---

## Phase 2: 集成测试 ⏳

### 目标
验证各组件协同工作，数据流完整。

### 测试用例

#### 2.1 STT → LLM Pipeline
```python
# 测试：语音输入 → 文字 → LLM 回复
async def test_stt_llm_pipeline():
    # 1. 模拟音频输入
    audio = generate_test_audio(duration=3.0)  # 3秒音频
    
    # 2. STT 转录
    text = await stt.transcribe(audio)
    assert len(text) > 0, "STT 未返回文本"
    
    # 3. LLM 生成回复
    response = await llm.generate(text)
    assert len(response) > 0, "LLM 未返回回复"
    
    print(f"✅ STT → LLM 通过: '{text}' → '{response[:50]}...'")
```

**前置条件**:
- [ ] TEN VAD 安装
- [ ] Agnes AI API Key 有效

**验收标准**:
- [ ] STT 准确识别测试音频
- [ ] LLM 返回有意义回复
- [ ] 端到端延迟 < 3s

---

#### 2.2 LLM → TTS Pipeline
```python
# 测试：LLM 回复 → TTS 音频
async def test_llm_tts_pipeline():
    # 1. LLM 生成文本
    text = await llm.generate("你好，介绍一下你自己")
    
    # 2. TTS 合成音频
    audio_url = await tts.synthesize(text)
    assert os.path.exists(audio_url), "TTS 未生成音频文件"
    
    # 3. 验证音频有效
    duration = get_audio_duration(audio_url)
    assert duration > 0, "音频时长异常"
    
    print(f"✅ LLM → TTS 通过: {len(text)} 字符 → {duration:.1f}s 音频")
```

**验收标准**:
- [ ] TTS 生成有效音频文件
- [ ] 音频格式正确 (WAV/MP3)
- [ ] 时长与文本长度匹配

---

#### 2.3 记忆系统测试
```python
# 测试：事实提取 + 关系推断
async def test_memory_extraction():
    # 1. 注入对话
    dialogues = [
        "我叫 Leo，我喜欢编程",
        "Leo 今天写了 100 行代码",
        "Leo 是人工智能爱好者"
    ]
    
    # 2. 触发记忆提取
    for d in dialogues:
        await memory.process_dialogue(d)
    
    # 3. 验证事实存储
    facts = await memory.get_facts("leo")
    assert len(facts) >= 2, f"期望 ≥2 条事实，实际 {len(facts)} 条"
    
    # 4. 验证关系强度
    relationships = await memory.get_relationships("leo")
    assert any(r["strength"] > 0 for r in relationships), "关系强度异常"
    
    print(f"✅ 记忆提取通过: {len(facts)} 条事实")
```

**验收标准**:
- [ ] 至少提取 2 条事实
- [ ] 关系强度合理递增
- [ ] change_log 记录完整

---

## Phase 3: 打断逻辑测试 ⏳

### 目标
验证三层打断机制正常工作。

### 测试用例

#### 3.1 状态机打断
```python
# 测试：AudioPlayQueue.interrupt()
async def test_state_machine_interrupt():
    # 1. 开始播放
    await audio_queue.play(generate_test_audio(5.0))
    assert audio_queue.is_playing, "音频未播放"
    
    # 2. 发送打断信号
    await event_bus.publish(EventType.INTERRUPT, {})
    
    # 3. 验证打断成功
    assert not audio_queue.is_playing, "打断后仍在播放"
    assert len(audio_queue.queue) == 0, "队列未清空"
    
    print("✅ 状态机打断通过")
```

**验收标准**:
- [ ] 播放立即停止
- [ ] 队列完全清空
- [ ] 无内存泄漏

---

#### 3.2 VAD 触发打断
```python
# 测试：新用户语音检测 → 触发打断
async def test_vad_interrupt():
    # 1. 播放音频期间，注入新用户语音
    await audio_queue.play(generate_test_audio(5.0))
    
    # 2. 模拟 VAD 检测到新语音
    vad_event = {"is_speech": True, "timestamp": time.time()}
    await event_bus.publish(EventType.VOICE_DETECTED, vad_event)
    
    # 3. 验证打断触发
    await asyncio.sleep(0.1)  # 等待事件处理
    assert agent_worker._interrupt_requested, "打断请求未设置"
    
    print("✅ VAD 打断通过")
```

**前置条件**:
- [ ] TEN VAD 已安装并运行

**验收标准**:
- [ ] 打断请求被设置
- [ ] 音频队列被清空
- [ ] LLM 生成被取消

---

#### 3.3 完整打断流程
```python
# 测试：用户说话 → VAD → STT → 打断 → 重新处理
async def test_full_interrupt_flow():
    # 1. Agent 正在播放回复
    task = asyncio.create_task(agent_worker.run())
    await asyncio.sleep(0.5)  # 等待启动
    
    # 2. 模拟用户打断
    interrupt_audio = generate_test_audio(2.0)
    await event_bus.publish(EventType.AUDIO_FRAME, interrupt_audio)
    
    # 3. 等待打断处理
    await asyncio.sleep(0.5)
    
    # 4. 验证状态重置
    assert agent_worker._interrupt_requested == False, "打断状态未重置"
    assert audio_queue.is_playing == False, "音频仍在播放"
    
    print("✅ 完整打断流程通过")
```

---

## Phase 4: 真实音频流测试 ⏳

### 目标
使用真实麦克风输入和扬声器输出测试。

### 测试用例

#### 4.1 录音质量测试
```python
# 测试：麦克风录音 → VAD 检测
async def test_mic_recording():
    # 1. 开始录音
    recorder = AudioRecorder(sample_rate=16000)
    await recorder.start()
    
    # 2. 录制 5 秒
    await asyncio.sleep(5.0)
    audio_data = await recorder.stop()
    
    # 3. 验证数据质量
    assert len(audio_data) > 0, "录音为空"
    assert audio_data.max() > 0.01, "录音音量过低"
    
    # 4. VAD 检测
    speech_segments = await vad.detect_speech(audio_data)
    assert len(speech_segments) > 0, "未检测到语音"
    
    print(f"✅ 录音测试通过: {len(audio_data)} samples, {len(speech_segments)} 段语音")
```

**前置条件**:
- [ ] PyAudio 安装
- [ ] 麦克风可用

**验收标准**:
- [ ] 录音清晰无噪音
- [ ] VAD 正确检测语音段落
- [ ] 无爆音/失真

---

#### 4.2 播放质量测试
```python
# 测试：TTS 音频 → 扬声器播放
async def test_speaker_playback():
    # 1. TTS 生成
    text = "这是一段测试音频"
    audio_path = await tts.synthesize(text)
    
    # 2. 播放音频
    player = AudioPlayer()
    await player.play(audio_path)
    
    # 3. 验证播放完成
    await asyncio.sleep(3.0)  # 等待播放
    assert not player.is_playing, "播放器未停止"
    
    print("✅ 播放测试通过")
```

**验收标准**:
- [ ] 音频清晰无杂音
- [ ] 播放完整无中断
- [ ] 音量适中

---

#### 4.3 端到端延迟测试
```python
# 测试：说话 → STT → LLM → TTS → 播放
async def test_end_to_end_latency():
    start_time = time.time()
    
    # 1. 语音输入（模拟）
    text_input = "你好"
    
    # 2. LLM + TTS（并行启动）
    llm_task = asyncio.create_task(llm.generate(text_input))
    tts_task = asyncio.create_task(tts.synthesize(await llm_task))
    
    # 3. 计时
    total_latency = time.time() - start_time
    
    # 4. 验证延迟在可接受范围
    assert total_latency < 5.0, f"延迟过高: {total_latency:.2f}s"
    
    print(f"✅ 端到端延迟: {total_latency:.2f}s")
```

**验收标准**:
- [ ] 总延迟 < 5s
- [ ] STT + LLM + TTS 各自 < 2s
- [ ] 无阻塞超时

---

## Phase 5: LiveKit Room 集成 ⏳

### 目标
测试真实的 WebRTC 音频传输。

### 测试用例

#### 5.1 Room 连接测试
```python
# 测试：连接到 LiveKit Room
async def test_livekit_room():
    # 1. 创建 Room
    room = Room("test-room")
    await room.connect(
        url="http://localhost:7880",
        token=generate_token("test-user"),
        auto_subscribe=True
    )
    
    # 2. 验证连接
    assert room.is_connected, "Room 未连接"
    assert len(room.participants) >= 1, "无参与者"
    
    # 3. 接收音频轨道
    audio_track = None
    for track in room.local Participant.track Publikations:
        if track.kind == TrackKind.AUDIO:
            audio_track = track
            break
    
    assert audio_track is not None, "无音频轨道"
    
    print(f"✅ Room 连接通过: {room.name}, {len(room.participants)} 参与者")
```

**前置条件**:
- [ ] LiveKit Server 运行中
- [ ] Token 生成器可用

**验收标准**:
- [ ] Room 连接成功
- [ ] 音频轨道可接收
- [ ] 无丢包/延迟

---

#### 5.2 Agent Worker Room 集成
```python
# 测试：Agent Worker 加入 Room
async def test_agent_in_room():
    # 1. 启动 Agent Worker
    await agent_worker.start()
    
    # 2. 加入 Room
    await agent_worker.join_room("test-room")
    
    # 3. 监听音频输入
    audio_frames = []
    agent_worker.on_audio_frame = lambda frame: audio_frames.append(frame)
    
    # 4. 模拟 Room 音频
    mock_audio = np.random.randint(-32768, 32767, 16000).astype(np.int16)
    await agent_worker.on_audio_received(mock_audio)
    
    # 5. 验证处理
    await asyncio.sleep(1.0)
    assert len(audio_frames) > 0, "未接收到音频帧"
    
    print(f"✅ Agent in Room 通过: 接收 {len(audio_frames)} 帧")
```

---

#### 5.3 多参与者测试
```python
# 测试：多个参与者 + Agent 协作
async def test_multi_participant():
    participants = []
    
    # 1. 创建 3 个参与者
    for i in range(3):
        p = await create_participant(f"user-{i}")
        participants.append(p)
    
    # 2. Agent 加入
    await agent_worker.join_room("test-room")
    
    # 3. 模拟对话轮转
    for p in participants:
        await p.say_hello()
        await asyncio.sleep(0.5)
    
    # 4. 验证 Agent 响应
    responses = await agent_worker.get_responses()
    assert len(responses) >= 1, "Agent 未响应"
    
    print(f"✅ 多参与者测试通过: {len(participants)} 用户, {len(responses)} 响应")
```

---

## Phase 6: 移动端测试 ⏳

### 目标
验证 React Native / Expo 客户端功能。

### 测试用例

#### 6.1 客户端连接测试
```typescript
// 测试：移动端连接到 LiveKit Room
async function test_mobile_connection() {
  // 1. 创建连接
  const room = new Room();
  await room.connect('http://localhost:7880', generateToken());
  
  // 2. 验证订阅
  assert(room.isConnected, '未连接');
  assert(room.participants.size > 0, '无参与者');
  
  console.log('✅ 移动端连接通过');
}
```

**前置条件**:
- [ ] Expo 开发服务器运行
- [ ] 移动端应用已构建

---

#### 6.2 语音通话测试
```typescript
// 测试：移动端语音输入 → Agent 响应
async function test_voice_call() {
  // 1. 开始录音
  await Audio.requestPermissionsAsync();
  await Audio.setAudioModeAsync({ playsSilentModeDuringRecording: true });
  
  const recording = new Audio.Recording();
  await recording.prepareToRecordAsync(recordingPresets);
  await recording.startAsync();
  
  // 2. 说话 3 秒
  await sleep(3000);
  await recording.stopAndUnloadAsync();
  
  // 3. 验证音频数据
  const uri = recording.getURI();
  const fileInfo = await FileSystem.getInfoAsync(uri);
  assert(fileInfo.exists, '录音失败');
  
  console.log('✅ 语音通话测试通过');
}
```

---

#### 6.3 打断交互测试
```typescript
// 测试：移动端打断 Agent 播放
async function test_mobile_interrupt() {
  // 1. Agent 正在播放
  const isPlaying = await agentWorker.isPlaying();
  assert(isPlaying, 'Agent 未播放');
  
  // 2. 移动端说话（触发打断）
  const interruptDetected = await simulateSpeechInterrupt();
  assert(interruptDetected, '未检测到打断');
  
  // 3. 验证停止
  const stillPlaying = await agentWorker.isPlaying();
  assert(!stillPlaying, '打断后仍在播放');
  
  console.log('✅ 打断交互测试通过');
}
```

---

## 📊 测试优先级

| 优先级 | 测试项 | 阻塞因素 |
|--------|--------|----------|
| **P0** | Phase 1 已完成 | - |
| **P0** | Phase 2.1-2.3 集成测试 | TEN VAD、API Key |
| **P1** | Phase 3 打断逻辑 | - |
| **P1** | Phase 4.1-4.3 真实音频 | PyAudio、麦克风 |
| **P2** | Phase 5 LiveKit 集成 | Room 服务器 |
| **P2** | Phase 6 移动端 | Expo 构建 |

---

## 🔧 测试环境准备

### 必做
```bash
# 1. 安装 TEN VAD
~/s2s-env/bin/pip install ten-vad

# 2. 更新 API Key
echo "OPENAI_API_KEY=[REDACTED]" > .env.dev
echo "OPENAI_BASE_URL=https://api.agnes-ai.cn/v1" >> .env.dev
```

### 可选
```bash
# 3. 安装 PyAudio（真实音频测试）
brew install portaudio
~/s2s-env/bin/pip install pyaudio

# 4. 启动 FunASR 服务
./scripts/start_funasr.sh

# 5. 启动 LiveKit Server
docker compose up -d livekit-server
```

---

## 📈 测试进度追踪

- [x] Phase 1: 单组件测试 (7/7 通过)
- [ ] Phase 2: 集成测试 (0/3)
- [ ] Phase 3: 打断逻辑 (0/3)
- [ ] Phase 4: 真实音频流 (0/3)
- [ ] Phase 5: LiveKit Room (0/3)
- [ ] Phase 6: 移动端 (0/3)

**总体进度**: 1/6 完成 (17%)