OpenAIのRealtime APIとは？使い方や料金！音声チャットと文字起こしをしてみた

具体的な手順の前に、今回作成するフォルダ構成を示します。

realtime-webrtc-app/
├── node_modules/   (npm install後に自動生成)
├── public/         (ブラウザからアクセスするファイル)
│   ├── index.html  (ページの骨格)
│   └── client.js   (ブラウザで動くJavaScript)
├── .env            (APIキーを保存するファイル)
├── package.json    (プロジェクト管理ファイル)
└── server.js       (サーバー側のNode.jsコード)

まずは、Ephemeral Keyと呼ばれる一時キーの発行をリクエストします。

このキーは短時間のみ有効であるため、セキュリティを保つことができます。

APIの利用は「セッション」という単位で管理され、バックエンドサーバーから、OpenAIのAPIキーを使ってセッション作成をリクエストします。

セッションが作成されると、Ephemeral Keyが発行されます。

ここまでの一連の流れが以下のコードです。

// server.js
import express from "express";

const app = express();
const apiKey = process.env.OPENAI_API_KEY; // 環境変数からAPIキーを読み込む

// セッションの設定
const sessionConfig = JSON.stringify({
  session: {
    type: "realtime",
    model: "gpt-realtime",
    audio: {
      output: {
        voice: "marin", // AIの声を指定
      },
    },
  },
});

// "/token" エンドポイント
app.get("/token", async (req, res) => {
  try {
    const response = await fetch(
      "https://api.openai.com/v1/realtime/client_secrets",
      {
        method: "POST",
        headers: {
          Authorization: `Bearer ${apiKey}`, // ★サーバーの標準APIキーを使用
          "Content-Type": "application/json",
        },
        body: sessionConfig,
      }
    );

    const data = await response.json();
    res.json(data); // クライアントに一時キーを含む情報を返す
  } catch (error) {
    console.error("Token generation error:", error);
    res.status(500).json({ error: "Failed to generate token" });
  }
});

app.listen(3000, () => console.log("Server listening on port 3000"));

次に、ブラウザ側でWebRTC接続を確立します。

Ephemeral Keyを取得してRTCPeerConnectionオブジェクトを作成し、データチャネルを作成してSDP (Session Description Protocol) の交換を行います。

これにより、WebRTCのピア接続が確立されます。

コード例は以下の通りです。

// client.js

async function connectToRealtimeAPI() {
    // 1. サーバーから一時キーを取得
    const tokenResponse = await fetch("/token");
    const data = await tokenResponse.json();
    const EPHEMERAL_KEY = data.value; // 一時キー

    // 2. RTCPeerConnectionを作成
    const pc = new RTCPeerConnection();
    const audioElement = document.createElement("audio");
    audioElement.autoplay = true;

    // 3. AIからの音声トラックを受信したときの処理
    pc.ontrack = (e) => (audioElement.srcObject = e.streams[0]);

    // 4. ユーザーのマイク音声トラックを追加
    const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
    pc.addTrack(stream.getTracks()[0]);

    // 5. イベント送受信用のデータチャネルを作成
    const dc = pc.createDataChannel("oai-events");

    // 6. SDPオファーを作成してOpenAI APIに送信
    const offer = await pc.createOffer();
    await pc.setLocalDescription(offer);

    const sdpResponse = await fetch("https://api.openai.com/v1/realtime/calls?model=gpt-realtime", {
        method: "POST",
        body: offer.sdp,
        headers: {
            Authorization: `Bearer ${EPHEMERAL_KEY}`, // ★一時キーを使用
            "Content-Type": "application/sdp",
        },
    });

    // 7. OpenAIからのアンサーSDPを設定
    const answer = {
        type: "answer",
        sdp: await sdpResponse.text(),
    };
    await pc.setRemoteDescription(answer);
    
    // これで接続が確立
    console.log("WebRTC connection established.");
}

接続の確立後、テキストの送信やセッションの状態変更などのやり取りは作成したデータチャネルを通じて行います。

サーバーからのイベント受信はデータチャネルのmessageイベント、クライアントからのイベント送信はdc.send()メソッドを使って行います。

// client.js (続き)

// データチャネルの準備
const dc = pc.createDataChannel("oai-events");

// サーバーからのイベントを受信
dc.addEventListener("message", (e) => {
    const event = JSON.parse(e.data);
    console.log("Received event from server:", event);
});

// クライアントからイベントを送信する例
dc.onopen = () => { // データチャネルが開通したら送信可能
    const event = {
        type: "conversation.item.create",
        item: {
            type: "message",
            role: "user",
            content: [
                {
                    type: "input_text",
                    text: "こんにちは！",
                },
            ],
        },
    };
    dc.send(JSON.stringify(event));
};

フォルダ構成内にあるその他のコードを以下に示します。

これらをコピーすることで、実際の動きを試すことが可能です。

package.jsonファイルの中身はこちらです。

