Grok Voice Think Fast 2.0 API ट्यूटोरियल: Python में रियल-टाइम वॉइस एजेंट बनाएं

SpaceXAI का Grok Voice Think Fast 2.0 एक स्पीच-टू-स्पीच मॉडल है। आप इसे WebSocket के ज़रिए ऑडियो भेजते हैं और यह ऑडियो वापस भेजता है, और बीच में यह तर्क कर सकता है और बोलते हुए भी बना रह सकता है जबकि वह फ़ंक्शन कॉल जिसे उसने चुन लिया है, पहले से चल रही होती है। कोई अलग स्पीच-टू-टेक्स्ट स्टेप नहीं, कोई अलग टेक्स्ट-टू-स्पीच स्टेप नहीं।

SpaceXAI ने 29 जुलाई, 2026 को Think Fast 2.0 की घोषणा की: पहले ऑडियो तक तेज़ पहुंच, अधिक स्थिर फुल-डुप्लेक्स व्यवहार (कड़क बारी-बारी लेने के बजाय बोलते समय सुनना), और टर्न के शुरुआती हिस्से में ट्रिगर होने वाली टूल कॉल्स। बेंचमार्क पर बात संक्षेप में रखूंगा, क्योंकि ट्यूटोरियल के लिए मायने रखता है कि आपके कोड में क्या बदलता है।

हम एक ऑनलाइन स्टोर के लिए कस्टमर-सपोर्ट वॉइस एजेंट बनाएंगे। कॉलर ऑर्डर के बारे में पूछ सकता है, डिलीवरी निर्देश बदल सकता है, कैंसल कर सकता है, एजेंट को बीच में रोक सकता है, और कनेक्शन टूटने के बाद बातचीत फिर से आगे बढ़ा सकता है। यह API वाला रास्ता है, नो-कोड Voice Agent Builder नहीं, जिसे हमारा Grok Voice Agent Builder ट्यूटोरियल समझाता है। कंसोल-फर्स्ट वर्शन के लिए वहीं से शुरू करें।

Grok Voice Think Fast 2.0 क्या है?

Grok Voice Think Fast 2.0 SpaceXAI का Speech to Speech API के लिए नवीनतम मॉडल है, वही प्रोडक्ट नाम जिसे अधिकांश लोग बस Grok Voice कहते हैं। अगर आप कंपनी को अब भी xAI के रूप में सोचते हैं, वही टीम: इसे 6 जुलाई, 2026 को SpaceX में समाहित करके SpaceXAI रीब्रांड किया गया। API ने रीब्रांड फॉलो नहीं किया, इसलिए नीचे हर आइडेंटिफायर अब भी xai कहता है, XAI_API_KEY वेरिएबल से लेकर api.x.ai होस्ट तक।

पारंपरिक वॉइस स्टैक तीन सेवाओं को जोड़ता है: स्पीच-टू-टेक्स्ट, एक लैंग्वेज मॉडल, फिर टेक्स्ट-टू-स्पीच—और हर होप लेटेंसी जोड़ता है और संदर्भ खोने की जगह बनाता है। Think Fast 2.0 इसे एक मॉडल में समेट देता है जो ऑडियो या टेक्स्ट इन में लेता है और उसी कनेक्शन पर ऑडियो या टेक्स्ट आउट देता है।

स्पीच-टू-स्पीच WebSocket बनाम तीन-सेवा पाइपलाइन आर्किटेक्चर। चित्र: लेखक।

ऐसे एजेंट के लिए जो केवल बात नहीं करता बल्कि काम भी करता है, असल मायने यह रखता है कि तर्क और वाणी समानांतर में चलें। SpaceXAI कहता है कि टूल कॉल्स "अक्सर" एजेंट के अपना पहला वाक्य पूरा करने से पहले ही चलने लगती हैं, और यह शब्द सचमुच अहम है।

SpaceXAI द्वारा उद्धृत Artificial Analysis के बेंचमार्क्स पर, Think Fast 2.0 ने Speech to Speech इंडेक्स पर 1.0 के 75.7% के मुकाबले 82.9% स्कोर किया, और पहले ऑडियो तक का समय 1.25 सेकंड से घटाकर 0.70 सेकंड कर दिया। किसी जनरल बेंचमार्क पर वेंडर के नंबर आपके कॉलर फ़्लो के बारे में एक परिकल्पना होते हैं, टेस्ट प्लान नहीं।

