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

Grok Voice Think Fast 2.0 de SpaceXAI es un modelo de voz a voz. Le envías audio por un WebSocket y te devuelve audio; entre medias puede razonar y seguir hablando mientras ya se está ejecutando una llamada a función que decidió hacer. Sin paso separado 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 al principio del turno. Resumo los benchmarks, porque en un tutorial lo que importa es qué cambia en tu código.

Vamos a crear un agente de voz de atención al cliente para una tienda online. Quien llama puede preguntar por un pedido, cambiar instrucciones de entrega, cancelar, interrumpir al agente a media frase y retomar la conversación tras una caída. Este es el camino vía API, no el creador sin código que explicamos en 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 de lo que la mayoría llama 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. La API no cambió sus identificadores con el rebranding, así que 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 riesgo de perder contexto. Think Fast 2.0 lo reduce a 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 en tres servicios. Imagen del autor.

Para un agente que actúa y no solo habla, lo clave es que razonamiento y síntesis de voz vayan en paralelo. SpaceXAI afirma que las llamadas a herramientas "normalmente" empiezan a ejecutarse antes de que el agente termine su primera frase, y ese normalmente importa.

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. Las cifras del proveedor en un benchmark general son una hipótesis sobre tu flujo de llamadas, no un plan de pruebas.

Verás tres strings de modelo: grok-voice-latest, grok-voice-think-fast-2.0 y grok-voice-think-fast-1.0. El alias es cómodo al prototipar y demasiado inestable para lo demás.

Cuando probé el 4 de agosto de 2026, grok-voice-latest seguía resolviendo 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 de precio tanto como de modelo: 0,08 $ por minuto de audio frente a 0,05 $ en 1.0; con un alias sin fijar, te sube el coste sin tocar una línea de 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 si no hay número, cambiar instrucciones de entrega, cancelar, abrir o comprobar un ticket y pasar la llamada a una persona. Por el camino habrá interrupciones y una conexión caída.

Son varios archivos pequeños, no un único script, porque cada pieza tiene un cometido y querrás probarlas por separado. Esta es la estructura:

• config.py carga la clave de API y guarda el string 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 tienda 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 en FastAPI que emite tokens efímeros

• app_streamlit.py pone el mismo cliente detrás de una llamada en el navegador; vuelvo a ello tras la parte de pruebas

La parte didáctica va desde terminal. La demo añade el micrófono.

Requisitos previos

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

Los ejemplos de inicio rápido de SpaceXAI usan el paquete websockets en crudo en lugar de un SDK dedicado, y nosotros también. La documentación no fija versión mínima de Python. Probé con 3.11.

Guarda la clave de 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 cubrimos en la sección de seguridad.

Configuración del 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 preparación. La conexión es lo interesante.

Entender la Realtime API de Grok Voice

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

El ciclo de vida de los eventos

La conexión sigue una forma fija: el servidor envía session.created y conversation.created en cuanto conectas; tú envías session.update para configurar voz y herramientas; el servidor confirma con session.updated, y desde ahí creas items de conversación y pides respuestas. Lo probé con una clave real y el orden calca la documentación.

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

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

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

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

• response.done (servidor) cierra el turno

Hay dos cosas que confunden. 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 canónica de eventos solo lista conversation.item.added, y es lo que recibí en todas mis pruebas, así que prográmalo así. También verás un ping no documentado a los pocos segundos de la mayoría de conexiones; lo menciono para que no lo tomes por un error.

Formatos de audio y transporte

Códec y transporte son decisiones separadas. El códec, configurado en audio.input.format y audio.output.format, puede ser audio/pcm (Linear16, por defecto 24000 Hz), 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 audio como texto 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 base64 a costa de ramificar el bucle de recepción según el tipo de mensaje

Empieza con JSON. Todos los ejemplos en la documentación lo usan, es trivial de inspeccionar y la sobrecarga base64 no es lo que limita un agente de soporte. Pasa a binario si mides un motivo.

Compatibilidad con la OpenAI Realtime API

Sáltate esto si nunca has usado la Realtime API de OpenAI. Para el resto, la Speech to Speech API sigue de cerca la Realtime API de OpenAI lo suficiente 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; algunos eventos de OpenAI no están soportados, y SpaceXAI añade extensiones propias: force_message para un aviso legal fijo, resumption para reconexiones y replace para corregir marcas mal pronunciadas antes del text‑to‑speech.

Crear el agente de voz en tiempo real

Basta de protocolo. Aquí está el cliente que habla con la API.

Conexión y configuración de la sesión

La conexión se abre con un token bearer y un parámetro de modelo en la query, y 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 lugar de portarlos tal cual. El mío indica respuestas breves, una sola pregunta cada vez y leer en voz alta cualquier cambio antes de ejecutarlo. La 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ó: un string de modelo no reconocido 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 extraño. Registra el campo session.model de session.created al arrancar y comprueba que recibiste lo que pediste.

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

Transmitir el audio del usuario

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

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 configurarlos mal es la forma más común de que un agente suene roto aunque no haya errores en logs. Ninguno aparece por defecto en el eco de session.updated, así que revísalos en la documentación, no los supongas.

