Grok Voice Agent Builder: Python में व्यावहारिक मार्गदर्शिका

xAI ने Voice Agent Builder जारी किया, जो वॉइस एजेंट बनाने के लिए एक कंसोल है। आप कॉल फ्लो का वर्णन करते हैं, दस्तावेज़ और टूल जोड़ते हैं, और एक आवाज़ चुनते हैं।

जब मैं किसी वॉइस एजेंट कंसोल का परीक्षण करता हूँ, तो मुझे लॉन्च नोट से कम और उन हिस्सों से ज़्यादा मतलब होता है जिन्हें मुझे कोड में वायर करना है: WebSocket सत्र कैसे कॉन्फ़िगर होता है, ऑडियो कैसे चलता है, टूल कॉल कहाँ होते हैं, कॉल की लागत क्या है, और कोई अन्य ऐप वर्कफ़्लो को कैसे कॉल करेगा।

नीचे दिया गया कोड उस फ्लो को सीधे Voice Agent API के साथ फिर से बनाता है। खास तौर पर, हम एक क्लिनिक अपॉइंटमेंट असिस्टेंट का उपयोग करेंगे जो उपलब्धता जाँचता है, आवाज़ में जवाब देता है, लागत ट्रैक करता है, टूल विफलताओं को संभालता है, और एक FastAPI एन्डपॉइंट उपलब्ध कराता है।

Grok Voice Agent Builder क्या है?

Voice Agent Builder, Grok Voice पर वॉइस एजेंट बनाने और डिप्लॉय करने के लिए xAI का कंसोल है। यह 1 जुलाई, 2026 को बीटा में लॉन्च हुआ। अलग-अलग स्पीच-टू-टेक्स्ट, लैंग्वेज मॉडल और टेक्स्ट-टू-स्पीच सेवाओं के बजाय, यह एक ही वॉइस मॉडल पथ का उपयोग करता है।

कंसोल में टेलीफोनी, दस्तावेज़ रिकवरी, टूल्स और कनेक्टर्स, गार्डरेल्स, रिमोट MCP सर्वर, और रिकॉर्डिंग, ट्रांसक्रिप्ट्स व ट्रेसेज़ के साथ कॉल लॉग शामिल हैं।

ऑडियो प्रति मिनट बिल होता है। कंसोल अभी बीटा में है, इसलिए हम सीधे API का उपयोग करते हैं।

Builder के तहत Grok Voice Agent API कैसे काम करता है

कंसोल के नीचे Voice Agent API है, जो एक रीयलटाइम WebSocket API है और वही रनटाइम उपलब्ध कराता है जो Builder उपयोग करता है।

Builder, Voice API के ऊपर बैठता है। चित्र: लेखक।

यहाँ उपयोग किया गया मॉडल है grok-voice-think-fast-1.0। grok-voice-latest उर्फ़ नवीनतम मॉडल की ओर इंगित करता है। मैं इसे यहाँ उपयोग करता हूँ, लेकिन डिप्लॉय किए गए ऐप के लिए मैं संस्करणित नाम पिन करूँगा। xAI इस मॉडल के लिए τ-voice Bench लीडरबोर्ड पर 67.3% स्कोर रिपोर्ट करता है; मैं इसे एक डेटा बिंदु मानता हूँ, कोई गारंटी नहीं।

संगतता नोट: यह API OpenAI Realtime API के साथ संगत है। यदि आपके पास ऐसा कोड है जो OpenAI के रीयलटाइम एन्डपॉइंट से बात करता है, तो आपको प्रायः केवल बेस URL और कुंजी बदलनी होती है।

प्रोजेक्ट अवलोकन: हम क्या बनाएँगे

क्लिनिक असिस्टेंट बोले गए इनपुट लेता है, जनरेटेड आवाज़ में जवाब देता है, फॉलो-अप सवाल पूछता है, स्लॉट देने से पहले उपलब्धता जाँचता है, और ज़रूरत पड़ने पर मानव को सौंप देता है। मुख्य उदाहरण एक टूल उपयोग करता है; Streamlit डेमो बुकिंग, ट्रांसफर और कॉल समाप्त करने की क्रियाएँ जोड़ता है।

मुख्य ट्यूटोरियल चार फाइलों में बाँटा गया है, प्रत्येक का एक काम:

• voice_client.py में WebSocket क्लाइंट, ऑडियो हेल्पर्स, और लागत ट्रैकिंग रहती है

