Tutorial da API Grok Voice Think Fast 2.0: crie um agente de voz em tempo real em Python

O Grok Voice Think Fast 2.0 da SpaceXAI é um modelo de fala para fala. Você envia áudio por um WebSocket e ele devolve áudio; no meio do caminho, consegue raciocinar e continuar falando enquanto uma chamada de função que ele decidiu fazer já está em execução. Sem etapa separada de speech-to-text, sem etapa separada de text-to-speech.

A SpaceXAI anunciou o Think Fast 2.0 em 29 de julho de 2026: primeiro áudio mais rápido, comportamento full-duplex mais estável (ouve enquanto fala em vez de turnos rígidos) e chamadas de ferramenta que disparam no início do turno. Vou manter os benchmarks curtos, porque o que importa num tutorial é o que muda no seu código.

Vamos criar um agente de voz para suporte ao cliente de uma loja online. Quem liga pode perguntar sobre um pedido, mudar instruções de entrega, cancelar, interromper o agente no meio da fala e retomar a conversa depois de uma queda de conexão. Este é o caminho via API, não o construtor no-code Voice Agent Builder que nosso tutorial do Grok Voice Agent Builder mostra. Comece por lá para a versão focada no console.

O que é o Grok Voice Think Fast 2.0?

O Grok Voice Think Fast 2.0 é o modelo mais novo da SpaceXAI para a Speech to Speech API, o nome do produto por trás do que a maioria só chama de Grok Voice. Se você ainda pensa na empresa como xAI, é a mesma: foi incorporada à SpaceX e renomeada para SpaceXAI em 6 de julho de 2026. A API não acompanhou o rebrand, então todo identificador abaixo ainda diz xai, da variável XAI_API_KEY ao host api.x.ai.

Uma stack de voz tradicional encadeia três serviços: speech-to-text, um modelo de linguagem e depois text-to-speech, e cada salto adiciona latência e um ponto para o contexto se perder. O Think Fast 2.0 comprime tudo em um único modelo que recebe áudio ou texto e produz áudio ou texto na mesma conexão.

WebSocket de fala para fala versus arquitetura em pipeline de três serviços. Imagem do autor.

Para um agente que age, e não só fala, o que importa é raciocínio e fala rodando em paralelo. A SpaceXAI diz que as chamadas de ferramenta "geralmente" começam a executar antes de o agente terminar a primeira frase — e esse “geralmente” faz diferença.

Nos benchmarks que a SpaceXAI cita do Artificial Analysis, o Think Fast 2.0 marca 82,9% no Speech to Speech Index contra 75,7% do 1.0, e reduz o tempo até o primeiro áudio de 1,25 s para 0,70 s. Números do fornecedor em um benchmark geral são uma hipótese sobre o seu fluxo de chamadas, não um plano de testes.

Há três strings de modelo que você vai ver: grok-voice-latest, grok-voice-think-fast-2.0 e grok-voice-think-fast-1.0. O alias é conveniente no protótipo e instável demais para qualquer outra coisa.

Quando testei em 4 de agosto de 2026, grok-voice-latest ainda apontava para grok-voice-think-fast-1.0, com as notas de versão da SpaceXAI prevendo a mudança para o Think Fast 2.0 no dia seguinte. Essa troca é uma mudança de preço tanto quanto de modelo, US$ 0,08 por minuto de áudio contra US$ 0,05 do 1.0, então um alias sem versão fica mais caro sem uma linha do seu código mudar. Fixe a versão em tudo que você colocar em produção.

O que vamos construir

O agente cobre o que uma linha de suporte recebe: consultar um pedido, encontrar por e-mail quando quem liga não tem número, alterar instruções de entrega, cancelar, abrir ou checar um ticket e transferir a chamada para uma pessoa. No caminho, aparecem interrupções e queda de conexão.

São alguns arquivos pequenos em vez de um script só, porque cada parte tem um papel diferente e você vai querer testá-las separadamente. Estrutura:

• config.py carrega a chave da API e guarda a string do modelo, taxa de amostragem e URLs de endpoint

• voice_client.py encapsula o WebSocket, acompanha a cobrança e expõe helpers de envio/recebimento

• tools.py define as funções de pedido e um pequeno repositório de pedidos em memória no lugar de um banco real

• assistant.py guarda o prompt do sistema, a configuração da sessão e o event loop que integra tudo

• token_server.py é um pequeno endpoint FastAPI que gera tokens efêmeros

• app_streamlit.py coloca o mesmo cliente por trás de uma chamada ao vivo no navegador; volto nele depois da seção de testes

O passo a passo didático roda no terminal. O demo adiciona o microfone.

Pré-requisitos

