Grok Voice Think Fast 2.0 APIチュートリアル：Pythonでリアルタイム音声エージェントを構築する

SpaceXAIのGrok Voice Think Fast 2.0は音声対音声モデルです。WebSocketで音声を送ると音声が返ってきます。その間に、モデルは推論し、発話を続けながら、すでに走り始めた関数呼び出しを進められます。別個の音声認識（speech-to-text）や音声合成（text-to-speech）の工程は不要です。

SpaceXAIは 2026年7月29日にThink Fast 2.0を発表しました。最初の音声までが速く、厳密なターン制御ではなく話しながら聞ける安定した全二重化、ターン早期のツール呼び出しが特長です。チュートリアルで重要なのはコードの変更点なので、ベンチマークの話は手短にします。

ここではECサイト向けのカスタマーサポート音声エージェントを作ります。発信者は注文の問い合わせ、配送指示の変更、キャンセル、エージェントの発話中の割り込み、切断後の会話再開ができます。これはAPIでの実装であり、Grok Voice Agent Builderのチュートリアルが扱うノーコードのビルダーではありません。コンソール中心で始めるなら、まずそちらから。

Grok Voice Think Fast 2.0とは？

Grok Voice Think Fast 2.0は、Speech to Speech API向けにSpaceXAIが提供する最新モデルです。多くの人がGrok Voiceと呼ぶプロダクト名の裏側にある正式名称です。会社をxAIと記憶している方も同じです。2026年7月6日にSpaceXへ統合されSpaceXAIに改称されました。APIはリブランディングに追随していないため、以下に出てくる識別子は xaiのままです。例えば XAI_API_KEY変数や api.x.aiホスト名など。

従来の音声スタックは、音声認識→言語モデル→音声合成の3つを連鎖させます。ホップごとに遅延とコンテキスト欠落のリスクが増えます。Think Fast 2.0はそれを1モデルに統合し、同一接続上で音声・テキストの入出力を扱います。

音声対音声WebSocketと3サービスのパイプラインアーキテクチャの比較。画像：著者作成。

「話すだけ」ではなく「行動する」エージェントにとって、推論と発話が並行して動くことが重要です。SpaceXAIによれば、ツール呼び出しは「通常」エージェントが最初の一文を話し終える前に実行開始します。この「通常」は重要な意味を持ちます。

SpaceXAIが引用するArtificial Analysisのベンチマークでは、Think Fast 2.0はSpeech to Speech Indexで1.0の75.7%に対して82.9%を記録し、最初の音声までの時間は1.25秒から0.70秒に短縮されています。ベンダーの一般ベンチマーク値は、実際の呼び出しフローに関する仮説であって、テスト計画ではありません。

目にするモデル文字列は3つあります。grok-voice-latest、grok-voice-think-fast-2.0、grok-voice-think-fast-1.0です。プロトタイピング時はエイリアスが便利ですが、それ以外には不安定です。

2026年8月4日のテスト時点では、grok-voice-latestはまだgrok-voice-think-fast-1.0に解決されていました。SpaceXAIのリリースノートでは翌日にThink Fast 2.0へ切り替える予定でした。この切り替えはモデル変更であると同時に価格変更でもあり、1.0の$0.05/分に対して$0.08/分です。エイリアスを固定しないと、コードを一行も変えずにコストが上がります。デプロイではバージョン付き文字列を固定してください。

作るもの

エージェントはサポート窓口に来る典型的な要件をカバーします。注文の照会、番号がない場合はメールからの検索、配送指示の変更、キャンセル、チケットの起票や照会、人へのエスカレーション。途中で割り込みや切断も発生します。

1つのスクリプトではなく小さなファイルの集合です。役割が異なるので、それぞれ個別にテストしたくなります。構成は次のとおりです。

• config.py：APIキーの読み込み、モデル文字列、サンプルレート、エンドポイントURLを保持

• voice_client.py ：WebSocketのラッパー、課金トラッキング、送受信ヘルパーを提供

• tools.py：注文機能群の定義と、実データベースの代役となる簡易インメモリ注文ストア

• assistant.py ：システムプロンプト、セッション設定、全体を結ぶイベントループ

• token_server.py：短期トークンを発行する小さなFastAPIエンドポイント

