１．はじめに

Azure OpenAI WhisperのAPIを活用したリアルタイム文字起こしツールのサンプルコードを作成してみました。このプロジェクトは、会議室での議事録作成の効率化を目的としています。従来の文字起こしサービスでは録音完了後に処理を行う必要があり、リアルタイムでの確認ができませんでした。さらに、長時間の会議では音声ファイルが大きくなりすぎ、処理速度が低下するという問題がありました。

このツールは、これらの課題を解決するため、音声の録音中にAzure OpenAI APIのレート制限に配慮しながら、25秒ごとのチャンクに分割してリアルタイムで文字起こしを実行します。
こちらで解説するコードはGitHubのpotofo/realtime-transcriptionにpush済みです。

２．技術的な課題

本プロジェクトの開発において、APIの利用制限とリアルタイム処理に関する複数の技術的課題に直面しました。これらの課題は、システムの安定性とパフォーマンスに直接的な影響を与えるため、適切な対策が必要でした。以下の表に主要な課題とその詳細をまとめます。

No	区分	課題	解説	備考
1	API制限	リクエスト制限	1分あたり3回のAPIリクエスト制限が存在	制限超過時は429エラーが発生
2	API制限	同時実行制限	複数リクエストの同時処理が制限される	キューイング処理で対応が必要
3	処理性能	録音の並行処理	音声録音と文字起こし処理の同時実行が必要	マルチスレッド処理で対応
4	処理性能	バッファ管理	音声データの一時保存と効率的な転送が必要	メモリ使用量の最適化が重要
5	システム	メモリ管理	長時間実行時のメモリリーク防止	定期的なガベージコレクションが必要
6	システム	エラーハンドリング	ネットワークエラーやAPI障害への対応	再試行機能の実装が必要

３．課題を考慮した設計

RealtimeTranscriberクラスを中心とした設計により、リアルタイムの音声録音と文字起こしを実現しています。環境変数による柔軟な設定と、各機能のモジュール化により、保守性と拡張性の高いシステムを構築しています。以下の表に実装された主要コンポーネントと処理フローをまとめます。

No	区分	実装内容	解説	備考
1	初期化設定	環境変数読み込み	Azure OpenAI API設定とモデル設定を.envから読み込み	openai.api_type, api_base, api_key, api_version
2	音声設定	録音パラメータ設定	サンプリングレート16kHz、モノラル、16bitで音声取得	FORMAT=paInt16, CHANNELS=1, RATE=16000
3	データ管理	音声チャンク制御	25秒単位でチャンクサイズを計算し、効率的な処理を実現	CHUNK = RATE × chunk_seconds
4	ファイル管理	保存ディレクトリ制御	audio/transcriptionディレクトリを自動生成	Path("audio"), Path("transcription")
5	音声処理	コールバック処理	音声データのキュー管理とファイル保存を非同期実行	audio_callback メソッド
6	デバイス制御	入力デバイス管理	デフォルト入力デバイスの自動検出と設定	get_default_input_device_index メソッド
7	API連携	Whisper API連携	音声ファイルの文字起こしをAzure OpenAI APIで実行	transcribe_audio メソッド
8	エラー処理	例外ハンドリング	各処理段階での例外捕捉と適切なエラーメッセージ表示	try-except blocks

４．実装の詳細

RealtimeTranscriberクラスを中心とした実装について、主要なコンポーネントごとに詳しく解説します。
こちらで解説するメソッドのコードはGitHubのpotofo/realtime-transcriptionの内容です。

4.1. 使用技術とライブラリ

Python 3.8+
非同期処理とモダンな言語機能の活用
筆者はPython-3.12.8で動作確認しています。
Azure OpenAI Whisper API
openaiモジュールはバージョン0.28.0を使用していますが、**高精度な音声認識エンジンです。
PyAudio
低レベルの音声入力制御
その他の依存パッケージ:
- pathlib
  ファイルパス操作
- dotenv
  環境変数管理
- queue
  音声データのバッファリング
- wave
  WAVファイル操作
- datetime
  タイムスタンプ生成

4.2. 初期化とシステム設定

