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

SpaceXAIのGrok Voice Think Fast 2.0は、音声対音声モデルです。WebSocketで音声を送ると音声が返ってきます。その途中で、モデルは推論を行い、すでに開始した関数呼び出しが実行中でも話し続けられます。別個の音声認識（STT）も、別個の音声合成（TTS）も不要です。

SpaceXAI は2026年7月29日にThink Fast 2.0を発表しました。最初の音声がより速く、ターン制ではなく話しながら聴くフルデュプレックス動作が安定し、ターンの早い段階でツール呼び出しが走ります。チュートリアルで重要なのはコードの変更点なので、ベンチマーク談義は最小限にします。

ここではオンラインストア向けのカスタマーサポート音声エージェントを作ります。発信者は注文の照会、メールからの検索（番号がない場合）、配達指示の変更、キャンセル、エージェントの話を途中で遮る、切断後に会話を再開する、といったことができます。これはAPIの道筋であり、ノーコードのVoice Agent Builderではありません。コンソール中心のやり方は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.05だった1.0に対し$0.08になります。固定していないエイリアスだと、コードを1行も変えずに高くなってしまいます。デプロイするものはバージョン付きの文字列を固定してください。

作るもの

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

1本のスクリプトではなく小さなファイルの集合にします。各コンポーネントの役割が異なり、個別にテストしたくなるからです。構成は次のとおりです。

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

• voice_client.py ：WebSocketのラッパー、課金の追跡、送受信ヘルパーを提供します

• tools.py：注文関連の関数を定義し、実データベースの代わりとなる小さなインメモリの注文ストアを用意します

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

• token_server.py：エフェメラルトークンを発行する小さなFastAPIエンドポイントです

• app_streamlit.py：同じクライアントをライブのブラウザ通話に載せます。これはテストの後に戻ってきます。

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

前提条件

APIキー付きのSpaceXAIアカウント、資金が入った課金設定（恒久的な無料枠はありません。新規アカウントのプロモーションクレジットは当てにできません）、そして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,000 Hz）、audio/pcmuまたは audio/pcma （G.711 8 kHz、電話回線向け）、あるいは audio/opus（24 kHz）が選べます。トランスポートはそれらのバイトを回線上でどう運ぶかです。

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

• binary ：WebSocketのバイナリフレームとしてコーデックの生バイトを送受信。base64のオーバーヘッドを省けますが、受信ループでメッセージタイプの分岐が必要に

まずはJSONから始めてください。ドキュメントの例はすべてこれを使い、検査も簡単です。サポートエージェント構築のボトルネックはbase64のオーバーヘッドではありません。必要があれば計測してからbinaryへ移行しましょう。

OpenAI Realtime APIとの互換性

OpenAIのRealtime APIを触ったことがないなら、この節は飛ばして構いません。それ以外の方へ。Speech to Speech APIはOpenAI Realtime APIを十分に追随しており、ベースURLとキーを替えるだけで多くのクライアントコードは移植できます。ただし完全なドロップインではありません。

トランスクリプトはOpenAIのdeltaではなく、こちらではconversation.item.input_audio_transcription.updatedとして届きます。いくつかのOpenAIイベントは未対応で、SpaceXAIは独自拡張も追加しています：スクリプトどおりの注意喚起を流すforce_message、再接続のresumption、誤読されるブランド名をTTS前に直す replaceなどです。

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

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

接続とセッション設定

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

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つずつ、書き込み操作は実行前に読み上げて確認する、と指示します。読み上げ確認は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が有効なら、バー ジインはサーバー側で自動です。発信者の再発話を検知した瞬間に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つ。リプレイは即時には届かず、接続が開いた瞬間に投げた質問がそれを先行してしまい、直前のターンを覚えていない返答になることがあります。再開を疑う前に少し待ってください。

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

これは自前のデータベースへの注文状態保存の代替にはなりません。キャッシュが期限切れになったり、翌日に折り返しが来た場合は文脈ゼロからの再開です。これは仕様です。

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

恒久的な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メーターで動きます。送受信の音声は前述の1分あたり$0.08（1時間$4.80）。さらに、音声でもfunction_call_outputでもないconversation.item.createは1件あたり$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は音声エージェントの会話履歴の永続化が落ちるため、再開機能とは両立しません。

通話が録音・AI対応である旨を開示するなら、前述のforce_message拡張を使います。モデルが言い換えるのではなく、書いたとおりの文言が再生されます。

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

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

• 正しい注文照会。返ってきた音声をレコードと照合し、レスポンス到着だけで合格にしない
• 割り込みレスポンス。再生停止と、新しい要求への適切な対応を確認
• 確認が必要な配達更新。注文レコードに反映されたかを確認
• 発送済みキャンセルなどの拒否。ルールの説明をさせ、謝罪で済ませない
• 未知の注文番号。ステータスを捏造せず「不明」と言えるか
• エラーを返すツール。エージェントが黙らずエラー内容を口頭で伝えるか
• 再接続と再開。先述のリプレイのタイミングも含めて検証
• 雑音の多い音声、速い話し方、数字や住所をスペルアウトする発信者

執筆中、これらの多くをライブキーで実行しました。興味深い失敗はエラーではなく挙動でした。前述の再開のタイミングや、範囲外のVADスレッショルドを拒否せず受け入れてしまうなど。ハッピーパスだけをテストすると、静かに壊れたまま出荷される類の問題です。多言語テストも加え、言語名の指定に関する注意点はFAQを参照してください。

このうち2つはタイピングではテストできません。 app_streamlit.pyはStreamlitのページで、ブラウザでライブ通話を実現します。マイクがWebRTC経由で同じWebSocketにストリーミングし、エージェントの声が戻り、ソケットは通話中開いたままです。

streamlit run app_streamlit.py