OpenAIにおける文字起こし（音声認識）の現在地

nokomoro3
2025.04.15
こんちには。
データ事業本部 インテグレーション部 機械学習チームの中村( @nokomoro3 )です。
本記事ではOpenAIで文字起こし（音声認識）を実現するための方法を整理して、最新の方法で実際に動かしてみたいと思います。
 冒頭まとめOpenAIで文字起こしをするには whisper-1 、gpt-4o-mini-transcribe 、gpt-4o-transcribe の3つのモデルが使用可能
いずれもRealtime APIでストリーミング入力に対応可能で、接続方式はWebSocket
本記事では実際にPythonを使って、wavファイルを分割したストリーミング入力で文字起こしを動かしてみている
 はじめに OpenAIにおける文字起こしモデルドキュメントにSpeech to Textとしてまとめられています。
Speech to text - OpenAI API
記載のとおり、長らく文字起こしは whisper-1 モデルのみが利用可能でした。
これがネイティブにマルチモーダルに対応した gpt-4o をベースとする、文字起こし用のモデル gpt-4o-mini-transcribe と gpt-4o-transcribe が2025年3月20日に発表され、より高品質な文字起こしが可能になっています。
Introducing next-generation audio models in the API | OpenAI
これにより、OpenAIとして文字起こしに使用できるモデルは３つとなりました。
 Realtime APIによるストリーミング入力対応また、 whisper-1 は長らく一括入力（バッチ入力）にしか対応しておらず、入力をマイクデータ等のストリーミングで与えることはできませんでした（出力レスポンスはストリーミングで得る方法がありましたが）。
こちらはRealtime APIを使うことによって、whisper-1 、gpt-4o-mini-transcribe 、gpt-4o-transcribe のいずれのモデルでもストリーミング入力での処理が可能となっています。
より具体的には、以下に言及があります。
https://platform.openai.com/docs/guides/speech-to-text#streaming-the-transcription-of-an-ongoing-audio-recording
Realtime APIは、音声認識と音声合成を組み合わせることにより、AIとの音声による自然な対話が可能となり、低遅延でマルチモーダルな体験をアプリに組み込めるようにするAPIです。
2024年10月にベータ版として発表され、現在でも公式ドキュメントではベータ版となっています。
Introducing the Realtime API | OpenAI
公式ドキュメントとしては、以下に記載があります。
Realtime API - OpenAI API
Realtime API自体は、WebRTCとWebScoketの2つの接続方式に対応しているようですが、文字起こしのストリーミング入力については公式ガイドでは以下のWebSocketのエンドポイントを使用するように案内があります。
wss://api.openai.com/v1/realtime?intent=transcription
ですので、今回はガイドに記載されている通り、WebSocket接続でやってみようと思います。
 WebSocketの使用方法 WebSocketへの接続Pythonを使います。以下のコードでWebSocketの接続を開始できます。（まだイベントの送受信を記載してないモック状態です）
import os
import json
import websocket

OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")

url = "wss://api.openai.com/v1/realtime?intent=transcription"
headers = [
    "Authorization: Bearer " + OPENAI_API_KEY,
    "OpenAI-Beta: realtime=v1"
]

def on_open(ws):
    print("Connected to server.")

    # TODO: ここにイベント送信処理を記載する

def on_message(ws, message):
    data = json.loads(message)
    print("Received event:", json.dumps(data, indent=2))

    # TODO: ここにイベント受信時の処理を記載

ws = websocket.WebSocketApp(
    url,
    header=headers,
    on_open=on_open,
    on_message=on_message,
)

ws.run_forever()
 WebSocketとやり取りするイベントについてWebSocket接続時のイベントは以下でその一覧を確認することができます。
Realtime conversations - OpenAI API
一覧では初期化時に session.update を使うよう記載されていますが、 文字起こしの場合は transcription_session.update を使う必要がありそうです。
そのため、おおむね以下のような流れでイベントを送信します。
初期化時
クライアントから送信するイベント
設定値の送信: transcription_session.update

サーバーから受信するイベント
transcription_session.created
transcription_session.updated


処理中
クライアントから送信するイベント
音声データの送信: input_audio_buffer.append|

サーバーから受信するイベント
音声区間の開始: input_audio_buffer.speech_started
音声区間の終了: input_audio_buffer.speech_stopped
音声区間の確定: input_audio_buffer.committed
音声認識処理の開始: conversation.item.created
音声認識の途中結果: conversation.item.input_audio_transcription.delta
音声認識の確定結果: conversation.item.input_audio_transcription.completed


