Руководство по API Grok Voice Think Fast 2.0: создаём голосового агента реального времени на Python

Grok Voice Think Fast 2.0 от SpaceXAI — это модель «речь-в-речь». Вы отправляете ей аудио по WebSocket, а она возвращает аудио; при этом модель может рассуждать и продолжать говорить, пока уже запущен выбранный ею вызов функции. Никаких отдельных шагов распознавания речи и синтеза речи.

SpaceXAI объявила о Think Fast 2.0 29 июля 2026 года: более быстрое первое аудио, более стабильный полнодуплексный режим (слушает, пока говорит, а не строго по очереди) и ранние вызовы инструментов в рамках реплики. На бенчмарках долго не задерживаюсь: для практикума важнее, что меняется в вашем коде.

Мы построим голосового агента службы поддержки интернет‑магазина. Звонящий может спросить о заказе, изменить инструкции по доставке, отменить, перебить агента посреди фразы и продолжить диалог после обрыва соединения. Это путь через API, а не конструктор без кода, который разбирает наш учебник по Grok Voice Agent Builder. Для версии через консоль начните оттуда.

Что такое Grok Voice Think Fast 2.0?

Grok Voice Think Fast 2.0 — новейшая модель SpaceXAI для Speech to Speech API — продуктового названия того, что большинство называет просто Grok Voice. Если вы всё ещё думаете о компании как о xAI — это та же команда: 6 июля 2026 года она была интегрирована в SpaceX и переименована в SpaceXAI. API названия не менял, поэтому все идентификаторы ниже по‑прежнему содержат xai — от переменной XAI_API_KEY до хоста api.x.ai.

Классический голосовой стек — это цепочка из трёх сервисов: распознавание речи, языковая модель, затем синтез речи. Каждый переход добавляет задержку и риск потери контекста. Think Fast 2.0 сводит всё к одной модели, которая принимает на вход аудио или текст и по тому же соединению выдаёт аудио или текст.

WebSocket «речь‑в‑речь» против трёхсервисной архитектуры. Иллюстрация автора.

Для агента, который действует, а не только говорит, важно, что рассуждение и речь идут параллельно. По словам SpaceXAI, вызовы инструментов «обычно» стартуют до того, как агент закончит первую фразу — и это «обычно» тут не случайно.

На бенчмарках, на которые ссылается SpaceXAI — Artificial Analysis — Think Fast 2.0 набирает 82,9% в Speech to Speech Index против 75,7% у 1.0 и сокращает время до первого аудио с 1,25 секунды до 0,70 секунды. Показатели вендора на общем бенчмарке — это гипотеза о вашем звонке, а не план тестирования.

Есть три строковых названия модели: grok-voice-latest, grok-voice-think-fast-2.0 и grok-voice-think-fast-1.0. Псевдоним удобен на этапе прототипирования и недостаточно надёжен для чего‑то большего.

Когда я тестировал 4 августа 2026 года, grok-voice-latest всё ещё разрешался в grok-voice-think-fast-1.0, а в заметках о релизах SpaceXAI перенос на Think Fast 2.0 был намечен на следующий день. Этот переключатель — смена не только модели, но и цены: $0,08 за минуту аудио против $0,05 у 1.0, так что незакреплённый псевдоним подорожает, хотя вы не меняли ни строки кода. В развёртывании закрепляйте версию явной строкой.

Что мы построим

Агент покрывает типичные запросы линии поддержки: найти заказ, найти его по email, если у звонящего нет номера, изменить инструкции по доставке, отменить, открыть или проверить тикет и передать звонок оператору. По пути будут перебивания и обрыв соединения.

Это несколько небольших файлов, а не один скрипт: у каждого своя задача, и вы захотите тестировать их по отдельности. Структура такая:

• config.py загружает ключ API и хранит строку модели, частоту дискретизации и URL конечных точек

• voice_client.py инкапсулирует WebSocket, ведёт учёт биллинга и предоставляет помощники отправки/приёма

• tools.py определяет функции для заказов и небольшой in-memory хранилище заказов вместо реальной базы

• assistant.py содержит системный промпт, конфигурацию сессии и цикл обработки событий, который всё связывает

• token_server.py — небольшой эндпоинт FastAPI, который выпускает эфемерные токены

• app_streamlit.py даёт тот же клиент в браузерном звонке в реальном времени; к нему вернусь после раздела о тестировании

Обучающий путь — из терминала. Демонстрация добавляет микрофон.

Предварительные требования

Вам нужна учётная запись SpaceXAI с ключом API, пополненный платёжный метод (постоянного бесплатного тарифа нет, промокредиты для новых аккаунтов не спасут) и достаточный уровень владения asyncio и WebSocket, чтобы следовать без построчного разбора await.

В примерах быстрого старта SpaceXAI используется чистый пакет websockets, а не отдельный SDK — так поступим и мы. В документации нет требований к версии 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 Realtime API

