Grok Voice Agent Builder: Python ile Uygulamalı Bir Rehber

xAI, sesli aracıları oluşturmak için bir konsol olan Voice Agent Builder’ı yayınladı. Çağrı akışını tarif ediyorsunuz, belgelere ve araçlara ekliyorsunuz ve bir ses seçiyorsunuz.

Bir sesli aracı konsolunu denerken, lansman notundan çok koda bağlamam gereken kısımlar ilgimi çeker: WebSocket oturumu nasıl yapılandırılır, ses nasıl taşınır, araç çağrıları nerede gerçekleşir, çağrının maliyeti nedir ve başka bir uygulama iş akışını nasıl çağırır.

Aşağıdaki kod, bu akışı doğrudan Voice Agent API’ye karşı yeniden kuruyor. Özellikle, uygunluk kontrolü yapan, sesle yanıt veren, maliyet takibi yapan, araç hatalarını yöneten ve bir FastAPI uç noktası sunan bir klinik randevu asistanı kullanacağız.

Grok Voice Agent Builder Nedir?

Voice Agent Builder, Grok Voice üzerinde sesli aracıları oluşturup dağıtmak için xAI’nin konsoludur. 1 Temmuz 2026’da beta olarak yayınlandı. Ayrı ayrı konuşmadan metne, dil modeli ve metinden konuşmaya hizmetleri kullanmak yerine tek bir ses modeli yolunu kullanır.

Konsol; telefon sistemi, belge getirimi, araçlar ve bağlayıcılar, koruma önlemleri, uzak MCP sunucuları ve kayıtlar, transkriptler ve izlerle birlikte çağrı günlüklerini içerir.

Ses dakika başına ücretlendirilir. Konsol hâlâ beta olduğundan, doğrudan API’yi kullanıyoruz.

Builder’ın Altındaki Grok Voice Agent API Nasıl Çalışır

Konsolun altında, Builder ile aynı çalışma zamanını sunan Voice Agent API vardır; bu, gerçek zamanlı bir WebSocket API’sidir.

Builder, Voice API’nin üstünde yer alır. Görsel: Yazar.

Burada kullanılan model grok-voice-think-fast-1.0. grok-voice-latest takma adı en yeni modele işaret eder. Burada onu kullanıyorum, ancak dağıtılmış bir uygulama için sürümlü adı sabitlerdim. xAI, bu model için τ-voice Bench liderlik tablosunda %67,3’lük bir puan bildiriyor; bunu bir veri noktası olarak görüyorum, garanti olarak değil.

Uyumluluk notu: API, OpenAI Realtime API ile uyumludur. OpenAI’nin gerçek zamanlı uç noktasına konuşan kodunuz varsa, çoğunlukla temel URL’yi ve anahtarı değiştirirsiniz.

Proje Genel Bakışı: Ne İnşa Edeceğiz

Klinik asistanı sesli girdi alır, üretilen bir sesle yanıt verir, takip soruları sorar, bir aralık önermeden önce uygunluğu kontrol eder ve gerektiğinde insana devreder. Çekirdek örnek bir araç kullanır; Streamlit demosu rezervasyon, aktarım ve çağrı sonlandırma eylemlerini ekler.

Çekirdek eğitim dört dosyaya ayrılır; her birinin tek bir işi vardır:

• voice_client.py WebSocket istemcisini, ses yardımcılarını ve maliyet takibini barındırır

• tools.py check_availability’i ve Streamlit tarafından kullanılan ek demo araçlarını barındırır

• assistant.py sistem komutunu, oturum yapılandırmasını ve iş akışını barındırır

• app.py tümünü FastAPI üzerinden sunar

Bu dört dosya, makaledeki yol haritasıdır. Depoda ayrıca görsel demo için app_streamlit.py ve bir Windows başlatıcısı olarak run.py bulunur, ancak çekirdek akış çalıştıktan sonra onlara döneceğiz.

Önkoşullar

Kodu çalıştırmadan önce Python 3.10 veya daha yenisi, bir xAI hesabı, console.x.ai’dan bir API anahtarı, ön ödemeli krediler ve ortam değişkenleri, JSON ve WebSocket’lerle temel düzeyde rahatlık gerekir.

Projeyi kurma

Bir klasör ve sanal ortam oluşturun, ardından paketleri yükleyin:

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

Taze bir checkout’ın aynı kurulumu kullanması için bu paketleri bir requirements.txt dosyasında sabitleyin.

.env dosyasını Python dosyalarının yanına oluşturun:

XAI_API_KEY=xai-your-key-here

.env dosyasını .gitignore’a ekleyin. API anahtarı sunucuda kalmalıdır.

