音声を自動でテキスト化！フルスタック文字起こしツールの作り方

はじめに

会議の議事録作成、インタビューの文字起こし、YouTube動画の字幕作成など、音声をテキストに変換する需要は日々増えています。

そこで今回、Whisper AIを使った自動文字起こしツール「VoiceScribe」を開発しました。ブラウザから音声ファイルをアップロードするだけで、高精度な文字起こしが可能です。

この記事では、フルスタック開発の流れと、技術的なポイントを詳しく解説します。

システム構成

アーキテクチャ全体図

Frontend (React + Vite)
    ↓ HTTP API
Backend (FastAPI + Python)
    ↓ データ保存
Database (PostgreSQL)
    ↓ タスクキュー
Redis + Celery Worker
    ↓ 文字起こし
Whisper AI

技術スタック

Frontend
– React 18 + TypeScript
– Vite（ビルドツール）
– TailwindCSS（スタイリング）
– Zustand（状態管理）

Backend
– FastAPI（Web API）
– SQLAlchemy + Alembic（ORM・マイグレーション）
– Celery（非同期タスク処理）
– Redis（キャッシュ・メッセージブローカー）

AI/ML
– OpenAI Whisper（音声認識）
– Claude API（要約機能・オプション）

インフラ
– Docker + Docker Compose
– PostgreSQL 16
– Redis 7

なぜこの構成にしたのか？

1. 非同期処理の必要性

音声ファイルの文字起こしは、数分〜数十分かかる重い処理です。同期的に処理すると：

❌ ユーザーがブラウザを閉じたら処理が中断
❌ 複数のリクエストで処理が詰まる
❌ タイムアウトエラーが発生しやすい

そこでCeleryを使った非同期処理を採用：

✅ バックグラウンドで処理が継続
✅ 複数のWorkerで並列処理が可能
✅ 処理状態をリアルタイムで確認可能

2. Dockerによる環境統一

開発環境と本番環境の差異をなくすため、全てのサービスをDocker化：

version: '3.8'

services:
  # PostgreSQL
  db:
    image: postgres:16-alpine
    ports:
      - "5433:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  # Redis
  redis:
    image: redis:7-alpine
    ports:
      - "6380:6379"

  # Backend API
  backend:
    build: ./backend
    ports:
      - "8001:8000"
    depends_on:
      - db
      - redis

  # Celery Worker
  worker:
    build: ./backend
    command: celery -A app.worker worker --loglevel=info
    depends_on:
      - db
      - redis

  # Frontend
  frontend:
    build: ./frontend
    ports:
      - "3001:5173"
    environment:
      - VITE_API_URL=http://localhost:8001/api/v1

メリット：
– docker-compose up一発で全サービスが起動
– 環境変数で設定を管理
– チーム開発でも環境差異が発生しない

実装のポイント

1. FastAPIによるAPI設計

FastAPIは、自動でAPI仕様書（Swagger UI）を生成してくれるのが最大の魅力。

from fastapi import FastAPI, UploadFile, File
from fastapi.responses import JSONResponse

app = FastAPI(
    title="VoiceScribe API",
    version="1.0.0",
    description="自動文字起こしAPI"
)

@app.post("/api/v1/transcriptions")
async def create_transcription(
    file: UploadFile = File(...),
    language: str = "ja"
):
    """
    音声ファイルをアップロードして文字起こしを開始

    Parameters:
    - file: 音声ファイル（mp3, wav, m4a対応）
    - language: 言語コード（ja, en, zh等）

    Returns:
    - task_id: 処理タスクID
    - status: 処理状態
    """

    # ファイルを保存
    file_path = save_upload_file(file)

    # Celeryタスクを起動
    task = transcribe_audio.delay(
        file_path=file_path,
        language=language
    )

    return JSONResponse({
        "task_id": task.id,
        "status": "processing",
        "message": "文字起こしを開始しました"
    })

@app.get("/api/v1/transcriptions/{task_id}")
async def get_transcription_status(task_id: str):
    """
    文字起こしの進捗状況を取得
    """
    task = AsyncResult(task_id)

    if task.state == 'PENDING':
        return {"status": "pending", "progress": 0}
    elif task.state == 'PROGRESS':
        return {
            "status": "processing",
            "progress": task.info.get('progress', 0)
        }
    elif task.state == 'SUCCESS':
        return {
            "status": "completed",
            "result": task.result
        }
    else:
        return {"status": "failed", "error": str(task.info)}

