Go to Qiita Advent Calendar Top
LoginSignup
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@Esperanto_kanjijigo(漢字化 世界語)

エスペラント音声のリアルタイム文字起こし＋日韓同時翻訳を実装した話

Posted at

はじめに

Zoom や Google Meet でのエスペラント会話を 低遅延でリアルタイム文字起こし し、さらに 日本語・韓国語への同時翻訳 をブラウザとDiscordに配信するシステムを構築しました。本記事では、その設計思想から実装の詳細、運用上のTipsまでを詳しく解説します。

対象読者

  • リアルタイム音声認識システムに興味がある方
  • マイナー言語（エスペラント等）の音声処理に取り組んでいる方
  • WebSocketベースの双方向通信やWeb UIの実装を学びたい方
  • 翻訳APIを組み込んだパイプライン設計に関心がある方

成果物の概要

  • 入力: PCのオーディオループバック（Zoom/Meet/Discord の音声）
  • 音声認識: Speechmatics Realtime STT（エスペラント eo 対応）
  • 翻訳: Google Cloud Translation API（日本語・韓国語）
  • 出力:
    • Web UI（ブラウザで最新発話＋履歴を表示）
    • Discord Webhook（原文＋翻訳をまとめて投稿）
    • Zoom Closed Caption API（Zoom画面内に字幕表示）
    • ログファイル（タイムスタンプ付き確定文）

1. システムアーキテクチャ

1.1 全体構成図（概念）

┌─────────────┐
│ Zoom/Meet   │──┐
│ (音声出力)  │  │
└─────────────┘  │
                 │  PCループバック
┌─────────────┐  │  （PipeWire/VoiceMeeter/BlackHole）
│ Discord     │──┤
│ (音声出力)  │  │
└─────────────┘  │
                 ▼
         ┌──────────────┐
         │ sounddevice  │  16kHz mono PCM
         │ (Python)     │
         └──────┬───────┘
                │
                ▼
    ┌───────────────────────┐
    │ TranscriptionPipeline │
    │  ・音声入力管理       │
    │  ・ASRバックエンド選択│
    │  ・翻訳サービス呼出  │
    │  ・出力先振り分け     │
    └───┬───────────────┬───┘
        │               │
        ▼               ▼
┌──────────────┐  ┌──────────────┐
│ Speechmatics │  │ Translation  │
│ Realtime WSS │  │ Service      │
│ (Bearer JWT) │  │ (Google/Libre)│
└──────┬───────┘  └──────┬───────┘
       │                 │
       │ 確定文          │ 翻訳結果
       ▼                 ▼
┌─────────────────────────────────┐
│        出力先（複数同時）        │
│  1. Web UI (WebSocket broadcast)│
│  2. Discord Webhook (batch post)│
│  3. Zoom CC API (throttled POST)│
│  4. ログファイル (append)        │
└─────────────────────────────────┘

1.2 各コンポーネントの役割

1.2.1 音声入力（transcriber/audio.py）

  • AudioChunkStream クラス
    • sounddevice.RawInputStream で 16kHz モノラル PCM を取得
    • 非同期イテレータで音声チャンクを yield
    • デバイス切替を 2 秒間隔で監視し、変更時は自動再接続
    • 無音/停止検知（5秒間チャンクが来なければリスタート）

主要メソッド:

class AudioChunkStream:
    async def __aiter__(self) -> AsyncGenerator[bytes, None]:
        """音声チャンクを非同期に生成"""
        # デバイス監視タスクを起動
        # ストリームを開始し、キューからチャンクを取り出してyield

ポイント:

  • OS レベルのデバイス切替（例: ヘッドセット抜き差し）にも追従
  • config.device_index が指定されていればそのデバイスを優先し、なければシステムのデフォルト入力を使用
  • ダウンミックス対応（ステレオ→モノラル変換）

1.2.2 音声認識バックエンド（transcriber/asr/）

