Grok Voice Agent Builder: praktyczny przewodnik w Pythonie

xAI udostępniło Voice Agent Builder — konsolę do tworzenia agentów głosowych. Opisujesz przebieg rozmowy, dołączasz dokumenty i narzędzia oraz wybierasz głos.

Gdy testuję konsolę do agentów głosowych, mniej obchodzi mnie notka o premierze, a bardziej to, co muszę podłączyć w kodzie: jak skonfigurować sesję WebSocket, jak przepływa audio, gdzie wywoływane są narzędzia, ile kosztuje połączenie i jak inna aplikacja wywoła ten workflow.

Poniższy kod odtwarza ten przepływ bezpośrednio przez Voice Agent API. Konkretnie, zbudujemy asystenta umawiania wizyt w klinice, który sprawdza dostępność, odpowiada głosem, śledzi koszt, obsługuje awarie narzędzi i wystawia endpoint FastAPI.

Czym jest Grok Voice Agent Builder?

Voice Agent Builder to konsola xAI do tworzenia i wdrażania agentów głosowych na Grok Voice. Wersja beta wystartowała 1 lipca 2026. Zamiast używać oddzielnie usług speech-to-text, modelu językowego i text-to-speech, korzysta z jednej ścieżki modelu głosowego.

Konsola obejmuje telefonię, wyszukiwanie w dokumentach, narzędzia i konektory, guardrails, zdalne serwery MCP oraz logi połączeń z nagraniami, transkrypcjami i trace’ami.

Audio jest rozliczane za minutę. Konsola jest wciąż w becie, więc korzystamy bezpośrednio z API.

Jak działa Grok Voice Agent API pod Builderem

Pod konsolą działa Voice Agent API — websocketowe API czasu rzeczywistego, które udostępnia ten sam runtime co Builder.

Builder opiera się na Voice API. Obraz: autor.

Model użyty tutaj to grok-voice-think-fast-1.0. Alias grok-voice-latest wskazuje na najnowszy model. Używam go tutaj, ale dla wdrożonej aplikacji przypiąłbym wersjonowaną nazwę. xAI raportuje wynik 67,3% dla tego modelu w rankingu τ-voice Bench; traktuję to jako jedną wskazówkę, nie gwarancję.

Uwaga dot. zgodności: API jest kompatybilne z OpenAI Realtime API. Jeśli masz kod mówiący do endpointu realtime OpenAI, w większości zmieniasz tylko bazowy URL i klucz.

Przegląd projektu: co zbudujemy

Asystent kliniki przyjmuje mowę, odpowiada wygenerowanym głosem, zadaje pytania uzupełniające, sprawdza dostępność przed zaproponowaniem terminu i przekazuje sprawę do człowieka, gdy trzeba. Przykład bazowy używa jednego narzędzia; demo w Streamlit dodaje akcje rezerwacji, przekazania i zakończenia rozmowy.

Rdzeń tutoriala dzieli się na cztery pliki, każdy z jedną rolą:

• voice_client.py zawiera klienta WebSocket, pomocniki audio i śledzenie kosztów

• tools.py zawiera check_availability plus dodatkowe narzędzia demo używane przez Streamlit

• assistant.py zawiera prompt systemowy, konfigurację sesji i workflow

• app.py serwuje całość przez FastAPI

Te cztery pliki wyznaczają ścieżkę przez artykuł. Repo zawiera też app_streamlit.py do wizualnego dema i run.py jako launcher dla Windows, ale wrócimy do nich po uruchomieniu rdzeniowego przepływu.

Wymagania wstępne

Zanim uruchomisz kod, potrzebujesz Pythona 3.10 lub nowszego, konta xAI, klucza API z console.x.ai, przedpłaconych środków oraz podstawowej swobody z zmiennymi środowiskowymi, JSON-em i WebSocketami.

Konfiguracja projektu

Utwórz folder i wirtualne środowisko, a potem zainstaluj pakiety:

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

Spnij wersje pakietów w requirements.txt, aby świeży checkout używał tej samej konfiguracji.

Utwórz plik .env obok plików Pythona:

XAI_API_KEY=xai-your-key-here

Dodaj .env do .gitignore. Klucz API powinien pozostać na serwerze.