処理中にサーバーから受信するイベントは少し複雑ですが、 input_audio_buffer.speech_started と同時にitem_idが割り当てられ、そのitem_idで以降のイベントが順次受信されます。
input_audio_buffer.committed で音声区間が確定後、音声認識処理が実行同時に順番が確定します。これ以降は認識処理の結果は、音声区間の検出と並列で実行されますので、各item_idの処理が入り混じることになります。
そのため、 conversation.item.input_audio_transcription.completed の受信タイミングは実際の音声区間と異なる可能性があり、 previous_item_id を使って上手く順序を整えてあげる後処理が必要そうです。
Received event: {
  "type": "input_audio_buffer.committed",
  "event_id": "event_BIFK3ijZSB6PgCDctNrEy",
  "previous_item_id": null,
  "item_id": "item_BIFK28t2v0YtjREkmywxD"
}
 イベントのペイロードここでは主要なイベントのペイロードの仕様を見ていきます。
各イベントのペイロードの詳細は、以下に記載されているようですので、そちらも合わせてご確認ください。
API Reference - OpenAI API
 transcription_session.updatetranscription_session.update については以下に記載されています。
https://platform.openai.com/docs/api-reference/realtime-client-events/transcription_session/update
このイベントはクライアントから送信するイベントで、初期化の際に実行し、音声のフォーマットや使用するモデル、言語指定などを設定できます。
また音声検出（VAD）や音声データに対するノイズ除去なども設定できます。
ノイズ除去は有効なケースもあるかと思いますが、以降で使用した音声では認識結果に悪影響があったため無効化しています。
具体的には以下のようなペイロードとなっています。
{
  # ここは固定
  "type": "transcription_session.update",

  "session": {
    # pcm16、g711_ulaw、g711-alawから選択、pcm16の場合、16-bit PCMでサンプルレートは24kHz、モノラル、Little-Endianに対応
    "input_audio_format": "pcm16",

    "input_audio_transcription": {

      # モデルの指定、gpt-4o-transcribe, gpt-4o-mini-transcribe, whisper-1から選択可能
      "model": "gpt-4o-transcribe",

      # 出力を制御するためのガイドとなるプロンプト
      "prompt": "",

      # 言語コード（ISO-639-1）
      "language": ""
    },

    # 発話の開始・終了を検出する設定
    "turn_detection": {
      "type": "server_vad",
      "threshold": 0.5,
      "prefix_padding_ms": 300,
      "silence_duration_ms": 500,
      "create_response": true,
    },

    # ノイズ除去設定
    "input_audio_noise_reduction": {
      "type": "near_field"
    },

    # トランスクリプションに含めるアイテムのセット
    "include": [
      "item.input_audio_transcription.logprobs",
    ]
  }
}
 input_audio_buffer.appendinput_audio_buffer.append については以下に記載されています。
https://platform.openai.com/docs/api-reference/realtime-client-events/input_audio_buffer/append
こちらもクライアントから送信するイベントで、音声データ送信用のイベントです。音声データはバイトデータをBase64形式にエンコードして送信する仕様となっています。
バイトデータの形式は、前述の transcription_session.update の input_audio_format で指定した形式に従う必要があります。
具体的には以下のようなペイロードとなっています。
{
    "type": "input_audio_buffer.append",
    "audio": "Base64EncodedAudioData"
}
 input_audio_buffer.committedinput_audio_buffer.committed については以下に記載されています。
https://platform.openai.com/docs/api-reference/realtime-server-events/input_audio_buffer/committed
このイベントはサーバーから受信するイベントで、確定した音声区間の結果が得られます。
具体的には以下のようなペイロードとなっています。
{
  "type": "input_audio_buffer.committed",
  "event_id": "event_BIFK3ijZSB6PgCDctNrEy",
  "previous_item_id": null,
  "item_id": "item_BIFK28t2v0YtjREkmywxD"
}
previous_item_id は一つ前の区間の音声区間を指しており、後方リンクリストのような構造となっています。次の音声区間の結果は以下のように得られます。
{
  "type": "input_audio_buffer.committed",
  "event_id": "event_BIFK39RBXur9t2ILnpSls",
  "previous_item_id": "item_BIFK28t2v0YtjREkmywxD",
  "item_id": "item_BIFK3pVFXKRPOG3bgBRo3"
}
 conversation.item.input_audio_transcription.completedconversation.item.input_audio_transcription.completed については以下に記載されています。
https://platform.openai.com/docs/api-reference/realtime-server-events/conversation/item/input_audio_transcription/completed
このイベントはサーバーから受信するイベントで、確定した認識結果が得られます。
具体的には以下のようなペイロードとなっています。
Received event: {
  "type": "conversation.item.input_audio_transcription.completed",
  "event_id": "event_BIFK3ZuDNrXXlccGZNpqX",
  "item_id": "item_BIFK28t2v0YtjREkmywxD",
  "content_index": 0,
  "transcript": "そうだね。"
}
元となった音声区間を示す item_id が含まれており、認識結果は transcript に含まれています。
 作ってみた 作成するコードについてとあるwavファイルが準備されている前提で、そのwavファイルを分割してストリーミングでの音声認識でのテストをします。
今回は以下の動画ファイルを、16-bit PCM、サンプルレートは24kHz、モノラル、Little-Endianで保存して tmp.wav として保存しています。
https://www.youtube.com/watch?v=prv49KDZBNQ
 Python実行環境uvで環境構築をします。Pythonやライブラリのバージョンは以下の通りです。
