Tutorial da API Grok Voice Think Fast 2.0: construa 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, ele consegue raciocinar e continuar falando enquanto uma chamada de função que ele decidiu executar já está rodando. 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 ferramentas que disparam ainda no começo do turno. Vou ser breve nos benchmarks, porque o que importa num tutorial é o que muda no seu código.

Vamos construir um agente de voz para suporte ao cliente de uma loja online. Quem liga pode perguntar sobre um pedido, alterar instruções de entrega, cancelar, interromper o agente no meio da frase 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 do nosso tutorial Grok Voice Agent Builder. Comece por lá para a versão via console.

O que é o Grok Voice Think Fast 2.0?

Grok Voice Think Fast 2.0 é o modelo mais novo da SpaceXAI para a Speech to Speech API, o nome de produto por trás do que a maioria chama simplesmente de Grok Voice. Se você ainda pensa na empresa como xAI, é a mesma: ela 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 usa xai, da variável XAI_API_KEY ao host api.x.ai.

Uma pilha de voz tradicional encadeia três serviços: speech‑to‑text, um modelo de linguagem e depois text‑to‑speech; cada salto adiciona latência e pontos de perda de contexto. 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 três serviços. Imagem do autor.

Para um agente que age, e não só fala, o que importa é que raciocínio e fala rodam em paralelo. A SpaceXAI diz que as chamadas de ferramentas "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 da Artificial Analysis, o Think Fast 2.0 marca 82,9% no Speech to Speech Index contra 75,7% da versão 1.0, e corta o tempo até o primeiro áudio de 1,25 s para 0,70 s. Números de fornecedor em um benchmark geral são uma hipótese sobre seu fluxo de chamadas, não um plano de testes.

Você vai ver três strings de modelo: grok-voice-latest, grok-voice-think-fast-2.0 e grok-voice-think-fast-1.0. O alias é útil no protótipo e instável 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 marcando a mudança para o Think Fast 2.0 para o dia seguinte. Essa troca é tanto uma mudança de modelo quanto de preço: US$ 0,08 por minuto de áudio contra US$ 0,05 da 1.0 — logo, um alias não fixado fica mais caro sem você alterar uma linha de código. Fixe a versão nas implantações.

O que vamos construir

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

São alguns arquivos pequenos, não um único script, porque cada parte tem um papel 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 em memória no lugar de um banco real

• assistant.py guarda o prompt de sistema, a configuração de sessão e o event loop que amarra tudo

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

• app_streamlit.py coloca o mesmo cliente por trás de uma chamada ao vivo no navegador, que retomarei após a seção de testes

O passo a passo roda pelo terminal. A demo adiciona o microfone.

Pré-requisitos

Você precisa de uma conta SpaceXAI com uma chave de API, cobrança ativa (não há camada gratuita permanente, e créditos promocionais de novas contas não vão sustentar seu uso) e familiaridade suficiente com asyncio e WebSockets para acompanhar sem explicação linha a linha de await.

Os exemplos de início rápido da SpaceXAI usam o pacote websockets cru, não um SDK dedicado — e nós também. A documentação não fixa uma versão mínima de Python. Testei no 3.11.

Mantenha a chave de 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, como explico na seção de segurança abaixo.

Configurando o projeto

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

XAI_API_KEY=xai-your-key-here

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

Entendendo a Realtime API do Grok Voice

Grok Voice é o nome do produto. O que você de fato usa no código é um endpoint WebSocket em wss://api.x.ai/v1/realtime, e a conversa inteira acontece como um fluxo de eventos JSON nesse único socket.

O ciclo de vida dos eventos

A 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 daí você cria itens da conversa e pede respostas. Testei com uma chave real e a ordem bateu exatamente com a doc.

• 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 o resultado de uma 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 ela é gerada

• response.done (servidor) encerra o turno

Duas coisas atrapalham. A página da Speech to Speech que linkei acima menciona um evento conversation.item.created durante retomada de sessão, mas a referência canônica só lista conversation.item.added, e foi o que chegou em todos os meus testes — então faça o código em cima dele. Você também verá um evento não documentado ping alguns segundos após a conexão, citado aqui só para você não achar que é 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). O transporte é como esses bytes viajam na rede:

• 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 crus do codec como frames binários do WebSocket, evitando a sobrecarga do base64 ao custo de um loop de recebimento que precisa tratar tipos de mensagem

Comece com JSON. Cada exemplo na doc usa isso, é trivial de inspecionar e a sobrecarga do base64 não é o gargalo de um agente de suporte. Passe para binário se você medir que vale a pena.

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 a maior parte do cliente portar trocando a base URL e a 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 são suportados, e a SpaceXAI adiciona extensões próprias: force_message para uma linha de divulgação roteirizada, 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 query 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 de sistema, e este modelo prefere prompts curtos. As notas de migração da SpaceXAI recomendam simplificar prompts escritos para modelos de voz da era GPT, não portar ao pé da letra. O meu diz ao agente para manter respostas curtas, fazer uma pergunta por vez e ler qualquer escrita de volta antes de agir. Confirmação falada é um agrado de UX, não um controle de segurança. Sua aplicação ainda faz a autorização na própria escrita.

