संक्षेप में

• Gemini 3.8 Live एक WebSocket पर दोनों दिशाओं में ऑडियो स्ट्रीम करता है, इसलिए आप ऐसा वॉयस असिस्टेंट बना सकते हैं जो बोलते समय सुनता भी रहे और इंटरप्शन संभाल सके।

• यह ट्यूटोरियल Python में चार asyncio वर्कर्स (माइक रिकॉर्डर, ऑडियो सेंडर, रिसीवर, स्पीकर) और दो क्यूज़ से जोड़कर इसे बनाता है।

• बर्ज-इन तब काम करता है जब Gemini interrupted भेजे, तो लोकल प्लेबैक क्यू को फ्लश कर दें।

• एक टूल (लाइव वेदर लुकअप) जोड़ने से दोनों मॉडलों में फर्क दिखता है: स्टैंडर्ड मॉडल टूल्स के चलते समय चुप रहता है, जबकि Extended Thinking बोलता रहता है।

• Extended Thinking के साथ, turn_complete के बजाय interaction_status == "IDLE" ट्रैक करें, और टूल कॉल्स को बैकग्राउंड टास्क के रूप में चलाएँ ताकि रिसीव लूप कभी ब्लॉक न हो।

Gemini 3.8 Live खास क्यों है?

Google का Gemini 3.8 Live एक नैटिव स्पीच-टू-स्पीच मॉडल है, जिसे खास तौर पर रियल-टाइम स्ट्रीमिंग और इंटरैक्टिव ऑडियो ऐप्लिकेशंस के लिए बनाया गया है। Gemini 3.8 Live मल्टीमॉडल इनपुट्स को सीधे एक पर्सिस्टेंट WebSocket कनेक्शन पर प्रोसेस करता है। 

यह द्विदिशीय स्ट्रीमिंग क्षमता डेवलपर्स को ऐसे फुल-डुप्लेक्स कन्वर्सेशनल एजेंट बनाने देती है जो एक साथ सुन और बोल सकें, और नैचुरल यूज़र इंटरप्शन तथा रियल-टाइम ऑडियो ट्रांसक्रिप्शन जैसी सुविधाएँ सपोर्ट करें।

एप्लिकेशन डेवलपमेंट के लिए, Gemini 3.8 Live असिंक्रोनस टूल कॉलिंग और बैकग्राउंड रीजनिंग पेश करता है, जिससे एजेंट बाहरी फ़ंक्शन कॉल्स चला सकते हैं या डेटा प्राप्त कर सकते हैं—साथ ही यूज़र से सक्रिय संवाद बनाए रखते हैं।

इसकी विशेषताओं, बेंचमार्क्स और कीमत का विस्तृत विवरण पाने के लिए हमारे Gemini 3.8 Live गाइड को देखें।

लाइव वॉयस असिस्टेंट कैसे काम करता है: 4 वर्कर्स और 2 क्यूज़

कोड में जाने से पहले, समझते हैं कि पर्दे के पीछे रियल-टाइम वॉयस असिस्टेंट कैसे चलता है।

स्टैंडर्ड Python स्क्रिप्ट्स में कोड लाइन-बाय-लाइन चलता है: फ़ंक्शन A खत्म, फिर फ़ंक्शन B चलता है। लेकिन लाइव वॉयस बातचीत में इंतज़ार करना नहीं चलता:

• जब हम बोल रहे हों, प्रोग्राम को आपकी आवाज़ Gemini तक रियल-टाइम में स्ट्रीम करनी चाहिए।
• जब Gemini जवाब दे रहा हो, प्रोग्राम को आते ही ऑडियो चंक्स स्पीकर पर प्ले करने चाहिए।
• सबसे ज़रूरी, Gemini के बोलते समय भी प्रोग्राम को सुनते रहना चाहिए ताकि हम बीच में टोक सकें (बर्ज-इन)।

इसे फ्रीज़ हुए बिना हासिल करने के लिए, हम Python के asyncio का उपयोग करते हैं ताकि 4 हल्के बैकग्राउंड टास्क ("वर्कर्स") चलें, जो दो asyncio.Queue बफ़र्स के ज़रिये संवाद करते हैं (इन्हें कन्वेयर बेल्ट समझें):

1. इनबाउंड कन्वेयर बेल्ट (input_queue):

• audio_recorder(): लगातार माइक्रोफोन सुनता है और ऑडियो स्लाइस बेल्ट पर डालता है।

• send_audio_loop(): बेल्ट से ऑडियो स्लाइस उठाकर Gemini को स्ट्रीम करता है।

2. आउटबाउंड कन्वेयर बेल्ट (audio_queue):

• receive_loop(): Gemini को सुनता है। टेक्स्ट आते ही प्रिंट करता है; स्पीच आते ही ऑडियो चंक्स बेल्ट पर डालता है।

• audio_player(): बेल्ट से ऑडियो चंक्स उठाकर स्पीकर या हेडफ़ोन पर चलाता है।

क्योंकि हर वर्कर सिर्फ अपने छोटे से काम पर ध्यान देता है, चारों Python के इवेंट लूप पर बिना एक-दूसरे में दखल दिए साथ-साथ चल सकते हैं।

इस ट्यूटोरियल में इस्तेमाल पूरा कोड इस GitHub रेपो में उपलब्ध है।

Gemini API Key कैसे जेनरेट और सेट अप करें

Gemini API का उपयोग करने के लिए, हमें एक API key बनानी और सेट करनी होगी ताकि हमारा कोड API से संवाद कर सके।

इसे करने का सबसे आसान तरीका:

• Google के AI Studio API key पेज पर जाएँ और लॉगिन करें।

• ऊपर-दाएँ कोने में Create API key बटन पर क्लिक करें।

