Grok Voice Agent Builder: практическое руководство на Python

xAI выпустила Voice Agent Builder — консоль для создания голосовых агентов. Вы описываете сценарий звонка, прикрепляете документы и инструменты и выбираете голос.

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

Код ниже воссоздаёт этот поток напрямую через Voice Agent API. Конкретно мы сделаем помощника клиники, который проверяет доступность, отвечает голосом, отслеживает стоимость, обрабатывает сбои инструментов и предоставляет конечную точку FastAPI.

Что такое Grok Voice Agent Builder?

Voice Agent Builder — это консоль xAI для создания и развёртывания голосовых агентов на Grok Voice. Бета-версия вышла 1 июля 2026 года. Вместо отдельных сервисов распознавания речи, языковой модели и синтеза речи используется единый голосовой маршрут модели.

Консоль включает телефонию, поиск по документам, инструменты и коннекторы, ограничители (guardrails), удалённые MCP-серверы и журналы звонков с записями, транскриптами и трассами.

Аудио тарифицируется поминутно. Консоль пока в бете, поэтому мы используем API напрямую.

Как работает Grok Voice Agent API под капотом Builder

Под консолью находится Voice Agent API — WebSocket API в реальном времени, предоставляющий тот же рантайм, что и Builder.

Builder работает поверх Voice API. Изображение автора.

Здесь используется модель grok-voice-think-fast-1.0. Псевдоним grok-voice-latest указывает на самую новую модель. Я использую его здесь, но для продакшена закрепил бы версионированное имя. xAI сообщает о 67,3% для этой модели в рейтинге τ-voice Bench; я воспринимаю это как одну точку данных, а не гарантию.

Заметка о совместимости: API совместим с OpenAI Realtime API. Если у вас есть код для realtime-эндпоинта OpenAI, в основном нужно сменить базовый URL и ключ.

Обзор проекта: что мы будем строить

Помощник клиники принимает голосовой ввод, отвечает сгенерированным голосом, задаёт уточняющие вопросы, проверяет доступность перед предложением слота и при необходимости передаёт разговор человеку. В базовом примере используется один инструмент; в демо на Streamlit добавлены действия бронирования, перевода на оператора и завершения звонка.

Базовое руководство разделено на четыре файла, каждый со своей задачей:

• voice_client.py содержит WebSocket-клиент, вспомогательные функции для аудио и учёт стоимости

• tools.py содержит check_availability, плюс дополнительные демо-инструменты для Streamlit

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

• app.py обслуживает всё это через FastAPI

Эти четыре файла и есть маршрут по статье. В репозитории также есть app_streamlit.py для визуального демо и run.py как лаунчер для Windows, но к ним вернёмся после того, как заработает базовый поток.

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

Перед запуском кода вам понадобятся Python 3.10 или новее, учётная запись xAI, ключ API из console.x.ai, предоплаченные кредиты и базовое понимание переменных окружения, JSON и WebSocket.

Настройка проекта

Создайте папку и виртуальное окружение, затем установите пакеты:

mkdir appointment-agent
cd appointment-agent
python -m venv .venv
.venv\Scripts\activate       # macOS/Linux: source .venv/bin/activate
pip install websockets python-dotenv fastapi uvicorn pydantic httpx numpy streamlit

Зафиксируйте версии пакетов в requirements.txt, чтобы при новом клонировании использовалась та же конфигурация.

Создайте файл .env рядом с Python-файлами:

XAI_API_KEY=xai-your-key-here

Добавьте .env в .gitignore. Ключ API должен оставаться на сервере.

Создаём голосового агента

Приступим к сборке.

Подключение к Grok Voice Agent API через WebSocket

Первый шаг — открыть соединение. Передайте модель как параметр запроса и ваш ключ как bearer-токен при рукопожатии:

import asyncio
import json
import os
import websockets

