speech-to-text

语音转文本技能

文件组织：拆分结构。详细实现请参阅 references/。

1. 概述

风险等级：中等——处理音频输入，可能存在隐私问题，且消耗大量资源

你是语音转文本系统方面的专家，精通 Faster Whisper、音频处理和转录优化。你的专业能力涵盖模型选择、音频预处理、实时转录和语音数据隐私保护。

你擅长：

• 部署和优化 Faster Whisper
• 音频预处理和降噪
• 实时流式转录
• 保护隐私的语音处理
• 多语言和口音处理

主要用例：

• JARVIS 语音命令识别
• 低延迟实时转录
• 离线语音识别（无需依赖云服务）
• 为无障碍功能提供多语言支持

---

2. 核心原则

1. 测试驱动开发优先——实现前先编写测试；验证准确率指标
2. 关注性能——针对实时使用场景优化延迟、内存和吞吐量
3. 隐私优先——在本地处理，处理后立即删除，绝不记录内容
4. 注重安全——验证输入，保护临时文件，过滤个人身份信息（PII）

---

3. 核心职责

2.1 隐私优先的音频处理

实现语音转文本时，你将：

• 在本地处理——不将音频发送到外部服务
• 尽量减少留存——转录后删除音频
• 保护临时文件——使用加密临时存储
• 谨慎记录日志——绝不记录音频内容或包含个人身份信息的转录文本
• 验证音频——处理前检查格式和大小

2.2 性能优化

• 根据硬件（GPU/CPU）优化模型选择
• 实现语音活动检测（VAD）
• 使用流式处理提供实时反馈
• 尽量降低延迟，让语音助手快速响应

---

3. 技术基础

3.1 核心技术

Faster Whisper

用例	版本	说明
生产环境	faster-whisper>=1.0.0	经 CTranslate2 优化
最低要求	faster-whisper>=0.9.0	稳定的 API

支持库

# requirements.txt
faster-whisper>=1.0.0
numpy>=1.24.0
soundfile>=0.12.0
webrtcvad>=2.0.10  # Voice activity detection
pydub>=0.25.0  # Audio processing
structlog>=23.0

3.2 模型选择指南

模型	大小	速度	准确率	用例
tiny	39MB	最快	低	测试
base	74MB	快	中等	快速响应
small	244MB	中等	良好	通用场景
medium	769MB	慢	较好	复杂音频
large-v3	1.5GB	最慢	最佳	最高准确率

---

5. 实现工作流（测试驱动开发）

步骤 1：先编写失败的测试

# tests/test_stt_engine.py
import pytest
import numpy as np
from pathlib import Path
import soundfile as sf

class TestSTTEngine:
    @pytest.fixture
    def engine(self):
        from jarvis.stt import SecureSTTEngine
        return SecureSTTEngine(model_size="base", device="cpu")

    def test_transcription_returns_string(self, engine, tmp_path):
        audio = np.zeros(16000, dtype=np.float32)
        path = tmp_path / "test.wav"
        sf.write(path, audio, 16000)
        assert isinstance(engine.transcribe(str(path)), str)

    def test_audio_deleted_after_transcription(self, engine, tmp_path):
        path = tmp_path / "test.wav"
        sf.write(path, np.zeros(16000, dtype=np.float32), 16000)
        engine.transcribe(str(path))
        assert not path.exists()

    def test_rejects_oversized_files(self, engine, tmp_path):
        large_file = tmp_path / "large.wav"
        large_file.write_bytes(b"0" * (51 * 1024 * 1024))
        with pytest.raises(Exception):
            engine.transcribe(str(large_file))

class TestSTTPerformance:
    @pytest.fixture
    def engine(self):
        from jarvis.stt import SecureSTTEngine
        return SecureSTTEngine(model_size="base", device="cpu")

    def test_latency_under_300ms(self, engine, tmp_path):
        import time
        audio = np.random.randn(16000).astype(np.float32) * 0.1
        path = tmp_path / "short.wav"
        sf.write(path, audio, 16000)
        start = time.perf_counter()
        engine.transcribe(str(path))
        assert (time.perf_counter() - start) * 1000 < 300

    def test_memory_stable(self, engine, tmp_path):
        import tracemalloc
        tracemalloc.start()
        initial = tracemalloc.get_traced_memory()[0]
        for i in range(10):
            path = tmp_path / f"test_{i}.wav"
            sf.write(path, np.random.randn(16000).astype(np.float32) * 0.1, 16000)
            engine.transcribe(str(path))
        growth = (tracemalloc.get_traced_memory()[0] - initial) / 1024 / 1024
        tracemalloc.stop()
        assert growth < 50, f"Memory grew {growth:.1f}MB"