• API key को उसी फ़ोल्डर में .env नामक फ़ाइल में कॉपी करें जहाँ Python कोड होगा, इस फ़ॉर्मेट में:

GEMINI_API_KEY=replace_with_api_key

ध्यान दें कि API का उपयोग प्रायः लागत लगाता है। फ्री टियर में दोनों Gemini 3.8 Live मॉडलों तक सीमित एक्सेस शामिल है, लेकिन फ्री-टियर डेटा Google के प्रोडक्ट्स को बेहतर बनाने में उपयोग होता है। प्रोडक्शन उपयोग या उच्च रेट लिमिट्स के लिए, Google के AI Studio बिलिंग पेज पर पेमेंट मेथड कॉन्फ़िगर होना चाहिए।

Gemini 3.8 Live के साथ वॉयस असिस्टेंट आर्किटेक्चर कैसे इम्प्लीमेंट करें

ये स्टेप्स एक लोकल Jupyter नोटबुक में चलाने के लिए डिज़ाइन किए गए हैं, जहाँ हर कोड स्निपेट एक नोटबुक सेल से मेल खाता है। हमें माइक्रोफोन और स्पीकर की एक्सेस चाहिए, इसलिए यह Google Colab जैसे ऑनलाइन नोटबुक पर बॉक्स से बाहर काम नहीं करेगा।

स्टेप 1: एनवायरनमेंट सेटअप और इम्पोर्ट्स

पहले यह सुनिश्चित करें कि ज़रूरी पैकेज इंस्टॉल हैं:

pip install google-genai sounddevice python-dotenv

ये पैकेज क्या करते हैं, इसका ब्योरा:

• google-genai: Gemini मॉडलों से इंटरैक्ट करने के लिए आधिकारिक Google पैकेज।

• sounddevice: ऑडियो हार्डवेयर हैंडल करने, माइक्रोफोन से रिकॉर्ड करने और स्पीकर के ज़रिये प्लेबैक के लिए।

• python-dotenv: हमारी .env फ़ाइल से Gemini API key लोड करने के लिए यूटिलिटी पैकेज।

अब हम एनवायरनमेंट वैरिएबल्स लोड कर सकते हैं, API key वेरिफ़ाई कर सकते हैं, और genai.Client इनिशियलाइज़ कर सकते हैं।

import asyncio
import os
import sys
from dotenv import load_dotenv
from google import genai
from google.genai import types
import sounddevice as sd

# Load environment variables from .env file
load_dotenv()

api_key = os.getenv("GEMINI_API_KEY")
if not api_key:
    raise ValueError("GEMINI_API_KEY not found. Please set it in your .env file or environment.")

# Initialize the Gemini Client
client = genai.Client(api_key=api_key)
print("Gemini Client initialized successfully!")

स्टेप 2: पहला रिक्वेस्ट बनाना

चलें, एक सिंगल टेक्स्ट टर्न भेजकर और स्ट्रीमिंग स्पीच व ट्रांसक्रिप्शन प्राप्त करके Gemini Live कनेक्शन का कोर लाइफ़साइकल समझते हैं। हम एक टेक्स्ट प्रॉम्प्ट भेजेंगे और टेक्स्ट व ऑडियो रिस्पॉन्स प्राप्त करेंगे। अभी हम ऑडियो प्ले नहीं करेंगे—फिलहाल, सिर्फ ऑडियो चंक्स इकट्ठा करने पर ध्यान दें।

Gemini Live API एक पर्सिस्टेंट WebSocket कनेक्शन का उपयोग करता है, जिसे client.aio.live.connect() से एक्सेस किया जाता है। स्पीच आउटपुट और रियल-टाइम ट्रांसक्रिप्शन कॉन्फ़िगर करने के लिए हम एक config डिक्शनरी सप्लाई करते हैं:

# Session configuration
config = {
    "response_modalities": ["AUDIO"],
    "output_audio_transcription": {},
}

• response_modalities: Gemini को स्पीच ऑडियो में जवाब देने के लिए ["AUDIO"] सेट करें।

• output_audio_transcription: वैल्यू {} Gemini को यह बताती है कि वह जो बोल रहा है उसका टेक्स्ट ट्रांसक्रिप्ट भी साथ में स्ट्रीम करे।

अब हम session.send_client_content() का उपयोग करके एक टेक्स्ट प्रॉम्प्ट भेजने और आने वाली टेक्स्ट ट्रांसक्रिप्शन को स्ट्रीम करने का परीक्षण कर सकते हैं।

print("Connecting to Gemini 3.8 Live API...")
async with client.aio.live.connect(model="gemini-3.8-live", config=config) as session:
    print("Connected! Sending text prompt...")
    
    await session.send_client_content(
        turns={"role": "user", "parts": [{"text": "Hello! In one short sentence, introduce yourself."}]},
        turn_complete=True,
    )
    
    print("\n[Gemini Transcription]: ", end="", flush=True)
    audio_chunks_received = 0
    total_audio_bytes = 0
    
    async for response in session.receive():
        server_content = response.server_content
        if server_content:
            # 1. Print real-time transcription as tokens arrive
            if server_content.output_transcription:
                print(server_content.output_transcription.text, end="", flush=True)
                
            # 2. Inspect audio chunks
            if server_content.model_turn:
                for part in server_content.model_turn.parts:
                    if part.inline_data and part.inline_data.data:
                        audio_chunks_received += 1
                        total_audio_bytes += len(part.inline_data.data)

    print(f"\n\nReceived {audio_chunks_received} audio chunks ({total_audio_bytes:,} bytes total).")

यह कोड चलाने पर हमें कुछ ऐसा दिखेगा:

Connecting to Gemini 3.8 Live API...
Connected! Sending text prompt...

[Gemini Transcription]: Hello, I am your helpful AI assistant designed to assist you with various tasks and answer your questions.

Received 22 audio chunks (304,800 bytes total).

कोड ने ऑडियो चंक्स कैप्चर किए, लेकिन हमारे पास ऑडियो प्लेयर सेटअप नहीं था, इसलिए हम सुन नहीं पाए। अब देखते हैं ऑडियो प्लेयर कैसे परिभाषित करें।

स्टेप 3: रियल-टाइम ऑडियो प्लेबैक

स्टेप 2 में हमने हजारों बाइट्स ऑडियो डेटा प्राप्त किया, पर कुछ सुना नहीं। अगर हम रिसीव लूप के भीतर सीधे ऑडियो हार्डवेयर पर लिखें, तो कोई भी नेटवर्क देरी ऑडियो में अटकन पैदा करेगी, और कोई भी प्लेबैक देरी नेटवर्क रिसेप्शन को ब्लॉक कर देगी।

नेटवर्क रिसीवर को ऑडियो प्लेबैक से ब्लॉक होने से रोकने के लिए, हम अपना पहला वर्कर इम्प्लीमेंट करते हैं: audio_player()।

आपको लो-लेवल ऑडियो इम्प्लीमेंटेशन डिटेल्स की चिंता करने की ज़रूरत नहीं है। हम सलाह देते हैं कि अभी इन्हें ब्लैक बॉक्स की तरह लें। 

OUTPUT_SAMPLE_RATE = 24000
CHANNELS = 1

async def audio_player(audio_queue: asyncio.Queue):
    """Plays raw 24kHz audio chunks from audio_queue through the speakers."""
    loop = asyncio.get_running_loop()
    with sd.RawOutputStream(
        samplerate=OUTPUT_SAMPLE_RATE, channels=CHANNELS, dtype="int16"
    ) as stream:
        while True:
            chunk = await audio_queue.get()
            if chunk is None:  # Sentinel value signaling end of stream
                audio_queue.task_done()
                break
            await loop.run_in_executor(None, stream.write, chunk)
            audio_queue.task_done()

print("Audio player defined!")

इसे टेस्ट करने के लिए, हम audio_player() को अपनी रिक्वेस्ट से जोड़ते हैं। इस बार, हम रियल-टाइम में Gemini को ज़ोर से बोलते हुए सुनेंगे और स्ट्रीम्ड ट्रांसक्रिप्शन भी देखेंगे:

audio_queue = asyncio.Queue()
player_task = asyncio.create_task(audio_player(audio_queue))

prompt_text = "Hello! In one short sentence, introduce yourself."
print(f"[User]: {prompt_text}")

async with client.aio.live.connect(model="gemini-3.8-live", config=config) as session:
    await session.send_client_content(
        turns={"role": "user", "parts": [{"text": prompt_text}]},
        turn_complete=True,
    )
    
    print("[Gemini]: ", end="", flush=True)
    async for response in session.receive():
        server_content = response.server_content
        if server_content:
            if server_content.output_transcription:
                print(server_content.output_transcription.text, end="", flush=True)

            if server_content.model_turn:
                for part in server_content.model_turn.parts:
                    if part.inline_data and part.inline_data.data:
                        await audio_queue.put(part.inline_data.data)

print()
# Signal the player to shut down and await completion
await audio_queue.put(None)
await player_task
print("Playback complete!")

यह स्निपेट चलाने पर अब हम Gemini का जवाब सुन पाएँगे।

स्टेप 4: यूज़र का ऑडियो इनपुट कैप्चर करना

Gemini से रियल-टाइम में बात करने के लिए, हमें माइक्रोफोन से अपनी आवाज़ लगातार कैप्चर करनी होगी।

हमारा दूसरा वर्कर है audio_recorder()। यह बैकग्राउंड में आपका माइक्रोफोन सुनता है, आने वाली स्पीच को छोटे चंक्स में काटता है और उन्हें input_queue पर रखता है। हम सैंपल रेट 16 kHz सेट करते हैं, जो Gemini की अपेक्षित स्टैंडर्ड स्पीच फ़ॉर्मेट है।

INPUT_SAMPLE_RATE = 16000  # Gemini Live expects 16kHz audio input
CHUNK_SIZE = 1024          # Number of samples per audio chunk

async def audio_recorder(input_queue: asyncio.Queue, stop_event: asyncio.Event):
    """Captures microphone input and puts raw audio chunks into the input queue."""
    loop = asyncio.get_running_loop()

    def record_loop():
        with sd.RawInputStream(
            samplerate=INPUT_SAMPLE_RATE,
            channels=CHANNELS,
            dtype="int16",
            blocksize=CHUNK_SIZE,
        ) as stream:
            while not stop_event.is_set():
                data, _ = stream.read(CHUNK_SIZE)
                loop.call_soon_threadsafe(input_queue.put_nowait, bytes(data))

    await asyncio.to_thread(record_loop)

print("Audio recorder defined!")

स्टेप 5: लगातार ऑडियो स्ट्रीम करने का फ़ंक्शन लिखना

स्टेप 2 में हमने send_client_content() से स्टैटिक टेक्स्ट वाला टर्न भेजा था। लगातार वॉयस स्ट्रीमिंग के लिए, Live API session.send_realtime_input() देता है।

हमारा तीसरा वर्कर है send_audio_loop()। यह input_queue पर नज़र रखता है और जैसे ही माइक्रोफोन से कोई ऑडियो chunk आता है, उसे खुले WebSocket पर Gemini को फॉरवर्ड कर देता है।