{
  "name": "realtime-webrtc-app",
  "version": "1.0.0",
  "description": "",
  "main": "server.js",
  "type": "module",
  "scripts": {
    "start": "node server.js"
  },
  "keywords": [],
  "author": "",
  "license": "ISC",
  "dependencies": {
    "dotenv": "^16.3.1",
    "express": "^4.18.2"
  }
}

.envファイルにはOpenAIのAPIキーを記述します。

OPENAI_API_KEY="sk-..."

index.htmlの例はこちらです。

<!DOCTYPE html>
<html lang="ja">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Realtime API with WebRTC</title>
    <style>
        body { font-family: sans-serif; text-align: center; margin-top: 50px; }
        button { font-size: 1.2rem; padding: 10px 20px; cursor: pointer; }
        #status { margin-top: 20px; color: #555; }
    </style>
</head>
<body>
    <h1>Realtime API WebRTC Demo</h1>
    <button id="connectButton">接続して会話を開始</button>
    <p id="status">待機中...</p>

    <script src="client.js"></script>
</body>
</html>

ターミナル（コマンドプロンプト）を開き、プロジェクトのルートフォルダrealtime-webrtc-app に移動して、以下のコマンドを実行します。

npm install

これによって、package.jsonに書かれたexpressとdotenvがnode_modulesフォルダにインストールされます。

同じターミナルで、以下のコマンドを実行してサーバーを起動します。

npm start

成功すると、Server is running at http://localhost:3000 と表示されます。

ターミナルでCtrlを押しながらURLをクリックするか、ブラウザで直接http://localhost:3000にアクセスしましょう。

今回のコードでは、以下のような画面が表示されます。

実際に動かしてみた動画がこちらです。

なお、実際の入力音声は字幕の通りです。

.envファイルを作成し、OpenAIのAPIキーを記述します。

OPENAI_API_KEY="sk-..."

まず、OpenAI Realtime APIのエンドポイント（wss://api.openai.com/v1/realtime?intent=transcription）に接続します。

接続時には、認証のためのAPIキーを含むヘッダー情報を付与します。

async with websockets.connect(WS_URL, additional_headers=HEADERS) as websocket:
    self._websocket = websocket
    logging.info("✅ WebSocket接続成功！")

WebSocket接続が確立したら、次にどのような文字起こしを行いたいかをサーバーに伝えます。

ここでは、使用するモデル（gpt-4o-transcribe）などをJSON形式で指定し、サーバーに送信します。

# 文字起こしセッションの設定を送信
session_config = {
    "type": "transcription_session.update",
    "session": {"input_audio_transcription": {"model": "gpt-4o-transcribe"}},
}
await self._websocket.send(json.dumps(session_config))

マイクからの音声をリアルタイムで取得し、細切れのデータ（チャンク）に分割します。

各チャンクはBase64形式にエンコードされ、input_audio_buffer.appendというイベントタイプを指定したJSONメッセージとして、WebSocketを通じてサーバーに送信され続けます。

この一連の処理は_audio_senderメソッドが担当しています。

async def _audio_sender(self):
    """マイクからの音声データをWebSocket経由で送信するタスク。"""
    logging.info("音声送信タスクを開始します。")
    while True:
        try:
            # マイクから音声データを読み込む
            audio_data = await asyncio.get_event_loop().run_in_executor(
                None, self._stream.read, CHUNK, False
            )
            
            # Base64にエンコードしてJSONメッセージを作成
            message = {
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(audio_data).decode("utf-8"),
            }
            
            await self._websocket.send(json.dumps(message))
        except websockets.exceptions.ConnectionClosed:
            logging.warning("WebSocket接続が閉じられたため、音声送信を停止します。")
            break
        except Exception as e:
            logging.error(f"音声送信中にエラーが発生しました: {e}")
            break
        # CPUを過度に消費しないように、他のタスクに制御を譲る
        await asyncio.sleep(0.01)

音声データを送信するのと並行して、サーバーからの応答を待ち受けます。

サーバーは文字起こしの進捗に応じて、中間結果（delta）と確定結果（completed）をJSONメッセージとして送り返してきます。

_transcript_receiverメソッドはこれらのメッセージを非同期で受信し、内容を解析してコンソールに出力します。

async def _transcript_receiver(self):
    """WebSocketから文字起こし結果を受信して表示するタスク。"""
    logging.info("文字起こし受信タスクを開始します。")
    full_transcript = []
    try:
        async for message_str in self._websocket:
            message = json.loads(message_str)
            msg_type = message.get("type")

            if msg_type == "conversation.item.input_audio_transcription.delta":
                # 中間結果をリアルタイムで表示
                print(message.get("delta", ""), end="", flush=True)
            elif msg_type == "conversation.item.input_audio_transcription.completed":
                # 発話の区切りで改行して表示
                transcript = message.get("transcript", "")
                full_transcript.append(transcript)
                print("\n>> ", end="", flush=True)
    except websockets.exceptions.ConnectionClosed as e:
        logging.info(f"WebSocket接続が正常に閉じられました: {e}")
    except Exception as e:
        logging.error(f"メッセージ受信中にエラーが発生しました: {e}")