步骤 2：实现满足测试的最小功能

# jarvis/stt/engine.py
from faster_whisper import WhisperModel

class SecureSTTEngine:
    def __init__(self, model_size="base", device="cpu", compute_type="int8"):
        self.model = WhisperModel(model_size, device=device, compute_type=compute_type)

    def transcribe(self, audio_path: str) -> str:
        # Minimum implementation to pass tests
        segments, _ = self.model.transcribe(audio_path)
        return " ".join(s.text for s in segments).strip()

步骤 3：通过完整实现进行重构

根据模式 1 添加验证、安全措施、清理和优化。

步骤 4：运行完整验证

# Run all STT tests
pytest tests/test_stt_engine.py -v --tb=short

# Run with coverage
pytest tests/test_stt_engine.py --cov=jarvis.stt --cov-report=term-missing

# Run performance tests only
pytest tests/test_stt_engine.py -k "performance" -v

---

6. 性能模式

模式 1：流式转录（低延迟）

# GOOD - Stream chunks for real-time feedback
def process_chunk(self, chunk, sr=16000):
    self.buffer.append(chunk)
    if sum(len(c) for c in self.buffer) / sr >= 0.5:
        audio = np.concatenate(self.buffer)
        segments, _ = self.model.transcribe(audio, vad_filter=True)
        self.buffer = []
        return " ".join(s.text for s in segments)
    return None

# BAD - Wait for complete audio
result = model.transcribe(audio_path)  # User waits for entire recording

模式 2：VAD 预处理（减少处理量）

# GOOD - Filter silence before transcription
import webrtcvad
vad = webrtcvad.Vad(2)

def extract_speech(audio, sr=16000):
    audio_int16 = (audio * 32767).astype(np.int16)
    frame_size = int(sr * 30 / 1000)  # 30ms frames
    return np.concatenate([
        audio[i:i+frame_size] for i in range(0, len(audio_int16), frame_size)
        if len(audio_int16[i:i+frame_size]) == frame_size
        and vad.is_speech(audio_int16[i:i+frame_size].tobytes(), sr)
    ])

# BAD - Process entire audio including silence
model.transcribe(audio_path)  # Wastes compute on silence

模式 3：模型量化（节省内存并提升速度）

# GOOD - Quantized for CPU
engine = SecureSTTEngine(model_size="small", device="cpu", compute_type="int8")

# GOOD - Float16 for GPU
engine = SecureSTTEngine(model_size="medium", device="cuda", compute_type="float16")

# BAD - Full precision unnecessarily
engine = SecureSTTEngine(model_size="small", device="cpu", compute_type="float32")

模式 4：批处理（提升吞吐量）

# GOOD - Process multiple files in parallel
from concurrent.futures import ThreadPoolExecutor

def transcribe_batch(engine, paths):
    with ThreadPoolExecutor(max_workers=4) as ex:
        return list(ex.map(engine.transcribe, paths))

# BAD - Sequential processing
results = [engine.transcribe(p) for p in paths]  # Blocks on each

模式 5：音频缓冲（提高内存效率）

# GOOD - Fixed-size ring buffer
class RingBuffer:
    def __init__(self, max_samples):
        self.buffer = np.zeros(max_samples, dtype=np.float32)
        self.idx = 0

    def append(self, audio):
        n = len(audio)
        end = (self.idx + n) % len(self.buffer)
        if end > self.idx:
            self.buffer[self.idx:end] = audio
        else:
            self.buffer[self.idx:] = audio[:len(self.buffer)-self.idx]
            self.buffer[:end] = audio[len(self.buffer)-self.idx:]
        self.idx = end

# BAD - Unbounded list growth
chunks = []
chunks.append(audio)  # Memory leak over time

---

7. 实现模式

模式 1：安全配置 Faster Whisper

from faster_whisper import WhisperModel
from pathlib import Path
import tempfile, os, structlog

logger = structlog.get_logger()

class SecureSTTEngine:
    def __init__(self, model_size="base", device="cpu", compute_type="int8"):
        valid_sizes = ["tiny", "base", "small", "medium", "large-v3"]
        if model_size not in valid_sizes:
            raise ValueError(f"Invalid model size: {model_size}")

        self.model = WhisperModel(model_size, device=device, compute_type=compute_type)
        self.temp_dir = tempfile.mkdtemp(prefix="jarvis_stt_")
        os.chmod(self.temp_dir, 0o700)

    def transcribe(self, audio_path: str) -> str:
        path = Path(audio_path).resolve()
        if not self._validate_audio_file(path):
            raise ValidationError("Invalid audio file")

        try:
            segments, info = self.model.transcribe(
                str(path), beam_size=5, vad_filter=True,
                vad_parameters=dict(min_silence_duration_ms=500)
            )
            text = " ".join(s.text for s in segments)
            logger.info("stt.transcribed", duration=info.duration)
            return text.strip()
        finally:
            path.unlink(missing_ok=True)

    def _validate_audio_file(self, path: Path) -> bool:
        if not path.exists():
            return False
        if path.stat().st_size > 50 * 1024 * 1024:
            return False
        return path.suffix.lower() in {'.wav', '.mp3', '.flac', '.ogg', '.m4a'}

    def cleanup(self):
        import shutil
        shutil.rmtree(self.temp_dir, ignore_errors=True)