• app_streamlit.py：同じクライアントをブラウザのライブ通話に載せます（テスト後に戻ります）

学習の流れはターミナルから進めます。デモでマイクを追加します。

前提条件

必要なもの：SpaceXAIアカウント（APIキー付き）、十分な資金のある請求設定（恒久的な無料枠はありません。新規アカウントのプロモーションクレジットだけでは足りません）、そしてasyncioやWebSocketに関する基礎知識（awaitの逐語的な説明なしで追える程度）。

SpaceXAIのクイックスタートは専用SDKではなく素のwebsocketsパッケージを使っています。ここでも同様です。必要なPythonバージョンは明記されていませんが、筆者は3.11でテストしました。

APIキーはサーバー側に保持してください。ブラウザやモバイルアプリがVoice APIに直接接続する場合は、実キーではなく短期トークンを使います（後述のセキュリティ節で説明）。

プロジェクトのセットアップ

以下のファイルはすべてプロジェクトリポジトリにあります。スニペットをコピペせずにクローンできます。

git clone https://github.com/KhalidAbdelaty/grok-voice-think-fast-2.0.git
cd grok-voice-think-fast-2.0
pip install -r requirements.txt

websocketsがリアルタイム接続を担い、 python-dotenvがキーを読み込みます。残りはトークンエンドポイントとブラウザデモ用です。キーは.envに記述します：

XAI_API_KEY=xai-your-key-here

セットアップの大半はこれで完了です。要点は接続部分です。

Grok VoiceリアルタイムAPIの理解

Grok Voiceは製品名です。実際にコードで相手をするのはwss://api.x.ai/v1/realtimeのWebSocketエンドポイントで、会話は1本のソケット上を流れるJSONイベントのストリームとしてやり取りされます。

イベントのライフサイクル

接続は一定の流れに従います。接続直後にサーバーから session.created と conversation.createdが届き、クライアントは session.update で音声やツールの設定を送ります。サーバーは session.updatedで確認し、その後は会話アイテムの作成と応答のリクエストを繰り返します。実キーで試したところ、順序はドキュメント通りでした。

• session.update （クライアント）：音声、インストラクション、ツール、音声フォーマットを設定

• conversation.item.create （クライアント）：ユーザーメッセージ、アシスタントメッセージ、ツール結果を追加

• response.create （クライアント）：モデルに発話を依頼。サーバーVADは自動でこれを送信

• response.output_audio.delta と response.output_audio_transcript.delta （サーバー）：生成中の返答をストリーム配信

• response.done （サーバー）：ターンを終了

つまずきがちな点が2つ。先のSpeech to Speechのドキュメントにはセッション再開時のconversation.item.createdが記載されていますが、正式なイベントリファレンスにはconversation.item.addedのみが掲載され、テストでも常にこちらが届きました。こちらに合わせて実装してください。また、数秒後に未ドキュメントのping イベントが来ることがあります。エラーと勘違いしないように。

音声フォーマットとトランスポート

コーデックとトランスポートは別設定です。コーデックはaudio.input.format と audio.output.formatで指定し、audio/pcm（Linear16、デフォルト24,000Hz）、audio/pcmuまたは audio/pcma （G.711、8kHz：電話回線向け）、あるいは audio/opus（24kHz）です。トランスポートは、これらのバイトを線上でどう運ぶかです。

• json（デフォルト）：input_audio_buffer.append や response.output_audio.delta内で音声をbase64文字列として送受信。ログやデバッグが容易

• binary ：WebSocketのバイナリフレームでコーデックの生バイトを送受信。base64のオーバーヘッドを避ける代わりに、受信ループでメッセージ型の分岐が必要

まずはJSONで始めてください。ドキュメントの例はすべてJSONで、検査が容易です。サポートエージェントの構築ではbase64のオーバーヘッドはボトルネックになりません。必要性を測定してからbinaryに移行しましょう。

OpenAI Realtime APIとの互換性

OpenAIのRealtime APIに触れたことがない場合は読み飛ばしてください。そうでない場合、Speech to Speech APIはOpenAI Realtime APIをかなり近く追従しており、ベースURLとキーを差し替えるだけでクライアントコードの大半は移植できます。ただし完全なドロップインではありません。