pyproject.toml
[project]
name = "openai-transcribe"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "openai>=1.69.0",
    "python-dotenv>=1.1.0",
    "websocket-client>=1.8.0",
]
 はじめにライブラリをインポートします。
main.py
import os
import json
import websocket
import wave
import base64
import time
import logging
import threading
from typing import Generator, Dict, Set, Optional, Any, List, Tuple
from dotenv import load_dotenv
また定数等を以下のように定義しておきます。
main.py
# ロガーの設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# .envファイルから環境変数を読み込む
load_dotenv()

# 設定

# 環境変数からOPENAI_API_KEYを取得
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")

# WebSocketのURL
WEBSOCKET_URL = "wss://api.openai.com/v1/realtime?intent=transcription"

# WebSocketのヘッダー
WEBSOCKET_HEADERS = [
    "Authorization: Bearer " + OPENAI_API_KEY,
    "OpenAI-Beta: realtime=v1"
]

# 音声ファイルを分割するチャンクの長さ [sec]
CHUNK_DURATION_SEC = 5

# 送信するチャンクの最大数
MAX_CHUNKS = 60

# 認識結果を待機する最大時間 [sec]
TRANSCRIPT_WAIT_TIMEOUT = 60
環境変数は .env から読み込むので、OPENAI_API_KEYを記載しておきます。
.env
OPENAI_API_KEY=sk-proj-xxxx
 WebSocket通信用クラス次にWebSocket通信用のクラスを以下のように作成します。（少し長いのですが…）