आप तीन मॉडल स्ट्रिंग्स देखेंगे: grok-voice-latest, grok-voice-think-fast-2.0, और grok-voice-think-fast-1.0। उर्फ़ (alias) प्रोटोटाइपिंग में सुविधाजनक है और उससे आगे के लिए स्थिर नहीं।

4 अगस्त, 2026 को मेरे टेस्ट में, grok-voice-latest अब भी grok-voice-think-fast-1.0 पर रिज़ॉल्व हो रहा था, जबकि SpaceXAI के रिलीज़ नोट्स अगले दिन Think Fast 2.0 पर स्विच का शेड्यूल दे रहे थे। यह स्विच मॉडल जितना ही कीमत का बदलाव भी है—1.0 के $0.05 प्रति मिनट के मुकाबले $0.08 प्रति मिनट—तो अनपिन्ड उर्फ़ आपके कोड की एक लाइन बदले बिना महंगा हो जाता है। किसी भी डिप्लॉयमेंट में वर्शन वाली स्ट्रिंग पिन करें।

हम क्या बनाएंगे

एजेंट सपोर्ट लाइन पर आने वाले सवाल कवर करता है: ऑर्डर देखना, जब कॉलर के पास नंबर न हो तो ईमेल से ढूंढना, डिलीवरी निर्देश बदलना, कैंसल करना, टिकट खोलना या चेक करना, और कॉल इंसान को सौंपना। बीच में इंटरप्शन और कनेक्शन ड्रॉप भी आएंगे।

यह एक स्क्रिप्ट की जगह कुछ छोटे फ़ाइलों का सेट है, क्योंकि हर हिस्सा अलग काम करता है और आप उन्हें अलग-अलग टेस्ट करना चाहेंगे। लेआउट यह है:

• config.py API कुंजी लोड करता है और मॉडल स्ट्रिंग, सैंपल रेट, और एंडपॉइंट URLs रखता है

• voice_client.py WebSocket को रैप करता है, बिलिंग ट्रैक करता है, और send/receive हेल्पर्स एक्सपोज़ करता है

• tools.py ऑर्डर फ़ंक्शंस डिफ़ाइन करता है और एक छोटा इन-मेमाेरी ऑर्डर स्टोर रखता है जो असली डेटाबेस की जगह खड़ा है

• assistant.py सिस्टम प्रॉम्प्ट, सेशन कॉन्फ़िग, और सब कुछ जोड़ने वाला इवेंट लूप रखता है

• token_server.py एक छोटा FastAPI एंडपॉइंट है जो अल्पकालिक टोकन बनाता है

• app_streamlit.py वही क्लाइंट लाइव ब्राउज़र कॉल के पीछे रखता है, जिस पर मैं टेस्टिंग सेक्शन के बाद वापस आऊंगा

पाठ्यक्रम का रास्ता टर्मिनल से चलता है। डेमो माइक्रोफ़ोन जोड़ता है।

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

आपको API कुंजी के साथ SpaceXAI खाता चाहिए, एक फंडेड बिलिंग व्यवस्था (कोई स्थायी फ्री टियर नहीं है, और नए खाते के प्रोमोशनल क्रेडिट आपको नहीं चलाएंगे), और asyncio और WebSockets के साथ इतना आराम कि आप await की लाइन-बाय-लाइन व्याख्या के बिना साथ चल सकें।

SpaceXAI के क्विक-स्टार्ट उदाहरण समर्पित SDK की जगह कच्चे websockets पैकेज का उपयोग करते हैं, और हम भी ऐसा ही करेंगे। डॉक्स ने कहीं आवश्यक Python वर्शन नहीं लिखा। मैंने 3.11 पर टेस्ट किया।

API कुंजी सर्वर पर रखें। अगर कोई ब्राउज़र या मोबाइल ऐप सीधे Voice API से बात करता है, तो उसे आपकी असली कुंजी की जगह अल्पकालिक टोकन मिलता है, जिसे नीचे सुरक्षा सेक्शन में कवर किया गया है।

