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 Index पर 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। एलियस प्रोटोटाइपिंग के दौरान सुविधाजनक है और उसके आगे के लिए स्थिर नहीं।

जब मैंने 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 उसी क्लाइंट को एक लाइव ब्राउज़र कॉल के पीछे रखता है, जिस पर मैं टेस्टिंग सेक्शन के बाद लौटूंगा

शिक्षण पथ टर्मिनल से चलता है। डेमो माइक्रोफोन जोड़ता है।

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

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

SpaceXAI के क्विक-स्टार्ट उदाहरण रॉ websockets पैकेज का उपयोग करते हैं, किसी समर्पित SDK का नहीं, और हम भी वही करेंगे। डॉक्स ने कहीं आवश्यक 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 (टेलिफ़ोनी के लिए 8 kHz पर G.711), या audio/opus (24 kHz) हो सकता है। ट्रांसपोर्ट वह है जिससे ये बाइट्स वायर पर जाती हैं:

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

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

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

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 पेज है जो ब्राउज़र में एक लाइव कॉल रखता है: माइक्रोफोन WebRTC के जरिए उसी WebSocket में स्ट्रीम होता है, एजेंट की आवाज़ वापस स्ट्रीम होती है, और सॉकेट पूरे समय खुला रहता है।

streamlit run app_streamlit.py