main.py
class TranscriptionSession:

    def __init__(self, wav_path: str) -> None:
        """
        OpenAIの文字起こしセッションを管理するクラス

        Args:
            wav_path (str): 音声認識を行うWAVファイルのパス
        """

        # WAVファイルのフォーマットをチェック
        check_wav_format(wav_path)

        self.transcription_items: Dict[str, Dict[str, Any]] = {}
        self.displayed_items: Set[str] = set()
        self.ws: Optional[websocket.WebSocketApp] = None
        self.is_running = False
        self.wav_path = wav_path

    def complete_check(self) -> bool:
        """
        すべての認識結果が得られたかどうかをチェックする

        Returns:
            bool: すべての認識結果が得られた場合はTrue、そうでない場合はFalse
        """
        wait_items = [item_id for item_id, item_dict in self.transcription_items.items() if "transcript" not in item_dict]

        logger.info(f"待機中の認識結果: {len(wait_items)}件")

        return len(wait_items) == 0

    def display_current_transcripts(self) -> None:
        """
        新しい認識結果のみを時系列順に表示する
        """
        # 認識結果を時系列順に並び替え
        items = self.get_ordered_transcripts()
        # logger.debug(f"認識結果: {len(items)}件")

        # 認識結果がない場合は何もしない
        if not items:
            return

        # 新しい認識結果を表示
        for item_index, item_id, item in items:

            # 既に表示済みの場合はスキップ
            if item_id in self.displayed_items:
                continue

            # 新しい認識結果が存在する場合はリストに追加
            if "transcript" in item:
                logger.info(f"認識結果[{item_index}]: {item["transcript"]}")
                self.displayed_items.add(item_id)
            else:
                break

    def get_ordered_transcripts(self) -> List[Tuple[int, str, Dict[str, Any]]]:
        """
        認識結果を時系列順に並び替える

        Returns:
            List[Tuple[int, str, Dict[str, Any]]]: 並べ替えた認識結果
        """
        # アイテムが空の場合は空のリストを返す
        if not self.transcription_items:
            return []

        # 最後のアイテムから開始して、previous_idを使って逆順にトラバース
        items = []
        current_id = list(self.transcription_items.keys())[-1]  # 最後のアイテムID

        # 最後のアイテムから開始して、previous_idを使って逆順にトラバース
        while current_id:
            # アイテムを取得
            item = self.transcription_items.get(current_id)

            # アイテムが存在する場合はリストに追加
            if item:
                items.append((current_id, item))
                current_id = item['previous_id']
            else:
                break

        # リストを反転して時系列順に並び替え
        return [(item_index, item[0], item[1]) for item_index, item in enumerate(reversed(items))]

    def on_open(self, ws: websocket.WebSocket) -> None:
        """
        WebSocket接続が確立されたときのコールバック

        Args:
            ws: WebSocket接続オブジェクト
        """
        logger.info("サーバーに接続しました。")
        self.is_running = True

        # スレッドを作成して音声認識セッションを開始
        threading.Thread(target=self.run).start()

    def on_message(self, ws: websocket.WebSocket, message: Any) -> None:
        """
        WebSocketからメッセージを受信したときのコールバック

        Args:
            ws: WebSocket接続オブジェクト
            message: 受信したメッセージ
        """
        try:
            # メッセージをJSONにパース
            data = json.loads(message)

            # バッファのコミットが完了した場合
            if data["type"] == "input_audio_buffer.committed":
                # 前の認識結果のID
                previous_item_id = data["previous_item_id"]

                # 認識結果のID
                item_id = data["item_id"]

                # リンク情報を保存
                self.transcription_items[item_id] = {
                    'previous_id': previous_item_id
                }
                logger.debug(f"バッファがコミットされました: {item_id}")

                logger.info(f"現在の音声区間数: {len(self.transcription_items)}件")

            # 認識結果が完了した場合
            if data["type"] == "conversation.item.input_audio_transcription.completed":
                # 認識結果のID
                item_id = data["item_id"]

                # 認識結果
                transcript = data["transcript"]

                # リンク情報を保存
                self.transcription_items[item_id] = {
                    **self.transcription_items.get(item_id, {'previous_id': None}),
                    "transcript": transcript,
                }
                logger.debug(f"認識結果: {transcript}")

                self.display_current_transcripts()
        except json.JSONDecodeError:
            logger.error(f"JSONのパースに失敗しました: {message}")
        except KeyError as e:
            logger.error(f"メッセージに必要なキーがありません: {e}")
        except Exception as e:
            logger.error(f"メッセージ処理中にエラーが発生しました: {e}")

    def on_error(self, ws: websocket.WebSocket, error: Any) -> None:
        """
        WebSocketでエラーが発生したときのコールバック

        Args:
            ws: WebSocket接続オブジェクト
            error: エラー情報
        """
        logger.error(f"WebSocketエラー: {error}")

    def on_close(self, ws: websocket.WebSocket, close_status_code: Any, close_msg: Any) -> None:
        """
        WebSocket接続が閉じられたときのコールバック

        Args:
            ws: WebSocket接続オブジェクト
            close_status_code: 閉じられた理由のステータスコード
            close_msg: 閉じられた理由のメッセージ
        """
        logger.info(f"WebSocket接続が閉じられました: {close_status_code} - {close_msg}")
        self.is_running = False

    def run(self) -> None:
        """
        音声認識セッションを実行する
        """
        try:
            # 音声認識セッションの設定を送信
            if self.ws:
                self.ws.send(create_transcription_session_config())
                logger.info("イベント送信: transcription_session.update")

                # 音声データをチャンクに分割して送信
                for i, payload in enumerate(generate_audio_buffer_payload(self.wav_path)):
                    # 最大チャンク数に達したら終了
                    if i >= MAX_CHUNKS:
                        break

                    # ペイロードを送信
                    self.ws.send(payload)
                    logger.debug(f"イベント送信: input_audio_buffer.append ({i+1}/{MAX_CHUNKS})")

                    # 各チャンク送信後に現在の認識結果を表示
                    self.display_current_transcripts()

                # 一応送信後は一旦待つ（音声区間検出前に全ての処理が終わったとみなさせるのを避けるため）
                # メモ：実際はマイク入力等においてオフボタンがあるはず
                time.sleep(5)

                # すべての認識結果が得られるまで待機
                logger.info("すべての認識結果を待機中...")
                transcript_wait_second = 0
                while transcript_wait_second < TRANSCRIPT_WAIT_TIMEOUT:
                    if self.complete_check():
                        logger.info("すべての認識結果が得られました")
                        break

                    time.sleep(1)
                    transcript_wait_second += 1

                    # 現在の認識結果を表示
                    self.display_current_transcripts()

                # タイムアウトした場合
                if transcript_wait_second >= TRANSCRIPT_WAIT_TIMEOUT:
                    logger.warning("一部の認識結果が得られませんでした")

                # 音声認識セッションをクローズ
                logger.info("音声認識セッションを終了します")
                self.ws.close()
        except Exception as e:
            logger.error(f"音声認識セッションの実行中にエラーが発生しました: {e}")
            if self.ws:
                self.ws.close()

    def start(self) -> None:
        """
        音声認識セッションを開始する
        """
        # 音声認識セッションを作成
        self.ws = websocket.WebSocketApp(
            WEBSOCKET_URL,
            header=WEBSOCKET_HEADERS,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )

        # 音声認識セッションを開始
        self.ws.run_forever()

    def stop(self) -> None:
        """
        音声認識セッションを停止する
        """
        if self.ws and self.is_running:
            logger.info("音声認識セッションを停止します")
            self.ws.close()
            self.is_running = False
on_open、on_message、on_erro、on_closeはそれぞれWebSocketのコールバックとして準備しています。
on_openは接続開始後に呼ばれるため、ここで音声データを送信する処理を別スレッドを起動しています。
        # スレッドを作成して音声認識セッションを開始
        threading.Thread(target=self.run).start()
このrun関数の中で以下を行います。
音声データを分割して送信
送信完了後は認識結果をすべての音声区間で得られるまで待つ
全ての結果が得られた場合はクローズ
また、 get_ordered_transcripts は音声区間を時系列順に並べてたものを得るための関数で、 display_current_transcripts は時系列順を維持しつつ、認識結果が得られているものを画面出力します。
 wavフォーマットチェック関数wavファイルのフォーマットチェックを以下のように行います。