こちらでは書き起こしがconversation.item.input_audio_transcription.updatedとして届き、OpenAIの deltaとは異なります。OpenAIのいくつかのイベントは未対応で、SpaceXAIは独自拡張としてforce_message（定型の開示文をそのまま再生）、 resumption（再接続）、 replace （TTS前に社名などの発音を補正）を提供しています。

リアルタイム音声エージェントの構築

プロトコルの説明はここまで。以下が実際に話すクライアントです。

接続とセッション設定

接続はベアラートークンとモデルのクエリパラメータで開始し、最初の送信メッセージでエージェントの挙動をすべて設定します。

import asyncio
import json
import os
import websockets

MODEL = "grok-voice-think-fast-2.0"  # pin the version, not grok-voice-latest

async def connect():
    url = f"wss://api.x.ai/v1/realtime?model={MODEL}"
    ws = await websockets.connect(
        url, additional_headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"}
    )
    await ws.send(json.dumps({
        "type": "session.update",
        "session": {
            "voice": "eve",
            "instructions": SYSTEM_PROMPT,
            "turn_detection": {"type": "server_vad"},
            "tools": ORDER_TOOLS,
            "resumption": {"enabled": True},
        }
    }))

return ws

instructionsはシステムプロンプトです。このモデルは短めのプロンプトを好みます。SpaceXAIの移行ノートでは、旧来のGPT系音声モデル向けプロンプトをそのまま移植せず、簡潔化するよう勧めています。筆者のプロンプトは、短く答えること、1度に1つの質問にすること、書き込み前に読み上げて確認することを指示します。口頭での確認はUX上の配慮であり、セキュリティ対策ではありません。実際の書き込みに対する認可はアプリケーション側で強制します。

驚いた点が1つ。未認識のモデル文字列を指定しても接続時にエラーは発生せず、黙って grok-voice-think-fast-1.0にフォールバックします。タイプミスで有料リクエストがダウングレードされるのに告知がないのは変わったデフォルトです。起動時にsession.createdの session.modelをログに出し、要求通りのモデルか確認してください。

接続後にsession.createdが表示されたターミナル出力。画像：著者作成。

ユーザー音声のストリーミング

turn_detection.typeをserver_vadにすると、やるべきことは音声を追加し続けるだけです。発信者の発話終了はサーバーが判断して応答を自動でトリガーします。nullにすると判定はクライアント側の責務になり、ターン終了と見なすタイミングでバッファを明示的にコミットします。

async def send_audio_chunk(ws, pcm_bytes: bytes):
    await ws.send(json.dumps({
        "type": "input_audio_buffer.append",
        "audio": base64.b64encode(pcm_bytes).decode(),
    }))

サーバーVADには3つの調整項目があり、ここを誤るとログにエラーがないのに「壊れて見える」音声エージェントになりがちです。いずれもデフォルトではsession.updatedのエコーに現れないため、推測ではなくドキュメントで確認してください。

• threshold（0.1〜0.9、デフォルト0.85）：音声として認識する音量の閾値。騒がしい環境では上げ、声が小さい人を取りこぼすなら下げる

• silence_duration_ms：無音がどれくらい続いたらターン終了とするか。短すぎると考え中に遮り、長すぎるともっさり感じる

• prefix_padding_ms（デフォルト333）：発話検出直前の音声を少し残し、語頭が切れないようにする

考え込む間に発話が頻繁に切られるなら、まずはsilence_duration_ms を調整してください。他の2つに手を出す前にここを触るのが定石です。

応答の受信と再生

音声はresponse.output_audio.deltaとして小さな塊で届きます。ストリーミングの要点は、response.doneを待たず、着弾した瞬間に再生することです。

async def play_response(ws):
    async for message in ws:
        event = json.loads(message)
        if event["type"] == "response.output_audio.delta":
            chunk = base64.b64decode(event["delta"])
            speaker.write(chunk)  # your playback call goes here
        elif event["type"] == "response.output_audio_transcript.delta":
            print(event["delta"], end="", flush=True)

本番でもトランスクリプトは保持してください。発信者が「変なことを言った」と報告したとき、最も安価で強力なデバッグ手段になります。

音声エージェントにツールを追加する

話すだけの音声エージェントは、マイク付きのチャットボットにすぎません。

注文ツールの作成

