Grok Voice Agent Builder: Panduan Praktis di Python

xAI merilis Voice Agent Builder, sebuah konsol untuk membuat agen suara. Anda menggambarkan alur panggilan, melampirkan dokumen dan tool, serta memilih suara.

Saat saya menguji konsol agen suara, saya kurang peduli pada catatan peluncuran dan lebih pada bagian yang harus saya hubungkan ke kode: bagaimana sesi WebSocket dikonfigurasi, bagaimana audio bergerak, di mana pemanggilan tool terjadi, berapa biaya panggilan, dan bagaimana aplikasi lain akan memanggil alur kerja tersebut.

Kode di bawah ini membangun ulang alur tersebut langsung terhadap Voice Agent API. Secara khusus, kita akan menggunakan asisten janji klinik yang memeriksa ketersediaan, membalas dengan suara, melacak biaya, menangani kegagalan tool, dan mengekspos endpoint FastAPI.

Apa Itu Grok Voice Agent Builder?

Voice Agent Builder adalah konsol xAI untuk membuat dan menerapkan agen suara di Grok Voice. Diluncurkan versi beta pada 1 Juli 2026. Alih-alih menggunakan layanan terpisah untuk speech to text, model bahasa, dan text to speech, produk ini memakai satu jalur model suara.

Konsol ini mencakup teleponi, pengambilan dokumen, tool dan konektor, guardrail, server MCP jarak jauh, serta log panggilan dengan rekaman, transkrip, dan jejak.

Audio ditagih per menit. Konsol masih beta, jadi kita gunakan API langsung.

Cara Kerja Grok Voice Agent API di Balik Builder

Di balik konsol terdapat Voice Agent API, sebuah API WebSocket realtime yang mengekspos runtime yang sama dengan yang digunakan Builder.

Builder berada di atas Voice API. Gambar oleh Penulis.

Model yang digunakan di sini adalah grok-voice-think-fast-1.0. Alias grok-voice-latest menunjuk ke model terbaru. Saya menggunakannya di sini, tetapi untuk aplikasi produksi saya akan mengunci ke nama versi. xAI melaporkan skor 67,3% untuk model ini pada papan peringkat τ-voice Bench; saya menganggapnya sebagai satu titik data, bukan jaminan.

Catatan kompatibilitas: API ini kompatibel dengan OpenAI Realtime API. Jika Anda memiliki kode yang berbicara ke endpoint realtime OpenAI, Anda umumnya hanya mengganti base URL dan kunci.

Gambaran Proyek: Apa yang Akan Kita Bangun

Asisten klinik menerima input lisan, membalas dengan suara yang dihasilkan, mengajukan pertanyaan lanjutan, memeriksa ketersediaan sebelum menawarkan slot, dan menyerahkan ke manusia saat diperlukan. Contoh inti menggunakan satu tool; demo Streamlit menambahkan aksi pemesanan, transfer, dan mengakhiri panggilan.

Tutorial inti terbagi menjadi empat berkas, masing-masing dengan satu tugas:

• voice_client.py memuat klien WebSocket, helper audio, dan pelacakan biaya

• tools.py memuat check_availability, plus tool demo tambahan yang digunakan oleh Streamlit

• assistant.py memuat sistem prompt, konfigurasi sesi, dan alur kerja

• app.py menyajikan seluruhnya melalui FastAPI

Keempat berkas itu menjadi jalur melalui artikel. Repo juga menyertakan app_streamlit.py untuk demo visual dan run.py sebagai peluncur Windows, tetapi kita akan kembali ke keduanya setelah alur inti berfungsi.

Prasyarat

Sebelum kode dijalankan, Anda memerlukan Python 3.10 atau lebih baru, akun xAI, kunci API dari console.x.ai, kredit prabayar, serta kenyamanan dasar dengan variabel lingkungan, JSON, dan WebSocket.

Menyiapkan proyek

Buat folder dan environment virtual, lalu instal paket-paket berikut:

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

Kunci versi paket ini dalam requirements.txt agar checkout baru menggunakan setup yang sama.

Buat berkas .env di sebelah berkas-berkas Python:

XAI_API_KEY=xai-your-key-here

Tambahkan .env ke .gitignore. Kunci API harus tetap berada di server.

