Tutorial de la API Grok Voice Think Fast 2.0: crea un agente de voz en tiempo real con Python

Grok Voice Think Fast 2.0 de SpaceXAI es un modelo de voz a voz. Le envías audio por WebSocket y te devuelve audio; entre medias, puede razonar y seguir hablando mientras ya se está ejecutando una llamada a función que decidió lanzar. Sin paso aparte de speech-to-text ni de text-to-speech.

SpaceXAI anunció Think Fast 2.0 el 29 de julio de 2026: primer audio más rápido, comportamiento full‑dúplex más estable (escucha mientras habla en lugar de turnos estrictos) y llamadas a herramientas que se disparan pronto en el turno. Iré rápido con los benchmarks: para un tutorial, lo que importa es qué cambia en tu código.

Vamos a crear un agente de voz de soporte para una tienda online. Quien llama puede consultar un pedido, encontrarlo por email si no tiene número, cambiar instrucciones de entrega, cancelar, interrumpir al agente a mitad de frase y retomar la conversación tras una caída de conexión. Este es el camino por API, no el constructor no‑code que cubre nuestro tutorial de Grok Voice Agent Builder. Empieza allí si prefieres la versión desde consola.

¿Qué es Grok Voice Think Fast 2.0?

Grok Voice Think Fast 2.0 es el modelo más reciente de SpaceXAI para la Speech to Speech API, el nombre de producto detrás de lo que la mayoría llama simplemente Grok Voice. Si aún piensas en la empresa como xAI, es la misma: se integró en SpaceX y pasó a llamarse SpaceXAI el 6 de julio de 2026. El API no cambió sus identificadores, así que abajo verás xai en todo, desde la variable XAI_API_KEY hasta el host api.x.ai.

Una pila de voz tradicional encadena tres servicios: speech‑to‑text, un modelo de lenguaje y text‑to‑speech, y cada salto añade latencia y puntos donde se pierde contexto. Think Fast 2.0 condensa todo en un solo modelo que recibe audio o texto y devuelve audio o texto por la misma conexión.

WebSocket de voz a voz frente a arquitectura de tres servicios. Imagen del autor.

Para un agente que actúa y no solo habla, lo clave es que el razonamiento y el habla corren en paralelo. SpaceXAI dice que las llamadas a herramientas "normalmente" empiezan a ejecutarse antes de que el agente termine su primera frase, y ese "normalmente" es importante.

En los benchmarks que SpaceXAI cita de Artificial Analysis, Think Fast 2.0 logra un 82,9% en el Speech to Speech Index frente al 75,7% de 1.0, y reduce el tiempo hasta el primer audio de 1,25 s a 0,70 s. Los números del proveedor en un benchmark general son una hipótesis sobre tu flujo de llamadas, no tu plan de pruebas.

Verás tres cadenas de modelo: grok-voice-latest, grok-voice-think-fast-2.0 y grok-voice-think-fast-1.0. El alias viene bien en prototipos y no es lo bastante estable para nada más.

Cuando probé el 4 de agosto de 2026, grok-voice-latest aún resolvía a grok-voice-think-fast-1.0, con las notas de versión de SpaceXAI programando el cambio a Think Fast 2.0 para el día siguiente. Ese cambio es tanto de modelo como de precio: 0,08 $ por minuto de audio frente a 0,05 $ en 1.0, así que un alias sin fijar se encarece sin tocar una línea de tu código. Fija la versión en todo lo que despliegues.

Qué vamos a construir

El agente cubre lo típico de un canal de soporte: consultar un pedido, encontrarlo por email cuando no hay número, cambiar instrucciones de entrega, cancelar, abrir o comprobar un ticket y pasar la llamada a una persona. A lo largo del camino habrá interrupciones y una conexión caída.

Son varios archivos pequeños en lugar de un único script, porque cada pieza tiene un cometido y querrás probarlas por separado. Estructura:

• config.py carga la clave del API y guarda la cadena de modelo, la frecuencia de muestreo y las URLs de los endpoints

• voice_client.py envuelve el WebSocket, controla la facturación y expone utilidades de envío/recepción

• tools.py define las funciones de pedidos y una pequeña “base” en memoria que sustituye a una base de datos real

• assistant.py contiene el prompt del sistema, la configuración de sesión y el bucle de eventos que lo une todo

• token_server.py es un pequeño endpoint FastAPI que emite tokens efímeros

• app_streamlit.py lleva el mismo cliente al navegador en vivo; volveré a él tras las pruebas

El recorrido didáctico va por terminal. La demo añade el micrófono.

Requisitos previos

