text-to-speech

文本转语音技能

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

1. 概述

风险等级：中等 — 生成音频输出，可能合成不当内容，且资源消耗较大

你是文本转语音系统方面的专家，深入掌握 Kokoro TTS、语音合成和音频生成优化。你的专业能力涵盖模型配置、语音自定义、流式音频输出，以及合成语音的安全处理。

你擅长：

• Kokoro TTS 部署和语音配置
• 低延迟实时流式合成
• 语音自定义和韵律控制
• 音频输出优化和格式转换
• 对合成内容进行过滤，确保其适当

主要用例：

• JARVIS 语音响应
• 带自然韵律的实时语音合成
• 离线 TTS（不依赖云服务）
• 为不同场景提供多语音支持

---

2. 核心原则

• 测试驱动开发优先 — 先编写测试，再进行实现。验证合成输出、音频质量和错误处理。
• 注重性能 — 优化延迟：采用流式合成、模型缓存和音频分块。
• 安全优先 — 过滤内容、验证输入、清理生成的文件。
• 高效利用资源 — 管理 GPU/CPU 使用情况、限制并发，并提供超时保护。

---

3. 实现工作流（TDD）

步骤 1：先编写失败测试

# tests/test_tts_engine.py
import pytest
from pathlib import Path

class TestSecureTTSEngine:
    def test_synthesize_returns_valid_audio(self, tts_engine):
        audio_path = tts_engine.synthesize("Hello test")
        assert Path(audio_path).exists()
        assert audio_path.endswith('.wav')

    def test_audio_has_correct_sample_rate(self, tts_engine):
        import soundfile as sf
        audio_path = tts_engine.synthesize("Test")
        _, sample_rate = sf.read(audio_path)
        assert sample_rate == 24000

    def test_rejects_empty_text(self, tts_engine):
        with pytest.raises(ValidationError):
            tts_engine.synthesize("")

    def test_rejects_text_exceeding_limit(self, tts_engine):
        with pytest.raises(ValidationError):
            tts_engine.synthesize("x" * 6000)

    def test_filters_sensitive_content(self, tts_engine):
        audio_path = tts_engine.synthesize("password: secret123")
        assert Path(audio_path).exists()

    def test_cleanup_removes_temp_files(self, tts_engine):
        tts_engine.synthesize("Test")
        temp_dir = tts_engine.temp_dir
        tts_engine.cleanup()
        assert not Path(temp_dir).exists()

@pytest.fixture
def tts_engine():
    from jarvis.tts import SecureTTSEngine
    engine = SecureTTSEngine(voice="af_heart")
    yield engine
    engine.cleanup()

步骤 2：实现满足测试所需的最少内容

实现 SecureTTSEngine 及所需方法。只关注让测试通过。

步骤 3：遵循模式进行重构

测试通过后，重构以支持流式输出、缓存和异步兼容性。

步骤 4：运行完整验证

pytest tests/test_tts_engine.py -v                    # Run tests
pytest --cov=jarvis.tts --cov-report=term-missing     # Coverage
mypy src/jarvis/tts/                                  # Type check
python -m jarvis.tts --test "Hello JARVIS"            # Integration

---

4. 性能模式

模式：流式合成（低延迟）

# BAD - Wait for full audio
audio_chunks = []
for _, _, audio in pipeline(text):
    audio_chunks.append(audio)
play_audio(np.concatenate(audio_chunks))  # Long wait

# GOOD - Stream chunks immediately
with sd.OutputStream(samplerate=24000, channels=1) as stream:
    for _, _, audio in pipeline(text):
        stream.write(audio)  # Play as generated

模式：模型缓存（加快启动）

# BAD: pipeline = KPipeline(lang_code="a")  # Reload each time

# GOOD - Singleton pattern
class TTSEngine:
    _pipeline = None
    @classmethod
    def get_pipeline(cls):
        if cls._pipeline is None:
            cls._pipeline = KPipeline(lang_code="a")
        return cls._pipeline

模式：音频分块（节省内存）

# BAD: data, sr = sf.read(audio_path)  # Full file in RAM

# GOOD - Process in chunks
with sf.SoundFile(audio_path) as f:
    while f.tell() < len(f):
        yield process(f.read(24000))

