文本转语音技能Skill text-to-speech

---

名称: 文本转语音 风险级别: 中 描述: “实现文本转语音的专业技能，使用Kokoro TTS。覆盖语音合成、音频生成、性能优化，以及安全处理生成的音频，用于JARVIS语音助手。” 模型: sonnet

文本转语音技能

文件组织: 分割结构。参见 references/ 获取详细实现。

1. 概述

风险级别: 中 - 生成音频输出，可能合成不当内容，资源密集

你是一个文本转语音系统的专家，精通Kokoro TTS、语音合成和音频生成优化。你的专长涵盖模型配置、语音定制、流式音频输出和安全处理合成语音。

你擅长:

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

主要用例:

• JARVIS 语音响应
• 实时语音合成与自然韵律
• 离线TTS（无云依赖）
• 多语音支持用于不同上下文

---

2. 核心原则

• TDD优先 - 先写测试再实现。验证合成输出、音频质量和错误处理。
• 性能意识 - 优化延迟：流式合成、模型缓存、音频分块。
• 安全第一 - 过滤内容、验证输入、清理生成文件。
• 资源高效 - 管理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                    # 运行测试
pytest --cov=jarvis.tts --cov-report=term-missing     # 覆盖率
mypy src/jarvis/tts/                                  # 类型检查
python -m jarvis.tts --test "Hello JARVIS"            # 集成测试

---

4. 性能模式

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

# 不好 - 等待完整音频
audio_chunks = []
for _, _, audio in pipeline(text):
    audio_chunks.append(audio)
play_audio(np.concatenate(audio_chunks))  # 长时间等待

# 好 - 立即流式分块
with sd.OutputStream(samplerate=24000, channels=1) as stream:
    for _, _, audio in pipeline(text):
        stream.write(audio)  # 生成时播放

模式: 模型缓存（更快启动）

# 不好: pipeline = KPipeline(lang_code="a")  # 每次重新加载

# 好 - 单例模式
class TTSEngine:
    _pipeline = None
    @classmethod
    def get_pipeline(cls):
        if cls._pipeline is None:
            cls._pipeline = KPipeline(lang_code="a")
        return cls._pipeline

模式: 音频分块（内存高效）

# 不好: data, sr = sf.read(audio_path)  # 全文件在RAM中

# 好 - 分块处理
with sf.SoundFile(audio_path) as f:
    while f.tell() < len(f):
        yield process(f.read(24000))

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

# 不好: audio = engine.synthesize(text)  # 阻塞事件循环

# 好 - 在执行器中运行
audio = await loop.run_in_executor(None, engine.synthesize, text)

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

# 不好: return SecureTTSEngine(voice=VOICES[voice_type])  # 冷启动

# 好 - 启动时预加载
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:
    """安全文本转语音，带内容过滤。"""

    def __init__(self, voice: str = "af_heart", lang_code: str = "a"):
        # 初始化Kokoro管道
        self.pipeline = KPipeline(lang_code=lang_code)
        self.voice = voice

        # 内容过滤模式
        self.blocked_patterns = [
            r"password\s*[:=]",
            r"api[_-]?key\s*[:=]",
            r"secret\s*[:=]",
        ]

        # 创建安全临时目录
        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:
        """将文本合成到音频文件。"""
        # 验证和过滤输入
        if not self._validate_text(text):
            raise ValidationError("无效文本输入")

        filtered_text = self._filter_sensitive(text)

        # 生成音频
        audio_path = Path(self.temp_dir) / f"{uuid.uuid4()}.wav"

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

        # 收集音频分块
        audio_chunks = []
        for _, _, audio in generator:
            audio_chunks.append(audio)

        if not audio_chunks:
            raise TTSError("未生成音频")

        # 拼接并保存
        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:
        """验证文本输入。"""
        if not text or not text.strip():
            return False

        # 长度限制（防止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:
        """从文本中过滤敏感内容。"""
        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):
        """清理临时文件。"""
        import shutil
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)

模式 2: 流式TTS

# 流式音频分块，低延迟
with sd.OutputStream(samplerate=24000, channels=1) as stream:
    for _, _, audio in pipeline(text, voice=voice):
        stream.write(audio)  # 立即播放

模式 3: 音频缓存

# 用哈希键缓存常见短语
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)  # 缓存命中
# 生成，保存到缓存，返回路径

模式 4: 语音管理器

# 按语音类型延迟加载引擎
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: 资源限制

# 并发信号量 + 超时保护
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:
    """在合成前过滤不当内容。"""

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

    def filter(self, text: str) -> tuple[str, bool]:
        """过滤文本并返回（过滤后的文本，是否被修改）。"""
        # 移除潜在命令注入
        text = text.replace(";", "").replace("|", "").replace("&", "")

        # 检查阻止模式
        for pattern in self.blocked_patterns:
            if re.search(pattern, text, re.IGNORECASE):
                return "[内容已过滤]", True

        return text, False

8.2 输入验证

def validate_tts_input(text: str) -> bool:
    """验证文本用于TTS合成。"""
    # 长度限制
    if len(text) > 5000:
        raise ValidationError("文本太长（最多5000字符）")

    # 字符验证
    if not all(c.isprintable() or c in '
\t' for c in text):
        raise ValidationError("文本中包含无效字符")

    return True

---

9. 常见错误

永不：直接合成不受信任的输入

# 不好 - 无过滤
def speak(user_input: str):
    engine.synthesize(user_input)

# 好 - 先过滤
def speak(user_input: str):
    filtered = content_filter.filter(user_input)
    engine.synthesize(filtered)

永不：无限生成

# 不好 - 可以生成长音频
engine.synthesize(long_text)  # 无限制

# 好 - 强制限制
if len(text) > 5000:
    raise ValidationError("文本太长")
engine.synthesize(text)

---

10. 实施前检查清单

写代码前

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

实施期间

• [ ] 每个方法实现后运行测试
• [ ] 实现低延迟的流式输出
• [ ] 添加输入验证（长度、字符）
• [ ] 实现敏感内容过滤
• [ ] 设置安全临时目录，权限0o700
• [ ] 添加并发限制（最多2个worker）
• [ ] 实现超时保护（默认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. 为性能缓存常见短语