Sesli Aracı Oluşturma

Başlamaya koyulalım.

Grok Voice Agent API’ye WebSocket ile bağlanma

İlk adım bağlantıyı açmaktır. Modeli bir sorgu parametresi olarak ve anahtarınızı el sıkışmada yetkilendirme belirteci olarak iletin:

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

Canlı bir anahtara karşı, gördüğünüz ilk olay session.created olur; bu da soketin açık ve yapılandırmaya hazır olduğu anlamına gelir.

Session created olayı bağlantıyı doğrular. Görsel: Yazar.

Ses oturumunu yapılandırma

Canlı bir soket, yapılandırılmış bir ajan değildir. Onu bir session.update olayı gönderip session nesnesiyle şekillendirirsiniz.

Ses, ses formatı ve yönergeler

En çok dokunduğunuz üç ayar sestir, ses formatıdır ve sistem komutudur. Gerçek zamanlı API beş isimli sesi sunar, eve, ara, rex, sal ve leo; ayrıca herhangi bir özel klon. Ses varsayılan olarak audio/pcm 24000 Hz’dir; giriş ve çıkış ayrı ayrı yapılandırılır.

İşte assistant.py’de derlenen asistanın kullandığı oturum yapılandırması:

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

instructions alanı sistem komutudur. Bu klinik komutu kısa tutuyoruz çünkü uzun sesli yanıtları takip etmek zordur:

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.

Yükseltme satırı klinik ajanını tıbbi tavsiyeden uzak tutar. Son iki satır, odağı korur ve arayan belirsiz olduğunda döngüleri durdurur. Yapılandırma ayrıca bugünün tarihini ekler; çünkü canlı testlerimde model, "6 Temmuz" gibi tarihler için yanlış yılı tahmin edebiliyordu.

Tur algılamayı ayarlama

Tur algılama, ajanın konuşmayı bıraktığınızı nasıl anladığıdır. turn_detection.type değerini server_vad olarak ayarlarsanız, sunucu sessizlikte turu bitirir. null bırakın ve turları ses arabelleğini onaylayarak siz kontrol edin; dosya akışı için benim kullandığım budur.

Sunucu VAD’inin bilmeye değer üç ayarı vardır: threshold, sesin konuşma sayılması için ne kadar yüksek olması gerektiğini; silence_duration_ms, bir duraklamanın turu bitirmesi için süresini; ve prefix_padding_ms, konuşma başlamadan önce biraz sesi korur. Ajanınız insanları yarıda kesiyorsa, önce silence_duration_ms değerini yükseltin.

Sese ajana gönderme

Şimdi arayanın sesini gönderiyoruz. Ses, oturum formatıyla eşleşmelidir: 24000 Hz’de mono 16 bit PCM; base64 kodlanmış ve parçalara ayrılarak gönderilir.

İstemci dosyayı dilimler halinde akar, ardından tur sonunu işaretlemek için arabelleği onaylar:

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)

Örnekleme hızınız veya kodlamanız session.update ile eşleşmiyorsa, temiz bir hata yerine cızırtı veya sessizlik alabilirsiniz. Ses, input_audio_buffer.append üzerinden geçer; bu nedenle mesaj başına değil, süreye göre ücretlendirilir.

Sesli yanıtları alma

Bir yanıt istediğinizde, ses response.output_audio.delta olarak gelir; transkript response.output_audio_transcript.delta olarak gelir; ve response.done turu kapatır.

İstemci bunların tümünü tek bir async döngüsünde toplar:

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

Ses deltalarını çözün, sırayla ekleyin ve sonucu bir response.wav dosyasına yazın. Arayanın kendi sözlerini yakalamak için audio.input.transcription ayarlayın ve conversation.item.input_audio_transcription.completed değerini okuyun.

Randevu Asistanı İş Akışını Kurma

Şimdi parçalar bir konuşma hâline geliyor: rezervasyon isteği, netleştirici soru, uygunluk kontrolü, sunulan aralıklar, onay. Turlar arasında bağlamı taşımak için her yeni tur, konuşma kimliğiyle yeniden bağlanır ve oturum sürdürmeyi kabul eder.

Sesli ajana araç çağrısı ekleme

Klinik için, ajan bir zaman sözü vermeden önce uygunluğu kontrol etmelidir. Özel araçlar, modelin kodunuza ulaşma yoludur: bir istek yayar, uygulamanız fonksiyonu çalıştırır ve sonucu geri gönderirsiniz.

Araç, oturum yapılandırmasına giren bir JSON şemasına sahip düz bir işlevdir. İşte tools.py’den şema:

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"],
    },
}