प्रोजेक्ट सेट करना

नीचे दी गई हर फ़ाइल प्रोजेक्ट रेपो में है, इसलिए आप स्निपेट कॉपी करने की बजाय इसे क्लोन कर सकते हैं:

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 रियलटाइम कनेक्शन संभालता है और python-dotenv आपकी कुंजी पढ़ता है। बाकी चीज़ें टोकन एंडपॉइंट और ब्राउज़र डेमो के लिए हैं। अपनी कुंजी .env में रखें:

XAI_API_KEY=xai-your-key-here

यही सेटअप का ज़्यादातर हिस्सा है। दिलचस्प भाग कनेक्शन है।

Grok Voice Realtime API को समझना

Grok Voice प्रोडक्ट का नाम है। जिस चीज़ के ख़िलाफ़ आप वास्तव में कोड लिखते हैं, वह wss://api.x.ai/v1/realtime पर एक WebSocket एंडपॉइंट है, और पूरी बातचीत उसी एक सॉकेट पर JSON ईवेंट्स की स्ट्रीम के रूप में होती है।

इवेंट लाइफ़साइकिल

कनेक्शन एक तय ढंग से चलता है: सर्वर session.created और conversation.created कनेक्ट होते ही भेज देता है, आप session.update भेजकर वॉइस और टूल्स कॉन्फ़िगर करते हैं, सर्वर session.updated से पुष्टि करता है, और वहाँ से आप बातचीत के आइटम बनाते हैं और रिस्पॉन्स मांगते हैं। मैंने इसे लाइव कुंजी पर चलाया और क्रम बिल्कुल डॉक्स जैसा ही था।

• session.update (क्लाइंट) वॉइस, निर्देश, टूल्स और ऑडियो फ़ॉर्मैट कॉन्फ़िगर करता है

• conversation.item.create (क्लाइंट) कोई यूज़र मैसेज, असिस्टेंट मैसेज, या टूल रिज़ल्ट जोड़ता है

• response.create (क्लाइंट) मॉडल से बोलने के लिए कहता है; सर्वर VAD यह आपके लिए अपने आप भेज देता है

• response.output_audio.delta और response.output_audio_transcript.delta (सर्वर) जवाब के जेनरेट होते ही उसे स्ट्रीम करते हैं

• response.done (सर्वर) टर्न को बंद करता है

दो बातें लोगों को उलझाती हैं। ऊपर लिंक किए Speech to Speech डॉक्स पेज में सेशन रेज़म्पशन के दौरान conversation.item.created इवेंट का ज़िक्र है, लेकिन कैनॉनिकल इवेंट रेफरेंस केवल conversation.item.added को सूचीबद्ध करता है, और मेरे हर टेस्ट में वही आया, तो उसी पर कोड लिखें। आपको कुछ सेकंड में एक अनडॉक्यूमेंटेड ping इवेंट भी दिखेगा; यह केवल इसलिए बताया है ताकि आप इसे एरर न समझ लें।

ऑडियो फ़ॉर्मैट्स और ट्रांसपोर्ट

कोडेक और ट्रांसपोर्ट अलग-अलग विकल्प हैं। कोडेक, जिसे audio.input.format और audio.output.format के तहत सेट किया जाता है, यह हो सकता है audio/pcm (Linear16, डिफ़ॉल्ट 24000 Hz), audio/pcmu या audio/pcma (G.711, 8 kHz, टेलीफोनी के लिए), या audio/opus (24 kHz)। ट्रांसपोर्ट वह है जिससे ये बाइट्स तार पर जाती हैं:

• json (डिफ़ॉल्ट) ऑडियो को base64 टेक्स्ट के रूप में input_audio_buffer.append और response.output_audio.delta में भेजता है—लॉग और डिबग में आसान

• binary WebSocket बाइनरी फ़्रेम्स के रूप में कच्चे कोडेक बाइट्स भेजता है, base64 ओवरहेड को छोड़ देता है, जिसकी कीमत है कि रिसीव लूप को मैसेज प्रकार पर ब्रांच करना पड़ता है