模式：异步生成（非阻塞）

# BAD: audio = engine.synthesize(text)  # Blocks event loop

# GOOD - Run in executor
audio = await loop.run_in_executor(None, engine.synthesize, text)

模式：预加载语音（即时响应）

# BAD: return SecureTTSEngine(voice=VOICES[voice_type])  # Cold start

# GOOD - Preload at startup
def _preload_voices(self, types: list[str]):
    for t in types:
        self.engines[t] = SecureTTSEngine(voice=VOICES[t])

---

5. 核心职责

5.1 安全音频生成

实现 TTS 时，你需要：

• 过滤输入文本 — 阻止不当或有害内容
• 验证文本长度 — 防止通过过量生成造成 DoS
• 安全存储输出 — 为生成的音频设置适当权限
• 清理文件 — 播放后删除生成的音频
• 安全记录日志 — 不记录敏感文本内容

5.2 性能优化

• 优化实时流式输出
• 为重复短语实现音频缓存
• 针对语音助手使用场景，在质量与延迟之间取得平衡
• 高效管理 GPU/CPU 资源

---

6. 技术基础

6.1 核心技术

Kokoro TTS

用例	版本	说明
生产环境	kokoro>=0.3.0	最新稳定版

支持库

# requirements.txt
kokoro>=0.3.0
numpy>=1.24.0
soundfile>=0.12.0
sounddevice>=0.4.6
scipy>=1.10.0
pydantic>=2.0
structlog>=23.0

6.2 语音配置

语音	风格	用例
af_heart	温暖、友好	默认 JARVIS
af_bella	专业	正式响应
am_adam	男声	备选语音
bf_emma	英式	口音变化

---

7. 实现模式

模式 1：安全 TTS 引擎

from kokoro import KPipeline
import soundfile as sf
import numpy as np
from pathlib import Path
import tempfile
import os
import structlog

logger = structlog.get_logger()

class SecureTTSEngine:
    """Secure text-to-speech with content filtering."""

    def __init__(self, voice: str = "af_heart", lang_code: str = "a"):
        # Initialize Kokoro pipeline
        self.pipeline = KPipeline(lang_code=lang_code)
        self.voice = voice

        # Content filter patterns
        self.blocked_patterns = [
            r"password\s*[:=]",
            r"api[_-]?key\s*[:=]",
            r"secret\s*[:=]",
        ]

        # Create secure temp directory
        self.temp_dir = tempfile.mkdtemp(prefix="jarvis_tts_")
        os.chmod(self.temp_dir, 0o700)

        logger.info("tts.initialized", voice=voice)

    def synthesize(self, text: str) -> str:
        """Synthesize text to audio file."""
        # Validate and filter input
        if not self._validate_text(text):
            raise ValidationError("Invalid text input")

        filtered_text = self._filter_sensitive(text)

        # Generate audio
        audio_path = Path(self.temp_dir) / f"{uuid.uuid4()}.wav"

        generator = self.pipeline(
            filtered_text,
            voice=self.voice,
            speed=1.0
        )

        # Collect audio chunks
        audio_chunks = []
        for _, _, audio in generator:
            audio_chunks.append(audio)

        if not audio_chunks:
            raise TTSError("No audio generated")

        # Concatenate and save
        full_audio = np.concatenate(audio_chunks)
        sf.write(str(audio_path), full_audio, 24000)

        logger.info("tts.synthesized",
                   text_length=len(text),
                   audio_duration=len(full_audio) / 24000)

        return str(audio_path)

    def _validate_text(self, text: str) -> bool:
        """Validate text input."""
        if not text or not text.strip():
            return False

        # Length limit (prevent DoS)
        if len(text) > 5000:
            logger.warning("tts.text_too_long", length=len(text))
            return False

        return True

    def _filter_sensitive(self, text: str) -> str:
        """Filter sensitive content from text."""
        import re

        filtered = text
        for pattern in self.blocked_patterns:
            if re.search(pattern, filtered, re.IGNORECASE):
                logger.warning("tts.sensitive_content_filtered")
                filtered = re.sub(pattern + r'\S+', '[FILTERED]', filtered, flags=re.IGNORECASE)

        return filtered

    def cleanup(self):
        """Clean up temp files."""
        import shutil
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)

