【第1弾：LLM×音声認識】Nottaが使いにくかったので、Whisper×PyAnnote×Gemini APIで音声文字起こしパイプラインを自作した話

はじめに：なぜ自作したのか

市販の音声文字起こしツール（例：Nottaなど）を利用する中で、次のような制限にもどかしさを感じたことはないでしょうか。

• 録音時間に関する制限（3分縛りなど）

• 月間利用時間の制限

• 完全クラウド依存によるプライバシーや仕様変更への不安

こうした制約を解消するため、今回はオープンソースのローカルモデル（Whisper、PyAnnote）とクラウドAPI（Gemini API）を組み合わせ、自分に合った音声認識・議事録生成パイプラインを構築しました。第1弾となる本エントリでは、構築したアーキテクチャや実際の使い心地、直面した課題などを共有します。

開発環境と使用したツール

本パイプラインの構築において使用した主な環境およびツール・ライブラリの構成は、以下の通りです。

• 実行環境: Google Colab（GPU：NVIDIA T4等を使用）

• プログラミング言語: Python

• 音声認識（文字起こし）: OpenAI Whisper

• 話者分離: Pyannote.audio （Hugging Face上のモデルおよびアクセストークンを利用）

• LLM（要約・議事録生成）: Google GenAI SDK / Gemini API

今回実装したパイプラインの全体像

本エントリで紹介する第1弾のパイプラインでは、音声データを以下のステップで処理していきます。

- Step1 [前処理]: 録音データを取り込み、無音区間やノイズを除去する

- Step2 [分割処理]: 長時間のデータは処理にかけやすいように分割する

- Step3 [文字起こし ＆ 話者分離]:

　　- 文字起こし: Whisperを使用

　　- 話者分離: Hugging Face（PyAnnote）を使用
　　 （※Hugging Faceのアクセストークン等を一部経由するため、完全なオフライン環境ではない）

- Step4 [統合]: 分割した音声データを統合して話者情報を繋ぎ合わせる

- Step5 [要約]: 統合したデータをGemini APIに渡し、議事録として出力する

各ステップでの実装コード例

• Step1 [前処理]: 録音データを取り込み、ノイズを除去する

import torch
import torchaudio
import noisereduce as nr

# 1. 音声ファイルの読み込み（例：無音区間を除去した音声ファイルなど）
input_file = "audio.wav"
audio_tensor, sample_rate = torchaudio.load(input_file)

# 2. noisereduceの入力用にPyTorchのテンソルからNumPy配列へ変換
audio_np = audio_tensor.numpy()

# 3. ノイズ除去の実行
# ※必要に応じて prop_decrease などのパラメータ調整も可能です
reduced_noise_np = nr.reduce_noise(y=audio_np, sr=sample_rate)

# 4. 処理後のデータを再度テンソルに変換し、ファイルとして保存
cleaned_tensor = torch.tensor(reduced_noise_np)
output_file = "enhanced_audio.wav"
torchaudio.save(output_file, cleaned_tensor, sample_rate)

print(f"ノイズ除去が完了しました！ 保存ファイル名: {output_file}")

• Step2 [分割処理]: 長時間のデータは処理にかけやすいように分割する

import os
from pydub import AudioSegment

# 1. パス設定
audio_path = "enhanced_audio.wav"
chunk_dir = "audio_chunks"
os.makedirs(chunk_dir, exist_ok=True)

print(f"音声をロード中: {audio_path}")
audio = AudioSegment.from_file(audio_path)

# 2. パラメータ設定（例：10分ごとに分割）
chunk_duration_ms = 10 * 60 * 1000  # 10分
chunk_files = []
audio_len_ms = len(audio)

start_ms = 0
idx = 0

while start_ms < audio_len_ms:
    # 終了位置を算出（音声全体の長さを超えないように調整）
    end_ms = min(start_ms + chunk_duration_ms, audio_len_ms)

    # チャンクの切り出しと保存
    chunk = audio[start_ms:end_ms]
    chunk_filename = os.path.join(chunk_dir, f"chunk_{idx:03d}.wav")
    chunk.export(chunk_filename, format="wav")

    chunk_files.append((idx, chunk_filename, start_ms / 1000.0))
    print(f"チャンク {idx:03d} 生成: {start_ms/1000.0:.1f}秒 〜 {end_ms/1000.0:.1f}秒")

    start_ms = end_ms
    idx += 1