Você precisa de uma conta SpaceXAI com uma chave de API, um arranjo de cobrança financiado (não há camada gratuita permanente, e créditos promocionais de contas novas não vão te levar longe), e conforto suficiente com asyncio e WebSockets para acompanhar sem uma explicação linha a linha de await.

Os exemplos de quick start da SpaceXAI usam o pacote websockets puro em vez de um SDK dedicado, e nós também. A documentação não declara versão mínima de Python. Testei no 3.11.

Mantenha a chave da API no servidor. Se um app web ou mobile falar direto com a Voice API, ele recebe um token efêmero em vez da sua chave real, coberto na seção de segurança abaixo.

Configurando o projeto

Cada arquivo abaixo está no repositório do projeto, então você pode clonar em vez de copiar trechos:

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 carrega a conexão em tempo real e python-dotenv lê sua chave. O resto cobre o endpoint de token e o demo no navegador. Coloque sua chave no .env:

XAI_API_KEY=xai-your-key-here

Isso é praticamente toda a configuração. A conexão é a parte interessante.

Entendendo a Grok Voice Realtime API

Grok Voice é o nome do produto. O que você realmente acessa é um endpoint WebSocket em wss://api.x.ai/v1/realtime, e toda a conversa acontece como um fluxo de eventos JSON por esse único socket.

O ciclo de vida dos eventos

Uma conexão segue um formato fixo: o servidor envia session.created e conversation.created assim que você conecta, você envia session.update para configurar voz e ferramentas, o servidor confirma com session.updated e, a partir daí, você cria itens de conversa e solicita respostas. Testei com uma chave real e a ordem bateu com a documentação.

• session.update (cliente) configura voz, instruções, ferramentas e formato de áudio

• conversation.item.create (cliente) adiciona uma mensagem do usuário, do assistente ou um resultado de ferramenta

• response.create (cliente) pede para o modelo falar; o VAD do servidor envia isso automaticamente para você

• response.output_audio.delta e response.output_audio_transcript.delta (servidor) transmitem a resposta conforme é gerada

• response.done (servidor) encerra o turno

Duas coisas atrapalham. A página de docs de Speech to Speech que linkei menciona um evento conversation.item.created durante a retomada de sessão, mas a referência canônica de eventos só lista conversation.item.added, e foi o que recebi em todos os testes, então codifique para ele. Você também verá um ping não documentado alguns segundos após a maioria das conexões — menciono apenas para você não ler como erro.

Formatos e transporte de áudio

Codec e transporte são escolhas separadas. O codec, definido em audio.input.format e audio.output.format, pode ser audio/pcm (Linear16, padrão 24000 Hz), audio/pcmu ou audio/pcma (G.711 a 8 kHz, para telefonia), ou audio/opus (24 kHz). Transporte é como esses bytes viajam no fio:

• json (padrão) envia áudio como texto base64 dentro de input_audio_buffer.append e response.output_audio.delta, fácil de logar e depurar

• binary envia bytes brutos do codec como frames binários do WebSocket, pulando a sobrecarga do base64 ao custo de um loop de recebimento que precisa ramificar por tipo de mensagem

Comece com JSON. Todos os exemplos na documentação usam isso, é trivial inspecionar e a sobrecarga do base64 não é o gargalo num agente de suporte. Migre para binário se você medir motivo para tal.

Compatibilidade com a OpenAI Realtime API

Pode pular se você nunca usou a Realtime API da OpenAI. Para o restante, a Speech to Speech API acompanha a OpenAI Realtime API de perto o suficiente para que a maior parte do código cliente seja portada trocando base URL e chave, mas não é um drop-in perfeito.

Transcrições chegam como conversation.item.input_audio_transcription.updated aqui em vez do delta da OpenAI; alguns eventos da OpenAI não têm suporte, e a SpaceXAI adiciona extensões próprias: force_message para uma fala obrigatória de divulgação, resumption para reconexões, e replace para corrigir pronúncia de marcas antes do text-to-speech.

Construindo o agente de voz em tempo real

Chega de protocolo. Aqui está o cliente que fala com ele.

Conectando e configurando a sessão

A conexão abre com um bearer token e um parâmetro de consulta do modelo, e a primeira mensagem que você envia configura tudo sobre como o agente se comporta:

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 é o prompt do sistema, e este modelo prefere prompts curtos. As notas de migração da SpaceXAI sugerem simplificar prompts escritos para modelos de voz da era GPT em vez de portar ao pé da letra. O meu pede para o agente manter respostas curtas, fazer uma pergunta por vez e ler qualquer gravação de escrita antes de agir. Uma confirmação falada é um agrado de UX, não um controle de segurança. Sua aplicação ainda precisa impor autorização na escrita em si.