JSON से शुरू करें। डॉक्स के हर उदाहरण में यही है, इसे देखना आसान है, और base64 ओवरहेड सपोर्ट-एजेंट बिल्ड में बॉटलनेक नहीं होता। केवल तभी बाइनरी पर जाएं जब आप माप लें कि वजह है।

OpenAI Realtime API से संगतता

अगर आपने कभी OpenAI की Realtime API नहीं छुई, तो आगे बढ़ें। बाकी सबके लिए, Speech to Speech API OpenAI Realtime API के इतना करीब है कि अधिकतर क्लाइंट कोड बेस URL और कुंजी बदलकर पोर्ट हो जाता है, लेकिन यह एक परफ़ेक्ट ड्रॉप-इन नहीं है।

यहाँ ट्रांसक्रिप्ट्स conversation.item.input_audio_transcription.updated के रूप में आते हैं, OpenAI के delta की जगह; OpenAI के कुछ इवेंट्स सपोर्टेड नहीं हैं, और SpaceXAI अपनी एक्सटेंशन्स जोड़ता है: force_message स्क्रिप्टेड डिस्क्लोज़र लाइन के लिए, resumption री-कनेक्ट्स के लिए, और replace ब्रांड नामों के गलत उच्चारण को टेक्स्ट-टू-स्पीच से पहले ठीक करने के लिए।

रियल-टाइम वॉइस एजेंट बनाना

प्रोटोकॉल काफी हुआ। यह रहा क्लाइंट जो इससे बात करता है।

कनेक्ट करना और सेशन कॉन्फ़िगर करना

कनेक्शन बेयरर टोकन और मॉडल क्वेरी पैरामीटर के साथ खुलता है, और जो पहला मैसेज आप भेजते हैं, वह एजेंट के व्यवहार की हर चीज़ कॉन्फ़िगर करता है:

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 सिस्टम प्रॉम्प्ट है, और यह मॉडल छोटे प्रॉम्प्ट चाहता है। SpaceXAI के माइग्रेशन नोट्स कहते हैं कि पुराने GPT-युग के वॉइस मॉडलों के लिए लिखे प्रॉम्प्ट को ज्यों-का-त्यों पोर्ट करने की बजाय सरल करें। मेरा एजेंट को बताता है कि जवाब छोटे रखें, एक बार में एक ही सवाल पूछें, और कोई भी लिखने की क्रिया करने से पहले उसे पढ़कर सुनाएं। बोला गया कन्फ़र्मेशन UX की शालीनता है, सुरक्षा नियंत्रण नहीं। आपका एप्लीकेशन अब भी खुद लिखी जाने वाली क्रिया पर ऑथराइज़ेशन लागू करता है।

एक बात जिसने मुझे चौंकाया: कोई अनपहचानी मॉडल स्ट्रिंग कनेक्ट टाइम पर एरर नहीं उठाती, चुपचाप grok-voice-think-fast-1.0 पर फॉलबैक कर जाती है। टाइपो की वजह से पेड रिक्वेस्ट को बिना बताए डाउनग्रेड करना अजीब डिफ़ॉल्ट है। स्टार्टअप पर एक बार session.created के session.model फ़ील्ड को लॉग करें और जाँचें कि आपको वही मिला जो आपने मांगा।

कनेक्ट करने के बाद session.created दिखाता टर्मिनल आउटपुट। चित्र: लेखक।

यूज़र ऑडियो स्ट्रीम करना

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

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

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

• threshold (0.1 से 0.9, डिफ़ॉल्ट 0.85): ऑडियो कितना तेज़ होना चाहिए ताकि वह स्पीच माने; शोरगुल वाले कमरे में इसे बढ़ाएं, शांत बोलने वालों के मिस होने पर घटाएं

• silence_duration_ms: कॉलर कितनी देर चुप रहे तब सर्वर उसका टर्न खत्म करे; बहुत कम होने पर लोग आधे विचार में कट जाते हैं, बहुत अधिक होने पर सुस्ती लगती है

• prefix_padding_ms (डिफ़ॉल्ट 333): स्पीच डिटेक्ट होने से ठीक पहले का ऑडियो स्लाइस संभालकर रखता है ताकि पहला अक्षर कट न जाए