print(f"\n分割完了: 全 {len(chunk_files)} 個のチャンクを生成しました。")

• Step3 [文字起こし ＆ 話者分離]:文字起こしにはWhisperを使用

import os
import json
import torchaudio
import whisper  # 音声認識（文字起こし）用のオープンソースモデル
from pyannote.audio import Pipeline  # 話者分離用のモデル

# 0. 事前準備：Hugging Faceのアクセストークンとモデルのロード
# ※PyAnnoteの利用には、Hugging Faceでの規約同意とアクセストークンが必要です
hf_token = "hf_xxxxxxxxxxxxxxxxxxxxxxxxxxxxx"  # ご自身のトークンに書き換えてください
diarization_pipeline = Pipeline.from_pretrained(
    "pyannote/speaker-diarization-3.1",
    token=hf_token
)

# 1. Whisperモデルのロード（例：ベースモデルやラージモデルなどを指定）
whisper_model = whisper.load_model("base")

# 2. 結果を格納するリストの初期化
all_transcript_segments = []

print(f"処理対象チャンク数: {len(chunk_files)} 個\n")

# 3. チャンクごとのループ処理
for idx, chunk_path, offset_sec in chunk_files:
    print(f"--- チャンク処理中: {chunk_path} (オフセット: {offset_sec:.1f}秒) ---")

    # 話者分離の実行
    diarization_result = diarization_pipeline(chunk_path)
    
    # 検出されたセグメントの取得
    segments_list = list(diarization_result.itertracks(yield_label=True))

    # チャンクの波形データをロード
    waveform, sample_rate = torchaudio.load(chunk_path)

    # 各話者セグメントごとに切り出してWhisperで文字起こし
    for turn, _, speaker in segments_list:
        start_sec = turn.start
        end_sec = turn.end

        # 短すぎるセグメント（例: 0.5秒未満）はスキップ
        if (end_sec - start_sec) < 0.5:
            continue

        # 全体のタイムスタンプに補正
        absolute_start = start_sec + offset_sec
        absolute_end = end_sec + offset_sec

        # 対象セグメントの音声をスライスして一時保存
        start_sample = int(start_sec * sample_rate)
        end_sample = int(end_sec * sample_rate)
        segment_waveform = waveform[:, start_sample:end_sample]
        
        temp_segment_path = "temp_segment.wav"
        torchaudio.save(temp_segment_path, segment_waveform, sample_rate)

        # Whisper（音声認識モデル）で文字起こしを実行
        result = whisper_model.transcribe(temp_segment_path, language="ja")
        text = result["text"].strip()

        if text:
            all_transcript_segments.append({
                "speaker": f"speaker_{speaker}",
                "start": round(absolute_start, 2),
                "end": round(absolute_end, 2),
                "text": text
            })

# 4. 結果をJSONとして保存
output_json = "diarized_transcript.json"
with open(output_json, "w", encoding="utf-8") as f:
    json.dump(all_transcript_segments, f, ensure_ascii=False, indent=2)

print(f"\n文字起こしと話者分離が完了しました！ 保存先: '{output_json}'")

• Step4 [統合]: 分割した音声データを統合して話者情報を繋ぎ合わせる

import json

# 1. 保存したJSONの読み込み
with open("diarized_transcript.json", "r", encoding="utf-8") as f:
    segments = json.load(f)

# 2. 開始時間の昇順にソート
segments_sorted = sorted(segments, key=lambda x: x["start"])

# 3. 連続する同一話者の発話をまとめる（マージ処理）
merged_segments = []
current = segments_sorted[0].copy()

for next_seg in segments_sorted[1:]:
    # 同一話者かつ、発話の間隔が近い（例: 2秒以内）場合は結合する
    if current["speaker"] == next_seg["speaker"] and (next_seg["start"] - current["end"]) < 2.0:
        current["end"] = next_seg["end"]
        current["text"] += " " + next_seg["text"]
    else:
        merged_segments.append(current)
        current = next_seg.copy()

merged_segments.append(current)

# 4. 統合結果をJSONとして上書き保存
output_json = "diarized_transcript.json"
with open(output_json, "w", encoding="utf-8") as f:
    json.dump(merged_segments, f, ensure_ascii=False, indent=2)