Azure OpenAI Whisperを活用したリアルタイム文字起こしシステムの初期化とシステム設定について解説します。このシステムでは、16kHzのサンプリングレート、16bitの音声フォーマット、モノラル録音の設定を基本とし、環境変数による柔軟な設定管理と効率的な音声データの処理を実現しています。また、音声データのバッファリング、ファイルパス操作、タイムスタンプ生成など、複数の依存パッケージを活用して堅牢なシステム構成を実現しています。

def __init__(self):
    # 録音設定
    self.FORMAT = pyaudio.paInt16    # 16bitの音声フォーマット
    self.CHANNELS = 1                # モノラル録音
    self.RATE = 16000               # サンプリングレート16kHz

    # チャンクサイズの計算
    chunk_seconds = float(os.getenv("AUDIO_CHUNK_SECONDS", "25"))
    self.CHUNK = int(self.RATE * chunk_seconds)  # 25秒分のフレーム数

    # キューとディレクトリの初期化
    self.is_recording = False
    self.audio_queue = queue.Queue()
    self.audio_dir = Path("audio")
    self.transcription_dir = Path("transcription")

4.3. 音声録音の実装

音声データの取得と保存を担当する3つの主要メソッド

①音声コールバック処理
音声コールバック処理は、リアルタイム文字起こしシステムの中核となる機能です。録音中の音声データをキューに格納し、チャンク単位でファイルに保存する処理を行います。このメソッドは、PyAudioのストリームコールバックとして機能し、連続的な音声データの取得と管理を効率的に実現しています。

def audio_callback(self, in_data, frame_count, time_info, status):
    if self.is_recording:
        self.audio_queue.put(in_data)
        filename = self.save_audio_chunk(in_data)
    return (in_data, pyaudio.paContinue)

②音声チャンクの保存
音声チャンクの保存メソッドは、録音された音声データをWAVファイル形式で保存する重要な機能を担っています。タイムスタンプを用いた一意のファイル名生成と、適切な音声パラメータ設定により、高品質な音声データの永続化を実現します。このメソッドは、音声チャンクごとに整理された形でデータを保存し、後続の文字起こし処理のための基盤となります。

def save_audio_chunk(self, audio_data):
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    filename = self.audio_dir / f"chunk_{timestamp}_{self.chunk_counter:04d}.wav"
    with wave.open(str(filename), 'wb') as wf:
        wf.setnchannels(self.CHANNELS)
        wf.setsampwidth(pyaudio.get_sample_size(self.FORMAT))
        wf.setframerate(self.RATE)
        wf.writeframes(audio_data)

③デバイス管理
音声入力デバイスの初期化と設定を担当するメソッドです。PyAudioを使用してデフォルトの入力デバイスを検出し、そのインデックスを取得します。デバイスの使用後は適切にリソースを解放することで、メモリリークを防止します。

def get_default_input_device_index(self):
    audio = pyaudio.PyAudio()
    default_device_index = audio.get_default_input_device_info()['index']
    audio.terminate()
    return default_device_index

4.4. 文字起こし処理の実装

Whisper APIを使用した文字起こしと文字起こし結果の保存を行います。

①音声ファイルの文字起こし
Azure OpenAI Whisperを活用したリアルタイム文字起こしシステムにおいて、音声ファイルの文字起こし処理は、環境変数で管理されたAPI設定を使用し、日本語に特化した高精度な文字認識を実現します。このメソッドは、バイナリ形式で音声ファイルを読み込み、Whisper APIに送信して文字起こしを行い、余分な空白を除去した結果を返却する効率的な実装となっています。

def transcribe_audio(self, audio_file):
    try:
        with open(audio_file, "rb") as audio:
            result = openai.Audio.transcribe(
                api_key=os.getenv("OPENAI_API_KEY"),
                model="whisper-1",
                deployment_id=os.getenv("AZURE_DEPLOYMENT_ID"),
                file=audio,
                language="ja"
            )
            return result["text"].strip()

②文字起こし結果の保存
文字起こしシステムにおいて、文字起こし結果の保存機能は、音声ファイルから生成されたテキストを効率的に管理します。このメソッドは、音声ファイル名をベースにテキストファイルを生成し、UTF-8エンコーディングで保存することで、文字起こし結果の永続化を実現します。