Budowa agenta głosowego

Zacznijmy budowę.

Łączenie z Grok Voice Agent API przez WebSocket

Pierwszy krok to otwarcie połączenia. Przekaż model jako parametr zapytania, a klucz jako bearer token w handshake’u:

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())

Przy ważnym kluczu pierwszym zdarzeniem będzie session.created, co oznacza, że socket jest otwarty i gotowy do konfiguracji.

Zdarzenie utworzenia sesji potwierdza połączenie. Obraz: autor.

Konfigurowanie sesji głosowej

Żywy socket to nie skonfigurowany agent. Kształtujesz go, wysyłając zdarzenie session.update z obiektem session.

Głos, format audio i instrukcje

Trzy ustawienia, które zmieniasz najczęściej, to głos, format audio i prompt systemowy. API realtime udostępnia pięć nazwanych głosów: eve, ara, rex, sal i leo, plus ewentualny własny klon. Domyślny format audio to audio/pcm w 24000 Hz, osobno dla wejścia i wyjścia.

Oto konfiguracja sesji używana przez asystenta, składana w 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],
    }

Pole instructions to prompt systemowy. Ten dla kliniki zostaje krótki, bo długich odpowiedzi głosowych trudno się słucha:

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.

Linia o eskalacji trzyma agenta kliniki z dala od porad medycznych. Dwie ostatnie linie utrzymują go w temacie i zapobiegają pętlom, gdy wypowiedź dzwoniącego jest niejasna. Konfiguracja dopisuje też dzisiejszą datę, bo w moich testach model potrafił zgadnąć zły rok przy datach typu „6 lipca”.

Strojenie wykrywania tur

Wykrywanie tur określa, kiedy agent uznaje, że skończyłeś mówić. Ustaw turn_detection.type na server_vad, a serwer zakończy turę po ciszy. Zostaw null, a kontrolujesz tury, commitując bufor audio — tak robię w przepływie plikowym.

Server VAD ma trzy istotne ustawienia: threshold określa głośność uznawaną za mowę, silence_duration_ms określa długość pauzy kończącej turę, a prefix_padding_ms zachowuje trochę audio sprzed startu mowy. Jeśli twój agent wchodzi ludziom w słowo, najpierw podnieś silence_duration_ms.

Wysyłanie audio do agenta

Teraz wysyłamy głos dzwoniącego. Audio musi odpowiadać formatowi sesji: mono PCM 16-bit w 24000 Hz, zakodowane base64 i wysyłane w kawałkach.

Klient strumieniuje plik w porcjach, a potem commit’uje bufor, by zaznaczyć koniec tury:

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)

Jeśli częstotliwość próbkowania lub kodowanie nie zgadzają się z session.update, możesz usłyszeć szum lub ciszę zamiast jasnego błędu. Audio przechodzi przez input_audio_buffer.append, więc jest rozliczane za czas trwania, a nie per wiadomość.

Odbieranie odpowiedzi głosowych

Po zażądaniu odpowiedzi audio przychodzi jako response.output_audio.delta, transkrypcja jako response.output_audio_transcript.delta, a response.done zamyka turę.

Klient zbiera to wszystko w jednej pętli asynchronicznej:

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

Dekoduj delty audio, sklej je w kolejności i zapisz wynik do pliku response.wav. Aby przechwycić słowa dzwoniącego, ustaw audio.input.transcription i czytaj conversation.item.input_audio_transcription.completed.

Budowa workflow asystenta umawiania wizyt

Teraz składamy elementy w rozmowę: prośba o rezerwację, pytanie doprecyzowujące, sprawdzenie dostępności, proponowane terminy, potwierdzenie. Aby nie tracić kontekstu między turami, każda nowa tura łączy się ponownie z identyfikatorem rozmowy i włącza wznawianie sesji.

Dodawanie wywołań narzędzi do agenta głosowego

W klinice agent musi sprawdzić dostępność, zanim obieca termin. Własne narzędzia to sposób, w jaki model sięga do twojego kodu: emituje żądanie, twoja aplikacja uruchamia funkcję i odsyłasz wynik.

Narzędzie to zwykła funkcja plus schemat JSON, który trafia do konfiguracji sesji. Oto schemat z 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"],
    },
}