各ツールはJSONスキーマと、こちら側の素のPython関数で構成します。モデルがデータベースに直接触れることはなく、関数の戻り値だけを見ます。

ORDER_TOOLS = [
    {
        "type": "function",
        "name": "check_order_status",
        "description": "Look up the status, ETA, and delivery instructions for an order.",
        "parameters": {
            "type": "object",
            "properties": {
                "order_number": {"type": "string", "description": "e.g. ORD-1042"},
            },
            "required": ["order_number"],
        },
    },
    # find_orders, update_delivery_instructions, cancel_order,
    # create_support_ticket, check_ticket_status and transfer_to_human
    # all follow the same shape
]

読み取り操作（check_order_statusなど）はタイムアウト時にリトライしても安全です。書き込み操作は違います。あいまいなタイムアウト後にupdate_delivery_instructionsを再試行すると、同じ変更が二重適用される恐れがあります。プロンプトの確認行では防げません。書き込みには冪等性キーや重複チェックを実装してください。

拒否ロジックも関数側に入れましょう。cancel_orderは、出荷済み注文についてはキャンセルせずに理由と代替案を返します。「出荷済みはキャンセルしないで」とプロンプトに書くのは提案に過ぎません。関数の拒否は提案ではありません。

ツール呼び出しループの処理

4つの手順があり、順序は見た目以上に重要です。モデルがresponse.function_call_arguments.doneを送信し、あなたのコードが関数を実行し、その結果をfunction_call_output アイテムとして返し、その後で初めてモデルに続きを依頼します。

async def handle_tool_call(ws, event):
    args = json.loads(event["arguments"])
    result = execute(event["name"], args)  # never raises; errors come back as {"error": ...}
    await ws.send(json.dumps({
        "type": "conversation.item.create",
        "item": {
            "type": "function_call_output",
            "call_id": event["call_id"],
            "output": json.dumps(result),
        },
    }))

1つの要求に複数のツールが必要な場合、音声が再生される前に複数のfunction_call_arguments.doneが発火します。すべてを解決し、全結果を返してから、response.createを1回だけ送ってください。早まると、未反映のツール結果を無視した回答になります。

SpaceXAIが注意喚起しており、筆者も初回で踏んだハマりどころがあります。ツール結果を送った直後にresponse.createを即送信すると、エージェントがまだ序文を話している最中の発話とかぶることがあります。ある実行では「ORD-1042のステータスをすぐ確認します」と話し始め、その文の途中でツールを呼びました。即時のレスポンス送信は前置きにかぶってしまいます。

現在ターンの音声が終わるのを待ち、その間は短い「考え中」状態を見せてください。

続行前のツール呼び出しフロー。画像：著者作成。

割り込みと会話状態の管理

ここには2つの問題があります。エージェントの発話中に発信者が話し始める場合と、WebSocketが切れて再開が必要な場合です。

自然な割り込みへの対応

server_vadが有効なら、割り込み（barge-in）はサーバー側で自動対応します。発信者の再発話を検知した瞬間にinput_audio_buffer.speech_startedが通知され、旧応答の生成を停止します。クライアント側の役割は、すでにキューに積まれた音声をクリアして、誰も求めていない文を最後まで再生しないようにすることです。

if event["type"] == "input_audio_buffer.speech_started":
    playback_queue.clear()

手動（非VAD）の場合は、response.cancelで同じことができます。さらに、実際に聞こえた範囲までアシスタントのアイテムを切り詰めるconversation.item.truncateもあります。ドキュメントには存在が示されていますが、ライブの割り込み中にいつ発火すべきかは明記されていません。タイミングは実際に検証してください。

配達指示の変更でテストしました。リクエストを開始し、エージェントの確認発話の途中で別の住所に割り込みます。大事なのは音声が止まったかではなく、訂正後の指示が適用されたかどうかです。サイレンスではなく注文レコードを検証してください。最後のブラウザデモで体験できます。

切断されたセッションの再開

セッションの再開はオプトインであり、記憶機能ではありません。 session.update でresumption.enabled: trueを設定し、conversation.createdのIDを控え、ソケットが落ちたらURLに?conversation_id=<id>を付けて再接続し、新しい接続でも再びオプトインします。