• tools.py में check_availability रहता है, साथ ही Streamlit द्वारा उपयोग किए गए अतिरिक्त डेमो टूल्स

• assistant.py में सिस्टम प्रॉम्प्ट, सत्र कॉन्फ़िग, और वर्कफ़्लो रहता है

• app.py पूरी चीज़ को FastAPI के माध्यम से सर्व करता है

ये चार फाइलें लेख के रास्ते का मानचित्र हैं। रिपो में app_streamlit.py भी है विजुअल डेमो के लिए और run.py Windows लॉन्चर के रूप में, पर हम उन पर तब लौटेंगे जब मुख्य फ्लो काम करने लगेगा।

पूर्वापेक्षाएँ

कोड चलाने से पहले, आपको Python 3.10 या नया संस्करण, xAI खाता, console.x.ai से API कुंजी, प्रीपेड क्रेडिट्स, और एन्वायरमेंट वेरिएबल्स, JSON, तथा WebSockets के साथ बुनियादी सहजता चाहिए।

प्रोजेक्ट सेटअप

एक फ़ोल्डर और एक वर्चुअल एन्वायरमेंट बनाएँ, फिर पैकेज इंस्टॉल करें:

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

इन पैकेजों को requirements.txt में पिन करें ताकि नया चेकआउट वही सेटअप उपयोग करे।

Python फाइलों के पास एक .env फाइल बनाएँ:

XAI_API_KEY=xai-your-key-here

.env को .gitignore में जोड़ें। API कुंजी सर्वर पर ही रहनी चाहिए।

वॉइस एजेंट बनाना

आइए बनाना शुरू करें।

WebSocket के माध्यम से Grok Voice Agent API से कनेक्ट करना

पहला कदम कनेक्शन खोलना है। मॉडल को एक क्वेरी पैरामीटर के रूप में और अपनी कुंजी को हैंडशेक पर बेयरर टोकन के रूप में पास करें:

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

लाइव कुंजी के साथ, पहला ईवेंट जो आप देखते हैं वह session.created है, जिसका मतलब है सॉकेट खुला है और कॉन्फ़िगर होने को तैयार है।

Session created ईवेंट कनेक्शन की पुष्टि करता है। चित्र: लेखक।

वॉइस सत्र कॉन्फ़िगर करना

लाइव सॉकेट एक कॉन्फ़िगर किया हुआ एजेंट नहीं है। आप इसे session.update ईवेंट भेजकर एक session ऑब्जेक्ट के साथ आकार देते हैं।

वॉइस, ऑडियो फ़ॉर्मेट, और निर्देश

तीन सेटिंग्स जिन्हें आप सबसे अधिक बदलते हैं: आवाज़, ऑडियो फ़ॉर्मेट, और सिस्टम प्रॉम्प्ट। रीयलटाइम API पाँच नामित आवाज़ें देता है, eve, ara, rex, sal, और leo, साथ ही कोई भी कस्टम क्लोन। ऑडियो डिफ़ॉल्ट रूप से audio/pcm 24000 Hz पर रहता है, इनपुट और आउटपुट अलग-अलग कॉन्फ़िगर होते हैं।

यह है सत्र कॉन्फ़िग जो असिस्टेंट उपयोग करता है, 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],
    }

instructions फ़ील्ड सिस्टम प्रॉम्प्ट है। यह क्लिनिक प्रॉम्प्ट छोटा रहता है क्योंकि लंबी वॉइस प्रतिक्रियाएँ फॉलो करना मुश्किल होता है:

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.

एस्केलेशन वाली पंक्ति क्लिनिक एजेंट को चिकित्सा सलाह से बाहर रखती है। आख़िरी दो पंक्तियाँ इसे दायरे में रखती हैं और जब कॉलर अस्पष्ट हो तो लूप रोकती हैं। कॉन्फ़िग आज की तारीख़ भी जोड़ता है क्योंकि, मेरी लाइव टेस्टिंग में, मॉडल "July 6th" जैसे दिनांक के लिए साल गलत अनुमान लगा सकता था।

टर्न डिटेक्शन ट्यून करना

टर्न डिटेक्शन यह तय करता है कि एजेंट कब मानता है कि आपने बोलना बंद कर दिया है। turn_detection.type को server_vad पर सेट करें और सर्वर साइलेंस पर टर्न खत्म कर देता है। इसे null छोड़ दें और आप ऑडियो बफर को कमिट करके टर्न नियंत्रित करते हैं, जो मैं फ़ाइल फ्लो के लिए उपयोग करता हूँ।