ここまでのコードをまとめると以下のようになります。

import asyncio
import base64
import json
import logging
import os

import pyaudio
import websockets
from dotenv import load_dotenv

# --- ロギング設定 ---
logging.basicConfig(
    level=logging.INFO, format="[%(asctime)s] [%(levelname)s] - %(message)s"
)

# --- 設定値 ---
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise ValueError("環境変数 'OPENAI_API_KEY' が設定されていません。")

# WebSocket接続先URLとヘッダー情報
WS_URL = "wss://api.openai.com/v1/realtime?intent=transcription"
HEADERS = {
    "Authorization": f"Bearer {OPENAI_API_KEY}",
    "OpenAI-Beta": "realtime=v1",
}

# PyAudio設定
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 24000  # OpenAI Realtime APIは24kHzを推奨
CHUNK = 2048


class RealtimeTranscriber:
    """
    マイクからの音声をリアルタイムで文字起こしするクラス。
    """

    def __init__(self):
        self._pyaudio_instance = pyaudio.PyAudio()
        self._stream = None
        self._websocket = None

    async def _audio_sender(self):
        """マイクからの音声データをWebSocket経由で送信するタスク。"""
        logging.info("音声送信タスクを開始します。")
        while True:
            try:
                # マイクから音声データを読み込む (ブロッキング処理のためExecutorを使用)
                audio_data = await asyncio.get_event_loop().run_in_executor(
                    None, self._stream.read, CHUNK, False
                )
                
                # Base64にエンコードしてJSONメッセージを作成
                message = {
                    "type": "input_audio_buffer.append",
                    "audio": base64.b64encode(audio_data).decode("utf-8"),
                }
                
                await self._websocket.send(json.dumps(message))
            except websockets.exceptions.ConnectionClosed:
                logging.warning("WebSocket接続が閉じられたため、音声送信を停止します。")
                break
            except Exception as e:
                logging.error(f"音声送信中にエラーが発生しました: {e}")
                break
            # CPUを過度に消費しないように、他のタスクに制御を譲る
            await asyncio.sleep(0.01)

    async def _transcript_receiver(self):
        """WebSocketから文字起こし結果を受信して表示するタスク。"""
        logging.info("文字起こし受信タスクを開始します。")
        full_transcript = []
        try:
            async for message_str in self._websocket:
                message = json.loads(message_str)
                msg_type = message.get("type")

                if msg_type == "conversation.item.input_audio_transcription.delta":
                    print(message.get("delta", ""), end="", flush=True)
                elif msg_type == "conversation.item.input_audio_transcription.completed":
                    transcript = message.get("transcript", "")
                    full_transcript.append(transcript)
                    print("\n>> ", end="", flush=True)
        except websockets.exceptions.ConnectionClosed as e:
            logging.info(f"WebSocket接続が正常に閉じられました: {e}")
        except Exception as e:
            logging.error(f"メッセージ受信中にエラーが発生しました: {e}")

    async def run(self):
        """リアルタイム文字起こし処理を開始する。"""
        try:
            self._stream = self._pyaudio_instance.open(
                format=FORMAT,
                channels=CHANNELS,
                rate=RATE,
                input=True,
                frames_per_buffer=CHUNK,
            )

            async with websockets.connect(WS_URL, additional_headers=HEADERS) as websocket:
                self._websocket = websocket
                logging.info("✅ WebSocket接続成功！")

                # 文字起こしセッションの設定を送信
                session_config = {
                    "type": "transcription_session.update",
                    "session": {"input_audio_transcription": {"model": "gpt-4o-transcribe"}},
                }
                await self._websocket.send(json.dumps(session_config))

                print("\n🎤 リアルタイム文字起こしを開始します。話しかけてください。(Ctrl+Cで終了)")
                print(">> ", end="", flush=True)

                # 音声送信と文字起こし受信タスクを並行して実行
                sender_task = asyncio.create_task(self._audio_sender())
                receiver_task = asyncio.create_task(self._transcript_receiver())
                await asyncio.gather(sender_task, receiver_task)

        except websockets.exceptions.NegotiationError as e:
            logging.error(f"WebSocket接続に失敗しました (ステータスコード: {e.status_code})。")
            logging.error("APIキーが正しいか、または組織の利用制限を確認してください。")
        except Exception as e:
            logging.error(f"予期せぬエラーが発生しました: {e}")
        finally:
            # 使用したリソースをクリーンアップ
            if self._stream and self._stream.is_active():
                self._stream.stop_stream()
            if self._stream:
                self._stream.close()
            self._pyaudio_instance.terminate()
            logging.info("リソースを解放し、プログラムを終了します。")


async def main():
    """メイン関数。"""
    transcriber = RealtimeTranscriber()
    await transcriber.run()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n👋 プログラムが中断されました。")

このコードを.envファイルと同じディレクトリで実行します。

python main.py

実際の文字起こしした結果が以下の通りです。