Grok Voice Agent Builder: Ghid practic în Python

xAI a lansat Voice Agent Builder, o consolă pentru crearea de agenți vocali. Descrii fluxul apelului, atașezi documente și unelte, și alegi o voce.

Când testez o consolă de agent vocal, mă interesează mai puțin nota de lansare și mai mult părțile pe care trebuie să le conectez în cod: cum e configurată sesiunea WebSocket, cum circulă audio-ul, unde au loc apelurile către unelte (tool calls), cât costă apelul și cum ar invoca altă aplicație fluxul.

Codul de mai jos reconstruiește acel flux direct peste Voice Agent API. Mai exact, vom folosi un asistent pentru programări la clinică ce verifică disponibilitatea, răspunde prin voce, urmărește costul, gestionează eșecurile uneltelor și expune un endpoint FastAPI.

Ce este Grok Voice Agent Builder?

Voice Agent Builder este consola xAI pentru a crea și lansa agenți vocali pe Grok Voice. A fost lansat în beta pe 1 iulie 2026. În loc să folosești servicii separate pentru speech-to-text, model de limbaj și text-to-speech, folosești un singur traseu de model vocal.

Consola include telefonie, regăsire de documente, unelte și conectori, guardrails, servere MCP la distanță și jurnale ale apelurilor cu înregistrări, transcrieri și trace-uri.

Audio-ul este tarifat pe minut. Consola este încă în beta, așa că folosim direct API-ul.

Cum funcționează Grok Voice Agent API sub Builder

Sub consolă se află Voice Agent API, un API WebSocket în timp real care expune același runtime folosit de Builder.

Builder rulează peste Voice API. Imagine de Autor.

Modelul folosit aici este grok-voice-think-fast-1.0. Aliasul grok-voice-latest indică cel mai nou model. Îl folosesc aici, dar pentru o aplicație în producție aș fixa numele versiunii. xAI raportează un scor de 67,3% pentru acest model pe topul τ-voice Bench; îl tratez ca pe un reper, nu ca pe o garanție.

Notă de compatibilitate: API-ul este compatibil cu OpenAI Realtime API. Dacă ai cod care vorbește cu endpointul realtime al OpenAI, de cele mai multe ori schimbi doar baza URL și cheia.

Prezentarea proiectului: ce vom construi

Asistentul clinicii primește input vocal, răspunde cu o voce generată, pune întrebări de clarificare, verifică disponibilitatea înainte să ofere un interval și transferă la un om când e nevoie. Exemplul de bază folosește o singură unealtă; demo-ul Streamlit adaugă acțiuni de rezervare, transfer și închidere a apelului.

Tutorialul de bază se împarte în patru fișiere, fiecare cu un rol:

• voice_client.py conține clientul WebSocket, helper-ele audio și urmărirea costurilor

• tools.py conține check_availability, plus unelte demo suplimentare folosite de Streamlit

• assistant.py conține promptul de sistem, configurarea sesiunii și workflow-ul

• app.py servește totul prin FastAPI

Aceste patru fișiere sunt traseul prin articol. Repo-ul include și app_streamlit.py pentru demo-ul vizual și run.py ca launcher pentru Windows, dar revenim la ele după ce fluxul de bază funcționează.

Prerechizite

Înainte de a rula codul, ai nevoie de Python 3.10 sau mai nou, un cont xAI, o cheie API de la console.x.ai, credite preplătite și confort de bază cu variabile de mediu, JSON și WebSocket-uri.

Configurarea proiectului

Creează un folder și un mediu virtual, apoi instalează pachetele:

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

Pinează aceste pachete într-un requirements.txt ca un checkout proaspăt să folosească aceeași configurare.

Creează un fișier .env lângă fișierele Python:

XAI_API_KEY=xai-your-key-here

Adaugă .env la .gitignore. Cheia API trebuie să rămână pe server.

Construirea agentului vocal

Hai să începem să construim.

Conectarea la Grok Voice Agent API prin WebSocket