Necesitas una cuenta de SpaceXAI con clave de API, un método de pago activo (no hay capa gratuita permanente y los créditos promocionales de alta no te cubrirán) y soltura con asyncio y WebSockets para seguir sin una explicación línea a línea de await.

Los ejemplos rápidos de SpaceXAI usan el paquete websockets sin SDK dedicado, y nosotros también. La documentación no fija versión de Python; yo probé con 3.11.

Guarda la clave del API en el servidor. Si un navegador o app móvil habla con la Voice API directamente, usa un token efímero en lugar de tu clave real; lo cubro en la sección de seguridad.

Configurar el proyecto

Todos los archivos están en el repositorio del proyecto, así que puedes clonarlo en lugar de copiar fragmentos:

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 lleva la conexión en tiempo real y python-dotenv lee tu clave. El resto cubre el endpoint de tokens y la demo en navegador. Pon tu clave en .env:

XAI_API_KEY=xai-your-key-here

Eso es casi toda la puesta a punto. La conexión es la parte interesante.

Entender la Realtime API de Grok Voice

Grok Voice es el nombre del producto. Contra lo que realmente programas es un endpoint WebSocket en wss://api.x.ai/v1/realtime, y toda la conversación sucede como un flujo de eventos JSON en ese único socket.

Ciclo de vida de eventos

La conexión sigue un patrón fijo: el servidor envía session.created y conversation.created nada más conectar; tú envías session.update para configurar voz y herramientas; el servidor confirma con session.updated y, desde ahí, creas elementos de conversación y pides respuestas. Probé con una clave real y el orden calca los docs.

• session.update (cliente) configura voz, instrucciones, herramientas y formato de audio

• conversation.item.create (cliente) añade un mensaje de usuario, del asistente o el resultado de una herramienta

• response.create (cliente) pide al modelo que hable; el VAD del servidor lo envía por ti automáticamente

• response.output_audio.delta y response.output_audio_transcript.delta (servidor) transmiten la respuesta mientras se genera

• response.done (servidor) cierra el turno

Dos cosas suelen liar. La página de Speech to Speech que enlacé menciona un evento conversation.item.created durante la reanudación de sesión, pero la referencia de eventos solo lista conversation.item.added y eso es lo que vi en todas mis pruebas, así que programa contra ese. También verás un ping no documentado a los pocos segundos de la mayoría de conexiones; lo menciono para que no lo leas como error.

Formatos y transporte de audio

Códec y transporte son elecciones separadas. El códec, bajo audio.input.format y audio.output.format, puede ser audio/pcm (Linear16, 24000 Hz por defecto), audio/pcmu o audio/pcma (G.711 a 8 kHz, para telefonía), o audio/opus (24 kHz). El transporte es cómo viajan esos bytes por el cable:

• json (por defecto) envía el audio en base64 dentro de input_audio_buffer.append y response.output_audio.delta, fácil de registrar y depurar

• binary envía bytes crudos del códec como frames binarios de WebSocket, evitando la sobrecarga de base64 a costa de un bucle de recepción que debe ramificar por tipo de mensaje

Empieza con JSON. Todos los ejemplos de la documentación lo usan, es trivial de inspeccionar y la sobrecarga de base64 no es lo que limita un agente de soporte. Pásate a binario si mides una razón.

Compatibilidad con la OpenAI Realtime API

Sáltate esto si nunca has tocado la Realtime API de OpenAI. Para el resto, la Speech to Speech API sigue de cerca la OpenAI Realtime API lo bastante como para portar la mayoría del cliente cambiando base URL y clave, pero no es un reemplazo perfecto.

Las transcripciones llegan como conversation.item.input_audio_transcription.updated aquí en lugar del delta de OpenAI; unos cuantos eventos de OpenAI no están soportados, y SpaceXAI añade extensiones propias: force_message para una locución obligatoria, resumption para reconexiones y replace para corregir marcas mal pronunciadas antes del TTS.

Construir el agente de voz en tiempo real

Basta de protocolo. Aquí está el cliente que habla con él.

Conectar y configurar la sesión

La conexión se abre con un bearer token y un parámetro de modelo; el primer mensaje que envías configura cómo se comporta el agente:

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 es el prompt del sistema, y este modelo prefiere prompts cortos. Las notas de migración de SpaceXAI recomiendan simplificar prompts escritos para modelos de voz de la era GPT en vez de portarlos tal cual. El mío pide respuestas breves, una sola pregunta cada vez y leer en voz alta cualquier cambio antes de aplicarlo. Una confirmación hablada mejora la UX, no es un control de seguridad. Tu aplicación sigue aplicando autorización en la escritura real.