ध्यान दें कि हमें Gemini को मैन्युअली यह बताने की ज़रूरत नहीं कि हम कब बोलना शुरू या बंद कर रहे हैं: Gemini में बिल्ट-इन Voice Activity Detection (VAD) है जो अपने-आप पता करता है।

async def send_audio_loop(session, input_queue: asyncio.Queue, stop_event: asyncio.Event):
    """Continuously streams microphone chunks from input_queue to Gemini."""
    while not stop_event.is_set():
        try:
            chunk = await asyncio.wait_for(input_queue.get(), timeout=0.1)
            await session.send_realtime_input(
                audio=types.Blob(data=chunk, mime_type=f"audio/pcm;rate={INPUT_SAMPLE_RATE}")
            )
            input_queue.task_done()
        except asyncio.TimeoutError:
            continue

print("send_audio_loop defined!")

जैसे हमने स्टेप 3 में टेक्स्ट प्रॉम्प्ट के साथ ऑडियो प्लेबैक टेस्ट किया था, वैसे ही अब हम माइक्रोफोन स्ट्रीमिंग को एंड-टू-एंड एक सिंगल बोले गए सवाल से टेस्ट कर सकते हैं।

नीचे वाला सेल चलाते समय, हम माइक्रोफोन में ज़ोर से एक सवाल पूछते हैं (उदाहरण: "What is the capital of France?")। Gemini हमारी आवाज़ को सीधे प्रोसेस करेगा और सिंथेसाइज़्ड स्पीच और रियल-टाइम ट्रांसक्रिप्शन में जवाब देगा:

audio_queue = asyncio.Queue()
input_queue = asyncio.Queue()
stop_event = asyncio.Event()

player_task = asyncio.create_task(audio_player(audio_queue))

print("Connecting to Gemini Live API...")
async with client.aio.live.connect(model="gemini-3.8-live", config=config) as session:
    print("Connected! Speak a question into your microphone (e.g. 'What is the capital of France?')...")
    recorder_task = asyncio.create_task(audio_recorder(input_queue, stop_event))
    sender_task = asyncio.create_task(send_audio_loop(session, input_queue, stop_event))

    print("\n[Gemini]: ", end="", flush=True)
    async for response in session.receive():
        server_content = response.server_content
        if server_content:
            # 1. As soon as Gemini starts replying, mute the microphone
            # so speaker audio cannot loop back into the mic and interrupt Gemini
            if not stop_event.is_set() and (server_content.output_transcription or server_content.model_turn):
                stop_event.set()

            # 2. Print transcription text as it streams
            if server_content.output_transcription:
                print(server_content.output_transcription.text, end="", flush=True)

            # 3. Queue audio parts for playback
            if server_content.model_turn:
                for part in server_content.model_turn.parts:
                    if part.inline_data and part.inline_data.data:
                        await audio_queue.put(part.inline_data.data)

            # 4. Turn complete
            if server_content.turn_complete:
                break

    # Clean up mic tasks cleanly
    stop_event.set()
    recorder_task.cancel()
    sender_task.cancel()
    await asyncio.gather(recorder_task, sender_task, return_exceptions=True)

# 5. Wait for playback queue to drain, then allow the soundcard buffer to finish playing
await audio_queue.join()
await asyncio.sleep(0.8)  # Prevents clipping the final syllables
await audio_queue.put(None)
await player_task

print("\nSingle-turn voice test complete!")

स्टेप 6: मल्टी-टर्न और इंटरप्शन

ऊपर के टेस्ट में देखें क्या हुआ: हमने माइक्रोफोन से सवाल पूछा, Gemini ने हमारी आवाज़ सीधे समझी और ज़ोर से जवाब दिया। पर अगर हम फॉलो-अप सवाल पूछने की कोशिश करें, तो सत्र पहले ही समाप्त हो चुका होगा। 

इसे हल करने के लिए हमें वास्तविक दुनिया के वॉयस असिस्टेंट के दो अहम पहलुओं को एड्रेस करना होगा: मल्टी-टर्न पर्सिस्टेंस और इंटरप्शन।

मल्टी-टर्न सत्र पर्सिस्टेंस:

google-genai SDK में, session.receive() एक टर्न के लिए async जेनरेटर है। Gemini जब अपना जवाब बोलना खत्म करता है, तो session.receive() भी खत्म हो जाता है। अगर इसे किसी आउटर लूप में न लपेटें, तो असिस्टेंट पहले रिस्पॉन्स के बाद ही बंद हो जाएगा।

कन्टिन्युअस मल्टी-टर्न बातचीत के लिए, हम session.receive() को एक आउटर while not stop_event.is_set(): लूप में लपेटते हैं:

while not stop_event.is_set():
    async for response in session.receive():
        ...

बर्ज-इन/इंटरप्शन और बफ़र फ्लशिंग:

Gemini 3.8 Live में नैटिव वॉयस एक्टिविटी डिटेक्शन और बर्ज-इन सपोर्ट है। अगर Gemini बोल रहा हो और आप बोलना शुरू करें, तो Gemini तुरंत ऑडियो जनरेट करना बंद कर देता है और एक फ़्लैग भेजता है: server_content.interrupted == True।

हालाँकि Gemini नया ऑडियो भेजना बंद कर देता है, हमारी लोकल audio_queue में अभी भी कुछ ऑडियो चंक्स बचे हो सकते हैं जो स्पीकर द्वारा चलने का इंतज़ार कर रहे हों। अगर हम इस क्यू को साफ़ नहीं करते, तो स्पीकर पिछले उत्तर को बजाते रहेंगे।

इसलिए, जैसे ही server_content.interrupted प्राप्त हो, हम क्यू फ्लश कर देते हैं ताकि प्लेबैक तुरंत रुक जाए:

