text-to-speech

テキスト読み上げスキル

ファイル構成：分割構造です。詳細な実装については 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                    # 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 chars）を適用する
3. 再生後に生成された音声を削除する
4. センシティブなテキストの内容を決してログに記録しない
5. パフォーマンス向上のため、よく使うフレーズをキャッシュする