Speechmatics Realtime（speechmatics_backend.py）

  • 認証フロー:

    1. 長期APIキーを管理プラットフォーム https://mp.speechmatics.com/v1/api_keys?type=rt に POST
    2. 短期JWT（key_value）を取得
    3. WebSocket 接続時に Authorization: Bearer <JWT> ヘッダを付与

  • プロトコル:

    1. StartRecognition メッセージ（JSON）を送信
    2. サーバから RecognitionStarted を待機
    3. 音声データ（PCM16）をバイナリで送信
    4. AddPartialTranscript（途中経過）と AddTranscript（確定文）を受信

接続URL構成:

# .env で SPEECHMATICS_CONNECTION_URL=wss://eu2.rt.speechmatics.com/v2
# .env で SPEECHMATICS_LANGUAGE=eo
# 実装内部で /eo を付与 → wss://eu2.rt.speechmatics.com/v2/eo

重要な実装上の注意点:

  • enable_punctuation パラメータはサーバスキーマと不一致のため送信しない（内部で自動句読点が付与される）
  • websockets v15 では additional_headers を使用（旧 extra_headers は非推奨）
  • Pydantic の HttpUrl は wss スキームを許容しないため、接続URLは str 型で扱う
Vosk（vosk_backend.py）
  • オフライン認識用のバックアップ
  • vosk-model-small-eo-0.42 以上のモデルを使用
  • ネットワーク不要、コストゼロ
Whisper（whisper_backend.py）
  • faster-whisper ライブラリを使用
  • GPU/Apple M シリーズ向け
  • セグメント単位の処理で sub-second 遅延を目指す

1.2.3 翻訳サービス（transcriber/translate/service.py）

  • TranslationService クラス
    • Google Cloud Translation API と LibreTranslate に対応
    • 非同期リクエスト（aiohttp）
    • TTL付きキャッシュ（OrderedDict でLRU風）

Google 認証:

# サービスアカウントJSON を使用する場合
self._google_credentials = service_account.Credentials.from_service_account_file(
    str(self.google_credentials_path),
    scopes=["https://www.googleapis.com/auth/cloud-translation"],
)
self._google_request = GoogleAuthRequest()

翻訳リクエスト例（Google）:

async def _translate_google(self, text: str, target: str) -> Optional[str]:
  """Google Cloud Translation API v2 を呼び出し"""
  assert self._session
  params: Dict[str, str] = {}
  headers: Dict[str, str] = {}

  if self.google_api_key:
    params["key"] = self.google_api_key
  elif self._google_credentials and self._google_request:
    await self._ensure_google_token()
    if not self._google_credentials.token:
      logging.error("Failed to obtain Google OAuth token for translation.")
      return None
    headers["Authorization"] = f"Bearer {self._google_credentials.token}"
  else:
    logging.error(
      "Google translation requested but neither GOOGLE_TRANSLATE_API_KEY nor valid credentials provided."
    )
    return None

  payload = {
    "q": text,
    "source": self.source_language,
    "target": target,
    "format": "text",
  }
  if self.google_model:
    payload["model"] = self.google_model

  url = "https://translation.googleapis.com/language/translate/v2"
  async with self._session.post(url, params=params, json=payload, headers=headers) as resp:
    if resp.status != 200:
      body = await resp.text()
      raise RuntimeError(f"HTTP {resp.status}: {body}")
    data = await resp.json()
    translations = data.get("data", {}).get("translations", [])
    if translations:
      return translations[0].get("translatedText")
    return None

キャッシュ機構:

  • 同一テキスト＋同一ターゲット言語の組み合わせをキャッシュ
  • TTL（既定では約32秒：timeout_seconds のデフォルト8秒×4）で古いエントリを削除
  • 最大サイズ（デフォルト128件）でLRU風に管理