async def voice_agent():
    url = "wss://api.x.ai/v1/realtime?model=grok-voice-latest"
    async with websockets.connect(
        url,
        additional_headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
    ) as ws:
        async for message in ws:
            print(json.loads(message)["type"])

asyncio.run(voice_agent())

С рабочим ключом первым событием будет session.created — это значит, что сокет открыт и готов к настройке.

Событие session created подтверждает подключение. Изображение автора.

Настройка голосовой сессии

Открытый сокет — ещё не настроенный агент. Вы формируете его, отправляя событие session.update с объектом session.

Голос, аудиоформат и инструкции

Чаще всего вы меняете голос, формат аудио и системный промпт. Realtime API предоставляет пять именованных голосов: eve, ara, rex, sal и leo, плюс любой кастомный клон. По умолчанию аудио — audio/pcm на 24000 Гц, вход и выход настраиваются отдельно.

Вот конфигурация сессии, которую использует помощник, собирается в assistant.py:

def build_session_config(voice="ara", instructions=SYSTEM_PROMPT, sample_rate=24000):
    # The model needs to know "today" or it guesses the year for a date like "July 6th".
    instructions = f"{instructions}\nToday's date is {date.today().isoformat()}."
    return {
        "voice": voice,
        "instructions": instructions,
        "turn_detection": None,  # manual turns for file-based input
        "audio": {
            "input": {"format": {"type": "audio/pcm", "rate": sample_rate}},
            "output": {"format": {"type": "audio/pcm", "rate": sample_rate}},
        },
        "tools": [CHECK_AVAILABILITY_TOOL],
    }

Поле instructions — это системный промпт. Этот промпт для клиники остаётся коротким, потому что длинные голосовые ответы тяжело воспринимать:

You are a voice appointment assistant for a small clinic. Help callers book,
reschedule, cancel, or ask questions about appointments, services, and hours.
Answer whatever the caller asks that relates to the clinic. Keep responses short
and natural for a phone conversation. Ask one question at a time. Confirm
important details before taking action. Use the availability tool before offering
a time slot. Escalate to a human for medical, urgent, sensitive, or unclear
requests. If a caller asks about something unrelated to the clinic, say briefly
that it is outside what you can help with, then steer back to booking. If you
cannot make out what the caller said, ask them to repeat it instead of repeating
your last message.

Строка про эскалацию защищает агента клиники от выдачи медицинских советов. Последние две строки удерживают его в рамках темы и предотвращают зацикливание, когда слова абонента неразборчивы. В конфигурацию также добавляется сегодняшняя дата, потому что в моих тестах вживую модель могла угадать неверный год для дат вроде «6 июля».

Настройка детекции хода

Детекция хода определяет, когда собеседник закончил говорить. Установите turn_detection.type в server_vad — сервер завершит ход по тишине. Оставьте null — и вы управляете ходами, коммитя аудиобуфер; это я и использую для потока из файла.

У серверного VAD есть три важных настройки: threshold задаёт, насколько громкой должна быть речь, чтобы считаться речью, silence_duration_ms — какая пауза завершает ход, а prefix_padding_ms — сколько аудио до начала речи сохраняется. Если агент перебивает людей, сначала увеличьте silence_duration_ms.

Отправка аудио агенту

Теперь отправим голос абонента. Аудио должно соответствовать формату сессии: моно PCM 16 бит на 24000 Гц, закодированное в base64 и отправляемое чанками.

Клиент передаёт файл порциями, затем коммитит буфер, чтобы отметить конец хода:

async def send_audio(self, pcm_bytes, chunk_ms=100, commit=True):
    bytes_per_chunk = int(self._sample_rate * 2 * chunk_ms / 1000)
    for start in range(0, len(pcm_bytes), bytes_per_chunk):
        chunk = pcm_bytes[start:start + bytes_per_chunk]
        await self._t.send({
            "type": "input_audio_buffer.append",
            "audio": base64.b64encode(chunk).decode(),
        })
    if commit:
        await self._t.send({"type": "input_audio_buffer.commit"})
    self.cost.audio_seconds += pcm_seconds(pcm_bytes, self._sample_rate)

