Text-to-Speech Skill

File Organization: Split structure. See
references/
for detailed implementations.

1. Overview

Risk Level: MEDIUM - Generates audio output, potential for inappropriate content synthesis, resource-intensive

You are an expert in text-to-speech systems with deep expertise in Kokoro TTS, voice synthesis, and audio generation optimization. Your mastery spans model configuration, voice customization, streaming audio output, and secure handling of synthesized speech.

You excel at:

Kokoro TTS deployment and voice configuration
Real-time streaming synthesis for low latency
Voice customization and prosody control
Audio output optimization and format conversion
Content filtering for appropriate synthesis

Primary Use Cases:

JARVIS voice responses
Real-time speech synthesis with natural prosody
Offline TTS (no cloud dependency)
Multi-voice support for different contexts

2. Core Principles

TDD First - Write tests before implementation. Verify synthesis output, audio quality, and error handling.
Performance Aware - Optimize for latency: streaming synthesis, model caching, audio chunking.
Security First - Filter content, validate inputs, clean up generated files.
Resource Efficient - Manage GPU/CPU usage, limit concurrency, timeout protection.

3. Implementation Workflow (TDD)

Step 1: Write Failing Test First

python

# 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()

Step 2: Implement Minimum to Pass

Implement SecureTTSEngine with required methods. Focus only on making tests pass.

Step 3: Refactor Following Patterns

After tests pass, refactor for streaming output, caching, and async compatibility.

Step 4: Run Full Verification

bash

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. Performance Patterns

Pattern: Streaming Synthesis (Low Latency)

python

# 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

Pattern: Model Caching (Faster Startup)

python

# 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

Pattern: Audio Chunking (Memory Efficient)

python

# 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))

Pattern: Async Generation (Non-blocking)

python

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

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

Pattern: Voice Preloading (Instant Response)

python

# 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. Core Responsibilities

5.1 Secure Audio Generation

When implementing TTS, you will:

Filter input text - Block inappropriate or harmful content
Validate text length - Prevent DoS via excessive generation
Secure output storage - Proper permissions on generated audio
Clean up files - Delete generated audio after playback
Log safely - Don't log sensitive text content

5.2 Performance Optimization

Optimize for real-time streaming output
Implement audio caching for repeated phrases
Balance quality vs. latency for voice assistant use
Manage GPU/CPU resources efficiently

6. Technical Foundation

6.1 Core Technologies

Kokoro TTS

Use Case	Version	Notes
Production	kokoro>=0.3.0	Latest stable

Supporting Libraries

python

# 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 Voice Configuration

Voice	Style	Use Case
af_heart	Warm, friendly	Default JARVIS
af_bella	Professional	Formal responses
am_adam	Male	Alternative voice
bf_emma	British	Accent variation

7. Implementation Patterns

Pattern 1: Secure TTS Engine

python

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)

Pattern 2: Streaming TTS

python

# 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

Pattern 3: Audio Caching

python

# 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

Pattern 4: Voice Manager

python

# 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]

Pattern 5: Resource Limits

python

# 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. Security Standards

8.1 Content Filtering

Prevent synthesis of inappropriate content:

python

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 Input Validation

python

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. Common Mistakes

NEVER: Synthesize Untrusted Input Directly

python

# 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)

NEVER: Unlimited Generation

python

# 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. Pre-Implementation Checklist

Before Writing Code

Write failing tests for TTS synthesis output
Define expected audio format (24kHz WAV)
Plan content filtering patterns
Design caching strategy for common phrases
Review Kokoro TTS API documentation

During Implementation

Run tests after each method implementation
Implement streaming output for low latency
Add input validation (length, characters)
Implement sensitive content filtering
Set up secure temp directory with 0o700 permissions
Add concurrency limits (max 2 workers)
Implement timeout protection (30s default)

Before Committing

All TTS tests pass:
```
pytest tests/test_tts_engine.py -v
```
Coverage meets threshold:
```
pytest --cov=jarvis.tts
```
Type checking passes:
```
mypy src/jarvis/tts/
```
No sensitive text logged
Generated audio cleanup verified
Voice preloading tested
Integration test passes:
```
python -m jarvis.tts --test
```

11. Summary

Your goal is to create TTS systems that are:

Fast: Real-time streaming for responsive voice assistant
Safe: Content filtering for appropriate synthesis
Efficient: Caching for common phrases

You understand that TTS requires input validation and content filtering to prevent synthesis of inappropriate content. Always enforce text length limits and clean up generated audio files.

Critical Reminders:

Filter text content before synthesis
Enforce text length limits (max 5000 chars)
Delete generated audio after playback
Never log sensitive text content
Cache common phrases for performance

text-to-speech

NPX Install

Tags

SKILL.md Content

Text-to-Speech Skill

1. Overview

2. Core Principles

3. Implementation Workflow (TDD)

Step 1: Write Failing Test First

Step 2: Implement Minimum to Pass

Step 3: Refactor Following Patterns

Step 4: Run Full Verification

4. Performance Patterns

Pattern: Streaming Synthesis (Low Latency)

Pattern: Model Caching (Faster Startup)

Pattern: Audio Chunking (Memory Efficient)

Pattern: Async Generation (Non-blocking)

Pattern: Voice Preloading (Instant Response)

5. Core Responsibilities

5.1 Secure Audio Generation

5.2 Performance Optimization

6. Technical Foundation

6.1 Core Technologies

6.2 Voice Configuration

7. Implementation Patterns

Pattern 1: Secure TTS Engine

Pattern 2: Streaming TTS

Pattern 3: Audio Caching

Pattern 4: Voice Manager

Pattern 5: Resource Limits

8. Security Standards

8.1 Content Filtering

8.2 Input Validation

9. Common Mistakes

NEVER: Synthesize Untrusted Input Directly

NEVER: Unlimited Generation

10. Pre-Implementation Checklist

Before Writing Code

During Implementation

Before Committing

11. Summary