अगर कॉलर सोचने के लिए रुकते हुए कटते जा रहे हों, तो पहले silence_duration_ms ट्यून करें। मैं बाकी दो को छूने से पहले इसी को पकड़ता हूँ।

रिस्पॉन्स प्राप्त करना और चलाना

ऑडियो छोटे टुकड़ों में response.output_audio.delta के रूप में आता है, और स्ट्रीमिंग का मतलब है कि आप हर टुकड़ा आते ही चला दें, 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)

प्रोडक्शन में भी ट्रांसक्रिप्ट साथ रखें। यह वह सबसे सस्ता डिबगिंग टूल है जो आपके पास तब होता है जब कोई कॉलर कहता है कि एजेंट ने "कुछ अजीब कहा"।

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

जो वॉइस एजेंट केवल बात करता है, वह माइक्रोफ़ोन वाला चैटबॉट है।

ऑर्डर टूल्स बनाना

हर टूल एक JSON स्कीमा और हमारी तरफ़ एक सादा Python फ़ंक्शन होता है। मॉडल कभी डेटाबेस को नहीं छूता, उसे केवल वही दिखता है जो हमारा फ़ंक्शन लौटाता है।

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
]

पढ़ने वाले ऑपरेशन जैसे check_order_status टाइमआउट होने पर रीट्राई के लिए सुरक्षित हैं। लिखने वाले नहीं: update_delivery_instructions को संदिग्ध टाइमआउट के बाद रीट्राई करना वही बदलाव दो बार लागू कर सकता है। प्रॉम्प्ट में कन्फ़र्मेशन लाइन इसे नहीं रोकती, इसलिए लिखने वालों को आइडेमपोटेंसी की या डुप्लिकेट चेक दें।

इंकार भी फ़ंक्शन में रखें। cancel_order शिप हो चुके ऑर्डर को कैंसल करने की बजाय एक कारण और विकल्प लौटाता है, क्योंकि "कभी शिप्ड ऑर्डर कैंसल न करें" कहने वाला प्रॉम्प्ट एक सुझाव है, और इंकार करने वाला फ़ंक्शन सुझाव नहीं।

टूल-कॉलिंग लूप संभालना

चार स्टेप, और क्रम जितना दिखता है उससे ज़्यादा मायने रखता है। मॉडल response.function_call_arguments.done भेजता है, आपका कोड फ़ंक्शन चलाता है, आप नतीजा function_call_output आइटम के रूप में वापस भेजते हैं, और तभी आप मॉडल से आगे जारी रखने को कहते हैं।

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

अगर मॉडल को एक रिक्वेस्ट के लिए एक से ज़्यादा टूल चाहिए, तो वह कोई ऑडियो चलने से पहले कई function_call_arguments.done इवेंट्स फायर करता है। उनमें से हर एक को पूरा करें और हर नतीजा भेजें, उसके बाद ही एक response.create भेजें। इसे जल्दी भेजने पर मॉडल उन कॉल्स के संदर्भ के बिना जवाब दे देता है जो अब भी चल रही हैं।

यहाँ एक पेच है जिसे SpaceXAI डॉक्यूमेंट करता है और मैं फिर भी पहली बार में फँस गया: आपका टूल रिज़ल्ट जाते ही response.create भेज देना उस शुरुआती वाक्य के साथ ओवरलैप कर सकता है जो एजेंट अभी भी बोल रहा है। एक रन में उसने "मैं अभी ऑर्डर ORD-1042 की स्थिति चेक करता हूँ" से ओपन किया और वाक्य के बीच में टूल कॉल कर दी, तो तुरंत रिस्पॉन्स एजेंट की अपनी शुरुआत पर ही बोल पड़ेगा।

वर्तमान टर्न का ऑडियो खत्म होने दें, और बीच में छोटा "सोच रहा है" स्टेट दिखाएँ।

रिस्पॉन्स जारी रखने से पहले टूल कॉल फ़्लो। चित्र: लेखक।

इंटरप्शन और बातचीत की स्थिति मैनेज करना

यहाँ दो अलग समस्याएँ हैं। कॉलर एजेंट के जवाब के बीच में बोल देता है, और एक WebSocket ड्रॉप हो जाता है जिसे फिर से उठाना पड़ता है।

प्राकृतिक इंटरप्शन का समर्थन