Primul pas este deschiderea conexiunii. Transmite modelul ca parametru de query și cheia ta ca bearer token în handshake:

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

Cu o cheie validă, primul eveniment pe care îl vezi este session.created, ceea ce înseamnă că socketul este deschis și gata de configurare.

Evenimentul Session created confirmă conexiunea. Imagine de Autor.

Configurarea sesiunii vocale

Un socket activ nu este un agent configurat. Îl modelezi trimițând un eveniment session.update cu un obiect session.

Voce, format audio și instrucțiuni

Cele trei setări pe care le atingi cel mai des sunt vocea, formatul audio și promptul de sistem. API-ul realtime expune cinci voci denumite, eve, ara, rex, sal și leo, plus orice clonă personalizată. Audio-ul are implicit audio/pcm la 24000 Hz, cu input și output configurate separat.

Iată configurarea sesiunii folosită de asistent, asamblată în 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],
    }

Câmpul instructions este promptul de sistem. Acest prompt pentru clinică rămâne scurt pentru că răspunsurile vocale lungi sunt greu de urmărit:

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 de escaladare ține agentul clinicii departe de oferirea de sfaturi medicale. Ultimele două rânduri îl mențin în scop și opresc buclele când apelantul nu e clar. Configurația adaugă și data de azi pentru că, în testele mele live, modelul putea ghici anul greșit pentru date ca „6 iulie”.

Ajustarea detecției de tură

Detecția de tură stabilește cum decide agentul că ai terminat de vorbit. Setează turn_detection.type la server_vad și serverul încheie tura la liniște. Lasă-l null și controlezi tu turele prin commit pe bufferul audio, ceea ce folosesc pentru fluxul pe fișier.

Server VAD are trei setări utile: threshold stabilește cât de tare trebuie să fie audio ca să conteze ca vorbire, silence_duration_ms stabilește cât de lungă e pauza care încheie tura, iar prefix_padding_ms păstrează puțin audio înainte de începerea vorbirii. Dacă agentul întrerupe oamenii, mărește silence_duration_ms mai întâi.

Trimiterea audio-ului către agent

Acum trimitem vocea apelantului. Audio-ul trebuie să se potrivească formatului sesiunii: mono PCM pe 16 biți la 24000 Hz, codat base64 și trimis pe bucăți.

Clientul transmite fișierul în felii, apoi face commit pe buffer pentru a marca finalul turei:

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)

Dacă rata de eșantionare sau codarea ta nu se potrivește cu session.update, e posibil să obții paraziți sau liniște în loc de o eroare clară. Audio-ul trece prin input_audio_buffer.append, deci este taxat după durată, nu per mesaj.

Primirea răspunsurilor vocale

După ce soliciți un răspuns, audio-ul sosește ca response.output_audio.delta, transcrierea sosește ca response.output_audio_transcript.delta, iar response.done închide tura.

Clientul colectează totul într-un singur loop async:

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

Decodează deltele audio, lipește-le în ordine și scrie rezultatul într-un fișier response.wav. Pentru a capta cuvintele apelantului, setează audio.input.transcription și citește conversation.item.input_audio_transcription.completed.

Construirea workflow-ului pentru asistentul de programări

Acum piesele devin o conversație: cerere de programare, întrebare de clarificare, verificarea disponibilității, intervale oferite, confirmare. Pentru a duce contextul între ture, fiecare tură nouă se reconectează cu id-ul conversației și optează pentru reluarea sesiunii.

Adăugarea apelării de unelte la agentul vocal

Pentru clinică, agentul trebuie să verifice disponibilitatea înainte să promită un interval. Uneltele personalizate sunt modul în care modelul ajunge la codul tău: emite o cerere, aplicația ta rulează funcția și trimiți înapoi rezultatul.

Unealta este o funcție simplă plus un schelet JSON care intră în configurarea sesiunii. Iată schema din 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"],
    },
}