Algo que me sorprendió: una cadena de modelo no reconocida no lanza error al conectar; hace fallback silencioso a grok-voice-think-fast-1.0. Degradar una petición de pago por un typo sin avisar es un valor por defecto raro. Registra el campo session.model de session.created al arrancar y comprueba que recibiste lo que pediste.

Salida en terminal mostrando session.created tras conectar. Imagen del autor.

Transmitir audio del usuario

Con turn_detection.type en server_vad, solo necesitas ir añadiendo audio. El servidor decide cuándo la persona ha dejado de hablar y dispara la respuesta. Si lo pones en null, esa decisión es tuya y confirmas el buffer explícitamente cuando crees que el turno acabó.

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

El VAD del servidor tiene tres ajustes, y equivocarlos es la forma más habitual de que un agente parezca roto sin errores en logs. No aparecen por defecto en el eco de session.updated, así que compruébalos en la documentación en lugar de asumir.

• threshold (0.1 a 0.9, por defecto 0.85): cuán alto debe ser el audio para contar como habla; súbelo en entornos ruidosos, bájalo si no detecta a voces suaves

• silence_duration_ms: cuánto tiempo en silencio antes de que el servidor cierre el turno; demasiado corto corta a la gente, demasiado largo se siente perezoso

• prefix_padding_ms (por defecto 333): un recorte de audio justo antes de detectar habla, para no comerse la primera sílaba

Ajusta primero silence_duration_ms si la gente se queda cortada al hacer una pausa. Es el primer control que toco antes que los otros dos.

Recibir y reproducir la respuesta

El audio llega en trozos pequeños como response.output_audio.delta, y la gracia del streaming es reproducir cada trozo al aterrizar, sin esperar a 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)

Conserva la transcripción incluso en producción. Es tu herramienta de depuración más barata cuando alguien dice que el agente "dijo algo raro".

Añadir herramientas al agente

Un agente que solo habla es un chatbot con micrófono.

Crear las herramientas de pedidos

Cada herramienta es un esquema JSON más una función Python del lado servidor. El modelo nunca toca la base de datos: solo ve lo que devuelve nuestra función.

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
]

Las operaciones de lectura como check_order_status son seguras de reintentar si algo expira. Las de escritura no: reintentar update_delivery_instructions tras un timeout ambiguo puede aplicar el mismo cambio dos veces. Una línea de confirmación en el prompt no evita eso; usa una clave de idempotencia o una comprobación de duplicados.

Incluye también las negativas en la función. cancel_order devuelve un motivo y una alternativa en lugar de cancelar un pedido enviado, porque un prompt que dice "nunca canceles pedidos enviados" es un consejo y una función que lo rechaza es vinculante.

Gestionar el bucle de llamadas a herramientas

Cuatro pasos, y el orden importa más de lo que parece. El modelo envía response.function_call_arguments.done, tu código ejecuta la función, devuelves el resultado como un elemento function_call_output y solo entonces pides al modelo que continúe.

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

Si el modelo necesita más de una herramienta, lanzará múltiples eventos function_call_arguments.done antes de que suene ningún audio. Resuelve todas y envía cada resultado antes de un solo response.create. Si lo mandas demasiado pronto, el modelo responde sin el contexto de las llamadas aún en curso.

Hay una trampa que SpaceXAI documenta y aun así me encontré: enviar response.create en cuanto sale tu resultado puede solaparse con la frase de apertura que el agente aún está reproduciendo. En una ejecución empezó con "Voy a comprobar el estado del pedido ORD-1042 ahora mismo" y llamó a la herramienta a mitad de frase, así que una respuesta inmediata habría hablado sobre su propio inicio.

Espera a que termine el audio del turno actual y muestra un breve estado de "pensando" entre medias.

Flujo de llamadas a herramientas antes de continuar la respuesta. Imagen del autor.

Gestionar interrupciones y estado de la conversación

Dos problemas distintos: quien llama interrumpe al agente a mitad de respuesta, y un WebSocket cae y hay que retomarlo.

Admitir interrupciones naturales

Con server_vad activado, el barge‑in es automático en el servidor: cuando detecta que la persona vuelve a hablar, envía input_audio_buffer.speech_started y deja de generar la respuesta anterior. Tu parte es el cliente: vaciar el audio en cola para que el agente se calle en lugar de terminar una frase que nadie pidió oír.

if event["type"] == "input_audio_buffer.speech_started":
    playback_queue.clear()

En sesiones manuales sin VAD, response.cancel hace lo mismo bajo demanda. También existe conversation.item.truncate para recortar un elemento del asistente a lo que realmente se oyó. La documentación confirma que existe pero no cuándo dispararlo en un barge‑in en vivo, así que prueba el timing tú mismo.