जब server_vad ऑन हो, तो बार्ज-इन सर्वर-साइड अपने आप होता है: जैसे ही वह कॉलर को फिर बोलते हुए डिटेक्ट करता है, वह input_audio_buffer.speech_started सिग्नल करता है और पुराने रिस्पॉन्स का जेनरेशन रोक देता है। आपका काम उस हैंडशेक का क्लाइंट वाला आधा है—क्यू में पड़ी ऑडियो साफ़ करना ताकि एजेंट चुप हो जाए, न कि वह वाक्य पूरा करे जिसे अब कोई सुनना नहीं चाहता।

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

मैनुअल, नॉन-VAD सेशन्स में, response.cancel रिक्वेस्ट पर वही काम करता है। conversation.item.truncate भी है ताकि असिस्टेंट आइटम को उतना ही रखें जितना वास्तविक में सुना गया। डॉक्स इसकी मौजूदगी की पुष्टि करते हैं, लेकिन लाइव बार्ज-इन के दौरान इसे कब फायर करना है, यह नहीं, तो टाइमिंग खुद टेस्ट करें।

मैंने इसे बीच-जवाब डिलीवरी-इंस्ट्रक्शन बदलाव के साथ टेस्ट किया: रिक्वेस्ट शुरू करें, एजेंट के कन्फ़र्मेशन के आधे में अलग पता बोलकर रोक दें। मायने यह रखता है कि एजेंट पुराना चुपचाप पूरा करने की बजाय सुधरे हुए निर्देश ही लागू करता है या नहीं—न कि ऑडियो रुका या नहीं। ऑर्डर रिकॉर्ड के खिलाफ़ सत्यापित करें, साइलेंस के नहीं। सबसे अंत का ब्राउज़र डेमो आप यह सुन पाएंगे।

डिस्कनेक्टेड सेशन को फिर से शुरू करना

सेशन रेज़म्पशन ऑप्ट-इन है और यह मेमोरी नहीं है। session.update पर resumption.enabled: true सेट करें, conversation.created इवेंट से ID पकड़ें, और अगर सॉकेट ड्रॉप हो, तो URL में ?conversation_id=<id> के साथ फिर कनेक्ट करें और नई कनेक्शन पर फिर से ऑप्ट-इन करें।

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

कैश किए गए टर्न, ट्रांसक्रिप्ट्स, टूल कॉल्स और टूल रिज़ल्ट्स आपके अगले सवाल से पहले रीप्ले हो जाते हैं, और 30 मिनट की निष्क्रियता के बाद कैश गायब हो जाता है। मैंने इसे किसी ऑर्डर के बारे में पूछकर, कनेक्शन गिराकर, और बिना दोहराए फॉलो-अप के लिए फिर से कनेक्ट होकर टेस्ट किया; एजेंट ने सही ETA फिर पकड़ लिया।

एक अनडॉक्यूमेंटेड बात: रीप्ले तुरंत नहीं आता, इसलिए सॉकेट खुलते ही दागा गया सवाल उससे आगे निकल सकता है और पिछले टर्न की मेमोरी के बिना लौट सकता है। रेज़म्पशन को दोष देने से पहले एक सेकंड दें।

पुनरारंभ किए सेशन का टर्मिनल लॉग। चित्र: लेखक।

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

एजेंट को सुरक्षित रखना और मॉनिटर करना

कभी भी स्थायी API कुंजी को ब्राउज़र या मोबाइल कोड में न रखें। अगर क्लाइंट सीधे कनेक्ट होता है, सर्वर से एक कम-अवधि का टोकन बनाएं:

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

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

सर्वर टोकन बनाता है, ब्राउज़र कॉल में जुड़ता है। चित्र: लेखक।

बिलिंग दो मीटर पर चलती है। भेजे या प्राप्त ऑडियो पर वही $0.08 प्रति मिनट लगता है जिसका मैंने ज़िक्र किया—यानी $4.80 प्रति घंटा—और हर वह conversation.item.create जो ऑडियो नहीं है और function_call_output नहीं है, उस पर $0.004 फ्लैट लगता है। response.create पर कोई बिलिंग नहीं है। हर response.done के साथ एक usage ऑब्जेक्ट आता है, जिसने मेरे टेस्ट में output_audio_seconds के साथ अलग से billable_audio_seconds रिपोर्ट किया। अनुमान से नहीं, इन्हीं से बिलिंग करें।