Grok Voice — это продуктовое имя. Фактически вы работаете с WebSocket-эндпоинтом по адресу wss://api.x.ai/v1/realtime, и весь диалог — это поток 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 (сервер) закрывает реплику

Две вещи часто мешают. На странице Speech to Speech, которую я привёл, упомянуто событие conversation.item.created при возобновлении сессии, но каноничный справочник событий содержит только conversation.item.added — и именно его я получал во всех тестах, так что ориентируйтесь на него. Вы также увидите недокументированное событие ping через несколько секунд после старта большинства соединений — упоминаю лишь для того, чтобы вы не приняли его за ошибку.

Аудиоформаты и транспорт

Кодек и транспорт выбираются отдельно. Кодек настраивается в audio.input.format и audio.output.format: audio/pcm (Linear16, по умолчанию 24000 Гц), audio/pcmu или audio/pcma (G.711 на 8 кГц, для телефонии), или audio/opus (24 кГц). Транспорт — это способ передачи байтов по сети:

• json (по умолчанию) отправляет аудио как base64‑текст внутри input_audio_buffer.append и response.output_audio.delta, удобно для логирования и отладки

• binary шлёт «сырые» байты кодека как бинарные кадры WebSocket, исключая накладные расходы base64 ценой усложнения цикла приёма (ветвление по типу сообщения)

Начните с JSON. Все примеры в документации используют его, он тривиально инспектируется, а накладные расходы base64 — не то, что станет узким местом для агента поддержки. Переходите на binary, только если замеры дадут повод.

Совместимость с OpenAI Realtime API

Пропустите, если вы никогда не работали с Realtime API OpenAI. Для остальных: Speech to Speech API достаточно близко следует OpenAI Realtime API, так что большую часть клиентского кода можно перенести, сменив базовый URL и ключ, но это не идеальная замена один‑к‑одному.

Транскрипты приходят как conversation.item.input_audio_transcription.updated вместо openai-шного delta, некоторые события OpenAI не поддерживаются, а SpaceXAI добавляет свои расширения: force_message для обязательной фразы раскрытия, resumption для переподключений и replace для исправления произношения брендов до синтеза речи.

Создание голосового агента в реальном времени

Хватит о протоколе. Вот клиент, который с ним говорит.

Подключение и настройка сессии

Соединение открывается с bearer‑токеном и параметром модели в запросе, а первое ваше сообщение задаёт всё поведение агента:

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, а не переносить их дословно. В моём агенте — короткие ответы, один вопрос за раз и обязательное зачитывание записи перед действием. Устное подтверждение — деталь UX, а не контроль безопасности. Авторизацию записи всё равно обеспечивает ваше приложение.

Что меня удивило: нераспознанная строка модели не даёт ошибку при подключении, а молча откатывается к grok-voice-think-fast-1.0. Платный запрос, «пониженный» из‑за опечатки без уведомления — странный дефолт. Логируйте поле session.model из session.created один раз при старте и проверяйте, что получили то, что просили.

Вывод терминала с 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 три регулятора, и неверные значения — самая частая причина, по которой голосовой агент «кажется сломанным», хотя в логах нет ошибок. По умолчанию они не отражаются в эхо session.updated, так что сверяйтесь с документацией, а не с предположениями.

• threshold (0,1–0,9, по умолчанию 0,85) — насколько громким должен быть звук, чтобы считаться речью; повышайте в шумных условиях, понижайте, если тихих говорящих не распознаёт

• silence_duration_ms — сколько тишины со стороны звонящего должно пройти, чтобы сервер завершил его реплику; слишком короткое — обрывает на полуслове, слишком длинное — ощущается медлительным

• prefix_padding_ms (по умолчанию 333) — фрагмент аудио непосредственно перед детекцией речи, чтобы не отрезать первый слог

В первую очередь настраивайте silence_duration_ms — если звонящих постоянно «обрывает» во время паузы на раздумье. Это мой первый регулятор до остальных двух.

Получение и воспроизведение ответа

Аудио приходит небольшими частями как 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 после неоднозначного таймаута может применить одно и то же изменение дважды. Фраза подтверждения в промпте это не предотвратит, поэтому давайте на записи idempotency‑ключ или проверку на дубликаты.

Отказы тоже закладывайте в функцию. cancel_order возвращает причину и альтернативу вместо отмены отправленного заказа, потому что «никогда не отменять отправленные заказы» в промпте — это пожелание, а отказ в функции — факт.

Обработка цикла вызова инструментов

Четыре шага, и порядок важнее, чем кажется. Модель присылает 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),
        },
    }))

Если модели нужно несколько инструментов для запроса, она инициирует несколько событий function_call_arguments.done до начала воспроизведения аудио. Разрешите их все и отправьте каждый результат прежде, чем посылать один‑единственный response.create. Если отправить его слишком рано, модель ответит без контекста незавершённых вызовов.