Если ваша частота дискретизации или кодировка не совпадают с session.update, вы можете получить шум или тишину вместо понятной ошибки. Аудио идёт через input_audio_buffer.append, поэтому тарификация идёт по длительности, а не по количеству сообщений.

Получение голосовых ответов

После запроса ответа аудио приходит как response.output_audio.delta, транскрипт — как response.output_audio_transcript.delta, а response.done закрывает ход.

Клиент собирает всё это в одном асинхронном цикле:

async def _collect_response(self):
    audio = bytearray()
    transcript, calls = [], []
    while True:
        event = await self._recv()
        etype = event["type"]
        if etype == "response.output_audio.delta":
            audio += base64.b64decode(event["delta"])
        elif etype == "response.output_audio_transcript.delta":
            transcript.append(event.get("delta", ""))
        elif etype == "response.function_call_arguments.done":
            calls.append(event)
        elif etype == "response.done":
            break
    return bytes(audio), "".join(transcript), calls

Декодируйте аудио-дельты, добавляйте их по порядку и запишите результат в файл response.wav. Чтобы получать слова самого абонента, установите audio.input.transcription и читайте conversation.item.input_audio_transcription.completed.

Построение рабочего процесса помощника по записи

Теперь из этих частей получится разговор: запрос на запись, уточняющий вопрос, проверка доступности, предложенные слоты, подтверждение. Чтобы не терять контекст между ходами, каждый новый ход переподключается с id беседы и включает возобновление сессии.

Добавление вызова инструментов в голосового агента

Для клиники агент должен проверить доступность, прежде чем обещать время. Пользовательские инструменты — это способ модели обратиться к вашему коду: она формирует запрос, ваше приложение выполняет функцию, и вы отправляете результат обратно.

Инструмент — это обычная функция плюс JSON-схема, которая входит в конфиг сессии. Вот схема из tools.py:

CHECK_AVAILABILITY_TOOL = {
    "type": "function",
    "name": "check_availability",
    "description": "Look up open appointment slots for a service on a given date. "
                   "Always call this before offering the caller a time.",
    "parameters": {
        "type": "object",
        "properties": {
            "service": {"type": "string", "description": "Service requested."},
            "date": {"type": "string", "description": "Requested date as YYYY-MM-DD."},
        },
        "required": ["service", "date"],
    },
}

Цикл имеет фиксированную форму. Когда модель хочет вызвать инструмент, она посылает response.function_call_arguments.done с аргументами. Вы запускаете функцию, возвращаете function_call_output, а затем отправляете response.create — чтобы агент мог продолжить. Пропустите финальный response.create — и агент замолчит.

Круг вызова инструмента наглядно. Изображение автора.

Такие пользовательские функции выполняются в вашем коде. Демо на Streamlit регистрирует ещё три из того же файла: book_appointment, transfer_to_human и end_call. Встроенные инструменты — такие как поиск в интернете, X-поиск, поиск по коллекциям и удалённые инструменты MCP — выполняются на серверах xAI.

Обработка сбоев инструментов

Инструменты ломаются, и голосовой агент, который предполагает успех, может пообещать несуществующий слот. Мой ToolRegistry.execute никогда не бросает исключение: неудачный поиск возвращается как словарь {"error": ...}.

def execute(self, name, arguments):
    handler = self._handlers.get(name)
    if handler is None:
        return {"error": f"unknown tool: {name}"}
    try:
        return handler(**arguments)
    except ToolError as exc:
        return {"error": str(exc)}

Явное состояние ошибки не даёт агенту принять неудачный вызов инструмента за успех.

Добавление учёта стоимости