Buclele au o formă fixă. Când modelul vrea unealta, trimite response.function_call_arguments.done cu argumentele. Rulezi funcția, returnezi un function_call_output și apoi trimiți response.create ca agentul să poată continua. Dacă omiți acel response.create final, agentul rămâne tăcut.

Turul dus-întors al apelului de unealtă, explicat. Imagine de Autor.

Funcțiile personalizate ca aceasta rulează în codul tău. Demo-ul Streamlit înregistrează încă trei din același fișier: book_appointment, transfer_to_human și end_call. Uneltele built-in, precum căutarea pe web, căutarea pe X, căutarea în colecții și unelte MCP la distanță, rulează pe serverele xAI.

Gestionarea eșecurilor uneltelor

Uneltele mai și eșuează, iar un agent vocal care presupune succesul poate promite un interval care nu există. Metoda mea ToolRegistry.execute nu aruncă niciodată excepții: o interogare eșuată se întoarce ca un dict {"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)}

Un stare explicită de eroare oprește agentul să trateze apelurile eșuate ca succes.

Adăugarea urmăririi costurilor

Înainte să servești asta cuiva, află cât costă un apel. Audio-ul este taxat cu 0,05 $ pe minut, contorizând atât ce trimiți, cât și ce primești. Evenimentele de input text sunt taxate cu 0,004 $ fiecare. Rezultatele function_call_output și evenimentele response.create nu sunt taxate.

Clientul le urmărește din mers, astfel că poți citi costul oricând ca proprietate:

@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

Un număr de telefon furnizat de xAI adaugă suprataxa de telefonie de 0,01 $ pe minut, pe care helper-ul o aplică atunci când setezi telephony=True. Uneltele găzduite de xAI se facturează separat: căutarea pe web și pe X sunt în jur de 5 $ la o mie de apeluri, iar căutarea în fișiere este aproximativ 2,50 $.

Gestionarea erorilor și a cazurilor limită

Cele mai multe eșecuri se încadrează într-o listă scurtă:

• Cheie API lipsă sau invalidă returnează 401 la handshake, deci verifică mai întâi cheia

• O echipă blocată returnează 403, iar un rate limit returnează 429, pe care îl retrimiți cu backoff

• Configurație de sesiune malformată returnează 400, de obicei o literă greșită în numele unui câmp

• Format audio neacceptat dă paraziți, nu o eroare, deci potrivește rata sesiunii

• Un response.create lipsă după un rezultat de unealtă lasă agentul în așteptare

• O încercare duplicată de rezervare poate cauza probleme reale, așa că nu retrimite orbește

Reluarea unei citiri eșuate precum check_availability este sigură, dar reluarea unei scrieri eșuate, precum o rezervare reală, poate duce la dublarea programării. Orice acțiune care schimbă date are nevoie mai întâi de o verificare a idempotentei.

Folosirea tokenurilor efemere pentru aplicații client

Tot ce am făcut până acum presupune că codul rulează pe serverul tău, unde aparține cheia API. Dacă un browser sau o aplicație mobilă se conectează direct, folosește tokenuri efemere.

Serverul tău apelează POST https://api.x.ai/v1/realtime/client_secrets cu cheia ta, primește un răspuns cu token și transmite valoarea tokenului către client. În rularea mea, răspunsul a inclus 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()

Browserele nu pot seta headere WebSocket personalizate, așa că tokenul merge în headerul sec-websocket-protocol cu prefix xai-client-secret..

Transformarea workflow-ului într-un endpoint FastAPI

Un endpoint permite unui frontend sau altui serviciu să apeleze workflow-ul. Ruta validează corpul cererii cu un model Pydantic, primește un mesaj tipizat sau o cale către audio și returnează transcrierea, audio-ul de răspuns, jurnalul uneltelor, latența și costul estimat.

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

Rulează cu uvicorn app:app --reload și deschide http://localhost:8000/docs. Citește XAI_API_KEY din mediul serverului și nu o accepta niciodată din corpul unei cereri.