ポイント：
– 型ヒント（TypeHint）で自動的にバリデーション
– 非同期処理（async/await）でパフォーマンス向上
– 詳細なドキュメント文字列でSwagger UIが充実

2. Celeryによる非同期処理

Celeryは、Pythonの定番非同期タスクライブラリ。

from celery import Celery
import whisper

# Celeryアプリケーション初期化
celery_app = Celery(
    'voicescribe',
    broker='redis://redis:6379/1',
    backend='redis://redis:6379/2'
)

@celery_app.task(bind=True)
def transcribe_audio(self, file_path: str, language: str = "ja"):
    """
    音声ファイルを文字起こし
    """
    try:
        # Whisperモデルを読み込み（初回のみ時間がかかる）
        model = whisper.load_model("medium")

        # 進捗を0%に更新
        self.update_state(
            state='PROGRESS',
            meta={'progress': 0, 'status': 'モデル読み込み中'}
        )

        # 文字起こし実行
        result = model.transcribe(
            file_path,
            language=language,
            verbose=False,
            task='transcribe'
        )

        # 進捗を50%に更新
        self.update_state(
            state='PROGRESS',
            meta={'progress': 50, 'status': 'テキスト整形中'}
        )

        # 結果を整形
        text = result["text"]
        segments = [
            {
                "start": seg["start"],
                "end": seg["end"],
                "text": seg["text"]
            }
            for seg in result["segments"]
        ]

        # 進捗を100%に更新
        return {
            "text": text,
            "segments": segments,
            "language": result["language"]
        }

    except Exception as e:
        self.update_state(
            state='FAILURE',
            meta={'error': str(e)}
        )
        raise

ポイント：
– bind=Trueでタスク自身のメソッドにアクセス可能
– update_state()でリアルタイム進捗更新
– エラー時は適切な例外を投げる

3. Whisper AIの活用

Whisperは、OpenAIが開発した高精度な音声認識モデル。

モデルサイズの選択：

モデル	パラメータ数	精度	処理速度
tiny	39M	★★☆☆☆	非常に速い
base	74M	★★★☆☆	速い
small	244M	★★★★☆	普通
medium	769M	★★★★★	遅い
large	1550M	★★★★★	非常に遅い

今回はmediumモデルを採用：
– 精度と速度のバランスが良い
– 日本語の認識率が高い
– GPU不要でCPUのみで動作

# Whisperの設定
result = model.transcribe(
    audio_file,
    language="ja",           # 言語を指定（自動検出も可）
    task="transcribe",       # 'transcribe'（文字起こし）or 'translate'（翻訳）
    temperature=0.0,         # 確信度の高い結果を優先
    beam_size=5,             # ビームサーチの幅
    best_of=5,               # 候補数
    word_timestamps=True     # 単語レベルのタイムスタンプ
)

4. React + Zustandによる状態管理

フロントエンドは、Zustandでシンプルな状態管理。

// stores/transcriptionStore.ts
import create from 'zustand'

interface TranscriptionState {
  taskId: string | null
  status: 'idle' | 'uploading' | 'processing' | 'completed' | 'error'
  progress: number
  result: string | null

  // アクション
  uploadFile: (file: File) => Promise<void>
  checkStatus: () => Promise<void>
  reset: () => void
}

export const useTranscriptionStore = create<TranscriptionState>((set, get) => ({
  taskId: null,
  status: 'idle',
  progress: 0,
  result: null,

  uploadFile: async (file: File) => {
    set({ status: 'uploading' })

    const formData = new FormData()
    formData.append('file', file)

    try {
      const res = await fetch('/api/v1/transcriptions', {
        method: 'POST',
        body: formData
      })

      const data = await res.json()
      set({
        taskId: data.task_id,
        status: 'processing'
      })

      // 状態チェックを開始
      get().checkStatus()
    } catch (error) {
      set({ status: 'error' })
    }
  },

  checkStatus: async () => {
    const { taskId } = get()
    if (!taskId) return

    try {
      const res = await fetch(`/api/v1/transcriptions/${taskId}`)
      const data = await res.json()

      if (data.status === 'completed') {
        set({
          status: 'completed',
          result: data.result.text,
          progress: 100
        })
      } else if (data.status === 'processing') {
        set({ progress: data.progress || 0 })

        // 3秒後に再チェック
        setTimeout(() => get().checkStatus(), 3000)
      } else if (data.status === 'failed') {
        set({ status: 'error' })
      }
    } catch (error) {
      set({ status: 'error' })
    }
  },

  reset: () => set({
    taskId: null,
    status: 'idle',
    progress: 0,
    result: null
  })
}))