main.py
def check_wav_format(wav_path: str) -> None:
    """
    WAVファイルのフォーマットをチェックする

    Args:
        wav_path (str): チェックするWAVファイルのパス

    Raises:
        FileNotFoundError: WAVファイルが見つからない場合
        wave.Error: WAVファイルの形式が不正な場合
        ValueError: 音声フォーマットが期待と異なる場合
    """
    if not os.path.exists(wav_path):
        raise FileNotFoundError(f"WAVファイルが見つかりません: {wav_path}")

    try:
        with wave.open(wav_path, 'rb') as wf:
            # サンプル幅のチェック
            sample_width = wf.getsampwidth()
            if sample_width != 2:
                raise ValueError(
                    f"不正なサンプル幅です: {sample_width * 8}-bit PCM"
                    " (期待値: 16-bit PCM)"
                )

            # サンプルレートのチェック
            frame_rate = wf.getframerate()
            if frame_rate != 24000:
                raise ValueError(
                    f"不正なサンプルレートです: {frame_rate}Hz"
                    " (期待値: 24000Hz)"
                )

            # チャンネル数のチェック
            channels = wf.getnchannels()
            if channels != 1:
                raise ValueError(
                    f"不正なチャンネル数です: {channels}チャンネル"
                    " (期待値: モノラル)"
                )

    except wave.Error as e:
        raise wave.Error(f"WAVファイルの形式が不正です: {e}")
後述のWebSocketの初期設定で、16-bit PCM、サンプルレートは24kHz、モノラル、Little-Endianとしているので、そのフォーマットに合致しているか確認をします。
 WebSocket向けのペイロード作成用関数WebSocket向けのペイロード作成用関数を別途以下のように定義します。
まずは初期化用のものです。
main.py
def create_transcription_session_config() -> str:
    """
    音声認識セッションの設定を作成する

    Returns:
        str: JSON形式の設定文字列
    """
    return json.dumps({
        # ここは固定
        "type": "transcription_session.update",

        "session": {
            # pcm16、g711_ulaw、g711-alawから選択、pcm16の場合、16-bit PCMでサンプルレートは24kHz、モノラル、Little-Endianに対応
            "input_audio_format": "pcm16",

            "input_audio_transcription": {

                # モデルの指定、gpt-4o-transcribe, gpt-4o-mini-transcribe, whisper-1から選択可能
                "model": "gpt-4o-transcribe",

                # 出力を制御するためのガイドとなるプロンプト
                "prompt": "",

                # 言語コード（ISO-639-1）
                "language": "ja"
            },

            # # 発話の開始・終了を検出する設定
            # "turn_detection": {
            #     "type": "server_vad",
            #     "threshold": 0.5,
            #     "prefix_padding_ms": 300,
            #     "silence_duration_ms": 500
            # },

            # # ノイズ除去設定
            # "input_audio_noise_reduction": {
            #     "type": "near_field"
            # }
        }
    })
ここの設定とwavフォーマットは合わせる必要があります。ファイルではなくマイク入力等を使用する場合は、マイク入力されたデータがこのフォーマットになっているか注意しましょう（ここを誤ると認識精度が極端に低下します）。
また、ノイズ除去設定やVADの設定は無効化しています。特にノイズ除去設定は実施すると認識精度に悪影響がでましたのでオフとしています。
次にチャンク分割された音声データのペイロードを作成する関数です。
main.py
def generate_audio_buffer_payload(wav_path: str, chunk_duration_sec: int = CHUNK_DURATION_SEC) -> Generator[str, None, None]:
    """
    音声データをチャンクに分割して送信するためのペイロードを作成する

    Args:
        wav_path (str): WAVファイルのパス
        chunk_duration_sec (int, optional): チャンクの長さ（秒）. デフォルトは5秒.

    Yields:
        str: JSON形式のペイロード文字列

    Raises:
        FileNotFoundError: WAVファイルが見つからない場合
        wave.Error: WAVファイルの形式が不正な場合
        AssertionError: 音声フォーマットが期待と異なる場合
    """
    try:

        # WAVファイルのフォーマットをチェック
        check_wav_format(wav_path)

        with wave.open(wav_path, 'rb') as wf:

            # チャンクのサイズを計算
            bytes_per_sec = wf.getframerate() * wf.getsampwidth() * wf.getnchannels()
            chunk_size = bytes_per_sec * chunk_duration_sec

            # チャンクを読み込んでbase64にエンコードしてペイロードを生成
            while True:
                # チャンクを読み込む
                data = wf.readframes(chunk_size // wf.getsampwidth())

                # チャンクが空の場合はループを抜ける
                if not data:
                    break

                # チャンクをbase64にエンコード
                encoded = base64.b64encode(data).decode('ascii')

                # ペイロードを生成
                yield json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": encoded
                })

    except FileNotFoundError:
        logger.error(f"WAVファイルが見つかりません: {wav_path}")
        raise
    except wave.Error as e:
        logger.error(f"WAVファイルの形式が不正です: {e}")
        raise
    except AssertionError as e:
        logger.error(f"音声フォーマットが期待と異なります: {e}")
        raise
    except Exception as e:
        logger.error(f"音声データの処理中にエラーが発生しました: {e}")
        raise