def save_transcription(self, text, audio_filename):
    txt_filename = self.transcription_dir / Path(audio_filename).stem
    filepath = txt_filename.with_suffix('.txt')
    with open(filepath, "w", encoding="utf-8") as f:
        f.write(text)

4.5. メインの処理フロー

音声録音から文字起こしまでのメインプロセスを実装します。音声ストリームの初期化後、録音状態を監視しながら一定間隔で音声ファイルを処理し、WhisperAPIを使用して文字起こしを行います。文字起こし結果は自動的にテキストファイルとして保存されます。

def process_audio(self):
    # 音声ストリームの初期化
    audio = pyaudio.PyAudio()
    stream = audio.open(
        format=self.FORMAT,
        channels=self.CHANNELS,
        rate=self.RATE,
        input=True,
        input_device_index=self.get_default_input_device_index(),
        frames_per_buffer=self.CHUNK,
        stream_callback=self.audio_callback
    )

    # 録音と文字起こしのメインループ
    while self.is_recording:
        time.sleep(chunk_seconds)
        try:
            latest_audio_file = sorted(self.audio_dir.glob("*.wav"))[-1]
            text = self.transcribe_audio(str(latest_audio_file))
            if text:
                self.save_transcription(text, str(latest_audio_file))

５．エラーハンドリング

リアルタイム文字起こしシステムの安定性を確保するため、以下のような多層的なエラーハンドリングを実装しています。
これらのエラーハンドリング実装により、以下の効果を実現しています。

システムの安定性向上
予期せぬエラーによるクラッシュを防止
デバッグ容易性
詳細なエラーメッセージによる問題特定の効率化
ユーザー体験の向上
エラー発生時の適切なフィードバック提供
リソースの確実な管理
異常終了時でもシステムリソースの適切な解放を保証

5.1. 音声録音時のエラー処理

デバイス初期化エラー
音声入力デバイスの初期化時に発生する可能性のあるエラーを包括的に処理します

def get_default_input_device_index(self):
    try:
        audio = pyaudio.PyAudio()
        default_device_index = audio.get_default_input_device_info()['index']
        audio.terminate()
        return default_device_index
    except Exception as e:
        print(f"⚠️ 入力デバイスの取得に失敗しました: {e}")
        raise

マイクデバイスが見つからない場合のエラー処理
デバイス情報の取得失敗時のエラーメッセージ表示
リソースの適切な解放処理

音声ストリームのエラー
音声ストリームの処理中に発生する可能性のあるエラーを包括的に管理します。音声データのキューイングやファイル保存時の例外を適切に捕捉し、エラーメッセージを表示しながらもストリームの継続性を保証します。

def audio_callback(self, in_data, frame_count, time_info, status):
    if self.is_recording:
        try:
            self.audio_queue.put(in_data)
            filename = self.save_audio_chunk(in_data)
            print(f"✅ 音声チャンクを保存しました: {filename}")
        except Exception as e:
            print(f"⚠️ 音声データの処理中にエラーが発生しました: {e}")
    return (in_data, pyaudio.paContinue)

音声データのキューイング失敗時の例外処理
ファイル保存処理のエラーハンドリング
ストリーム継続の保証

5.2. ファイル操作のエラー処理

WAVファイル保存時のエラー
音声データをWAVファイルとして保存する際のエラーハンドリングを実装しています。タイムスタンプベースのファイル名生成、適切なオーディオパラメータの設定、そしてファイル書き込み処理を包括的に管理し、エラーが発生した場合は適切なエラーメッセージを表示します。

def save_audio_chunk(self, audio_data):
    try:
        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = self.audio_dir / f"chunk_{timestamp}_{self.chunk_counter:04d}.wav"
        with wave.open(str(filename), 'wb') as wf:
            wf.setnchannels(self.CHANNELS)
            wf.setsampwidth(pyaudio.get_sample_size(self.FORMAT))
            wf.setframerate(self.RATE)
            wf.writeframes(audio_data)
        return filename
    except Exception as e:
        print(f"❌ 音声ファイルの保存中にエラーが発生しました: {e}")
        return None

ディレクトリアクセス権限エラーの処理
ファイル書き込みエラーのハンドリング
タイムスタンプ生成エラーの対応