if server_content.interrupted:
    print("\n[Interrupted!]")
    while not audio_queue.empty():
        audio_queue.get_nowait()
        audio_queue.task_done()

सब कुछ साथ में रखना

यह रहा हमारा चौथा और अंतिम वर्कर: receive_loop()। यह मल्टी-टर्न पर्सिस्टेंस, रियल-टाइम ट्रांसक्रिप्शन, और इंस्टेंट इंटरप्शन को एक साथ जोड़ता है:

async def receive_loop(session, audio_queue: asyncio.Queue, stop_event: asyncio.Event):
    """Receives transcription and audio output from Gemini across multiple turns."""
    first_chunk_received = False
    try:
        while not stop_event.is_set():
            async for response in session.receive():
                if stop_event.is_set():
                    break

                server_content = response.server_content
                if server_content:
                    # 1. Handle user interruption (barge-in)
                    if server_content.interrupted:
                        print("\n[Interrupted!]")
                        # Flush remaining unplayed audio so speakers go silent immediately
                        while not audio_queue.empty():
                            try:
                                audio_queue.get_nowait()
                                audio_queue.task_done()
                            except asyncio.QueueEmpty:
                                break
                        first_chunk_received = False
                        print("\n[Listening... Speak now]")

                    # 2. Print real-time transcription
                    if server_content.output_transcription:
                        if not first_chunk_received:
                            print("\n[Gemini]: ", end="", flush=True)
                            first_chunk_received = True
                        print(server_content.output_transcription.text, end="", flush=True)

                    # 3. Enqueue synthesized audio for playback
                    if server_content.model_turn:
                        for part in server_content.model_turn.parts:
                            if part.inline_data and part.inline_data.data:
                                await audio_queue.put(part.inline_data.data)

                    # 4. Interaction complete: wait for audio to finish playing before prompt
                    # In Gemini 3.8, interaction_status tracks when the overall exchange is finished
                    is_done = False
                    if server_content.interaction_status is not None:
                        is_done = str(server_content.interaction_status).endswith("IDLE") or server_content.interaction_status == "IDLE"
                    elif server_content.turn_complete:
                        is_done = True

                    if is_done:
                        print()
                        await audio_queue.join()
                        first_chunk_received = False
                        print("\n[Listening... Speak now]")
    except asyncio.CancelledError:
        pass
    except Exception as e:
        print(f"\n[Receive Error]: {e}", file=sys.stderr)
        stop_event.set()

print("receive_loop defined!")

स्टेप 7: पूरे वॉयस असिस्टेंट को असेंबल करना

अब हम अपने चारों कॉन्करेंट वर्कर्स को run_voice_assistant में ऑर्केस्ट्रेट करते हैं:

• audio_player(): audio_queue से कंज्यूम करता है और स्पीकर्स पर लिखता है।

• audio_recorder(): माइक्रोफोन से पढ़ता है और ऑडियो को input_queue में पुश करता है।

• send_audio_loop(): input_queue से कंज्यूम करता है और session.send_realtime_input() के ज़रिये Gemini को स्ट्रीम करता है।

• receive_loop(): session.receive() से Gemini का आउटपुट कंज्यूम करता है, ट्रांसक्रिप्शन प्रिंट करता है और प्लेबैक के लिए audio_queue में ऑडियो पुश करता है।

async def run_voice_assistant():
    """Runs the full-duplex interactive voice assistant."""
    audio_queue: asyncio.Queue[bytes | None] = asyncio.Queue()
    input_queue: asyncio.Queue[bytes] = asyncio.Queue()
    stop_event = asyncio.Event()

    player_task = asyncio.create_task(audio_player(audio_queue))

    print("Connecting to Gemini Live API...")
    async with client.aio.live.connect(model="gemini-3.8-live", config=config) as session:
        print("[Listening... Speak now]")

        recorder_task = asyncio.create_task(audio_recorder(input_queue, stop_event))
        sender_task = asyncio.create_task(send_audio_loop(session, input_queue, stop_event))
        receiver_task = asyncio.create_task(receive_loop(session, audio_queue, stop_event))

        try:
            while not stop_event.is_set():
                await asyncio.sleep(0.5)
        except (asyncio.CancelledError, KeyboardInterrupt):
            print("\nStopping voice assistant...")
        finally:
            stop_event.set()
            recorder_task.cancel()
            sender_task.cancel()
            receiver_task.cancel()
            await asyncio.gather(recorder_task, sender_task, receiver_task, return_exceptions=True)

    # Terminate player
    await audio_queue.put(None)
    await player_task
    print("\nSession finished cleanly.")

print("run_voice_assistant is ready to run!")

स्टेप 8: लाइव असिस्टेंट चलाना

नोटबुक में वॉयस असिस्टेंट ऐसे चलाएँ:

await run_voice_assistant()

नोट्स:

• हेडफ़ोन का कड़ा सुझाव है। अगर Gemini की आवाज़ लैपटॉप स्पीकर से बजती है, तो माइक्रोफोन उसे उठा लेगा और Gemini सोचेगा आप उसे टोक रहे हैं।
• असिस्टेंट को रोकने के लिए, बस नोटबुक का इंटरप्ट बटन (■) क्लिक करें।
• अगर हम नोटबुक चलते समय हेडफ़ोन कनेक्ट/डिसकनेक्ट करते हैं, तो साउंड डिवाइस सेटिंग्स बदल सकती हैं और ऑडियो एरर आ सकता है। ऐसे में, नोटबुक कर्नेल रीस्टार्ट करें और सेल्स को क्रम में फिर से चलाएँ।

Gemini 3.8 Live Extended Thinking के साथ उन्नत उपयोग