yieldを使用してジェネレータで実装をしています。バイナリデータはbase64にエンコードしてから送信する必要があります。
 動かしてみた以下で動かすことができます。
main.py
# 音声認識セッションを作成して開始
session = TranscriptionSession(wav_path="tmp.wav")
session.start()
得られる結果は以下のようになりました。
2025-04-15 16:00:42,918 - websocket - INFO - Websocket connected
2025-04-15 16:00:42,919 - __main__ - INFO - サーバーに接続しました。
2025-04-15 16:00:42,923 - __main__ - INFO - イベント送信: transcription_session.update
2025-04-15 16:00:44,170 - __main__ - INFO - 認識結果[0]: あるのね。
2025-04-15 16:00:45,542 - __main__ - INFO - 認識結果[1]: 皆さん、こんばんは。
2025-04-15 16:00:45,543 - __main__ - INFO - 認識結果[2]: はい。
2025-04-15 16:00:47,121 - __main__ - INFO - 認識結果[3]: ラスベガスからre:Inventの様子をお届けします、Developers.IO in ラスベガス2022ということで始めさせていただきます。どうぞよろしくお願いします。
2025-04-15 16:00:47,122 - __main__ - INFO - 認識結果[4]: 日本は今ですね、12月1日のお昼12時の時間帯かと思いますけれども、こちらはまだ11月30日の夜7時ということで。
2025-04-15 16:00:47,642 - __main__ - INFO - 認識結果[5]: 夜でお疲れもあって変なテンションになるかもしれないですけども、聞いていただければと思います。
2025-04-15 16:00:48,638 - __main__ - INFO - 認識結果[6]: ちなみに今何人ぐらい入ってらっしゃるんですか、視聴者の方は?
2025-04-15 16:00:48,639 - __main__ - INFO - 認識結果[7]: ありがとうございます。
2025-04-15 16:00:48,639 - __main__ - INFO - 認識結果[8]: この配信ではですね。
2025-04-15 16:00:48,733 - __main__ - INFO - 認識結果[9]: AWS re:Invent 2022でたくさんのアップデートだったりとかセッションが
2025-04-15 16:00:49,780 - __main__ - INFO - 認識結果[10]: キーノートがありましたので、ここまでで出ている情報を基にですね、クラスメソッドのエンジニアが注目した内容とか、そういったものをですね、エンジニア
メンバーから
2025-04-15 16:00:51,019 - __main__ - INFO - 認識結果[11]: 今回今やっていたのが前半のメンバーということで、だいたい30分ぐらいを目処で後半のメンバーにチェンジしてお送りしていきたいと思います。まずですね、
今出ている前半メンバーの自己紹介をしたいと思いますので、よろしくお願いします。
2025-04-15 16:00:51,428 - __main__ - INFO - 認識結果[12]: はい、エイラクス情報本部のコンサルティング部で働いてますモンメツと申します。
2025-04-15 16:00:51,680 - __main__ - INFO - 認識結果[13]: devioとかTwitterとかはもこっていう名前でやってます。よろしくお願いします。
2025-04-15 16:00:51,681 - __main__ - INFO - 認識結果[14]: 拍手
2025-04-15 16:00:51,681 - __main__ - INFO - 認識結果[15]: イェーイ
2025-04-15 16:00:51,682 - __main__ - INFO - 認識結果[16]: はい。
2025-04-15 16:00:51,682 - __main__ - INFO - 認識結果[17]: クラスメソッドのプリズマティクス事業部というところでエンジニアをしています寺岡と申します。
2025-04-15 16:00:51,682 - __main__ - INFO - 認識結果[18]: インターネットではとばしという名前で存在しています。よろしくお願いします。
2025-04-15 16:00:52,237 - __main__ - INFO - 認識結果[19]: クラスメソッドのCXD本部っていうところにいます。
2025-04-15 16:00:52,836 - __main__ - INFO - 認識結果[20]: 主にサーバーサイドエンジニアとかバックエンドのエンジニアやっています。
2025-04-15 16:00:52,836 - __main__ - INFO - 認識結果[21]: えっと。
2025-04-15 16:00:53,636 - __main__ - INFO - 認識結果[22]: あと、普段の活動としてはAWS UGのCDK支部みたいなところで運営とかをやったりしてます。今回CDKのアップデートとか気になってきたりします。よろしくお願
いします。
2025-04-15 16:00:53,636 - __main__ - INFO - 認識結果[23]: おーい。
2025-04-15 16:00:53,636 - __main__ - INFO - 認識結果[24]: AWS事業本部コンサルティンググループの高栗と申します。
2025-04-15 16:00:53,637 - __main__ - INFO - 認識結果[25]: はい、私はコンテナを追い続けてここに来ています。よろしくお願いします。
2025-04-15 16:00:54,613 - __main__ - INFO - 認識結果[26]: ということで前半はですね
2025-04-15 16:00:54,614 - __main__ - INFO - 認識結果[27]: コンテナサーバーですとかインフラだったりセキュリティそのあたりにフォーカスしてるメンバーの
2025-04-15 16:00:54,614 - __main__ - INFO - 認識結果[28]: お送りしていきたいと思います。AWS事業本部の菊池です。よろしくお願いします。
2025-04-15 16:00:54,615 - __main__ - INFO - 認識結果[29]: では早速聞いていきたいと思いますが、まず門別さん。
2025-04-15 16:00:54,847 - __main__ - INFO - 認識結果[30]: 今回のリインベント。
2025-04-15 16:00:54,864 - __main__ - INFO - 認識結果[31]: 失礼。
2025-04-15 16:00:55,175 - __main__ - INFO - 認識結果[32]: 今見ててる中で、これは熱いぞというのとかあれば?
2025-04-15 16:00:55,176 - __main__ - INFO - 認識結果[33]: どうぞ。
2025-04-15 16:00:55,215 - __main__ - INFO - 認識結果[34]: そうですね。
2025-04-15 16:00:55,991 - __main__ - INFO - 認識結果[35]: 初日のマンデーナイトライブで出た、ナイトロV5とハーゲット3E。それに伴うC7GNインスタンス
2025-04-15 16:00:56,142 - __main__ - INFO - 認識結果[36]: 今のところ僕の中ではホッとかな。
2025-04-15 16:00:58,746 - __main__ - INFO - 認識結果[37]: グラビトンシリーズってこれまでメモリ最適化インスタンスみたいなところあったはずなんですけど、ネットワークに最適化されているファミリーっていうのが
なくて、それがGraviton3が出てものすごく早くなって。
2025-04-15 16:00:58,747 - __main__ - INFO - 認識結果[38]: で、確か上下で200Gbpsぐらい出るみたいな話が出てたので、
2025-04-15 16:00:58,747 - __main__ - INFO - 認識結果[39]: グラビトンシリーズでもHPCであったりとか使えるようになったのがすごいアップデートかなと思ってます。
2025-04-15 16:00:58,747 - __main__ - INFO - 認識結果[40]: そうですね。
2025-04-15 16:00:58,747 - __main__ - INFO - 認識結果[41]: アイドルグループとか。
2025-04-15 16:00:58,747 - __main__ - INFO - 認識結果[42]: グラビトン3E
2025-04-15 16:00:58,748 - __main__ - INFO - 認識結果[43]: 今までNKの
2025-04-15 16:00:58,748 - __main__ - INFO - 認識結果[44]: あの
2025-04-15 16:00:59,008 - __main__ - INFO - 認識結果[45]: インスタンスタイプの変更が。
2025-04-15 16:00:59,210 - __main__ - INFO - 認識結果[46]: 5Nとか付いてる本あったよね。
2025-04-15 16:00:59,211 - __main__ - INFO - 認識結果[47]: さらにそれが。
2025-04-15 16:00:59,211 - __main__ - INFO - 認識結果[48]: あわふして。
2025-04-15 16:00:59,507 - __main__ - INFO - 認識結果[49]: そうですね。
2025-04-15 16:00:59,508 - __main__ - INFO - 認識結果[50]: はい
2025-04-15 16:00:59,508 - __main__ - INFO - 認識結果[51]: しかし、
2025-04-15 16:00:59,509 - __main__ - INFO - 認識結果[52]: ありがとうございます。じゃあ続いて、戸鉢さんはどうですか?
2025-04-15 16:00:59,510 - __main__ - INFO - 認識結果[53]: そうですね、ここが
2025-04-15 16:00:59,942 - __main__ - INFO - 認識結果[54]: 今のところ注目しているのは二つありまして、
2025-04-15 16:01:00,726 - __main__ - INFO - 認識結果[55]: 昨日のキーノートで発表されたAmazon OpenSearch Serviceのサーバーレス対応
2025-04-15 16:01:01,149 - __main__ - INFO - 認識結果[56]: もう一つは、これが確か直前の日曜日ぐらいに発表されたと思うんですけど。
2025-04-15 16:01:01,150 - __main__ - INFO - 認識結果[57]: Amazon ECSの
2025-04-15 16:01:01,151 - __main__ - INFO - 認識結果[58]: 新しいネットワークの機能でAmazon ECS Service Connectというものが。
2025-04-15 16:01:01,152 - __main__ - INFO - 認識結果[59]: 発表されていて、
2025-04-15 16:01:01,152 - __main__ - INFO - 認識結果[60]: 次アップデートなのかなという。
2025-04-15 16:01:01,668 - __main__ - INFO - すべての認識結果が得られました
2025-04-15 16:01:01,668 - __main__ - INFO - 音声認識セッションを終了します
2025-04-15 16:01:01,813 - __main__ - INFO - WebSocket接続が閉じられました: None - None
ストリーミング入力でもなかなか良い精度を出していそうです。
参考までにファイル入力した場合の結果は以下です。
皆さんこんばんは。
クラスメソッドがラスベガスからリーンベントの様子をお届けします。
デベロッパーズI.O. in ラスベガス2022ということで始めさせていただきます。
どうぞよろしくお願いします。
日本は今12月1日のお昼12時の時間帯かと思いますけれども、こちらはまだ11月30日の夜7時ということで。
ちょっと夜でですね、お疲れもあって大変なテンションになるかもしれないですけども、暖かく聞いていただければと思います。
ちなみに今何人くらい入っていらっしゃるんですかね、視聴者の方は。
104名です。
素晴らしいですね。
ありがとうございます。
この配信ではですね、AWSイベント2022でですね、たくさんのアップデートだったりとかですね、セッション、 キーノートとありましたので、ここまでで出ている情報をもとにですね、クラスメソッドのエンジニアが注目した内容とか、そういったものをですね、エンジニアメンバーから聞いていきたいと思います。
今回ですね、人数の 都合上ですね、二部制となっておりまして、今回今出ているのが前半のメンバーということで、大体30分くらい目処でですね、後半のメンバーにチェンジしてお送りしていきたいと思います。
まずですね、今出ている前半メン バーの自己紹介をしたいと思いますので、順によろしくお願いします。
はい、エラシジオン本部のコンサルティング部で働いてます、モンベツと申します。
Dev.ioとかTwitterとかはモコっていう名前でやってます。
よろしくお願いします。
クラスメソッドのプリズマティクス事業部というところでエンジニアをしています、寺岡と申します。
インターネットではトバシという名前で存在しています。
よろしくお願いします。
クラスメソッドのCX事業本 部というところにいます。
主にサーバーサイドエンジニアとかバックエンドのエンジニアやっています。
あと普段の活動としてはJaws-UGのCDK支部みたいなところで運営とかをやったりしています。
今回CDKのアップデートとか気になってきてやります。
よろしくお願いします。
AWS事業本部コンサルティング部内ソリューションアクティブでおります高国と申します。
私はコンテナを追い続けてここに来ています。
よろしくお願いします。
ということで前半はコンテナサーバーレスとかインフラだったりセキュリティそのあたりにフォーカスしているメンバーの中心にお送りしていきたいと思います。
申し遅れました。
私AWS事業本部の菊池です。
よろしくお願いします。
早速聞いていきたいと思いますが、まず文別さん。
今回のリーンベントですね。
今見ている中でこれは熱いぞというのとかあればどうぞ。
初日のマンデーナイトライブで出たナイトロV5とハーゲット3E。
それに伴うC7GNインスタンス の登場が今のところ僕の中ではホットなトピックです。
それどの辺が注目ポイントですか?そうですね。
グラビトンシリーズってこれまでメモリ最適化インスタンスみたいなところあったはずなんですけど、ネットワークに最適化されているファミリーっていうのがなくて、それがグラビトン3Eが出てものすごく速くなって、確か上限で200Gbpsくらい出るみたいな話が出ていたので、すごいグラビトンシリーズでもHPCであってとか、トラフィック食う ようなワークロードとか使えるようになったのがすごいいいアップデートかなと思っています。
はい、そうですね。
ナイトロV5とかグラビトン3Eですね。
今までN系のインスタンスタイプの後に何とか5Nとかついているのがあったり、ネットワークというのがありましたけれども、さらにそれがパワーアップして出るという感じですかね。
そうですね。
はい、ありがとうございます。
じゃあ続いて、とばちさんはどうですか。
そうですね。
僕が今のとこ ろ注目しているのは2つありまして、1つは昨日のキーノートで発表されたAmazonオープンサーチサービスのサーバーレス対応が1つと、もう1つはこれが確か直前の日曜日ぐらいに発表されたと思うんですけど、Amazon ECSの新 しいネットワークの機能でAmazon ECSサービスコネクトというものが発表されていて、こちらも結構厚いアップデートなのかなというふうに思っています。
なるほど。
まず最初の1つ目のオープンサーチサービスがサーバーレス。
コードは以下でできます。
import os
from dotenv import load_dotenv
from openai import OpenAI

# .envファイルから環境変数を読み込む
load_dotenv()

# 環境変数からOPENAI_API_KEYを取得
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")

client = OpenAI()
audio_file= open("tmp.wav", "rb")

transcription = client.audio.transcriptions.create(
    model="gpt-4o-transcribe",
    file=audio_file
)

print(transcription.text)
さすがにファイル入力の方が高精度になっているようです。
 まとめいかがでしたでしょうか。本記事がご参考になれば幸いです。