print(f"マージ完了！ 統合されたセグメント数: {len(merged_segments)}件")

# 5. 「MM:SS」形式のタイムスタンプと話者名付きテキストに整形して保存
output_txt = "transcript_with_timestamps.txt"
with open(output_txt, "w", encoding="utf-8") as f:
    for seg in merged_segments:
        speaker = seg["speaker"]
        
        # 秒数を 「MM:SS」 形式へ変換
        start_min, start_sec = divmod(int(seg["start"]), 60)
        end_min, end_sec = divmod(int(seg["end"]), 60)

        # 出力行の作成と書き込み
        line = f"[{start_min:02d}:{start_sec:02d} - {end_min:02d}:{end_sec:02d}] {speaker}: {seg['text']}\n"
        f.write(line)

print(f"整形テキストの保存が完了しました: '{output_txt}'")

• Step5 [要約]: 統合したデータをGemini APIに渡し、議事録として出力する

import json
from google import genai
from google.colab import userdata

# 0. 事前準備：Gemini APIキーの設定とクライアントの初期化
gemini_api_key = "AIzaSyXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"  # ご自身のAPIキーに書き換えてください

# 1. APIクライアントの初期化
client = genai.Client(api_key=gemini_api_key)

# 2. タイムスタンプ・話者付きJSONの読み込み
with open("diarized_transcript.json", "r", encoding="utf-8") as f:
    segments = json.load(f)

# 3. プロンプト用にテキストを組み立て
formatted_text_lines = []
for seg in segments:
    start_min, start_sec = divmod(int(seg["start"]), 60)
    end_min, end_sec = divmod(int(seg["end"]), 60)
    formatted_text_lines.append(f"[{start_min:02d}:{start_sec:02d} - {end_min:02d}:{end_sec:02d}] {seg['speaker']}: {seg['text']}")

full_text = "\n".join(formatted_text_lines)

prompt = f"""
以下の会議の文字起こしデータをもとに、ビジネス用の洗練された議事録を作成してください。
以下の構成（Markdown）に従って分かりやすく整理してください。

- 概要
- 決定事項
- 主要な議論の流れ
- TODO・ネクストアクション

【会議の文字起こしデータ】
{full_text}
"""

print("Gemini APIで議事録を生成中...")

# 4. Gemini APIで議事録を生成
response = client.models.generate_content(
    model='gemini-2.5-flash',
    contents=prompt,
)

# 5. マークダウンファイルとして保存
output_filename = "meeting_minutes.md"
with open(output_filename, "w", encoding="utf-8") as f:
    f.write(response.text)

print(f"議事録の生成が完了しました！ 保存先: '{output_filename}'")

実際の使い心地と検証結果

約50分・6人以上の多人数という、比較的負荷の高い条件下で実装を試みました。

• 使用感: 非常に快適に使用でき、出力される議事録のクオリティも高い水準に仕上がりました。文字起こしの段階ではまだまだ誤字脱字も少なくなかったのですが、LLMの部分で吸収してくれて議事録としての完成度はかなり高いと感じました。

• 不満点（課題）: 話者分離の精度については、かなり不満が強かったです。事前に話者人数を指定していなかったり、声紋の事前学習をはさむことでまだ改善の余地を感じます。

• GPUの壁（重要なポイント）:
  50分音源の処理を行った際、Google Colabの無料GPU枠の制限に到達してしまいました。「それではCPUで処理できないか」と試してみたものの、絶望的に処理が重く、実用に耐えないという結果になりました。そのため、この規模の処理にはGPUが不可欠であると痛感しております。

  • ※運用コストの肌感としては、「半日おきに1時間分の生データの処理」という程度であれば、Colabの無料枠の範囲内で十分運用できそうな手応えでした。

まとめと次回予告

今回はHugging Face上のモデルとGemini APIを組み合わせたクラウド併用型により、実用に十分耐えうるパイプラインを構築することができました。権利やライセンス周りのクリアランスには留意が必要ですが、ツールとしての実用性は非常に高いものに仕上がったと感じました。

次回の第2弾では、クラウドAPI（Gemini）の部分を「ローカルLLM」に置き換えて比較検証していきたいと思っています。

📬 お仕事・開発のご相談について

音声処理やAIを活用した業務効率化・自動化システムの開発ご相談、お見積もりは大歓迎です。どうぞお気軽にご連絡ください。