Gemini 3.8 Live दो वर्ज़न में आता है:

• स्टैंडर्ड (gemini-3.8-live): अल्ट्रा-लो लेटेंसी डायरेक्ट स्पीच-टू-स्पीच बातचीत के लिए ऑप्टिमाइज़्ड। टूल्स कॉल करते समय यह जवाब देने से पहले चुपचाप इंतज़ार करता है।

• Extended Thinking (gemini-3.8-live-extended-thinking): बैकग्राउंड रीजनिंग और पैरेलल कन्वर्सेशनल फिलर्स देता है। यह टूल्स बैकग्राउंड में चलते हुए नैचुरल अपडेट्स बोल सकता है (जैसे, "Let me look that up for you...")।

यह रहा दोनों के बीच अंतर का सार:

Gemini 3.8 Live बनाम 3.8 Live Extended Thinking कब उपयोग करें

अगर आपको तय नहीं कि किस वर्ज़न का उपयोग करें, तो मेरा निर्णय ढांचा यह है। कन्वर्सेशनल एजेंट बनाते समय:

• gemini-3.8-live का उपयोग डायरेक्ट प्रश्नोत्तर और तेज़ वॉयस कमांड्स के लिए करें जहाँ लेटेंसी कम रखना सर्वोच्च प्राथमिकता है।

• gemini-3.8-live-extended-thinking का उपयोग रिच कन्वर्सेशनल असिस्टेंट्स और ऐसे एजेंट्स के लिए करें जो मल्टी-स्टेप रीजनिंग, बाहरी डेटा रिट्रीवल, या API कॉल्स करते हुए यूज़र के साथ सक्रिय, स्वाभाविक संवाद बनाए रखें।

Gemini 3.8 Live के साथ टूल कॉलिंग कैसे इम्प्लीमेंट करें

मॉडल के एक्सटेंडेड-थिंकिंग वर्ज़न की ताकतों में से एक यह है कि यह बातचीत बनाए रखते हुए बैकग्राउंड में रीजनिंग और टूल्स चला सकता है। 

कोड में जाने से पहले, इसे एक्शन में देखते हैं। मैंने बेस मॉडल को मौसम जांचने के टूल से लैस किया। यहाँ न्यूयॉर्क का मौसम पूछते हुए मेरा वीडियो है; ध्यान दें कि उत्तर कंप्यूट होते समय मॉडल चुप रहता है:

यही इंटरैक्शन एक्सटेंडेड थिंकिंग के साथ:

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

असिस्टेंट में उपयोग करने के लिए टूल बनाना

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

Gemini 3.8 Live में कस्टम टूल इंटीग्रेट करने के लिए हमें अपने लोकल कोड और मॉडल के रीजनिंग इंजन को ब्रिज करना होता है। इसके लिए ज़रूरी हैं:

• Execution Logic: एक स्टैंडर्ड Python फ़ंक्शन परिभाषित करें जो असल काम करे और परिणाम लौटाए।

• Tool Mapping: एक डिक्शनरी (tool_map) बनाएँ जो फ़ंक्शन के स्ट्रिंग नाम को executable Python ऑब्जेक्ट से जोड़ती है। 

• Function Declaration: एक FunctionDeclaration बनाएँ जो टूल का इंस्ट्रक्शन मैनुअल हो। नाम, विवरण, और पैरामीटर स्कीमा (टाइप्स और आवश्यक फ़ील्ड्स सहित) साफ़-साफ़ परिभाषित करके हम Gemini को सिखाते हैं कि टूल कब और कैसे उपयोग करना है। हम behavior="NON_BLOCKING" भी सेट करते हैं, जो Extended Thinking के लिए ज़रूरी है, ताकि टूल चलते समय मॉडल बोलता रहे।

• Session Configuration: डिक्लेरेशन को सत्र के tools_config पेलोड में इंजेक्ट करें। 

इसे दर्शाने के लिए, हम एक वेदर लुक-अप टूल बनाते हैं:

import urllib.request
import urllib.parse
import json
import asyncio

async def get_current_weather(location: str) -> str:
    """Fetch live real-time weather for any city in the world using Open-Meteo's free API."""
    def fetch():
        # 1. Geocode city name to lat/lon coordinates
        geo_url = f"https://geocoding-api.open-meteo.com/v1/search?name={urllib.parse.quote(location)}&count=1"
        req = urllib.request.Request(geo_url, headers={"User-Agent": "VoiceAssistantTutorial/1.0"})
        with urllib.request.urlopen(req, timeout=5) as r:
            geo_data = json.loads(r.read().decode("utf-8"))
            if not geo_data.get("results"):
                return f"Could not find coordinates for '{location}'."
            loc = geo_data["results"][0]
            lat, lon = loc["latitude"], loc["longitude"]
            city_name = loc.get("name", location)
            country = loc.get("country", "")

        # 2. Fetch current temperature
        weather_url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&current=temperature_2m"
        req2 = urllib.request.Request(weather_url, headers={"User-Agent": "VoiceAssistantTutorial/1.0"})
        with urllib.request.urlopen(req2, timeout=5) as r:
            weather_data = json.loads(r.read().decode("utf-8"))
            temp = weather_data.get("current", {}).get("temperature_2m")
            return f"The current temperature in {city_name}, {country} is {temp}°C."

    try:
        return await asyncio.to_thread(fetch)
    except Exception as e:
        return f"Error retrieving weather for {location}: {e}"

tool_map = {
    "get_current_weather": get_current_weather,
}

weather_tool = types.FunctionDeclaration(
    name="get_current_weather",
    description="Get the current live weather and temperature for a given city or location.",
    behavior="NON_BLOCKING", 
    parameters=types.Schema(
        type="OBJECT",
        properties={
            "location": types.Schema(
                type="STRING",
                description="The city or location name (e.g. Tokyo, Paris, New York).",
            )
        },
        required=["location"],
    ),
)