async def reconnect(conversation_id):
    url = f"wss://api.x.ai/v1/realtime?model={MODEL}&conversation_id={conversation_id}"
    ws = await websockets.connect(url, additional_headers=auth_header)
    await ws.send(json.dumps({"type": "session.update", "session": {"resumption": {"enabled": True}}}))
    return ws

キャッシュされたターン、書き起こし、ツール呼び出しとその結果が次の質問前に再生されます。30分無操作でキャッシュは消えます。注文を尋ね、接続を落とし、言い直さずに再接続して追問するテストでは、ETAを正しく引き継ぎました。

未ドキュメントの注意点が1つ。再生は即時には届きません。ソケットを開いた直後に投げた質問が先行し、直前のターンを覚えていない応答が返ることがあります。再開を疑う前に1秒待ってください。

再開したセッションのターミナルログ。画像：著者作成。

これは自前のデータベースへの注文状態の保存の代用ではありません。キャッシュが期限切れになったり翌日に再コールされたりすると、コンテキストはゼロからになります。これは仕様です。

エージェントのセキュリティと監視

恒久的なAPIキーをブラウザやモバイルのコードに埋め込まないでください。クライアントがサーバーを介さずに直接接続するなら、短期トークンを発行します。

from fastapi import FastAPI
import httpx, os

app = FastAPI()

@app.post("/session")
async def create_session():
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.x.ai/v1/realtime/client_secrets",
            headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
            json={"expires_after": {"seconds": 300}},
        )
    return response.json()  # {"value": "xai-realtime-client-secret-...", "expires_at": ...}

ブラウザはWebSocketハンドシェイクでカスタムのAuthorizationヘッダーを設定できないため、sec-websocket-protocolヘッダーにxai-client-secret.を接頭してトークンを渡します。

サーバーがトークンを発行し、ブラウザが通話に参加。画像：著者作成。

課金は2系統です。送受信の音声は先述の$0.08/分（= $4.80/時）で、音声以外かつfunction_call_outputではないconversation.item.createは一律$0.004です。response.create自体は課金されません。各response.doneにはusageオブジェクトが付き、テストではoutput_audio_secondsと別にbillable_audio_secondsが報告されました。推定ではなくこの値を元に請求してください。

Speech to Speech APIの制限は、チームあたり同時セッション10件、セッション上限120分（いずれもus-east-1）と記されています。Voice Agent APIの数値とは異なるため、容量計画の参考にしないでください。

プライバシーについては正確に。SpaceXAIのセキュリティFAQによれば、APIのリクエストとレスポンスは不正対策のため30日間暗号化保存され、許可なく学習には使用されません。チームはZero Data Retentionを有効化できますが、ZDRは音声エージェントの会話履歴の永続化を無効化するため、再開（resumption）とは両立しません。

通話が録音・AI対応である旨の開示には、先述のforce_message拡張を使います。モデルの言い換えではなく、記述した文がそのまま再生されます。

音声エージェントのテスト

WebSocketハンドシェイクの200は、エージェントが正しく動作したかを何も示しません。接続ではなく結果をテストしてください。

• 注文照会の正常系。音声応答の到着有無ではなく、レコードとの一致を確認
• 割り込み発話。再生が止まり、新リクエストに対応することを確認
• 要確認の配送更新。注文レコードへの反映で検証
• 拒否ケース（出荷済みのキャンセルなど）。謝罪ではなく規則の説明ができるか
• 不明な注文番号。ステータスを捏造せず「不明」と言えるか
• エラーを返すツール。エージェントが固まらず、エラーを口頭で伝えるか
• 再接続と再開。上で触れたリプレイのタイミングも含めて
• ノイズの多い音、早口、数字や住所を綴りで伝える発信者

執筆中に大半を実キーで走らせました。興味深い失敗はエラーではなく挙動でした。先述の再開タイミング、そして範囲外のVADしきい値が拒否されずに受け入れられた件などです。ハッピーパスだけを試すと見逃される類です。多言語テストも追加し、言語名の指定に関する注意点はFAQを参照してください。

タイプだけでは試せない項目が2つあります。 app_streamlit.pyは Streamlitのページで、ブラウザ上でライブ通話を実現します。マイクがWebRTC経由で同じWebSocketにストリームし、エージェントの声が戻り、ソケットは通話中ずっと開いたままです。

streamlit run app_streamlit.py