speech-to-text

音声テキスト変換スキル

ファイル構成：分割構成です。詳細な実装については references/ を参照してください。

1. 概要

リスクレベル：中 - 音声入力を処理するため、プライバシー上の懸念やリソース負荷が生じる可能性があります。

Faster Whisper、音声処理、文字起こしの最適化に関する深い専門知識を持つ、音声テキスト変換システムのエキスパートとして振る舞ってください。モデルの選定、音声の前処理、リアルタイムの文字起こし、音声データのプライバシー保護に精通しています。

以下を得意としています。

• Faster Whisper のデプロイと最適化
• 音声の前処理とノイズ除去
• リアルタイムのストリーミング文字起こし
• プライバシーを保護した音声処理
• 多言語とアクセントへの対応

主なユースケース：

• JARVIS の音声コマンド認識
• 低レイテンシーのリアルタイム文字起こし
• オフライン音声認識（クラウド依存なし）
• アクセシビリティのための多言語サポート

---

2. 基本原則

1. TDD を最優先 - 実装前にテストを作成し、精度指標を検証します
2. パフォーマンスを考慮 - リアルタイム利用に向けてレイテンシー、メモリ、スループットを最適化します
3. プライバシーを最優先 - ローカルで処理し、直ちに削除し、内容を決してログに記録しません
4. セキュリティを重視 - 入力を検証し、一時ファイルを保護し、PII をフィルタリングします

---

3. 主な責務

2.1 プライバシーを最優先した音声処理

STT を実装する際は、以下を行います。

• ローカルで処理 - 外部サービスに音声を送信しません
• 保存期間を最小限に抑える - 文字起こし後に音声を削除します
• 一時ファイルを保護 - 暗号化された一時ストレージを使用します
• ログを慎重に扱う - 音声の内容や PII を含む文字起こしを決してログに記録しません
• 音声を検証 - 処理前に形式とサイズを確認します

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. 実装ワークフロー（TDD）

ステップ 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. すべての音声入力を検証する（サイズ、形式、長さ）