メリット：
– Reduxより圧倒的にシンプル
– TypeScript完全対応
– React Contextより高速

パフォーマンス最適化

1. ファイルサイズ制限

大きすぎる音声ファイルは、処理時間とストレージを圧迫するため制限：

MAX_FILE_SIZE = 100 * 1024 * 1024  # 100MB

@app.post("/api/v1/transcriptions")
async def create_transcription(file: UploadFile = File(...)):
    # ファイルサイズチェック
    content = await file.read()
    if len(content) > MAX_FILE_SIZE:
        raise HTTPException(
            status_code=413,
            detail="ファイルサイズは100MB以下にしてください"
        )

    # 処理を続行...

2. Whisperモデルのキャッシュ

Whisperモデルは初回読み込みに数秒かかるため、メモリにキャッシュ：

_whisper_model_cache = {}

def get_whisper_model(model_name: str = "medium"):
    """Whisperモデルをキャッシュから取得"""
    if model_name not in _whisper_model_cache:
        print(f"Loading Whisper model: {model_name}")
        _whisper_model_cache[model_name] = whisper.load_model(model_name)
    return _whisper_model_cache[model_name]

これにより、2回目以降の文字起こしが約5秒高速化しました。

3. Redis接続プール

Redisへの接続を毎回作成すると遅いため、接続プールを使用：

from redis import ConnectionPool, Redis

redis_pool = ConnectionPool.from_url(
    "redis://redis:6379/0",
    max_connections=20
)

def get_redis() -> Redis:
    return Redis(connection_pool=redis_pool)

苦労した点と解決策

1. 日本語の認識精度

問題：
Whisperは英語に特化しており、日本語の認識率が低かった。

解決策：
– language="ja"を明示的に指定
– temperatureを0.0に設定（確実性重視）
– 音声ファイルを前処理（ノイズ除去）

2. メモリ不足

問題：
大きな音声ファイルを処理すると、Workerがメモリ不足でクラッシュ。

解決策：
– 音声ファイルを10分ごとにチャンク分割
– 各チャンクを順次処理して結合
– Celeryの--max-memory-per-childオプションで制限

def transcribe_long_audio(file_path: str):
    """長い音声を分割して処理"""
    chunks = split_audio_file(file_path, chunk_duration=600)  # 10分ごと

    results = []
    for chunk in chunks:
        result = model.transcribe(chunk)
        results.append(result["text"])

        # メモリ解放
        del result
        gc.collect()

    return " ".join(results)

3. Docker Composeの起動順序

問題：
BackendがDB起動前に起動して、接続エラーが発生。

解決策：
healthcheckとdepends_onで起動順序を制御：

backend:
  depends_on:
    db:
      condition: service_healthy
    redis:
      condition: service_healthy

今後の拡張計画

• [ ] リアルタイム文字起こし（WebSocket）
• [ ] 話者分離（誰が話したかを識別）
• [ ] 自動要約機能（Claude API統合）
• [ ] 字幕ファイルエクスポート（SRT, VTT形式）
• [ ] スマホアプリ（React Native）

まとめ

フルスタック開発で学んだこと：

✅ 非同期処理は必須
→ 重い処理はCeleryで切り離す

✅ Docker Composeで環境統一
→ チーム開発・本番デプロイが楽に

✅ AI APIは思ったより簡単
→ Whisperは数行で高精度な文字起こしが可能

✅ 状態管理はシンプルに
→ Zustandで十分、複雑なReduxは不要

特にCeleryの非同期処理は、他のプロジェクトでも応用できる重要な技術でした。バッチ処理、画像変換、メール送信など、様々な場面で活用できます。

---

技術スタック: React, FastAPI, PostgreSQL, Redis, Celery, Whisper AI
開発期間: 約1ヶ月
コード行数: 約5,000行
対応音声形式: mp3, wav, m4a, flac