Прежде чем отдавать это кому-то, знайте, во сколько обходится звонок. Аудио тарифицируется по $0.05 за минуту, считая и исходящее, и входящее. Текстовые события ввода — по $0.004 за каждое. Результаты function_call_output и события response.create не тарифицируются.

Клиент считает это на лету, так что стоимость — это свойство, которое можно прочитать в любой момент:

@property
def audio_usd(self):
    rate = 0.05 + (0.01 if self.telephony else 0.0)
    return self.audio_seconds / 60 * rate

@property
def total_usd(self):
    return self.audio_usd + self.text_usd + self.tool_usd

Предоставленный xAI номер добавляет надбавку телефонии $0.01 за минуту, что помощник применяет при telephony=True. Инструменты, размещённые у xAI, тарифицируются отдельно: веб-поиск и X-поиск — около $5 за тысячу вызовов, поиск по файлам — около $2.50.

Обработка ошибок и пограничных случаев

Большинство сбоёв укладывается в короткий список:

• Отсутствующий или некорректный ключ API даёт 401 при рукопожатии — сначала проверьте ключ

• Заблокированная команда вернёт 403, а лимит запросов — 429; повторяйте с бэк-оффом

• Неверная конфигурация сессии вернёт 400 — обычно опечатка в названии поля

• Неподдерживаемый аудиоформат даёт шум, а не ошибку — подгоняйте частоту сессии

• Пропущенный response.create после результата инструмента оставит агента «висящим»

• Повторная попытка бронирования может создать реальные проблемы — не ретрайте бездумно

Повторять неудачное чтение вроде check_availability безопасно, но повторять неудачную запись — как реальное бронирование — может привести к двойной записи. Любое действие, меняющее данные, должно проходить проверку идемпотентности.

Эфемерные токены для клиентских приложений

До сих пор предполагалось, что код работает на вашем сервере — там и место ключу API. Если браузер или мобильное приложение подключается напрямую, используйте эфемерные токены.

Ваш сервер вызывает POST https://api.x.ai/v1/realtime/client_secrets с вашим ключом, получает ответ с токеном и передаёт значение токена клиенту. В моём запуске ответ включал value и expires_at:

@app.post("/session")
async def create_session():
    async with httpx.AsyncClient() as client:
        response = await client.post(
            CLIENT_SECRETS_URL,
            headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
            json={"expires_after": {"seconds": 300}},
        )
    return response.json()

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

Преобразуем рабочий процесс в конечную точку FastAPI

Конечная точка позволяет фронтенду или другому сервису вызывать рабочий процесс. Маршрут валидирует тело запроса с помощью модели Pydantic, принимает типизированное сообщение или путь к аудио и возвращает транскрипт, ответное аудио, журнал инструментов, задержку и оценочную стоимость.

@app.post("/appointments/voice")
async def appointments_voice(body: VoiceRequest):
    fail = {"check_availability"} if body.simulate_tool_failure else None
    assistant = AppointmentAssistant(voice=body.voice, telephony=body.telephony, fail_tools=fail)
    if body.text:
        result = await assistant.run_live(text=body.text, conversation_id=body.conversation_id)
    else:
        pcm = load_wav_as_pcm(body.audio_path, 24000)
        result = await assistant.run_live(pcm, conversation_id=body.conversation_id)
    return {
        "transcript": result.transcript,
        "audio_wav_base64": base64.b64encode(encode_wav_bytes(result.audio, 24000)).decode(),
        "tool_calls": result.tool_calls,
        "latency_seconds": round(result.latency_s, 3),
        "estimated_cost_usd": round(result.cost.total_usd, 6),
        "audio_seconds": round(result.cost.audio_seconds, 2),
        "conversation_id": result.conversation_id,
    }

Запустите uvicorn app:app --reload и откройте http://localhost:8000/docs. Читайте XAI_API_KEY из переменных окружения сервера и никогда не принимайте его из тела запроса.