Pętla ma stały kształt. Gdy model chce narzędzie, wysyła response.function_call_arguments.done z argumentami. Uruchamiasz funkcję, zwracasz function_call_output, a następnie wysyłasz response.create , aby agent mógł kontynuować. Jeśli pominiesz końcowe response.create, agent zamilknie.

Runda wywołania narzędzia wyjaśniona. Obraz: autor.

Takie własne funkcje działają w twoim kodzie. Demo Streamlit rejestruje trzy kolejne z tego samego pliku: book_appointment, transfer_to_human i end_call. Wbudowane narzędzia, takie jak wyszukiwanie w sieci, wyszukiwanie w X, wyszukiwanie w kolekcjach oraz zdalne narzędzia MCP, uruchamiane są na serwerach xAI.

Obsługa awarii narzędzi

Narzędzia zawodzą, a agent głosowy, który zakłada sukces, może obiecać termin, którego nie ma. Moje ToolRegistry.execute nigdy nie rzuca wyjątków: nieudane wyszukanie zwraca słownik {"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)}

Jawny stan błędu powstrzymuje agenta przed traktowaniem nieudanych wywołań jako sukcesu.

Dodawanie śledzenia kosztów

Zanim podasz to komuś, wiedz, ile kosztuje rozmowa. Audio rozliczane jest po 0,05 USD za minutę, licząc to, co wysyłasz i co odbierasz. Zdarzenia wejścia tekstowego kosztują 0,004 USD każde. Wyniki function_call_output i zdarzenia response.create nie są rozliczane.

Klient śledzi to na bieżąco, więc koszt możesz odczytać w dowolnym momencie:

@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

Numer telefoniczny przydzielony przez xAI dodaje dopłatę 0,01 USD za minutę, co pomocnik uwzględnia, gdy ustawisz telephony=True. Narzędzia hostowane przez xAI są rozliczane osobno: wyszukiwanie w sieci i w X to ok. 5 USD za tysiąc wywołań, a wyszukiwanie w plikach ok. 2,50 USD.

Obsługa błędów i przypadków brzegowych

Większość awarii mieści się na krótkiej liście:

• Brak lub nieprawidłowy klucz API zwraca 401 przy handshake’u — najpierw sprawdź klucz

• Zablokowany zespół zwraca 403, a limit tempa 429 — wznawiaj z backoffem

• Błędna konfiguracja sesji daje 400, zwykle literówka w nazwie pola

• Nieobsługiwany format audio daje szum zamiast błędu — dopasuj częstotliwość sesji

• Brak response.create po wyniku narzędzia zawiesza agenta

• Próba podwójnej rezerwacji może narobić kłopotów — nie retry’uj bezrefleksyjnie

Ponowienie nieudanego odczytu, jak check_availability, jest bezpieczne, ale ponowienie nieudanego zapisu, jak faktyczna rezerwacja, może zdublować termin. Każda akcja zmieniająca dane wymaga najpierw sprawdzenia idempotencji.

Używanie tokenów efemerycznych w aplikacjach klienckich

Dotąd zakładaliśmy, że kod działa na twoim serwerze — tam powinien być klucz API. Jeśli przeglądarka lub aplikacja mobilna łączy się bezpośrednio, użyj tokenów efemerycznych.

Twój serwer wywołuje POST https://api.x.ai/v1/realtime/client_secrets z twoim kluczem, dostaje odpowiedź z tokenem i przekazuje wartość tokena klientowi. U mnie odpowiedź zawierała value i 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()

Przeglądarki nie mogą ustawiać własnych nagłówków WebSocket, więc token jedzie w nagłówku sec-websocket-protocol z prefiksem xai-client-secret..

Zamiana workflow na endpoint FastAPI

Endpoint pozwala frontendowi lub innej usłudze wywołać workflow. Trasa waliduje body żądania modelem Pydantic, przyjmuje wiadomość tekstową lub ścieżkę do audio i zwraca transkrypt, odpowiedź audio, log narzędzi, opóźnienie i szacowany koszt.

@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,
    }

Uruchom komendą uvicorn app:app --reload i otwórz http://localhost:8000/docs. Odczytuj XAI_API_KEY ze środowiska serwera i nigdy nie przyjmuj go z body żądania.