音声を自動でテキスト化！フルスタック文字起こしツールの作り方
音声を自動でテキスト化！フルスタック文字起こしツールの作り方
はじめに
会議の議事録作成、インタビューの文字起こし、YouTube動画の字幕作成など、音声をテキストに変換する需要は日々増えています。
そこで今回、Whisper AIを使った自動文字起こしツール「VoiceScribe」を開発しました。ブラウザから音声ファイルをアップロードするだけで、高精度な文字起こしが可能です。
この記事では、フルスタック開発の流れと、技術的なポイントを詳しく解説します。
システム構成
アーキテクチャ全体図
Frontend (React + Vite)
↓ HTTP API
Backend (FastAPI + Python)
↓ データ保存
Database (PostgreSQL)
↓ タスクキュー
Redis + Celery Worker
↓ 文字起こし
Whisper AI技術スタック
Frontend
Backend
AI/ML
インフラ
なぜこの構成にしたのか？
1. 非同期処理の必要性
音声ファイルの文字起こしは、数分〜数十分かかる重い処理です。同期的に処理すると：
そこでCeleryを使った非同期処理を採用：
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メリット：
実装のポイント
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)}ポイント：
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ポイント：
3. Whisper AIの活用
Whisperは、OpenAIが開発した高精度な音声認識モデル。
モデルサイズの選択：
| モデル | パラメータ数 | 精度 | 処理速度 |
|——–|————|——|———|
| tiny | 39M | ★★☆☆☆ | 非常に速い |
| base | 74M | ★★★☆☆ | 速い |
| small | 244M | ★★★★☆ | 普通 |
| medium | 769M | ★★★★★ | 遅い |
| large | 1550M | ★★★★★ | 非常に遅い |
今回はmediumモデルを採用：
# 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
checkStatus: () => Promise
reset: () => void
}
export const useTranscriptionStore = create((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
})
})) メリット：
パフォーマンス最適化
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は英語に特化しており、日本語の認識率が低かった。
解決策：
2. メモリ不足
問題：
大きな音声ファイルを処理すると、Workerがメモリ不足でクラッシュ。
解決策：
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今後の拡張計画
まとめ
フルスタック開発で学んだこと：
→ 重い処理はCeleryで切り離す
→ チーム開発・本番デプロイが楽に
→ Whisperは数行で高精度な文字起こしが可能
→ Zustandで十分、複雑なReduxは不要
特にCeleryの非同期処理は、他のプロジェクトでも応用できる重要な技術でした。バッチ処理、画像変換、メール送信など、様々な場面で活用できます。
—
技術スタック: React, FastAPI, PostgreSQL, Redis, Celery, Whisper AI
開発期間: 約1ヶ月
コード行数: 約5,000行
対応音声形式: mp3, wav, m4a, flac