Speech to Speech API पर डॉक्यूमेंटेड लिमिट्स हैं: प्रति टीम 10 समवर्ती सेशन और 120-मिनट सेशन कैप, दोनों us-east-1 में। क्षमता की योजना Voice Agent API के नंबरों से न बनाएं, जो अलग हैं।

प्रायवेसी पर, सटीक रहें। SpaceXAI के सिक्योरिटी FAQ के अनुसार API रिक्वेस्ट्स और रिस्पॉन्स 30 दिनों तक एन्क्रिप्टेड रखे जाते हैं दुरुपयोग मॉनिटरिंग के लिए और बिना अनुमति ट्रेनिंग में इस्तेमाल नहीं होते, और टीमें Zero Data Retention ऑन कर सकती हैं—हालाँकि ZDR पर्सिस्टेड वॉइस-एजेंट बातचीत इतिहास हटा देता है और इसलिए रेज़म्पशन के साथ काम नहीं करता।

अगर आप बता रहे हैं कि कॉल रिकॉर्ड की जाती है या AI-हैंडल्ड है, तो इसके लिए वही force_message एक्सटेंशन है जिसका मैंने पहले ज़िक्र किया। यह लाइन बिल्कुल लिखी हुई तरह चलती है, न कि जैसा मॉडल इसे पैराफ्रेज़ करे।

वॉइस एजेंट का परीक्षण

WebSocket हैंडशेक पर 200 स्टेटस आपको यह नहीं बताता कि एजेंट ने सही काम किया या नहीं। कनेक्शन नहीं, नतीजे को टेस्ट करें।

• साफ़-सुथरा ऑर्डर लुकअप—सिर्फ़ रिस्पॉन्स आने से नहीं, रिकॉर्ड के खिलाफ़ बोले गए उत्तर की जाँच
• बीच में रोका गया जवाब—यह पक्का करना कि प्लेबैक रुकता है और एजेंट नई रिक्वेस्ट को संबोधित करता है
• कन्फ़र्मेशन मांगने वाला डिलीवरी अपडेट—ऑर्डर रिकॉर्ड के खिलाफ़ जाँचा हुआ
• इंकार—जैसे शिप्ड ऑर्डर कैंसल करना—जहाँ एजेंट को नियम समझाने हैं, माफ़ी नहीं मांगनी
• अज्ञात ऑर्डर नंबर—पक्का करें कि एजेंट यह कहता है, स्टेटस गढ़ता नहीं
• एसा टूल जो एरर लौटाए—जाँचे कि एजेंट इसे बोलता है, अटकता नहीं
• रीकनेक्शन और रेज़म्पशन—उस रीप्ले विंडो सहित जिसका मैंने ऊपर सामना किया
• शोरगुल वाला ऑडियो, तेज़ बोलना, और कॉलर जो नंबर और पते स्पेल करता है

इनमें से ज़्यादातर मैंने लिखते वक्त लाइव कुंजी पर चलाए। दिलचस्प फेल्योर एरर नहीं, बिहेवियरल थे: ऊपर वाला रेज़म्पशन टाइमिंग, और आउट-ऑफ-रेंज VAD थ्रेशहोल्ड का स्वीकार हो जाना न कि रिजेक्ट—वही तरह की बातें जो अगर आप केवल हैप्पी पाथ टेस्ट करते हैं तो चुपचाप टूटी हुई शिप हो जाती हैं। एक बहुभाषी टेस्ट भी जोड़ें, और भाषाओं के नामकरण में एक पेच के लिए FAQs देखें।

इनमें से दो को आप टाइप करके टेस्ट नहीं कर सकते। app_streamlit.py एक Streamlit पेज है जो ब्राउज़र में लाइव कॉल रखता है: माइक्रोफ़ोन उसी WebSocket में WebRTC के ज़रिए स्ट्रीम होता है, एजेंट की आवाज़ वापस स्ट्रीम होती है, और सॉकेट पूरे समय खुला रहता है।

streamlit run app_streamlit.py