Membangun Voice Agent

Mari mulai membangun.

Menyambung ke Grok Voice Agent API via WebSocket

Langkah pertama adalah membuka koneksi. Lewatkan model sebagai parameter query dan kunci Anda sebagai bearer token pada 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())

Dengan kunci aktif, event pertama yang Anda lihat adalah session.created, yang berarti soket terbuka dan siap dikonfigurasi.

Event session created mengonfirmasi koneksi. Gambar oleh Penulis.

Mengonfigurasi sesi suara

Soket aktif bukan berarti agen telah terkonfigurasi. Anda membentuknya dengan mengirim event session.update dengan objek session.

Suara, format audio, dan instruksi

Tiga pengaturan yang paling sering disentuh adalah suara, format audio, dan system prompt. Realtime API mengekspos lima suara bernama, eve, ara, rex, sal, dan leo, plus klon kustom apa pun. Audio default ke audio/pcm pada 24000 Hz, dengan input dan output dikonfigurasi terpisah.

Berikut konfigurasi sesi yang digunakan asisten, disusun di 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],
    }

Bidang instructions adalah system prompt. Prompt klinik ini tetap singkat karena jawaban suara yang panjang sulit diikuti:

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.

Baris eskalasi menjaga agen klinik agar tidak memberi nasihat medis. Dua baris terakhir menjaganya tetap pada lingkup dan menghentikan loop ketika penelepon tidak jelas. Konfigurasi juga menambahkan tanggal hari ini karena, dalam uji langsung saya, model bisa menebak tahun yang salah untuk tanggal seperti "6 Juli".

Menyetel turn detection

Turn detection adalah cara agen memutuskan Anda sudah berhenti berbicara. Setel turn_detection.type ke server_vad dan server akan mengakhiri giliran pada saat hening. Biarkan null dan Anda mengontrol giliran dengan meng-commit buffer audio, yang saya gunakan untuk alur berbasis berkas.

Server VAD memiliki tiga pengaturan yang perlu diketahui: threshold menetapkan seberapa keras audio agar dihitung sebagai ucapan, silence_duration_ms menetapkan lamanya jeda yang mengakhiri giliran, dan prefix_padding_ms mempertahankan sedikit audio sebelum ucapan dimulai. Jika agen Anda sering menyela orang, naikkan silence_duration_ms terlebih dahulu.

Mengirim audio ke agen

Sekarang kita kirim suara penelepon. Audio harus sesuai format sesi: mono 16 bit PCM pada 24000 Hz, dikodekan sebagai base64 dan dikirim dalam potongan.

Klien melakukan streaming berkas dalam irisan, lalu meng-commit buffer untuk menandai akhir giliran:

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)

Jika laju sampel atau enkoding Anda tidak cocok dengan session.update, Anda mungkin mendapat statis atau hening alih-alih error yang jelas. Audio melalui input_audio_buffer.append, jadi penagihan berdasarkan durasi, bukan per pesan.

Menerima respons suara

Setelah Anda meminta respons, audio datang sebagai response.output_audio.delta, transkrip datang sebagai response.output_audio_transcript.delta, dan response.done menutup giliran.

Klien mengumpulkan semuanya dalam satu loop async berikut:

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

Dekode delta audio, gabungkan berurutan, dan tulis hasilnya ke berkas response.wav. Untuk menangkap ucapan penelepon sendiri, setel audio.input.transcription dan baca conversation.item.input_audio_transcription.completed.

Membangun Alur Kerja Asisten Janji

Kini potongan-potongan menjadi percakapan: permintaan pemesanan, pertanyaan klarifikasi, pemeriksaan ketersediaan, penawaran slot, konfirmasi. Untuk membawa konteks lintas giliran, setiap giliran baru menyambung kembali dengan id percakapan dan memilih melanjutkan sesi.

Menambahkan pemanggilan tool ke agen suara

Untuk klinik, agen harus memeriksa ketersediaan sebelum menjanjikan waktu. Tool kustom adalah cara model menjangkau kode Anda: model memancarkan permintaan, aplikasi Anda menjalankan fungsi, dan Anda mengirimkan hasilnya kembali.

