1. はじめに
前回の記事「Realtime API 正式版 ( gpt-realtime ) を試す」では、Realtime APIの機能や可能性を検証しました。
今回は、実際に動くアプリケーションを作り、Realtime APIの実装方法について、接続確立からイベント処理、音声データの送受信まで、全体像をまとめました。
サンプルコード
ソースコードとセットアップ方法はGitHubで公開しています。
2. サンプルアプリの紹介
2.1 サンプルアプリ概要
ユーザーがマイクに向かって話すと、AIがリアルタイムで音声とテキストで応答するチャットアプリケーションです。
画面の録音開始ボタンを押すと音声で会話できるようにしています。
ユーザーが話すと、ユーザーの音声が自動的に文字起こしされて表示されます。RealtimeAPIが音声とテキストでリアルタイムに応答します。
2.2 主な機能
サンプルアプリでは以下のような機能を実装しています。
-
✅ リアルタイム音声対話（Speech-to-Speech）
- Realtime APIによる低遅延の音声対話
- ユーザの発話の終端は、VADで自動検知
- ※STT（音声→テキスト）とTTS（テキスト→音声）を経由せず、音声で直接処理
-
✅ ユーザー音声の文字起こし
- 内部的にWhisper-1を使用（
input_audio_transcription設定） - ※現状Whisper-1ではストリーミング表示されず、完了後に一括表示
- 内部的にWhisper-1を使用（
-
✅ Realtime APIの応答ストリーミング
- Realtime APIのテキスト応答はリアルタイムでストリーミング表示
- 音声も同時にストリーミング再生
-
✅ テキスト入力にも対応
- 音声なしでもチャット可能
2.3 技術スタックと全体アーキテクチャ
┌─────────────┐ ┌──────────────┐ ┌─────────────────┐
│ ブラウザ │◄───────►│ Chainlit │◄────────►│ Realtime API │
│ │ HTTP/WS │ (app.py) │ WebSocket│ (Azure OpenAI) │
└─────────────┘ └──────────────┘ └─────────────────┘
│ │ │
マイク入力 イベント処理 Speech-to-Speech
音声再生 状態管理 + 文字起こし(Whisper-1)
データフロー：
- ユーザーがマイクで話す → PCM16音声データ
- ChainlitがWebSocketでRealtime APIに送信
- サーバーサイドVADが発話を検出
- Speech-to-Speech処理：音声入力→音声出力
- 並行して文字起こし：ユーザ音声をWhisper-1でテキスト化（完了後に表示）
- Realtime APIが応答生成 → テキストと音声を同時配信（ストリーミング）
- Chainlitが画面に表示 + 音声再生
3. Realtime APIの実装：4つのステップ
Realtime APIの実装を4つのステップでまとめます。
3.1 WebSocket接続の確立
Realtime APIは常時接続の双方向チャネルで、音声とイベントをリアルタイムに送受信します。サンプルアプリではWebSocketベースで実装しています。
実装コード
def _create_azure_client() -> AsyncAzureOpenAI:
"""Azure OpenAI クライアントを作成"""
return AsyncAzureOpenAI(
azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
api_key=os.environ["AZURE_OPENAI_API_KEY"],
api_version="2025-04-01-preview",
)
async def _setup_azure_realtime():
"""WebSocket接続を確立"""
client = _create_azure_client()
# WebSocket接続
cm: AsyncRealtimeConnectionManager = client.beta.realtime.connect(
model=AZURE_DEPLOYMENT
)
connection: AsyncRealtimeConnection = await cm.__aenter__()
# セッションに保存
cl.user_session.set(KEY_CONN, connection)
cl.user_session.set(KEY_CM, cm)
3.2 セッション設定：音声対話の基本設定
WebSocket接続が確立したら、まず session.update で会話のモードや音声設定を指定します。
ここで、Realtime API が どんな形式でやり取りするか を設定します。
各パラメータの解説
| パラメータ | 説明 |
|---|---|
modalities |
AIが返す応答の形式を指定します。["text"]のみ(音声なし)、["audio"]のみ（テキストなし）、または両方（音声とテキストを同時に返す）。今回のアプリは音声チャットアプリなので ["text","audio"] を指定します。 |
input_audio_format / output_audio_format
|
音声データのフォーマットを指定します。サーバとクライアントでサンプルレートやビット深度が一致しないと、音声が速く/遅く再生されたり、認識精度が落ちるのでエンドツーエンドで統一させます。Chainlitのconfig.tomlもこれにあわせます。 |
turn_detection.type |
ユーザーの発話終端をどう検出するかを決めます。server_vadにするとサーバー側で無音検出して自動で応答を開始します。noneまたは未設定の場合は、クライアント側でユーザーの発話が終わったことをinput_audio_buffer.commit()をCallして明示します。 |
turn_detection.threshold |
音声検出感度（0.5〜0.7） |
turn_detection.silence_duration_ms |
無音判定時間（700ms前後が自然）短すぎると途中で切れる、長すぎると応答が遅い |
turn_detection.prefix_padding_ms |
発話開始前のバッファ（300ms程度） |
input_audio_transcription |
ユーザー音声をテキスト化する設定です。 |
interrupt_response |
AI応答中にユーザーが話し始めたら、応答を中断するかどうかを指定します。falseだと割り込み無効で、trueだと割り込み有効になります。 |
instructions |
AIの振る舞いを決めるプロンプトを設定します。会話のトーンや制約をここで定義します。 |
実装コード
await connection.session.update(
session={
# モダリティ：テキストと音声の両方を有効化
"modalities": ["text", "audio"],
# 音声フォーマット
"input_audio_format": "pcm16", # 入力（マイク）
"output_audio_format": "pcm16", # 出力（スピーカー）
# 音声認識設定（ユーザー発話の文字起こし）
"input_audio_transcription": {
"model": "whisper-1",
"language": "ja", # 日本語を明示的に指定
},
# ターン検出（発話の自動検出）
"turn_detection": {
"type": "server_vad", # サーバー側で音声検出
"threshold": 0.6, # 音声検出の感度
"prefix_padding_ms": 300, # 発話開始前のバッファ
"silence_duration_ms": 700, # 無音が続いたら発話終了
"interrupt_response": False, # AI応答中の割り込み無効
"create_response": True, # 自動応答生成
},
# システムプロンプト
"instructions": "Always respond in Japanese. ユーザーとは日本語で会話してください。",
}
)
3.3 イベント処理：テキスト・音声・完了・エラーの処理
Realtime APIはイベント駆動で動作します。サーバーから送られてくるイベントを非同期で受信し、タイプごとに処理します。
イベントの流れ（典型的な会話の例）
1. ユーザー発話開始
↓
2. input_audio_transcription.delta（繰り返し）
（処理はストリーミング対応しているが、現状はストリーミング表示にはなっていない）
↓
3. input_audio_transcription.completed
最終文字起こし：「こんにちは」
↓
4. response.text.delta（繰り返し）
「こんに」→「こんにちは」→「こんにちは！」
↓
5. response.audio.delta（繰り返し）
音声データのチャンク送信
↓
6. response.done
応答完了
主要イベント一覧
| イベントタイプ | 発生タイミング | 用途 |
|---|---|---|
conversation.item.input_audio_transcription.delta |
ユーザーが話している間 | 文字起こしをリアルタイム表示 |
conversation.item.input_audio_transcription.completed |
ユーザーの発話終了 | 最終的な文字起こし確定 |
response.text.delta |
AIが応答生成中 | テキストをストリーミング表示 |
response.audio_transcript.delta |
AIが音声生成中 | 音声の文字起こし（テキストがない場合） |
response.audio.delta |
AIが音声生成中 | 音声データをスピーカーに送信 |
response.done |
AI応答完了 | 状態をリセット |
error |
エラー発生 | エラーハンドリング |
実装コード
接続確立後、下記のイベントループを開始して、サーバーからのイベントを受信できるように、asyncio.create_taskでバックグラウンドタスクとして実行しておきます。
async def _receive_events_loop(connection: AsyncRealtimeConnection):
"""メインイベントループ"""
try:
async for event in connection:
event_type = getattr(event, "type", "")
# イベントタイプに応じて処理を分岐
if event_type == "response.text.delta":
# AIのテキスト応答（差分）
await _handle_agent_text_delta(...)
elif event_type == "response.audio.delta":
# AIの音声データ（差分）
await _handle_agent_audio_delta(...)
elif event_type == "conversation.item.input_audio_transcription.delta":
# ユーザー音声の文字起こし（差分）
await _handle_user_transcription_delta(...)
elif event_type == "response.done":
# 応答完了
await _handle_response_done()
# ... 他のイベント処理
except asyncio.CancelledError:
pass # 正常終了
3.4 音声/テキストデータの送受信
3.4.1 マイクからRealtime APIへ
ユーザーの音声をRealtime APIに送信する実装です。WebSocketはテキストベースのプロトコルのため、バイナリデータはBase64エンコードが必要です。ブラウザから届いた PCM16 を Base64 文字列に変換して input_audio_buffer.append に流し込みます。
データフロー
ブラウザのマイク
↓ (PCM16バイナリ)
Chainlitが受信
↓ (Base64エンコード)
Realtime APIの音声バッファ
↓ (サーバーサイドVADが自動検出)
文字起こし + AI応答生成
実装コード
@cl.on_audio_chunk
async def on_audio_chunk(chunk: cl.InputAudioChunk):
"""ブラウザから音声チャンクを受信"""
connection = cl.user_session.get(KEY_CONN)
if connection is None:
return
# PCM16データをBase64エンコード
b64_audio = base64.b64encode(chunk.data).decode("ascii")
# Realtime APIの音声バッファに追加
await connection.input_audio_buffer.append(audio=b64_audio)
3.4.2 ユーザー音声の文字起こし表示
受信したイベントをChainlitのUIに反映します。
実装コード
async def _handle_user_transcription_delta(delta: Optional[str]):
"""ユーザー発話の文字起こし（差分）を画面に表示"""
if isinstance(delta, str) and delta:
user_msg = cl.user_session.get(KEY_CURRENT_USER_MSG)
if user_msg is None:
# 初回：新しいメッセージバブルを作成
user_msg = cl.Message(
content="",
author="user",
type="user_message"
)
await user_msg.send()
cl.user_session.set(KEY_CURRENT_USER_MSG, user_msg)
# 差分を追加（ストリーミング表示）
await user_msg.stream_token(delta)
async def _handle_user_transcription_completed(final_text: str):
"""文字起こし完了：メッセージを確定"""
user_msg = cl.user_session.get(KEY_CURRENT_USER_MSG)
if user_msg is None:
# ストリーミングがなかった場合、最終テキストで作成
if isinstance(final_text, str) and final_text.strip():
await cl.Message(
content=final_text,
author="user",
type="user_message",
).send()
else:
# ストリーミング中のメッセージを確定
await user_msg.update()
cl.user_session.set(KEY_CURRENT_USER_MSG, None)
3.4.3 AIのテキスト応答表示
Realtime APIは2種類のテキストイベントを送信します。
-
response.text.delta（テキスト応答） -
response.audio_transcript.delta（音声の文字起こし）
UIに両方のイベントを表示すると重複して表示されるため、クライアント側でロックの仕組みをつくり制御しました。
実装コード
async def _handle_agent_text_delta(delta: Optional[str]):
"""AIのテキスト応答（差分）を処理"""
if isinstance(delta, str) and delta:
# テキストストリームロックを設定（重複防止）
cl.user_session.set(KEY_TEXT_STREAM_LOCKED, True)
await _stream_agent_text(delta)
async def _stream_agent_text(delta: str):
"""AIのテキストをUIにストリーミング表示"""
current_msg = cl.user_session.get(KEY_CURRENT_RESPONSE_MSG)
if current_msg is None:
# 初回：新しいメッセージバブルを作成
current_msg = cl.Message(content="", author="エージェント")
await current_msg.send()
cl.user_session.set(KEY_CURRENT_RESPONSE_MSG, current_msg)
# 差分を追加
await current_msg.stream_token(delta)
async def _finalize_agent_message():
"""AIメッセージを確定"""
current_msg = cl.user_session.get(KEY_CURRENT_RESPONSE_MSG)
if current_msg is not None:
await current_msg.update()
cl.user_session.set(KEY_CURRENT_RESPONSE_MSG, None)
3.4.4 AIの音声データ再生
OpenAI Realtime APIでは、複数の音声ストリームが同時に存在する可能性があるため、各応答ごとに一意なトラックIDを割り当てることで、UIが個別の音声ストリームを正しく区別・再生できるようにしました。
同じトラックIDを使うことで連続した音声として再生されます。トラックIDを変えると音声が中断されるため、割り込み時に新しいIDを生成します。
データフロー
Realtime API
↓ (Base64エンコード済みPCM16)
_handle_agent_audio_delta
↓ (Base64デコード)
Chainlit音声API
↓ (WebSocketでブラウザに送信)
ブラウザのスピーカー
実装コード
async def _handle_agent_audio_delta(raw_b64: Optional[Any]):
"""AIの音声データを再生"""
if isinstance(raw_b64, (str, bytes)) and raw_b64:
# Base64 → PCM16バイトに変換
audio_bytes = base64.b64decode(raw_b64)
# 再生中フラグを設定
if not cl.user_session.get(KEY_IS_PLAYING):
cl.user_session.set(KEY_IS_PLAYING, True)
# Chainlitの音声APIで再生
await cl.context.emitter.send_audio_chunk(
cl.OutputAudioChunk(
mimeType="pcm16",
data=audio_bytes,
track=_get_track_id(), # 音声トラックID
)
)
4. 設定・チューニングのポイント
4.1 turn_detection
サーバー側でVAD制御をする場合は、ユーザーの発話状況にあわせてパラメータチューニングができます。
例えばsilence_duration_msが短すぎると、ユーザーが
「今日の天気は...（考え中）...どうですか？」
と考えている間に、「今日の天気は」だけで切れてAIが応答開始してしまいます。
ユーザー側の想定が難しい場合は、クライアント側でユーザーの発話が終わったことをinput_audio_buffer.commit()をCallして明示する選択肢も考えられます。
turn_detectionのパラメータ
"turn_detection": {
"type": "server_vad",
"threshold": 0.6, # 音声検出の感度（0.0～1.0）
"prefix_padding_ms": 300, # 発話開始前のバッファ
"silence_duration_ms": 700, # 無音判定時間
}
パラメータの効果
| パラメータ | 小さい値 | 大きい値 | 推奨値 |
|---|---|---|---|
threshold |
雑音も拾いやすい | 小さい声を拾いにくい | 0.5～0.7 |
silence_duration_ms |
途中で切れやすい | 応答が遅くなる | 500～1000 |
prefix_padding_ms |
発話の頭が切れる | 余分な音を拾う | 200～500 |
パラメータ設定例
# 静かな環境向け（高感度）
"turn_detection": {
"threshold": 0.4,
"silence_duration_ms": 500,
}
# 騒がしい環境向け（低感度）
"turn_detection": {
"threshold": 0.8,
"silence_duration_ms": 1000,
}
4.2 voice
Realtime APIは複数の声質をサポートしています。キャラクターにあわせて選択しましょう。
参考：Voice Options
声質の選び方
-
alloy: 中性的、クリア（デフォルト） -
echo: 男性的、落ち着いた -
fable: やや高め、親しみやすい -
nova: 女性的、明るい
voiceの設定
await connection.session.update(
session={
"voice": "alloy", # または "echo", "fable", "onyx", "nova", "shimmer"
# その他の設定...
}
)
4.3 instructions
Realtime API用のプロンプトガイドがでていますので、こちらを参考にしましょう。
Realtime Prompting Guide
5. おわりに
Realtime APIとChainlitの組み合わせは、音声対話アプリケーションの開発を大きく簡略化してくれました。
次はToolCallを試してみたいと思います。
Discussion