tools_config = {
    "response_modalities": ["AUDIO"],
    "output_audio_transcription": {},
    "tools": [
        {"function_declarations": [weather_tool]},
    ],
}

print("Tool and configuration defined!")

टूल कॉल्स को असिंक्रोनसली हैंडल करना

टूल चलते समय बोलते रहने के लिए दो बातें जरूरी हैं। सर्वर साइड पर, NON_BLOCKING डिक्लेरेशन Extended Thinking को रिज़ल्ट का इंतज़ार करने के बजाय बोलते रहने देता है। क्लाइंट साइड पर, हमारा कोड भी ब्लॉक नहीं होना चाहिए। अगर हम रिसीव लूप के अंदर सीधे टूल चलाएँ, तो 1.5-सेकंड की API कॉल पूरी होने तक हम Gemini के फिलर ऑडियो और इंटरप्शन सिग्नल्स पढ़ नहीं पाएँगे।

सचमुच "execute करते हुए बात" करने के लिए, हम receive_loop_with_tools() को दो मुख्य डिज़ाइन चॉइसेज़ के साथ अपडेट करते हैं:

1. नॉन-ब्लॉकिंग एक्ज़ीक्यूशन: हम handle_tool_call को asyncio.create_task() के ज़रिये कंकरेन्ट बैकग्राउंड टास्क के रूप में लॉन्च करते हैं। इससे रिसीव लूप बिना बाधा के Gemini की स्पीच प्रोसेस और प्ले करता रहता है, जबकि Python पैरलल में वेदर लाता है।

2. इंटरैक्शन स्टेटस ट्रैकिंग: Extended Thinking में, Gemini इंटरमीडिएट फिलर फ़्रेज़ (जैसे, "Checking the weather for you...") खत्म करते समय turn_complete: True इमिट करता है। अगर कोड केवल turn_complete देखे, तो टूल चलते समय ही असिस्टेंट जल्दबाज़ी में [Listening... Speak now] दिखा देगा! server_content.interaction_status == "IDLE" चेक करके, क्लाइंट तब तक इंतज़ार करता है जब तक सारी बैकग्राउंड रीजनिंग, टूल कॉल्स और अंतिम स्पीच सच में पूरी न हो जाएँ।

यह रहा receive_loop_with_tools()। यह receive_loop() जैसा ही है, बस नए handle_tool_call() हेल्पर और ब्लॉक 1 के साथ, जो टूल कॉल्स डिस्पैच करता है:

async def receive_loop_with_tools(session, audio_queue: asyncio.Queue, stop_event: asyncio.Event):
    """Receives transcription and audio from Gemini, and automatically handles tool calls asynchronously."""
    first_chunk_received = False

    async def handle_tool_call(tool_call):
        """Executes tool calls in the background without blocking the audio receive loop."""
        try:
            function_responses = []
            for fc in tool_call.function_calls:
                print(f"\n[Tool Requested]: {fc.name}({fc.args})")
                fn = tool_map.get(fc.name)
                if fn:
                    if asyncio.iscoroutinefunction(fn):
                        result = await fn(**fc.args)
                    else:
                        result = fn(**fc.args)
                else:
                    result = f"Error: Unknown tool {fc.name}"
                print(f"[Tool Result]: {result}")
                function_responses.append(
                    types.FunctionResponse(
                        id=fc.id,
                        name=fc.name,
                        response={"result": result},
                    )
                )
            await session.send_tool_response(function_responses=function_responses)
        except Exception as e:
            print(f"\n[Tool Execution Error]: {e}", file=sys.stderr)

    try:
        while not stop_event.is_set():
            async for response in session.receive():
                if stop_event.is_set():
                    break

                # 1. Handle tool calls asynchronously (non-blocking)
                if response.tool_call:
                    asyncio.create_task(handle_tool_call(response.tool_call))

                server_content = response.server_content
                if server_content:
                    # 2. Handle user interruption (barge-in)
                    if server_content.interrupted:
                        print("\n[Interrupted!]")
                        while not audio_queue.empty():
                            try:
                                audio_queue.get_nowait()
                                audio_queue.task_done()
                            except asyncio.QueueEmpty:
                                break
                        first_chunk_received = False
                        print("\n[Listening... Speak now]")

                    # 3. Print real-time transcription
                    if server_content.output_transcription:
                        if not first_chunk_received:
                            print("\n[Gemini]: ", end="", flush=True)
                            first_chunk_received = True
                        print(server_content.output_transcription.text, end="", flush=True)

                    # 4. Enqueue synthesized audio for playback
                    if server_content.model_turn:
                        for part in server_content.model_turn.parts:
                            if part.inline_data and part.inline_data.data:
                                await audio_queue.put(part.inline_data.data)

                    # 5. Check if the interaction is complete
                    is_done = False
                    if server_content.interaction_status is not None:
                        is_done = str(server_content.interaction_status).endswith("IDLE") or server_content.interaction_status == "IDLE"
                    elif server_content.turn_complete:
                        is_done = True

                    if is_done:
                        print()
                        await audio_queue.join()
                        first_chunk_received = False
                        print("\n[Listening... Speak now]")
    except asyncio.CancelledError:
        pass
    except Exception as e:
        print(f"\n[Receive Error]: {e}", file=sys.stderr)
        stop_event.set()

print("receive_loop_with_tools defined!")