सर्वर VAD में तीन सेटिंग्स जानने लायक हैं: threshold तय करता है कि आवाज़ को स्पीच मानने के लिए कितना तेज़ होना चाहिए, silence_duration_ms तय करता है कि कितना लंबा विराम टर्न समाप्त करता है, और prefix_padding_ms स्पीच शुरू होने से पहले थोड़ा ऑडियो रखता है। यदि आपका एजेंट लोगों की बात काटता है, तो पहले silence_duration_ms बढ़ाएँ।

एजेंट को ऑडियो भेजना

अब हम कॉलर की आवाज़ भेजते हैं। ऑडियो को सत्र फ़ॉर्मेट से मेल खाना चाहिए: मोनो 16-बिट PCM 24000 Hz पर, base64 में एन्कोडेड और हिस्सों में भेजा गया।

क्लाइंट फ़ाइल को स्लाइस में स्ट्रीम करता है, फिर टर्न के अंत को चिह्नित करने के लिए बफ़र कमिट करता है:

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)

यदि आपका सैंपल रेट या एन्कोडिंग session.update से मेल नहीं खाता, तो आपको साफ़ त्रुटि की जगह स्टैटिक या साइलेंस मिल सकता है। ऑडियो input_audio_buffer.append के जरिए जाता है, इसलिए यह प्रति संदेश नहीं बल्कि अवधि के हिसाब से बिल होता है।

वॉइस प्रतिक्रियाएँ प्राप्त करना

जब आप प्रतिक्रिया का अनुरोध करते हैं, तो ऑडियो response.output_audio.delta के रूप में आता है, ट्रांसक्रिप्ट response.output_audio_transcript.delta के रूप में आता है, और response.done टर्न बंद करता है।

क्लाइंट यह सब एक async लूप में इकट्ठा करता है:

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

ऑडियो डेल्टा डिकोड करें, क्रम में जोड़ें, और परिणाम को response.wav फाइल में लिखें। कॉलर के अपने शब्द कैप्चर करने के लिए, audio.input.transcription सेट करें और conversation.item.input_audio_transcription.completed पढ़ें।

अपॉइंटमेंट असिस्टेंट वर्कफ़्लो बनाना

अब ये हिस्से एक बातचीत बनते हैं: बुकिंग अनुरोध, स्पष्टीकरण सवाल, उपलब्धता जाँच, दिए गए स्लॉट, पुष्टि। टर्न्स के बीच संदर्भ ले जाने के लिए, प्रत्येक नया टर्न बातचीत आईडी के साथ फिर से कनेक्ट होता है और सत्र पुनरारंभ के लिए सहमति देता है।

वॉइस एजेंट में टूल कॉलिंग जोड़ना

क्लिनिक के लिए, एजेंट को समय का वादा करने से पहले उपलब्धता जाँचनी चाहिए। कस्टम टूल्स वह तरीका हैं जिससे मॉडल आपके कोड तक पहुँचता है: यह एक अनुरोध निकालता है, आपका एप्लिकेशन फ़ंक्शन चलाता है, और आप परिणाम वापस भेजते हैं।

टूल एक सादा फ़ंक्शन होता है और एक JSON स्कीमा, जो सत्र कॉन्फ़िग में जाता है। यहाँ 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"],
    },
}

लूप का एक तयशुदा आकार होता है। जब मॉडल टूल चाहता है, तो वह आर्ग्युमेंट्स के साथ response.function_call_arguments.done भेजता है। आप फ़ंक्शन चलाते हैं, एक function_call_output वापस करते हैं, और फिर response.create भेजते हैं ताकि एजेंट आगे बढ़ सके। वह अंतिम response.create छूट गया तो एजेंट चुप हो जाता है।

टूल कॉल राउंड ट्रिप समझाया गया। चित्र: लेखक।

इस तरह के कस्टम फ़ंक्शंस आपके कोड में चलते हैं। Streamlit डेमो उसी फाइल से तीन और रजिस्टर करता है: book_appointment, transfer_to_human, और end_call। बिल्ट-इन टूल्स, जैसे web search, X सर्च, collections सर्च, और रिमोट MCP टूल्स, xAI के सर्वरों पर चलते हैं।

टूल विफलताओं को संभालना