Uma coisa me surpreendeu: uma string de modelo não reconhecida não gera erro na conexão; ela recai silenciosamente para grok-voice-think-fast-1.0. Fazer downgrade de uma requisição paga por causa de um typo, sem avisar, é um padrão estranho. Faça log do campo session.model do session.created uma vez na inicialização e confira se você recebeu o que pediu.

Saída do terminal mostrando session.created após conectar. Imagem do autor.

Transmitindo o áudio do usuário

Com turn_detection.type definido como server_vad, você só precisa continuar anexando áudio. O servidor decide quando a pessoa parou de falar e dispara a resposta automaticamente. Defina como null e essa decisão é sua, finalizando o buffer explicitamente quando achar que o turno acabou.

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

O VAD do servidor tem três controles, e errá-los é a forma mais comum de um agente de voz parecer quebrado sem nada dar erro nos logs. Nenhum deles aparece no eco do session.updated por padrão, então confira na documentação em vez de assumir.

• threshold (0,1 a 0,9, padrão 0,85): quão alto o áudio precisa ser para contar como fala; aumente em ambientes barulhentos, reduza se falantes mais baixos não forem detectados

• silence_duration_ms: quanto tempo de silêncio antes de o servidor encerrar o turno; curto demais corta pessoas no meio do raciocínio, longo demais deixa lento

• prefix_padding_ms (padrão 333): trecho de áudio mantido de pouco antes da detecção de fala, para não cortar a primeira sílaba

Ajuste silence_duration_ms primeiro se quem liga estiver sendo cortado ao pausar para pensar. É o primeiro que eu mexo antes dos outros dois.

Recebendo e reproduzindo a resposta

O áudio chega em pequenos pedaços como response.output_audio.delta, e a lógica de streaming é tocar cada pedaço assim que chegar, em vez de esperar o 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)

Guarde a transcrição mesmo em produção. É a ferramenta de debug mais barata quando alguém diz que o agente "falou algo estranho".

Adicionando ferramentas ao agente de voz

Um agente de voz que só fala é um chatbot com microfone.

Criando as ferramentas de pedidos

Cada ferramenta é um schema JSON mais uma função Python simples do nosso lado. O modelo nunca toca o banco; ele só vê o que nossa função retorna.

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
]

Operações de leitura como check_order_status são seguras para repetir se algo der timeout. Escritas não são: repetir update_delivery_instructions depois de um timeout ambíguo pode aplicar a mesma mudança duas vezes. Uma linha de confirmação no prompt não evita isso, então dê às escritas uma chave de idempotência ou uma checagem de duplicidade.

Coloque as recusas na função também. cancel_order devolve um motivo e uma alternativa em vez de cancelar um pedido já enviado, porque um prompt dizendo "nunca cancele pedidos enviados" é sugestão, e uma função que recusa não é.

Tratando o loop de chamadas de ferramenta

Quatro passos, e a ordem importa mais do que parece. O modelo envia response.function_call_arguments.done, seu código executa a função, você manda o resultado de volta como um item function_call_output e só então pede para o modelo continuar.

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

Se o modelo precisar de mais de uma ferramenta para um pedido, ele dispara vários eventos function_call_arguments.done antes de qualquer áudio. Resolva todos e envie todos os resultados antes de um único response.create. Enviar cedo demais faz o modelo responder sem o contexto das chamadas ainda em andamento.

Tem uma pegadinha que a SpaceXAI documenta e eu ainda bati na primeira passada: enviar response.create no instante em que o resultado da ferramenta sai pode sobrepor a frase introdutória que o agente ainda está falando. Em uma execução, ele abriu com "Vou verificar o status do pedido ORD-1042 agora mesmo" e chamou a ferramenta no meio da frase; uma resposta imediata teria falado por cima da abertura.

Espere o áudio do turno atual terminar e mostre um curto estado de “pensando” nesse intervalo.

Fluxo de chamada de ferramenta antes de continuar a resposta. Imagem do autor.

Gerenciando interrupções e estado da conversa

Dois problemas distintos aqui. Quem liga fala por cima do agente no meio da resposta, e um WebSocket cai e precisa ser retomado.

Suportando interrupções naturais

Com server_vad ligado, o barge-in é automático no servidor: no momento em que detecta quem liga falando de novo, ele sinaliza input_audio_buffer.speech_started e para de gerar a resposta antiga. Seu papel é a metade cliente desse aperto de mãos, limpando o que já está na fila de áudio para o agente silenciar em vez de terminar uma frase que ninguém quer ouvir.

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

Para sessões manuais, sem VAD, response.cancel faz o mesmo sob demanda. Há também conversation.item.truncate para reduzir um item do assistente ao que foi realmente ouvido. A documentação confirma que existe, mas não quando dispará-lo durante um barge-in ao vivo, então teste o timing você mesmo.