अंत में, हम run_voice_assistant_with_tools() इम्प्लीमेंट करते हैं। tools_config देने के अलावा, यह फ़ंक्शन हमें स्टैंडर्ड मॉडल और एक्सटेंडेड थिंकिंग मॉडल में चुनने देता है। क्योंकि Extended Thinking मॉडल को thinking_level ("low", "medium", या "high") बताने वाली thinking_config डिक्शनरी चाहिए, हम इसे कंडीशनली सत्र कॉन्फ़िगरेशन में इंजेक्ट करते हैं:

async def run_voice_assistant_with_tools(
    model: str = "gemini-3.8-live-extended-thinking",
    thinking_level: str = "low",
):
    """Runs the interactive voice assistant with tool calling enabled.
    
    Supports both:
    - 'gemini-3.8-live-extended-thinking' (requires thinking_level: 'low', 'medium', or 'high')
    - 'gemini-3.8-live' (standard, ultra-low latency, no thinking_level)
    """
    audio_queue: asyncio.Queue[bytes | None] = asyncio.Queue()
    input_queue: asyncio.Queue[bytes] = asyncio.Queue()
    stop_event = asyncio.Event()

    player_task = asyncio.create_task(audio_player(audio_queue))

    # Extended Thinking models require thinking_config with thinking_level
    session_config = dict(tools_config)
    if "extended-thinking" in model:
        session_config["thinking_config"] = {
            "thinking_level": thinking_level,
        }

    print(f"Connecting to Gemini Live API with tools (model: {model})...")
    async with client.aio.live.connect(model=model, config=session_config) as session:
        print("[Listening... Speak now.]")

        recorder_task = asyncio.create_task(audio_recorder(input_queue, stop_event))
        sender_task = asyncio.create_task(send_audio_loop(session, input_queue, stop_event))
        receiver_task = asyncio.create_task(receive_loop_with_tools(session, audio_queue, stop_event))

        try:
            while not stop_event.is_set():
                await asyncio.sleep(0.5)
        except (asyncio.CancelledError, KeyboardInterrupt):
            print("\nStopping voice assistant...")
        finally:
            stop_event.set()
            recorder_task.cancel()
            sender_task.cancel()
            receiver_task.cancel()
            await asyncio.gather(recorder_task, sender_task, receiver_task, return_exceptions=True)

    # Terminate player
    await audio_queue.put(None)
    await player_task
    print("\nSession finished cleanly.")

print("run_voice_assistant_with_tools is ready to run!")

टूल-इनेबल्ड असिस्टेंट चलाना

अब हम अपना टूल-इनेबल्ड वॉयस असिस्टेंट चला सकते हैं और दोनों मॉडलों के लाइव व्यवहार की तुलना कर सकते हैं।

पहले, असिस्टेंट को Extended Thinking के साथ टेस्ट करें:

await run_voice_assistant_with_tools("gemini-3.8-live-extended-thinking")

जब असिस्टेंट सुन रहा हो, कोई ऐसा सवाल पूछें जिसे लाइव डेटा चाहिए, उदाहरण के लिए: 

"What's the weather like in Tokyo right now?"

क्योंकि इंटरनेट पर Open-Meteo API क्वेरी करने में ~1.5 सेकंड लगते हैं, हम बैकग्राउंड रीजनिंग को एक्शन में देखेंगे:

1. Gemini तुरंत हमारी रिक्वेस्ट स्वीकारते हुए ज़ोर से बोलता है: "Let me check the current weather in Tokyo for you..."
2. जब Gemini बोल रहा होता है, हमारा बैकग्राउंड टास्क पैरलल में लाइव वेदर डेटा फ़ेच करता है।
3. जैसे ही टूल रिस्पॉन्स आता है, Gemini लाइव टेम्परेचर पढ़ने लगता है।

अब वही असिस्टेंट स्टैंडर्ड Gemini 3.8 Live मॉडल के साथ चलाएँ:

await run_voice_assistant_with_tools("gemini-3.8-live")

जब हम वही सवाल स्टैंडर्ड मॉडल से पूछते हैं, इस केस में मॉडल लगभग ~1.5 सेकंड तक नेटवर्क पर टूल के रिस्पॉन्स का इंतज़ार करते हुए बिल्कुल चुप रहता है, और फिर बिना कोई फिलर फ़्रेज़ बोले सीधे टेम्परेचर बताता है।

पूरा प्रोजेक्ट देखने के लिए, संबंधित GitHub रेपो देखें।

निष्कर्ष

इस ट्यूटोरियल में, हमने Python और Gemini 3.8 Live के साथ एक पूरा फुल-डुप्लेक्स वॉयस असिस्टेंट बनाया। तीन विशेषताएँ जो इसे रियल-टाइम काम के लिए खास तौर पर उपयोगी बनाती हैं:

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

• बैकग्राउंड टूल कॉलिंग: टूल एक्ज़ीक्यूशन को नॉन-ब्लॉकिंग बैकग्राउंड टास्क्स (asyncio.create_task) के रूप में लॉन्च करना Gemini 3.8 Live Extended Thinking को बात करते हुए रीजनिंग और बाहरी फ़ंक्शंस execute करने देता है।

• स्टेट मैनेजमेंट: interaction_status == "IDLE" ट्रैक करना सुनिश्चित करता है कि असिस्टेंट तभी फिर से सुनना शुरू करे जब सारी बैकग्राउंड रीजनिंग, टूल कॉल्स, और अंतिम स्पीच टर्न्स खत्म हो जाएँ।

अगर आप AI इंजीनियरिंग में अपना करियर शुरू करने के इच्छुक हैं, तो मैं सिफारिश करूँगा कि आप हमारे AI Engineer for Developers कैरियर ट्रैक से शुरुआत करें, जो OpenAI API, Hugging Face, MCP, और बहुत कुछ सिखाता है!