Tool berupa fungsi biasa plus skema JSON yang dimasukkan ke konfigurasi sesi. Berikut skema dari 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"],
    },
}

Loop-nya memiliki bentuk tetap. Saat model menginginkan tool, ia mengirim response.function_call_arguments.done dengan argumen. Anda menjalankan fungsi, mengembalikan function_call_output, lalu mengirim response.create agar agen dapat melanjutkan. Jika Anda melewatkan response.create terakhir itu, agen akan diam.

Perjalanan bolak-balik pemanggilan tool dijelaskan. Gambar oleh Penulis.

Fungsi kustom seperti ini berjalan di kode Anda. Demo Streamlit mendaftarkan tiga lagi dari berkas yang sama: book_appointment, transfer_to_human, dan end_call. Tool bawaan, seperti pencarian web, X search, pencarian koleksi, dan tool MCP jarak jauh, dieksekusi di server xAI.

Menangani kegagalan tool

Tool bisa gagal, dan agen suara yang menganggap semua berhasil dapat menjanjikan slot yang tidak ada. ToolRegistry.execute saya tidak pernah melempar exception: pencarian yang gagal kembali sebagai 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)}

Status error eksplisit menghentikan agen memperlakukan pemanggilan tool yang gagal sebagai keberhasilan.

Menambahkan pelacakan biaya

Sebelum Anda menyajikan ini ke siapa pun, ketahui berapa biaya panggilan. Audio ditagih $0,05 per menit, menghitung baik yang Anda kirim maupun yang Anda terima. Event input teks ditagih $0,004 masing-masing. Hasil function_call_output dan event response.create tidak ditagih.

Klien melacaknya sepanjang jalan, jadi biaya adalah properti yang bisa Anda baca kapan saja:

@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

Nomor yang diprovisikan xAI menambahkan biaya tambahan teleponi $0,01 per menit, yang diterapkan helper saat Anda menyetel telephony=True. Tool yang di-host oleh xAI ditagih terpisah: pencarian web dan X sekitar $5 per seribu panggilan, dan pencarian berkas sekitar $2,50.

Menangani error dan edge case

Sebagian besar kegagalan jatuh ke daftar pendek ini:

• Kunci API hilang atau tidak valid mengembalikan 401 saat handshake, jadi periksa kunci terlebih dahulu

• Tim yang diblokir mengembalikan 403, dan batas laju mengembalikan 429, yang perlu Anda retry dengan backoff

• Konfigurasi sesi yang salah bentuk mengembalikan 400, biasanya salah ketik pada nama kolom

• Format audio yang tidak didukung menghasilkan statis, bukan error, jadi samakan laju sesi

• Kehilangan response.create setelah hasil tool membuat agen menggantung

• Upaya pemesanan duplikat bisa menimbulkan masalah nyata, jadi jangan sekadar retry membabi buta

Melakukan retry untuk pembacaan yang gagal seperti check_availability aman, tetapi retry untuk penulisan yang gagal seperti pemesanan nyata bisa menggandakan pemesanan penelepon. Setiap tindakan yang mengubah data perlu pemeriksaan idempoten terlebih dahulu.

Menggunakan token sementara untuk aplikasi klien

Semua sejauh ini mengasumsikan kode berjalan di server Anda, tempat kunci API berada. Jika browser atau aplikasi seluler tersambung langsung, gunakan token sementara.

Server Anda memanggil POST https://api.x.ai/v1/realtime/client_secrets dengan kunci Anda, menerima respons token, dan meneruskan nilai token ke klien. Dalam percobaan saya, respons menyertakan value dan 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()

Browser tidak dapat menetapkan header WebSocket kustom, jadi token dibawa dalam header sec-websocket-protocol dengan awalan xai-client-secret..

Mengubah Alur Kerja Menjadi Endpoint FastAPI

Sebuah endpoint memungkinkan frontend atau layanan lain memanggil alur kerja. Rute memvalidasi body permintaan dengan model Pydantic, menerima pesan bertipe atau path audio, dan mengembalikan transkrip, audio respons, log tool, latensi, serta estimasi biaya.

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

Jalankan dengan uvicorn app:app --reload dan buka http://localhost:8000/docs. Baca XAI_API_KEY dari environment server dan jangan pernah menerimanya dari body permintaan.