Testei isso com uma alteração de instrução de entrega no meio da resposta: começar o pedido, interromper com um endereço diferente enquanto o agente confirma. O que importa é se o agente aplica a instrução corrigida em vez de terminar a antiga em silêncio, não se o áudio parou. Faça assert no registro do pedido, não no silêncio. O demo no navegador ao fim permite ouvir isso.

Retomando uma sessão desconectada

Retomada de sessão é opt-in e não é memória. Defina resumption.enabled: true no session.update, pegue o ID do evento conversation.created e, se o socket cair, reconecte com ?conversation_id=<id> na URL e opte de novo na conexão nova.

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

Os turnos em cache, transcrições, chamadas de ferramenta e resultados são reproduzidos antes da sua próxima pergunta, e o cache expira após 30 minutos de inatividade. Testei perguntando sobre um pedido, derrubando a conexão e reconectando para um follow-up sem me repetir; o agente retomou corretamente o ETA.

Um detalhe não documentado: a reprodução não cai instantaneamente, então uma pergunta enviada no exato momento em que o socket abre pode “ganhar” dela e voltar sem memória do turno anterior. Dê um segundo antes de culpar a retomada.

Log de terminal de uma sessão retomada. Imagem do autor.

Não use isso no lugar de salvar o estado do pedido no seu próprio banco. Se o cache expirar ou a pessoa ligar amanhã, você começa do zero — por design.

Protegendo e monitorando o agente

Nunca coloque uma chave de API permanente em código de navegador ou mobile. Se um cliente conecta direto em vez de passar pelo seu servidor, gere um token de curta duração:

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

Um navegador não consegue definir um header Authorization personalizado no handshake do WebSocket, então ele passa o token pelo header sec-websocket-protocol, prefixado com xai-client-secret..

Servidor emite token, navegador entra na chamada. Imagem do autor.

A cobrança tem dois medidores. Áudio, enviado ou recebido, corre à taxa de US$ 0,08 por minuto que mencionei, o que dá US$ 4,80 por hora, e cada conversation.item.create que não é áudio e não é function_call_output custa US$ 0,004 fixos. response.create não é faturado. Cada response.done traz um objeto usage que, no meu teste, reportou output_audio_seconds junto com billable_audio_seconds. Fature por eles, não por estimativas.

Limites documentados na Speech to Speech API são 10 sessões simultâneas por time e um teto de 120 minutos por sessão, ambos em us-east-1. Não planeje capacidade pelos números da Voice Agent API, que são diferentes.

Sobre privacidade, seja preciso. O FAQ de segurança da SpaceXAI diz que requisições e respostas de API são retidas criptografadas por 30 dias para monitoramento de abuso e não são usadas para treinamento sem permissão, e que times podem ativar Zero Data Retention; porém, ZDR elimina o histórico persistido de conversas do agente de voz e, portanto, não funciona com retomada.

Se você vai informar que a chamada é gravada ou tratada por IA, é para isso que serve a extensão force_message que citei. A fala toca exatamente como escrita, em vez do que o modelo parafrasearia.

Testando o agente de voz

Um status 200 no handshake do WebSocket não diz nada sobre se o agente fez a coisa certa. Teste o resultado, não só a conexão.

• Uma consulta limpa de pedido, conferindo a resposta falada com o registro, não só se houve resposta
• Uma resposta interrompida, confirmando que a reprodução para e o agente atende ao novo pedido
• Uma atualização de entrega que exige confirmação, conferida no registro do pedido
• Uma recusa, como cancelar um pedido já enviado, em que o agente explica a regra em vez de se desculpar
• Um número de pedido desconhecido, garantindo que o agente diga isso em vez de inventar um status
• Uma ferramenta que retorna erro, checando se o agente fala o erro em vez de travar
• Reconexão e retomada, incluindo aquela janela de replay que eu encontrei
• Áudio ruidoso, fala rápida e alguém soletrando números e endereços

Executei a maioria desses com uma chave real enquanto escrevia. As falhas interessantes foram comportamentais, não erros: o timing da retomada acima, e um threshold de VAD fora do intervalo aceito em vez de rejeitado — o tipo de coisa que vai a produção silenciosamente quebrada se você só testa o caminho feliz. Adicione também um teste multilíngue, e veja as FAQs por um detalhe sobre nomear o idioma.

Dois desses você não consegue testar digitando. app_streamlit.py é uma página Streamlit que coloca uma chamada ao vivo no navegador: o microfone transmite para o mesmo WebSocket via WebRTC, a voz do agente volta em streaming e o socket fica aberto o tempo todo.

streamlit run app_streamlit.py