文字起こし結果の保存エラー
文字起こしの結果をテキストファイルとして保存する際のエラーハンドリングを実装しています。ファイルパスの生成、書き込み権限、エンコーディングに関する例外を捕捉し、エラーメッセージとともに処理中のテキストを表示します。

def save_transcription(self, text, audio_filename):
    try:
        txt_filename = self.transcription_dir / Path(audio_filename).stem
        filepath = txt_filename.with_suffix('.txt')
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(text)
        print(f"✅ 保存完了: {filepath}\\n")
    except Exception as e:
        print(f"\\n❌ 文字起こし結果の保存中にエラーが発生しました: {e}")
        print(f"テキスト: {text}")

ファイルパス生成エラーの処理
書き込み権限エラーのハンドリング
エンコーディングエラーの対応

5.3. API連携のエラー処理

Whisper API呼び出し時のエラー
音声ファイルの文字起こし処理時に発生する可能性のある様々なエラーを包括的に処理します。API認証、ネットワーク接続、レート制限、応答形式などの異常を検知し、適切なエラーメッセージを表示しながら、システムの安定性を確保します。

def transcribe_audio(self, audio_file):
    try:
        with open(audio_file, "rb") as audio:
            result = openai.Audio.transcribe(
                api_key=os.getenv("OPENAI_API_KEY"),
                model="whisper-1",
                deployment_id=os.getenv("AZURE_DEPLOYMENT_ID"),
                file=audio,
                language="ja"
            )
            return result["text"].strip()
    except Exception as e:
        print(f"\n❌ 文字起こし処理中にエラーが発生しました: {e}")
        return None

API認証エラーの処理
ネットワーク接続エラーのハンドリング
レート制限エラーの対応
応答形式エラーの処理

5.4. メインプロセスのエラー処理

プロセス終了時の制御
メインプロセスの終了処理では、音声録音の停止、リソースの解放、エラー処理を適切に行い、システムの安全な終了を保証します。以下のコードでは、キーボード割り込みや例外発生時の処理、およびPyAudioリソースのクリーンアップを実装しています。

def process_audio(self):
    try:
        while self.is_recording:
            time.sleep(chunk_seconds)
            try:
                latest_audio_file = sorted(self.audio_dir.glob("*.wav"))[-1]
                text = self.transcribe_audio(str(latest_audio_file))
                if text:
                    self.save_transcription(text, str(latest_audio_file))
            except IndexError:
                print("\n⚠️ 音声ファイルが見つかりません")
            except Exception as e:
                print(f"\n❌ 処理中にエラーが発生しました: {e}")
    except KeyboardInterrupt:
        print("\n⏹️ 録音を停止します...")
        self.is_recording = False
    finally:
        # PyAudioのクリーンアップ処理
        try:
            stream.stop_stream()
            stream.close()
            audio.terminate()
            print("✅ 録音リソースを解放しました")
        except Exception as e:
            print(f"⚠️ リソース解放中にエラーが発生しました: {e}")

キーボード割り込み時の適切な終了処理
音声ファイル不在時のエラー処理
リソース解放失敗時のエラーハンドリング
プロセス状態の適切な管理

６．セットアップと実行方法

環境構築手順
設定ファイルの説明
- API認証情報
- 音声パラメータ
基本的な使い方
トラブルシューティング

6.1. リポジトリのクローン

git clone https://github.com/potofo/realtime-transcription.git
cd realtime-transcription

6.2. 仮想環境の作成

py -3.12 -m venv venv
venv\Scrips\activate.ps1

6.3. 依存パッケージのインストール

pip install -r requirements.txt

6.4. 環境変数の設定

.env_templateを.envにコピー

以下の項目を設定
※Azure OpenAIのwhisperにはレイト制限(1分間あたりの要求制限＝3)があるため、AUDIO_CHUNK_SECONDSを21秒以下にするとAPIがエラーを応答します。

OPENAI_API_TYPE=azure
OPENAI_API_HOST=your_endpoint_here
OPENAI_API_KEY=your_api_key_here
OPENAI_API_VERSION=2024-06-01
AZURE_DEPLOYMENT_ID=whisper

# 音声録音設定
AUDIO_CHUNK_SECONDS=25  # チャンクサイズ（秒）