Döngünün sabit bir şekli vardır. Model aracı istediğinde, argümanlarla birlikte response.function_call_arguments.done gönderir. Fonksiyonu çalıştırır, bir function_call_output döndürür ve ardından ajanın devam edebilmesi için response.create gönderirsiniz. O son response.create adımı atlayın; ajan sessiz kalır.

Araç çağrısı gidiş-dönüşü açıklanıyor. Görsel: Yazar.

Bunun gibi özel işlevler kodunuzda çalışır. Streamlit demosu, aynı dosyadan üç tanesini daha kaydeder: book_appointment, transfer_to_human ve end_call. Dahili araçlar, örneğin web araması, X araması, koleksiyonlar araması ve uzak MCP araçları, xAI’nin sunucularında çalışır.

Araç hatalarını yönetme

Araçlar başarısız olur ve başarı varsayan bir sesli ajan, var olmayan bir aralığı söz verebilir. Benim ToolRegistry.execute hiç istisna fırlatmaz: başarısız bir arama {"error": ...} sözlüğü olarak döner.

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

Açık bir hata durumu, ajanın başarısız araç çağrılarını başarı olarak değerlendirmesini engeller.

Maliyet takibi ekleme

Bunu birine sunmadan önce, bir çağrının maliyetini bilin. Ses, gönderdiğiniz ve aldığınız her iki taraf da sayılarak, dakika başına 0,05 $ üzerinden ücretlendirilir. Metin giriş olayları, olay başına 0,004 $’dır. function_call_output sonuçları ve response.create olayları ücretlendirilmez.

İstemci, giderken bunları takip eder; böylece maliyet, herhangi bir noktada okuyabileceğiniz bir özelliktir:

@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

xAI tarafından tahsis edilen bir numara, dakika başına 0,01 $ telekom ek ücretini ekler; yardımcı, telephony=True ayarladığınızda bunu uygular. xAI tarafından barındırılan araçlar ayrı ücretlendirilir: web araması ve X araması bin çağrı başına yaklaşık 5 $, dosya araması ise yaklaşık 2,50 $’dır.

Hataları ve köşe durumlarını ele alma

Çoğu hata kısa bir listeye girer:

• Eksik veya geçersiz API anahtarı, el sıkışmada 401 döner; bu yüzden önce anahtarı kontrol edin

• Engellenmiş bir ekip 403 döner ve hız sınırı 429 döner; geri alma (backoff) ile yeniden deneyin

• Hatalı oturum yapılandırması 400 döner; genellikle bir alan adında yazım hatasıdır

• Desteklenmeyen ses formatı hata vermez, cızırtı üretir; bu yüzden oturum hızını eşleştirin

• Bir araç sonucundan sonra eksik bir response.create, ajanı askıda bırakır

• Yinelenen bir rezervasyon denemesi gerçek sorunlara yol açabilir; bu yüzden körlemesine yeniden denemeyin

Başarısız bir okuma olan check_availability’i yeniden denemek güvenlidir; ancak gerçek bir rezervasyon gibi başarısız bir yazmayı yeniden denemek, bir arayanı iki kez rezerve edebilir. Veriyi değiştiren herhangi bir eylem, önce bir idempotency kontrolü gerektirir.

İstemci uygulamalar için geçici belirteçler kullanma

Şimdiye kadar her şey, kodun API anahtarının ait olduğu sunucunuzda çalıştığını varsaydı. Bir tarayıcı veya mobil uygulama doğrudan bağlanıyorsa, geçici belirteçler kullanın.

Sunucunuz POST https://api.x.ai/v1/realtime/client_secrets çağrısını anahtarınızla yapar, bir belirteç yanıtı alır ve belirteç değerini istemciye iletir. Benim çalışmamda, yanıt value ve expires_at içeriyordu:

@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()

Tarayıcılar özel WebSocket üstbilgileri ayarlayamaz; bu nedenle belirteç, sec-websocket-protocol üstbilgisinde xai-client-secret. önekiyle taşınır.

İş Akışını Bir FastAPI Uç Noktasına Dönüştürme

Bir uç nokta, bir ön yüzün veya başka bir hizmetin iş akışını çağırmasına izin verir. Rota, istek gövdesini bir Pydantic modeliyle doğrular, yazılı bir mesaj veya bir ses yolu alır ve transkript, yanıt sesi, araç günlüğü, gecikme ve tahmini maliyeti döndürür.

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

uvicorn app:app --reload ile çalıştırın ve http://localhost:8000/docs adresini açın. XAI_API_KEY ’i sunucu ortamından okuyun ve asla istek gövdesinden kabul etmeyin.