1.2.4 Web UI（transcriber/display/webui.py）

  • CaptionWebUI クラス
    • aiohttp.web でHTTPサーバとWebSocketサーバを兼ねる
    • ポート 8765（デフォルト）で起動、衝突時は次のポートを試行（最大5回）

エンドポイント:

  • GET / → web/index.html（静的ページ）
  • GET /ws → WebSocket 接続（双方向通信）
  • GET /config → 翻訳ターゲット設定をJSONで返す
  • GET /static/* → CSS/JSファイル

ブロードキャスト:

async def broadcast(self, message: Dict) -> None:
    """接続中の全クライアントにメッセージを送信"""
    # message = {"type": "partial"|"final", "text": "...", "translations": {...}}
    payload = json.dumps(message)
    for ws in list(self._clients):
        await ws.send_str(payload)

1.2.5 Discord 連携（transcriber/discord/）

DiscordNotifier（notifier.py）
  • Webhook URL に POST リクエストを送信
  • content フィールドにメッセージを格納
DiscordBatcher（batcher.py）
  • バッチング機構:
    • 確定文を 2 秒間隔でバッファに蓄積
    • 文字数が 350 字を超えたら即座に投稿
    • タイマー満了時にまとめて投稿

フォーマット例:

Esperanto: Bonvenon al nia kunsido.
日本語: 私たちの会議へようこそ。
한국어: 우리 회의에 오신 것을 환영합니다.

1.2.6 パイプライン統合（transcriber/pipeline.py）

  • TranscriptionPipeline クラス
    • 音声入力・ASR・翻訳・出力先をオーケストレーション
    • 確定文ごとに各出力先へ並列配信

主要メソッド:

async def run(self) -> None:
    """パイプライン本体"""
    # 1. Web UI 起動
    # 2. 翻訳サービス初期化
    # 3. ASR バックエンド接続
    # 4. 音声ストリームを開始
    # 5. 確定文を受信 → 翻訳 → 各出力先へ配信

確定文処理フロー:

async def _handle_final(self, segment: TranscriptSegment) -> None:
    """確定文を受信した時の処理"""
    # 1. テキスト正規化（空白・句読点整形）
    # 2. 翻訳サービス呼び出し
    # 3. ログファイル書き込み
    # 4. Web UI へブロードキャスト
    # 5. Discord へバッチ投稿
    # 6. Zoom CC API へ送信

2. 実装の詳細

2.1 音声入力のデバイス切替対応

課題:

  • OS の音声設定変更やデバイス抜き差しでパイプラインが停止してしまう

解決策:

  • 2 秒ごとに現在のデフォルト入力デバイスを確認
  • デバイス変更を検知したら古いストリームをクローズし、新デバイスで再接続
  • 5 秒間音声が来なければストリームが死んでいると判断してリスタート

実装:

async def _monitor_device_changes(self) -> None:
    """デバイス変更を監視し、必要に応じて再接続"""
    while not self._stopped.is_set():
        await asyncio.sleep(self._check_interval)
        new_device = self._get_effective_device()
        if new_device != self._current_device:
            logging.info("Audio device changed: %s -> %s", self._current_device, new_device)
            await self._reconnect_stream(new_device)

2.2 Speechmatics の認証とエラーハンドリング

JWT 自動取得（リージョン指定が必須）:

from urllib.parse import urlparse

async def _authorize_jwt(self) -> Optional[str]:
  """APIキーから短期JWTを取得"""
  auth_url = "https://mp.speechmatics.com/v1/api_keys"
  headers = {"Authorization": f"Bearer {self.config.api_key}"}

  # 接続先ホスト（例: eu2.rt.speechmatics.com）から region を推定
  parsed_ws = urlparse(self.config.connection_url)
  region = infer_region(parsed_ws.hostname)  # eu / us / ca / ap

  params = {"type": "rt", "sm-sdk": "python-custom"}
  body = {"ttl": int(self.config.jwt_ttl_seconds), "region": region}

  async with aiohttp.ClientSession() as session:
    async with session.post(auth_url, params=params, headers=headers, json=body) as resp:
      resp.raise_for_status()
      data = await resp.json()
      return data["key_value"]  # 短期JWT

infer_region はホスト名のプレフィックス（eu2, us2 など）から eu / us / ca / ap のいずれかを返すヘルパー関数です。Speechmatics の管理プラットフォームは region 指定が必須のため、これを送らないと 4xx が返って JWT を発行できません。

再接続ロジック:

async def connect(self) -> None:
    """最大 max_reconnect_attempts 回までリトライ"""
    attempt = 0
    backoff = self.config.reconnect_backoff_seconds
    while attempt <= self.config.max_reconnect_attempts:
        try:
            await self._open_connection(ws_url, headers)
            return
        except SpeechmaticsRealtimeError:
            await asyncio.sleep(backoff * (2 ** attempt))
            attempt += 1
    raise SpeechmaticsRealtimeError("Failed to connect after all retries")

よくあるエラーと対処:

  • 404 path not found: JWT発行エンドポイントのパスが違う → https://mp.speechmatics.com/v1/api_keys?type=rt を使用
  • 1003 unsupported data: StartRecognition の形式不一致 → スキーマを最新版に合わせる
  • 401/403 Unauthorized: APIキーが無効 → Portalでキーを再取得
  • 429 Too Many Requests: 連続接続のレート超過 → リトライ間隔を 5〜10 秒空ける

2.3 翻訳サービスの並列リクエスト

並列翻訳:

from .service import TranslationResult

async def translate(self, text: str) -> TranslationResult:
  """複数ターゲット言語を並列リクエスト"""
  if not self.enabled or not text.strip():
    return TranslationResult(text=text, translations={})

  cache_key = self._cache_key(text)
  cached = self._get_cached(cache_key)
  if cached is not None:
    return TranslationResult(text=text, translations=cached)

  translations: Dict[str, str] = {}
  async with self._lock:
    cached = self._get_cached(cache_key)
    if cached is not None:
      return TranslationResult(text=text, translations=cached)

    if self._session is None or self._session.closed:
      self._session = aiohttp.ClientSession(timeout=self._timeout)

    tasks = [self._translate_single(text, target) for target in self.targets]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    for target, result in zip(self.targets, results):
      if isinstance(result, Exception):
        logging.error("Translation to %s failed: %s", target, result)
      elif result:
        translations[target] = result

    self._store_cache(cache_key, translations)

  return TranslationResult(text=text, translations=translations)

Google API の認証トークン更新:

async def _get_google_token(self) -> str:
    """サービスアカウントからアクセストークンを取得"""
    if self._google_credentials.expired or not self._google_credentials.token:
        await asyncio.get_running_loop().run_in_executor(
            None, self._google_credentials.refresh, self._google_request
        )
    return self._google_credentials.token

2.4 Web UI のリアルタイム更新

フロントエンド（web/static/client.js）:

// WebSocket 接続
const ws = new WebSocket(`ws://${window.location.host}/ws`);

ws.onmessage = (event) => {
  const msg = JSON.parse(event.data);
  if (msg.type === 'final') {
    // 最新発話を更新
    document.getElementById('final').textContent = msg.text;
    
    // 翻訳を表示
    const translationsEl = document.getElementById('translations');
    translationsEl.innerHTML = '';
    for (const [lang, text] of Object.entries(msg.translations || {})) {
      if (translationVisibility[lang]) {
        const div = document.createElement('div');
        div.className = 'translation-item';
        div.textContent = `${labelForLang(lang)}: ${text}`;
        translationsEl.appendChild(div);
      }
    }
    
    // 履歴に追加（最新が上）
    addHistoryEntry(msg.text, msg.translations);
  } else if (msg.type === 'partial') {
    // 途中経過を表示（設定でON/OFFできる）
    if (showPartialEl.checked) {
      document.getElementById('partial').textContent = msg.text;
    }
  }
};

設定の永続化:

function saveSettings() {
  const payload = {
    fontSize: fontSizeEl.value,
    darkMode: darkModeEl.checked,
    showPartial: showPartialEl.checked,
    translationVisibility: translationVisibility,
  };
  localStorage.setItem(STORAGE_KEY, JSON.stringify(payload));
}

履歴のコピー/保存:

copyHistoryBtn.addEventListener('click', () => {
  const text = historyEntries.map(e => e.text).join('\n\n');
  navigator.clipboard.writeText(text);
});

downloadHistoryBtn.addEventListener('click', () => {
  const text = historyEntries.map(e => e.text).join('\n\n');
  const blob = new Blob([text], { type: 'text/plain;charset=utf-8' });
  const url = URL.createObjectURL(blob);
  const a = document.createElement('a');
  a.href = url;
  a.download = `transcript-${Date.now()}.txt`;
  a.click();
});

2.5 Discord バッチング

バッファリングとフラッシュ:

async def add_entry(self, text: str, translations: Dict[str, str]) -> None:
    """確定文を追加（バッファが満杯なら即座に投稿）"""
    entry = self._format_message(text, translations)
    async with self._lock:
        if len("\n\n".join(self._buffer + [entry])) > self._max_chars:
            await self._flush_locked()
        self._buffer.append(entry)
        self._schedule_flush()

async def _delayed_flush(self) -> None:
    """指定秒数後にフラッシュ"""
    await asyncio.sleep(self._flush_interval)
    async with self._lock:
        await self._flush_locked()

メッセージフォーマット:

def _format_message(self, text: str, translations: Dict[str, str]) -> str:
    """原文＋翻訳を整形"""
    parts = [f"Esperanto: {text}"]
    for lang_code, translated in translations.items():
        label = LANG_LABELS.get(lang_code, lang_code.upper())
        parts.append(f"{label}: {translated}")
    return "\n".join(parts)

3. セットアップと運用

3.1 環境構築

必要な環境:

  • Python 3.11
  • 仮想オーディオデバイス（PipeWire/VoiceMeeter/BlackHole）
  • Speechmatics アカウント（Realtime 有効）
  • Google Cloud Translation API（サービスアカウント or APIキー）

インストール:

git clone git@github.com:Takatakatake/esperanto_onsei_mojiokosi.git
cd esperanto_onsei_mojiokosi
python3.11 -m venv .venv311
source .venv311/bin/activate
pip install --upgrade pip
pip install -r requirements.txt

システム依存パッケージ（Ubuntu）:

sudo apt update
sudo apt install -y build-essential libsndfile1-dev libportaudio2 portaudio19-dev ffmpeg

3.2 設定ファイル（.env）

# 音声認識
TRANSCRIPTION_BACKEND=speechmatics
SPEECHMATICS_API_KEY=<あなたのAPIキー>
SPEECHMATICS_CONNECTION_URL=wss://eu2.rt.speechmatics.com/v2
SPEECHMATICS_LANGUAGE=eo
AUDIO_DEVICE_INDEX=8  # python -m transcriber.cli --list-devices で確認

# Web UI
WEB_UI_ENABLED=true
WEB_UI_PORT=8765
WEB_UI_OPEN_BROWSER=true

# 翻訳
TRANSLATION_ENABLED=true
TRANSLATION_PROVIDER=google
TRANSLATION_TARGETS=ja,ko
TRANSLATION_DEFAULT_VISIBILITY=ja:on,ko:off
GOOGLE_TRANSLATE_CREDENTIALS_PATH=/path/to/service-account.json
GOOGLE_TRANSLATE_MODEL=nmt

# Discord
DISCORD_WEBHOOK_ENABLED=true
DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/...
DISCORD_BATCH_FLUSH_INTERVAL=2.0
DISCORD_BATCH_MAX_CHARS=350

# Zoom（任意）
ZOOM_CC_ENABLED=true
ZOOM_CC_POST_URL=https://wmcc.zoom.us/closedcaption?...

# ログ
TRANSCRIPT_LOG_ENABLED=true
TRANSCRIPT_LOG_PATH=logs/esperanto-caption.log

3.3 起動

デバイス確認:

python -m transcriber.cli --list-devices

通常起動:

python -m transcriber.cli --log-level=INFO

バックエンド切替:

python -m transcriber.cli --backend=vosk --log-file=logs/offline.log

翻訳テスト:

scripts/test_translation.py "Bonvenon al nia kunsido."

3.4 Web UI の安定起動

ポート解放スクリプト:

install -Dm755 scripts/run_transcriber.sh ~/bin/run-transcriber.sh
source .venv311/bin/activate
~/bin/run-transcriber.sh

手動でポート 8765 を解放:

pkill -f "python -m transcriber.cli" || true
sleep 0.2
lsof -t -iTCP:8765 | xargs -r kill || true
sleep 0.5 && lsof -iTCP:8765 || true

3.5 PipeWire のループバック固定（Linux）

問題: WirePlumber が既定入力を物理マイクに戻してしまう

解決策:

install -Dm755 scripts/wp-force-monitor.sh ~/bin/wp-force-monitor.sh
~/bin/wp-force-monitor.sh
cp systemd/wp-force-monitor.{service,path} ~/.config/systemd/user/
systemctl --user daemon-reload
systemctl --user enable --now wp-force-monitor.service wp-force-monitor.path

4. トラブルシューティング

4.1 音声が取れない

症状: Recognition did not start in time. または無音

原因と対処:

  1. デバイス番号が間違っている → --list-devices で確認
  2. ループバックが設定されていない → pactl load-module module-loopback を実行
  3. デバイスが切り替わった → 2秒待てば自動再接続されるはず

4.2 Speechmatics 接続エラー

症状: 404 path not found / 1003 unsupported data

対処:

  • JWT発行URLが https://mp.speechmatics.com/v1/api_keys?type=rt であることを確認
  • StartRecognition 形式が最新スキーマに一致しているか確認
  • リージョン（eu2/us2）と言語サフィックス（/eo）が正しいか確認

4.3 翻訳が表示されない

症状: Translation to ja failed ログ

対処:

  1. Google Cloud Translation の認証を確認
    • サービスアカウントJSONのパスが正しいか
    • サービスアカウントに Cloud Translation API 権限があるか
    • 課金設定が有効か
  2. LibreTranslate を使う場合
    • エンドポイント疎通を確認: curl https://libretranslate.de/languages
    • API キーが必要な場合は LIBRETRANSLATE_API_KEY を設定

4.4 Web UI のポートが埋まっている

症状: Address already in use

対処:

  • scripts/prep_webui.sh を実行してからCLIを起動
  • または手動で lsof -t -iTCP:8765 | xargs -r kill を実行

5. 運用のベストプラクティス

5.1 事前準備チェックリスト

  • 仮想オーディオの疎通確認（テスト音声→録音できるか）
  • .env を最新のキー/リージョン/デバイスに更新
  • 参加者へ字幕/録音の実施を事前告知
  • Discord Webhook の投稿先チャンネル/権限を確認
  • 翻訳APIのエンドポイント疎通・利用制限を確認

5.2 会議中のモニタリング

  • パイプライン起動後 Recognition started. を確認
  • Web UI を開いて画面共有（タブ共有が安定）
  • 翻訳とDiscord投稿が期待通りに反映されているか確認
  • CPU/GPU/ネットの負荷、ログ（Final/Partial）の進行を監視
  • Zoom: CC が表示されているか参加者に確認

5.3 会議後の整理

  • ログ（logs/）・Discord投稿を整理（必要に応じて削除/まとめ）
  • 誤認識単語を辞書・メモに追加して次回チューニング
  • Speechmatics/翻訳APIの利用量・コストを確認

5.4 セキュリティ対策

  • .env と logs/ は .gitignore に追加（機微情報をコミットしない）
  • Googleサービスアカウント JSON は絶対にリポジトリに入れない
  • 必要に応じてAPIキーのローテーションを実施
  • 参加者への録音・字幕の事前通知を徹底

6. パフォーマンスと拡張性

6.1 遅延の最適化

現状のE2E遅延: Zoom/Meet → 文字起こし → Web UI 表示まで約 1〜2 秒

ボトルネック:

  1. Speechmatics の WebSocket 往復（0.3〜0.5秒）
  2. 翻訳APIリクエスト（0.2〜0.5秒）
  3. ネットワーク遅延
  • 翻訳キャッシュの TTL を調整（既定では約32秒。要件に応じて短縮/延長）
  • Speechmatics の operating_point を enhanced にして精度と速度のバランスを調整
  • 翻訳を部分文でも先行実行（ただし確定前の訳は不安定）

6.2 スケーラビリティ

複数会議の並列処理:

  • 各会議ごとに独立した TranscriptionPipeline インスタンスを起動
  • Web UI ポートを会議ごとに変える（8765, 8766, ...）
  • Discord Webhook を会議ごとに分ける

大規模運用の検討事項:

  • systemd/pm2 で常駐化し、クラッシュ時の自動再起動
  • Prometheus などでメトリクス収集（遅延・リクエスト数・エラー率）
  • ログローテーション（logrotate or logging.handlers.RotatingFileHandler）

6.3 他言語への対応

Speechmatics 対応言語:

  • 英語（en）、スペイン語（es）、フランス語（fr）、ドイツ語（de）など30言語以上
  • .env の SPEECHMATICS_LANGUAGE を変更するだけ

翻訳ターゲットの追加:

TRANSLATION_TARGETS=ja,ko,en,zh
TRANSLATION_DEFAULT_VISIBILITY=ja:on,ko:on,en:off,zh:off

7. まとめ

7.1 達成したこと

  • リアルタイム文字起こし: Speechmatics を使ってエスペラント会話を1〜2秒遅延で文字化
  • 多言語同時翻訳: Google Cloud Translation で日本語・韓国語へ並列翻訳
  • 多様な出力先: Web UI / Discord / Zoom CC / ログファイルへ同時配信
  • 運用の安定性: デバイス切替自動追従、ポート解放スクリプト、PipeWire 固定化

7.2 今後の展開

  • カスタム辞書登録 UI: 固有名詞の誤認を削減するためのWeb管理画面
  • Meet 向け字幕オーバーレイ: Electron/OBS で Google Meet 画面に直接字幕を重畳
  • Whisper バックエンドの最適化: GPU/M シリーズで sub-second 遅延を実現
  • 自動リトライ・再接続の強化: ネットワーク断絶時の冗長化
  • 翻訳の高度化: 用語集サポート/キャッシュ戦略の改善/複数翻訳APIの冗長化

7.3 学んだこと

  • WebSocket ベースのリアルタイム通信は asyncio との相性が良い
  • 音声認識APIは認証・プロトコルのバージョン互換に注意が必要
  • 翻訳APIは並列リクエストとキャッシュでレイテンシを大幅削減できる
  • ユーザー体験（Web UI のトグル、履歴コピー等）は小さな工夫の積み重ねが重要

参考資料

おわりに

本記事で紹介したシステムは、エスペラント会話に特化していますが、設計思想やコンポーネント構成は他言語や他用途（会議録音、リアルタイム翻訳サービスなど）にも応用できます。特に、非同期I/O（asyncio）を活用した並列処理と、複数の外部APIを組み合わせたパイプライン設計は、様々なリアルタイムシステムで再利用可能です。

ぜひ、本記事を参考に、あなた自身のリアルタイム音声処理システムを構築してみてください！

リポジトリ: esperanto_onsei_mojiokosi

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?