Есть подводный камень, о котором SpaceXAI пишет, и который я всё равно поймал с первого раза: отправка response.create сразу после результата инструмента может наложиться на ещё проигрываемое вступительное предложение агента. В одном прогона он начал с «Сейчас проверю статус заказа ORD‑1042» и вызвал инструмент посреди фразы, так что мгновенный ответ заговорил бы поверх вступления.

Дождитесь завершения текущего аудио и покажите короткое состояние «думает» между ними.

Поток вызова инструмента перед продолжением ответа. Иллюстрация автора.

Управление перебиваниями и состоянием беседы

Здесь две разные задачи. Звонящий перебивает агента в середине ответа, и 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 для обрезки элемента ассистента до реально услышанного. В документации подтверждается его наличие, но не указано, когда именно его вызывать при живом барж‑ине — подберите тайминг опытным путём.

Я проверял это на изменении инструкции по доставке в середине ответа: начать запрос, перебить другой деталью адреса во время подтверждения агента. Важно, чтобы агент применил скорректированную инструкцию, а не тихо довёл старую, а не то, остановилось ли аудио. Сверяйтесь с записью заказа, а не с тишиной. В конце есть браузерная демо, где это можно услышать.

Возобновление оборванной сессии

Возобновление сессии — по выбору и это не память. Поставьте resumption.enabled: true в session.update, возьмите ID из события conversation.created, и если сокет оборвётся — переподключайтесь с ?conversation_id=<id> в URL и снова включайте опцию на новом соединении.

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.

Есть недокументированная особенность: повтор не прилетает мгновенно, поэтому вопрос, отправленный в момент открытия сокета, может «обогнать» воспроизведение и вернуться без памяти о предыдущей реплике. Дайте секунду, прежде чем винить механизм возобновления.

Журнал терминала с возобновлённой сессией. Иллюстрация автора.

Не используйте это вместо сохранения состояния заказа в собственной базе. Если кэш истечёт или звонящий наберёт завтра, вы начнёте с нуля — так задумано.

Безопасность и мониторинг агента

Никогда не размещайте постоянный ключ 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": ...}

Браузер не может задать пользовательский заголовок Authorization в рукопожатии WebSocket, поэтому он передаёт токен через заголовок sec-websocket-protocol с префиксом xai-client-secret..

Сервер выпускает токен, браузер присоединяется к звонку. Иллюстрация автора.

Биллинг ведётся по двум счётчикам. Аудио, отправленное или полученное, тарифицируется по $0,08 за минуту, о чём я упоминал выше (то есть $4,80 в час), а каждый conversation.item.create, который не аудио и не function_call_output, стоит фиксированные $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 — они другие.

Что до приватности — будьте точны. В FAQ по безопасности SpaceXAI говорится, что запросы и ответы API хранятся в зашифрованном виде 30 дней для мониторинга злоупотреблений и не используются для обучения без разрешения, а также что команды могут включить Zero Data Retention, но ZDR отключает историю разговоров голосового агента и потому несовместим с возобновлением.

Если вы сообщаете, что звонок записывается или обрабатывается ИИ, для этого и служит расширение force_message, которое я упоминал. Фраза проигрывается ровно как написана, а не как её перефразирует модель.

Тестирование голосового агента

Статус 200 на рукопожатии WebSocket ничего не говорит о том, корректно ли агент выполнил задачу. Тестируйте результат, а не только соединение.

• Чистый поиск заказа — сверяйте озвученный ответ с записью, а не просто факт ответа
• Прерванный ответ — подтвердите, что воспроизведение остановилось и агент переключился на новый запрос
• Обновление доставки с подтверждением — сверяйте с записью заказа
• Отказ — например, отмена отправленного заказа: агент должен объяснить правило, а не извиняться
• Неизвестный номер заказа — убедитесь, что агент так и говорит, а не выдумывает статус
• Инструмент, вернувший ошибку — проверьте, что агент её озвучивает, а не «зависает»
• Переподключение и возобновление — включая окно воспроизведения, с которым я столкнулся
• Шумное аудио, быстрая речь и звонящий, диктующий номера и адреса по буквам

Большинство из этого я прогнал на живом ключе, пока писал статью. Интересные сбои — поведенческие, не ошибки: описанная выше задержка при возобновлении и выход за пределы допустимого порога VAD, принятый вместо отклонения — то, что тихо уедет в прод, если тестировать только «счастливый путь». Добавьте многоязычный тест и посмотрите раздел FAQ — там нюанс с указанием языка.

Два сценария вы не проверите печатая. app_streamlit.py — это страница на Streamlit, которая даёт живой звонок в браузере: микрофон стримит в тот же WebSocket по WebRTC, голос агента стримится обратно, и сокет открыт всё время.

streamlit run app_streamlit.py