• threshold (0.1 a 0.9, por defecto 0.85): qué volumen cuenta como voz; súbelo en entornos ruidosos, bájalo si se pierden voces bajas

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

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

Ajusta primero silence_duration_ms si cortas a quien llama cuando hace pausas. Es mi primer recurso antes de tocar los otros dos.

Recibir y reproducir la respuesta

El audio llega en trozos como response.output_audio.delta, y la gracia del streaming es reproducir cada pieza según llega, en lugar de 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 la forma más barata de depurar cuando alguien dice que el agente "dijo algo raro".

Añadir herramientas al agente de voz

Un agente de voz 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 de nuestro lado. El modelo nunca toca la base de datos: solo ve lo que nuestra función devuelve.

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 hay timeout. 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 lo evita: añade 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 una sugerencia y una función que se niega 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 item 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, dispara varios eventos function_call_arguments.done antes de reproducir audio. Resuelve todos y envía cada resultado antes de un solo response.create. Si lo envías demasiado pronto, el modelo responde sin el contexto de las llamadas aún en vuelo.

Aquí hay una trampa que SpaceXAI documenta y aun así me pilló: enviar response.create justo cuando sale tu resultado puede solaparse con la frase introductoria que el agente sigue reproduciendo. En una ejecución abrió con "Voy a comprobar el estado del pedido ORD-1042 ahora mismo" y llamó a la herramienta a media frase, así que una respuesta inmediata habría pisado 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 el estado de la conversación

Aquí hay dos problemas distintos: quien llama interrumpe al agente a mitad de respuesta, y un WebSocket que cae y hay que retomar.

Permitir interrupciones naturales

Con server_vad activado, el barge‑in es automático en el servidor: en cuanto detecta que la persona vuelve a hablar, envía input_audio_buffer.speech_started y deja de generar la respuesta anterior. Tu parte es la mitad cliente de ese apretón de manos: vaciar lo que ya esté encolado para que el agente se calle en lugar de terminar una frase que nadie pidió.

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 item de asistente a lo que realmente se oyó. La documentación confirma que existe pero no cuándo dispararlo durante un barge‑in en vivo, así que prueba el timing tú mismo.

Probé esto con un cambio de instrucciones de entrega a media respuesta: inicias la solicitud, interrumpes con una dirección distinta a mitad de la confirmación del agente. Lo que importa es si el agente aplica la instrucción corregida en lugar de terminar en silencio la anterior, no si el audio se detuvo. Comprueba el registro del pedido, no el silencio. La demo del navegador al final te deja oírlo.

Reanudar una sesión desconectada

La reanudación de sesión es opcional y no es memoria. Pon resumption.enabled: true en session.update, guarda 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, transcripciones, llamadas a herramientas y resultados en caché se reproducen antes de tu siguiente pregunta, y la caché caduca tras 30 minutos inactiva. Lo probé preguntando por un pedido, dejando caer la conexión y reconectando para repreguntar sin repetir; el agente recuperó bien la ETA.

Un detalle no documentado: la reproducción no llega al instante, así que una pregunta lanzada 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 base de datos. Si la caché expira o la persona vuelve a llamar mañana, empiezas sin contexto, y es así a propósito.

Seguridad y monitorización del agente

Nunca pongas una clave de API permanente en código de navegador o móvil. Si un cliente conecta directamente en lugar de 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 establecer un header Authorization personalizado en el handshake de WebSocket, así que pasa el token por el header sec-websocket-protocol, con el prefijo xai-client-secret.

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

La facturación corre por dos contadores. El audio, enviado o recibido, va a 0,08 $ por minuto como dije antes (4,80 $/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 informaba de output_audio_seconds y billable_audio_seconds por separado. Factura con esos valores, no con 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 planees capacidad con los números de la Voice Agent API, que son diferentes.

En privacidad, sé preciso. El FAQ de seguridad de SpaceXAI dice que las peticiones y respuestas de la 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 persistente de conversaciones y por tanto no funciona con la reanudación.

Si vas a informar de que la llamada se graba o la gestiona IA, para eso está la extensión force_message que mencioné: la línea se reproduce tal cual en lugar de lo que 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 haya respuesta
• Una respuesta interrumpida, confirmando que se para la reproducción y el agente atiende la nueva petición
• Una actualización de entrega que exige confirmación, verificada contra el registro del pedido
• Una negativa, como cancelar un pedido ya enviado, donde el agente explica la norma en lugar de disculparse
• Un número de pedido desconocido, asegurando que el agente lo indica y no inventa un estado
• Una herramienta que devuelve error, comprobando que el agente lo dice en voz alta en lugar de quedarse colgado
• Reconexión y reanudación, incluido ese margen de reproducción que me encontré
• Audio ruidoso, habla rápida y alguien que deletrea números y direcciones

Hice 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 VAD fuera de rango aceptado en lugar de rechazado; es el tipo de cosa que sale rota en silencio si solo pruebas el camino feliz. Añade también una prueba multilingüe 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 en Streamlit que pone una llamada en vivo en el navegador: el micro transmite al mismo WebSocket por WebRTC, la voz del agente vuelve en streaming y el socket permanece abierto.

streamlit run app_streamlit.py