6.5. 実行方法

プログラムの起動は以下のコマンドで行います。

python realtime_transcription.py

プログラム起動後、自動的に録音が開始されます
音声は設定された時間間隔（デフォルト25秒）でチャンクに分割されます
各チャンクは自動的に文字起こしされ、結果が表示されます
終了するにはCtrl+Cを押してください

７．利用シナリオと運用

会議室での利用シナリオ
- 最適な設置位置
  会議室での音声文字起こしシステムの最適な設置位置について、マイクを会議参加者の中心に配置することで、均一な音声収集が可能になります。
- 音声入力の調整
  音声入力の調整について、マイクを会議参加者の中心に配置することで均一な音声収集が可能になります。
- 長時間利用時
  ディスク容量の管理が重要で、例えば4時間の会議で約2GBの音声データが生成されるため、定期的なクリーンアップが必要です
パフォーマンスチューニング
- メモリ使用量の最適化
  古い音声データを自動的に削除し、チャンク処理時のバッファサイズを1MBに制限することで、長時間の会議でも安定した動作を実現しています。
- CPU負荷の軽減
  Azure OpenAI APIでWhisperを利用することで、重い生成AI処理をクラウド側で実現し、メインスレッドのCPU負荷を軽減しています。
運用上の注意点
- APIコストの管理
  APIコストの管理について、Whisper APIのレート制限（1分間あたり3リクエスト）に対応するため、音声チャンクを25秒間隔で処理することで、APIコストを最適化しています。
- データの取り扱い
  音声データと文字起こし結果には機密情報が含まれる可能性があるため、Azure OpenAI APIは閉域化した環境化で利用することを推奨します。

８．今後改善可能な点

8.1. 改善可能な点

UI/UXの向上
ユーザーインターフェースとユーザー体験の改善を通じて、より使いやすいシステムを目指します。
精度の改善
文字起こしの精度をさらに向上させることで、より正確な議事録作成を実現します。
スケーラビリティの向上
システムの拡張性を改善し、より大規模な利用に対応できるようにします。

8.2. 追加機能の検討

話者の識別
個々の発言者を区別できる機能を追加し、より詳細な議事録作成を可能にします。
但し、現在のWhisperやgpt-4o-realtime-previewモデルでは話者識別は残念ながら対応できていません。
OpenAI ChatGPTのAdvanced Voice Modeでは出来ていそうなので、今後に期待です。
要約機能の追加
文字起こしされたテキストの自動要約機能を実装し、効率的な情報把握を支援します。
文字起こしされたテキストは会議の最後にサマリーとして、ファシリテーターが付議内容、決定事項、持ち帰り事項などを容易に整理できるようにしたいです。

９．まとめ

9.1. 実装のポイント

音声データの効率的な分割処理
- 25秒ごとのチャンク処理による適切なAPI利用
- バッファサイズの1MB制限による安定性確保
堅牢なエラーハンドリング
- 音声ファイル不在時の例外処理
- 適切なリソース解放処理
- キーボード割り込み時の終了処理
最適化されたリソース管理
- 古い音声データの自動削除機能
- クラウドでのAI処理による負荷分散

9.2. 得られた知見

API制限への対応
- Whisper APIの1分間あたり3リクエスト制限を考慮した設計
- 25秒以上のチャンク間隔による安定運用
システムリソースの管理
- 4時間の会議で約2GBの音声データが生成されることを確認
- 定期的なデータクリーンアップの重要性
実運用上の考慮点
- 会議室での最適なマイク配置の重要性
- 機密情報保護のための閉域環境での運用推奨

9.3. 技術的な知見

APIのレート制限（1分間あたり3リクエスト）に対応するには、25秒以上のチャンク間隔が必要
長時間の会議（4時間）では約2GBの音声データが生成される
安定した動作のために
- チャンク処理時のバッファサイズを1MBに制限する必要がありそう
- 古い音声データの自動削除機能の実装が必要
重いAI処理はAzure OpenAI APIでクラウド側で処理することでCPU負荷を軽減可能
エラーハンドリングの実装ポイント
- 音声ファイル不在の例外処理
- キーボード割り込み時の終了処理
- リソース解放時の例外処理