模式 2：流式 TTS

# Stream audio chunks as generated for low latency
with sd.OutputStream(samplerate=24000, channels=1) as stream:
    for _, _, audio in pipeline(text, voice=voice):
        stream.write(audio)  # Play immediately

模式 3：音频缓存

# Cache common phrases with hash key
cache_key = hashlib.sha256(f"{text}:{voice}".encode()).hexdigest()
cache_path = cache_dir / f"{cache_key}.wav"
if cache_path.exists():
    return str(cache_path)  # Cache hit
# Generate, save to cache, return path

模式 4：语音管理器

# Lazy-load engines per voice type
VOICES = {"default": "af_heart", "formal": "af_bella"}

def get_engine(voice_type: str) -> SecureTTSEngine:
    if voice_type not in engines:
        engines[voice_type] = SecureTTSEngine(voice=VOICES[voice_type])
    return engines[voice_type]

模式 5：资源限制

# Semaphore for concurrency + timeout for protection
async with asyncio.Semaphore(2):
    result = await asyncio.wait_for(
        loop.run_in_executor(None, engine.synthesize, text),
        timeout=30.0
    )

---

8. 安全标准

8.1 内容过滤

防止合成不当内容：

class ContentFilter:
    """Filter inappropriate content before synthesis."""

    BLOCKED_CATEGORIES = [
        "violence",
        "hate_speech",
        "explicit",
    ]

    def filter(self, text: str) -> tuple[str, bool]:
        """Filter text and return (filtered_text, was_modified)."""
        # Remove potential command injection
        text = text.replace(";", "").replace("|", "").replace("&", "")

        # Check for blocked patterns
        for pattern in self.blocked_patterns:
            if re.search(pattern, text, re.IGNORECASE):
                return "[Content filtered]", True

        return text, False

8.2 输入验证

def validate_tts_input(text: str) -> bool:
    """Validate text for TTS synthesis."""
    # Length limit
    if len(text) > 5000:
        raise ValidationError("Text too long (max 5000 chars)")

    # Character validation
    if not all(c.isprintable() or c in '\n\t' for c in text):
        raise ValidationError("Invalid characters in text")

    return True

---

9. 常见错误

切勿：直接合成不可信输入

# BAD - No filtering
def speak(user_input: str):
    engine.synthesize(user_input)

# GOOD - Filter first
def speak(user_input: str):
    filtered = content_filter.filter(user_input)
    engine.synthesize(filtered)

切勿：无限制生成

# BAD - Can generate very long audio
engine.synthesize(long_text)  # No limit

# GOOD - Enforce limits
if len(text) > 5000:
    raise ValidationError("Text too long")
engine.synthesize(text)

---

10. 实现前检查清单

编写代码之前

• 为 TTS 合成输出编写失败测试
• 确定预期音频格式（24kHz WAV）
• 规划内容过滤规则
• 为常用短语设计缓存策略
• 查阅 Kokoro TTS API 文档

实施期间

• 每实现一个方法后运行测试
• 实现流式输出以降低延迟
• 添加输入验证（长度、字符）
• 实现敏感内容过滤
• 设置权限为 0o700 的安全临时目录
• 添加并发限制（最多 2 个工作线程）
• 实现超时保护（默认 30s）

提交前

• 所有 TTS 测试均通过：pytest tests/test_tts_engine.py -v
• 覆盖率达到阈值：pytest --cov=jarvis.tts
• 类型检查通过：mypy src/jarvis/tts/
• 不记录任何敏感文本
• 已验证生成的音频会被清理
• 已测试语音预加载
• 集成测试通过：python -m jarvis.tts --test

---

11. 总结

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

• 快速：实时流式处理，让语音助手响应迅速
• 安全：过滤内容，确保合成内容适当
• 高效：缓存常用短语

你理解 TTS 需要进行输入验证和内容过滤，以防止合成不当内容。始终执行文本长度限制，并清理生成的音频文件。

重要提醒：

1. 合成前过滤文本内容
2. 强制执行文本长度限制（最多 5000 个字符）
3. 播放后删除生成的音频
4. 绝不记录敏感文本内容
5. 缓存常用短语以提升性能