こんにちは！
KDDIアイレットの取り組みとして6月22日〜7月3日の期間で開催中の「Google Cloud Next '26 / Google I/O やってみた系ブログリレー」の投稿です。
はじめに
今年の「Google I/O 2026」において、個別の「Speech-to-Text API」自体のメジャーアップデートに関する発表はありませんでしたが、Googleのエコシステム全体で、AIによる音声認識・音声入力機能が大幅に強化・統合されるという非常に重要な方向性が示されました。
これは、「高精度な音声解析はもはや特別な技術ではなく、あらゆるアプリケーションの標準機能になる」という強力なメッセージでもあります。インタビューやポッドキャスト、複数人のミーティング録音などを、「誰が」「いつ」「何を」話したかまで自動でテキスト化するニーズは、今後さらに加速していくでしょう。
この記事では、FastAPI と Google Cloud の最新音声認識 API である Speech-to-Text v2（Chirp 3 モデル） を組み合わせ、実運用に耐えうる「会話の文字起こし＆話者分離」のバックエンドをスマートに実装する方法を解説します。
1. 技術スタック と 全体フロー
API のレスポンスをブロックせずに裏側で重い音声解析を実行するため、FastAPI の BackgroundTasks を活用した非同期アーキテクチャを採用しています。
技術スタック
- 言語/フレームワーク: Python 3.10+ / FastAPI
-
音声認識: Google Cloud Speech-to-Text v2 (SpeechAsyncClient,
chirp_3モデル) - ストレージ: Cloud Storage (音声ファイルおよび文字起こしテキストの保存)
-
非同期実行: FastAPI
BackgroundTasks+asyncio
全体フロー
[クライアント]
│ 音声ファイルをPOST
▼
[FastAPI]
│ 1. 音声ファイルを Cloud Storage (GCS) に保存
│ 2. DBのオーディオURLを更新
▼
[FastAPI]
│ BackgroundTasks に解析ジョブを投入
▼ (即座にクライアントへ 202 Accepted 相当を返却)
[バックグラウンド処理]
│ ① HTTPS URL を Speech-to-Text 用の gs:// URI に変換
│ ② Speech-to-Text v2 (BatchRecognize) を非同期実行
│ ③ 返ってきた単語列を話者ごとに区切ってセグメント化
│ ④ 話者ラベルを実名にマッピングしてDBへ保存 / テキストをGCSに保存
▼
[完了]
2. Speech-to-Text v2 + Chirp 3 を選んだ理由
複数人のリアルな会話を高精度に処理したい場合、Google Cloud の最新構成には圧倒的な優位性があります。
-
日常会話に強い日本語精度
最新のchirp_3モデルは、従来の音声認識モデル（v1系）と比較して、日本語特有の相槌やくだけた表現、複数人が交差して話すような会話の認識精度が飛躍的に向上しています。 -
長尺音声に対する高い安定性
対談やインタビューは数十分から1時間以上に及ぶことが一般的です。v2 の「バッチ認識（BatchRecognize）」を活用することで、長時間の重い音声ファイルでもシステムをタイムアウトさせることなく、安定して処理を行うことが可能です。
3. 【コア実装】単語列から「会話セグメント」へ再構築する
Speech-to-Text v2 で話者分離（Speaker Diarization）を有効にすること自体は簡単ですが、API から返ってくる結果は「文章ごと」ではなく、「単語ごと」に分割されたデータとして返ってきます。
API からは、「こんにちは（話者1）」「今日は（話者1）」「よろしく（話者2）」のように、細かい単語単位でタイムスタンプと話者ラベルが付与されて返ってくるため、バックエンド側で「話者が切り替わるタイミング」を検知して文章にまとめるロジックが必要です。
この状態管理のロジックは愚直に書くと長くなりがちですが、Python の itertools.groupby を使って実装できます。
話者分離のパースを最適化した文字起こしコード
from itertools import groupby
from google.api_core.client_options import ClientOptions
from google.cloud.speech_v2 import SpeechAsyncClient
from google.cloud.speech_v2.types import cloud_speech
from typing import TypedDict
class _SegmentData(TypedDict):
speaker_label: str # STTが付与する "1", "2" などの話者ラベル
text: str # その話者の発話テキスト
start_time_ms: int # 開始時刻 (ms)
end_time_ms: int # 終了時刻 (ms)
confidence: float | None # 信頼度
async def _transcribe_audio(gcs_uri: str, project_id: str) -> list[_SegmentData]:
"""Speech-to-Text v2 BatchRecognize で音声を文字起こしし、話者ごとにセグメント化する"""
# リージョン（今回はEU）のエンドポイントを明示的に指定してクライアントを初期化
client = SpeechAsyncClient(
client_options=ClientOptions(api_endpoint="eu-speech.googleapis.com")
)
request = cloud_speech.BatchRecognizeRequest(
recognizer=f"projects/{project_id}/locations/eu/recognizers/_", # `_` はアドホック認識器
config=cloud_speech.RecognitionConfig(
auto_decoding_config=cloud_speech.AutoDetectDecodingConfig(),
language_codes=["ja-JP"],
model="chirp_3",
features=cloud_speech.RecognitionFeatures(
enable_word_time_offsets=True,
diarization_config=cloud_speech.SpeakerDiarizationConfig(), # 話者分離を有効化
),
),
files=[cloud_speech.BatchRecognizeFileMetadata(uri=gcs_uri)],
recognition_output_config=cloud_speech.RecognitionOutputConfig(
inline_response_config=cloud_speech.InlineOutputConfig(),
),
)
# 非同期でAPIを呼び出し、長尺音声に備えてタイムアウトを長め（10分）に設定して待機
response = await (await client.batch_recognize(request=request)).result(timeout=600)
segments: list[_SegmentData] = []
for file_result in response.results.values():
for res in file_result.transcript.results:
# セイウチ演算子で空チェックと代入をスマートに記述
if not (alts := res.alternatives) or not (words := alts[0].words):
continue
conf = float(alts[0].confidence) if alts[0].confidence else None
# 【コア改善】groupby で連続する同じ話者の単語を自動でグループ化
for speaker, group in groupby(words, key=lambda w: w.speaker_label or "1"):
word_list = list(group)
segments.append({
"speaker_label": speaker,
"text": "".join(w.word for w in word_list),
"start_time_ms": int(word_list[0].start_offset.total_seconds() * 1000),
"end_time_ms": int(word_list[-1].end_offset.total_seconds() * 1000),
"confidence": conf,
})
return segments
手動で「前の話者」を記録する変数（フラグ）を管理する代わりに、groupby を適用することで、複雑な条件分岐を一掃し、保守性の高いコードを実現しています。
4. バックグラウンド実行と堅牢なエラーハンドリング
文字起こし処理は数分かかるため、FastAPI 側はジョブを BackgroundTasks に投入して即座にレスポンスを返します。
裏側で走るタスク（run_analysis）では、外部 API 連携に伴う例外を確実にキャッチし、失敗時にはステータスを FAILED に倒して Cloud Logging 等に例外を残す設計が実運用では不可欠です。
import logging
from fastapi import BackgroundTasks
logger = logging.getLogger(__name__)
async def run_analysis(meeting_id: int) -> None:
"""BackgroundTasks から呼び出すエントリポイント"""
try:
# DBステータスを PROCESSING に変更し、文字起こしやGCS・DB保存を実行
await _do_analysis(meeting_id)
except Exception:
# 例外発生時は確実にエラーログを残し、ステータスを FAILED に更新する
logger.exception("解析処理に失敗しました: meeting_id=%d", meeting_id)
await _mark_failed(meeting_id)
# FastAPIのエンドポイント側での呼び出し方はこれだけです
# background_tasks.add_task(run_analysis, meeting_id)
補足：話者ラベルを実名にマッピングする
API から取得した "1" や "2" といった無機質な話者ラベルは、保存時にアプリケーション上の実名（例：ホスト名、ゲスト名など）へマッピングしてあげると、フロントエンドでの表示が非常に楽になります。
# Speech-to-Text の "1" / "2" を人名にマッピング
speaker_name_map = {"1": user_name, "2": guest_name}
transcript_segments = [
TranscriptSegment(
meeting_id=meeting_id,
speaker_label=speaker_name_map.get(seg["speaker_label"], seg["speaker_label"]),
text=seg["text"],
start_time_ms=seg["start_time_ms"],
end_time_ms=seg["end_time_ms"],
)
for seg in segments
]
5. まとめ
Google のエコシステム全体で AI による音声技術の統合が進む中、独自プロダクトにおいても高精度な音声解析をシームレスに組み込むことは、今後のスタンダードになっていきます。
最新の AI モデルも、FastAPI のようなモダンなフレームワークや Python の強力な標準ライブラリ（itertools.groupby など）と組み合わせることで、初めて美しく実用的なバックエンドとして機能します。ぜひ皆さんのプロダクトでも試してみてください！