टूल विफल होते हैं, और सफलता मानने वाला वॉइस एजेंट ऐसा स्लॉट वादा कर सकता है जो है ही नहीं। मेरा ToolRegistry.execute कभी एक्सेप्शन नहीं उठाता: विफल लुकअप {"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)}

एक स्पष्ट त्रुटि स्थिति एजेंट को विफल टूल कॉल्स को सफलता मानने से रोकती है।

लागत ट्रैकिंग जोड़ना

किसी को सर्व करने से पहले, जान लें कि एक कॉल की लागत क्या है। ऑडियो $0.05 प्रति मिनट पर बिल होता है, जो आप भेजते हैं और जो आप प्राप्त करते हैं, दोनों गिने जाते हैं। टेक्स्ट इनपुट ईवेंट $0.004 प्रति इवेंट पर बिल होते हैं। function_call_output परिणाम और response.create ईवेंट बिल नहीं होते।

क्लाइंट चलते-चलते इसे ट्रैक करता है, इसलिए लागत किसी भी बिंदु पर पढ़ी जा सकती है:

@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 द्वारा प्रोविज़न किया गया नंबर $0.01 प्रति मिनट टेलीफोनी सरचार्ज जोड़ता है, जिसे हेल्पर तब लागू करता है जब आप telephony=True सेट करते हैं। xAI द्वारा होस्ट किए गए टूल अलग से बिल होते हैं: वेब सर्च और X सर्च लगभग $5 प्रति हज़ार कॉल्स, और फ़ाइल सर्च लगभग $2.50।

त्रुटियाँ और किनारे के मामले संभालना

अधिकांश विफलताएँ एक छोटी सूची में आती हैं:

• लापता या अमान्य API कुंजी हैंडशेक पर 401 लौटाती है, इसलिए पहले कुंजी जाँचें

• ब्लॉक की गई टीम 403 लौटाती है, और रेट लिमिट 429 लौटाती है, जिसे आप बैकऑफ़ के साथ पुनः आज़माते हैं

• खराब सत्र कॉन्फ़िग 400 लौटाता है, आमतौर पर किसी फ़ील्ड नाम में टाइपो

• असमर्थित ऑडियो फ़ॉर्मेट त्रुटि नहीं देता बल्कि स्टैटिक देता है, इसलिए सत्र रेट मिलाएँ

• टूल परिणाम के बाद response.create गायब होने पर एजेंट अटका रह जाता है

• डुप्लीकेट बुकिंग प्रयास वास्तविक समस्याएँ पैदा कर सकता है, इसलिए बिना सोचे-समझे रिट्राई न करें

किसी check_availability जैसी विफल रीड को पुनः आज़माना सुरक्षित है, लेकिन वास्तविक बुकिंग जैसी विफल राइट को रिट्राई करना कॉलर की डबल बुकिंग करा सकता है। कोई भी एक्शन जो डेटा बदलता है, उसे पहले idempotency जाँच चाहिए।

क्लाइंट ऐप्स के लिए इफीमरल टोकन उपयोग करना

अब तक सब मानता है कि कोड आपके सर्वर पर चलता है, जहाँ API कुंजी होनी चाहिए। यदि ब्राउज़र या मोबाइल ऐप सीधे कनेक्ट करता है, तो ephemeral tokens उपयोग करें।

आपका सर्वर POST https://api.x.ai/v1/realtime/client_secrets को अपनी कुंजी के साथ कॉल करता है, एक टोकन प्रतिक्रिया पाता है, और टोकन मान क्लाइंट को पास करता है। मेरी रन में, प्रतिक्रिया में value और 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()

ब्राउज़र कस्टम WebSocket हेडर सेट नहीं कर सकते, इसलिए टोकन sec-websocket-protocol हेडर में xai-client-secret. प्रीफिक्स के साथ जाता है।

वर्कफ़्लो को FastAPI एन्डपॉइंट में बदलना

एक एन्डपॉइंट फ्रंटएंड या किसी अन्य सेवा को वर्कफ़्लो कॉल करने देता है। रूट एक Pydantic मॉडल से अनुरोध बॉडी को मान्य करता है, टाइप किया हुआ संदेश या ऑडियो पाथ लेता है, और ट्रांसक्रिप्ट, प्रतिक्रिया ऑडियो, टूल लॉग, लैटेंसी, और अनुमानित लागत लौटाता है।

@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 से चलाएँ और http://localhost:8000/docs खोलें। XAI_API_KEY सर्वर एन्वायरमेंट से पढ़ें और इसे कभी भी अनुरोध बॉडी से स्वीकार न करें।