Учебник по 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, это та же команда: её включили в SpaceX и переименовали в SpaceXAI 6 июля 2026 года. 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.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 три настройки, и ошибки в них — самая частая причина «поломки» голосового агента при отсутствии ошибок в логах. По умолчанию они не эхоятся в 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 после неоднозначного тайм-аута может применить одно и то же изменение дважды. Строка подтверждения в промпте это не остановит, так что давайте записям идемпотентный ключ или проверку дубликатов.

Отказы тоже закладывайте в функцию. 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.

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

Лог терминала возобновлённой сессии. Изображение автора.

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

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

Никогда не кладите постоянный ключ 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