Probé esto con un cambio de instrucciones de entrega a mitad de respuesta: inicia la petición, interrumpe con una dirección distinta a mitad de la confirmación. Lo que importa es si el agente aplica la corrección en el registro del pedido, no si el audio se paró. Valida contra el registro, no contra el silencio. La demo en navegador al final te deja escucharlo.

Reanudar una sesión desconectada

La reanudación de sesión es optativa y no es memoria. Pon resumption.enabled: true en session.update, toma el ID del evento conversation.created y, si el socket cae, reconecta con ?conversation_id=<id> en la URL y vuelve a activar la opción en la nueva conexión.

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

Los turnos en caché, transcripciones, llamadas a herramientas y resultados se reproducen antes de tu siguiente pregunta, y la caché desaparece tras 30 minutos de inactividad. Lo probé preguntando por un pedido, cortando la conexión y reconectando para una repregunta sin repetirme; el agente retomó bien la ETA.

Un detalle no documentado: la reproducción no llega al instante, así que una pregunta enviada justo al abrir el socket puede adelantarse y volver sin memoria del turno anterior. Dale un segundo antes de culpar a la reanudación.

Registro en terminal de una sesión reanudada. Imagen del autor.

No uses esto en lugar de guardar el estado del pedido en tu propia base. Si la caché expira o te llaman mañana, empiezas de cero, y es así a propósito.

Asegurar y monitorizar el agente

Nunca pongas una clave permanente de API en código de navegador o móvil. Si un cliente conecta directo sin pasar por tu servidor, emite un token de corta duración:

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": ...}

Un navegador no puede fijar un header Authorization personalizado en el handshake del WebSocket, así que pasa el token por el header sec-websocket-protocol, con prefijo xai-client-secret..

El servidor emite el token y el navegador se une a la llamada. Imagen del autor.

La facturación va por dos contadores. El audio, enviado o recibido, se factura a 0,08 $ por minuto como dije (4,80 $ la hora), y cada conversation.item.create que no sea audio ni function_call_output cuesta 0,004 $ fijos. response.create no se factura. Cada response.done incluye un objeto usage que en mis pruebas reportó output_audio_seconds junto con billable_audio_seconds. Factura desde ahí, no desde estimaciones.

Los límites documentados de la Speech to Speech API son 10 sesiones concurrentes por equipo y 120 minutos por sesión, ambos en us-east-1. No planifiques capacidad desde los números de la Voice Agent API, que son distintos.

En privacidad, sé preciso. El FAQ de seguridad de SpaceXAI dice que las peticiones y respuestas del API se conservan cifradas 30 días para monitorizar abusos y no se usan para entrenamiento sin permiso, y que los equipos pueden activar Zero Data Retention, aunque ZDR elimina el historial persistido de conversaciones y por tanto no funciona con reanudación.

Si vas a avisar de que la llamada se graba o está gestionada por IA, para eso sirve la extensión force_message que cité: la frase se reproduce tal cual, no como el modelo la parafrasee.

Probar el agente de voz

Un 200 en el handshake del WebSocket no te dice si el agente hizo lo correcto. Prueba el resultado, no solo la conexión.

• Una consulta limpia de pedido, contrastando la respuesta hablada con el registro, no solo que llegó algo
• Una respuesta interrumpida, confirmando que la reproducción se detiene y el agente atiende la nueva petición
• Una actualización de entrega que requiere confirmación, verificada en el registro del pedido
• Una negativa, como cancelar un pedido ya enviado, donde el agente explica la norma en vez de disculparse
• Un número de pedido desconocido, asegurando que el agente lo reconoce y no se inventa un estado
• Una herramienta que devuelve error, comprobando que el agente lo verbaliza y no se queda colgado
• Reconexión y reanudación, incluyendo esa ventana de reproducción que comenté
• Audio ruidoso, habla rápida y alguien que deletrea números y direcciones

Probé la mayoría con una clave real mientras escribía. Los fallos interesantes fueron de comportamiento, no de errores: el timing de reanudación de arriba y un umbral VAD fuera de rango que aceptó en vez de rechazar, lo típico que se cuela si solo pruebas el camino feliz. Añade una prueba multilingüe también, y mira las FAQs por un matiz en cómo indicar el idioma.

Dos de esas no puedes probarlas tecleando. app_streamlit.py es una página de Streamlit que pone una llamada en vivo en el navegador: el micrófono transmite al mismo WebSocket sobre WebRTC, la voz del agente vuelve en streaming y el socket permanece abierto.

streamlit run app_streamlit.py