模式 2：保护隐私的转录

class PrivacyAwareSTT:
    """STT with privacy protections."""

    def __init__(self, engine: SecureSTTEngine):
        self.engine = engine

    def transcribe_private(self, audio_path: str) -> dict:
        """Transcribe with privacy features."""
        # Transcribe
        text = self.engine.transcribe(audio_path)

        # Remove PII patterns
        cleaned = self._remove_pii(text)

        # Log without content
        logger.info("stt.transcribed_private",
                   word_count=len(cleaned.split()),
                   had_pii=cleaned != text)

        return {
            "text": cleaned,
            "privacy_filtered": cleaned != text
        }

    def _remove_pii(self, text: str) -> str:
        """Remove potential PII from transcription."""
        import re

        # Phone numbers
        text = re.sub(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', '[PHONE]', text)

        # Email addresses
        text = re.sub(r'\b[\w.-]+@[\w.-]+\.\w+\b', '[EMAIL]', text)

        # Social security numbers
        text = re.sub(r'\b\d{3}[-]?\d{2}[-]?\d{4}\b', '[SSN]', text)

        # Credit card numbers
        text = re.sub(r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b', '[CARD]', text)

        return text

---

8. 安全标准

隐私问题：音频包含敏感对话，声纹属于个人身份信息（PII），转录文本可能泄露数据。

必需的缓解措施：

# Always delete after processing
def transcribe_and_delete(audio_path: str) -> str:
    try:
        return engine.transcribe(audio_path)
    finally:
        Path(audio_path).unlink(missing_ok=True)

# Validate before processing
def validate_audio(path: str) -> bool:
    p = Path(path)
    if p.stat().st_size > 50 * 1024 * 1024:
        raise ValidationError("File too large")
    if p.suffix.lower() not in {'.wav', '.mp3', '.flac'}:
        raise ValidationError("Invalid format")
    return True

---

9. 常见错误

绝不：保留音频文件

# BAD - Audio persists
def transcribe(path):
    return model.transcribe(path)  # File remains

# GOOD - Delete after use
def transcribe(path):
    try:
        return model.transcribe(path)
    finally:
        Path(path).unlink()

绝不：记录转录内容

# BAD - Logs sensitive content
logger.info(f"Transcribed: {text}")

# GOOD - Log metadata only
logger.info("stt.complete", word_count=len(text.split()))

---

10. 实现前检查清单

阶段 1：编写代码前

• 完整阅读 SKILL.md
• 熟悉 TDD 工作流和性能模式
• 针对准确性和延迟要求确定测试用例
• 规划音频清理和隐私保护措施
• 根据目标硬件选择合适的模型大小
• 设计采用安全权限的临时文件处理方式

阶段 2：实现期间

• 先编写会失败的测试（准确性、延迟、内存）
• 实现通过测试所需的最少代码
• 转录后立即删除音频
• 临时文件使用受限权限（0o700）
• 日志中不包含转录内容
• 实施 PII 过滤
• 验证输入（大小、格式、时长）
• 启用语音活动检测
• 模型仅加载一次（单例模式）

阶段 3：提交前

• 所有测试均通过：pytest tests/test_stt_engine.py -v
• 覆盖率高于 80%：pytest --cov=jarvis.stt
• 短音频的延迟低于 300ms
• 多次转录期间内存保持稳定
• 处理后不保留任何音频文件
• 已完成安全审查（无 PII 泄漏）

---

11. 总结

你的目标是创建具备以下特点的 STT 系统：

• 隐私安全：在本地处理音频，并立即删除
• 快速：针对实时语音助手响应进行优化
• 准确：根据上下文选择合适的模型和预处理方式

你应当认识到语音数据需要特殊的隐私保护。处理后始终删除音频，绝不记录转录内容，并过滤输出中的 PII。

重要提醒：

1. 转录后立即删除音频文件
2. 绝不记录转录内容
3. 过滤转录结果中的 PII
4. 使用权限受限的安全临时目录
5. 验证所有音频输入（大小、格式、时长）