Uma surpresa: uma string de modelo desconhecida não gera erro ao conectar; ela regride silenciosamente para grok-voice-think-fast-1.0. Rebaixar uma requisição paga por causa de um typo, sem avisar, é um padrão estranho. Faça log do campo session.model de 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 quem liga parou de falar e dispara a resposta para você. Defina como null e essa decisão é sua, finalizando o buffer explicitamente quando você julgar 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 ajustes, e errá-los é a forma mais comum de um agente de voz parecer quebrado mesmo sem erros no log. Nenhum deles aparece no eco de session.updated por padrão, então confira na doc, não assuma.

• threshold (0,1 a 0,9; padrão 0,85): quão alto o áudio precisa ser para contar como fala; aumente em ambientes ruidosos, diminua se vozes baixas não forem detectadas

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

• prefix_padding_ms (padrão 333): um 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. É a primeira alavanca que eu mexo antes das outras.

Recebendo e reproduzindo a resposta

O áudio chega em pedaços pequenos como response.output_audio.delta, e o ponto do streaming é você reproduzir cada pedaço assim que ele chega, 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)

Mantenha a transcrição mesmo em produção. É a ferramenta de depuração 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 pedido

Cada ferramenta é um schema JSON mais uma função Python simples do nosso lado. O modelo nunca toca o banco de dados; ele só vê o que a 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 após um timeout ambíguo pode aplicar a mesma mudança duas vezes. Uma linha de confirmação no prompt não impede isso; dê às escritas uma chave de idempotência ou verificação de duplicidade.

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

Tratando o loop de chamadas de ferramentas

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ê devolve o resultado 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 function_call_arguments.done antes de tocar qualquer áudio. Resolva todos e envie cada resultado antes de um único response.create. Se você enviar cedo demais, o modelo responde sem o contexto das chamadas ainda em andamento.

Há uma pegadinha que a SpaceXAI documenta e mesmo assim eu enfrentei: enviar response.create no exato instante em que seu resultado sai pode se sobrepor à frase introdutória que o agente ainda está falando. Numa execução, ele abriu com "Vou conferir o status do pedido ORD‑1042 agora mesmo" e chamou a ferramenta no meio da frase; uma resposta imediata teria falado por cima da própria 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 separados aqui. Quem liga fala por cima do agente no meio da resposta, e um WebSocket cai e precisa ser retomado.

Dando suporte a interrupções naturais

Com server_vad ativado, o barge‑in é automático no servidor: no momento em que detecta a pessoa falando de novo, ele sinaliza input_audio_buffer.speech_started e para de gerar a resposta antiga. Seu papel é a metade do cliente: limpar o áudio já enfileirado para o agente ficar em silêncio 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 a pedido. Existe também conversation.item.truncate para reduzir um item do assistente ao que foi realmente ouvido. A doc confirma que existe, mas não quando disparar durante um barge‑in ao vivo, então teste o timing você mesmo.

Testei isso num ajuste de instrução de entrega no meio da resposta: iniciar o pedido, interromper com um endereço diferente no meio da confirmação do agente. O que importa é se o agente aplica a instrução corrigida em vez de terminar a antiga silenciosamente, não se o áudio parou. Valide no registro do pedido, não no silêncio. A demo no navegador no fim deixa você ouvir isso.

Retomando uma sessão desconectada

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

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, transcrições, chamadas e resultados de ferramentas são reexecutados antes da sua próxima pergunta, e o cache some 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 o ETA corretamente.

Uma ressalva não documentada: a reexecução não cai instantaneamente, então uma pergunta enviada no exato momento em que o socket abre pode chegar antes e voltar sem memória do turno anterior. Dê um segundo antes de culpar a retomada.

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

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

Segurança e monitoramento do agente

Nunca coloque uma chave de API permanente em código de navegador ou mobile. Se o cliente conecta direto em vez de passar pelo seu servidor, emita 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 customizado 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 roda em dois medidores. Áudio, enviado ou recebido, vai na tarifa de US$ 0,08 por minuto que citei — US$ 4,80 por hora — e cada conversation.item.create que não for áudio e não for function_call_output custa fixo US$ 0,004. response.create não é cobrado. Cada response.done traz um objeto usage que, no meu teste, reportou output_audio_seconds e um billable_audio_seconds separado. Fature por eles, não por estimativas.

Os limites documentados da Speech to Speech API são 10 sessões concorrentes por time e uma sessão máxima de 120 minutos, 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 da API são retidas criptografadas por 30 dias para monitoramento de abuso e não são usadas para treino sem permissão, e que times podem ativar Zero Data Retention — embora o ZDR elimine o histórico de conversas do agente e, portanto, não funcione com retomada.

Se você informa que a chamada é gravada ou atendida por IA, é para isso que serve a extensão force_message que citei antes. A linha é reproduzida 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 chegou uma 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, validada no registro do pedido
• Uma recusa, como cancelar um pedido já enviado, em que o agente precisa explicar 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, verificando se o agente fala o erro em vez de travar
• Reconexão e retomada, incluindo aquela janela de reexecução que citei
• Áudio ruidoso, fala acelerada e alguém que soletra números e endereços

Rodei a maioria contra uma chave real enquanto escrevia. As falhas interessantes foram comportamentais, não erros: o timing da retomada acima e um threshold de VAD fora da faixa sendo aceito em vez de rejeitado — o tipo de coisa que passa quebrado em silêncio se você só testa o caminho feliz. Adicione também um teste multilíngue e veja as FAQs para um detalhe de como nomear o idioma.

Dois desses você não testa 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