Pe scurt

• Gemini 3.8 Live transmite audio în ambele sensuri printr-un singur WebSocket, astfel încât poți construi un asistent vocal care ascultă în timp ce vorbește și gestionează întreruperile.

• Tutorialul îl construiește în Python cu patru workeri asyncio (înregistrator microfon, expeditor audio, receptor, difuzor) legați de două cozi.

• Barge-in funcționează prin golirea cozii locale de redare când Gemini trimite interrupted.

• Adăugarea unei unelte (o interogare meteo live) arată diferența dintre cele două modele: modelul standard tace cât timp rulează uneltele, în timp ce Extended Thinking continuă să vorbească.

• Cu Extended Thinking, urmărește interaction_status == "IDLE” în loc de turn_complete și rulează apelurile de unelte ca taskuri de fundal, ca bucla de recepție să nu se blocheze niciodată.

Ce e special la Gemini 3.8 Live?

Gemini 3.8 Live de la Google este un model nativ speech-to-speech construit special pentru streaming în timp real și aplicații audio interactive. Gemini 3.8 Live procesează intrări multimodale direct printr-o conexiune WebSocket persistentă. 

Această capabilitate de streaming bidirecțional permite dezvoltatorilor să creeze agenți conversaționali full-duplex care pot asculta și vorbi simultan, cu suport pentru întreruperi naturale ale utilizatorului și transcriere audio în timp real.

Pentru dezvoltarea de aplicații, Gemini 3.8 Live introduce apelarea asincronă a uneltelor și raționamentul în fundal, permițând agenților să execute apeluri către funcții externe sau să recupereze date în timp ce mențin dialogul activ cu utilizatorul.

Pentru o prezentare cuprinzătoare a funcțiilor, benchmark-urilor și prețurilor, consultă ghidul Gemini 3.8 Live.

Cum funcționează un asistent vocal live: 4 workeri și 2 cozi

Înainte să intrăm în cod, hai să înțelegem cum funcționează, în culise, un asistent vocal în timp real.

În scripturi Python standard, codul rulează linie cu linie: funcția A se termină, apoi rulează funcția B. Dar într-o conversație vocală live, a aștepta nu funcționează:

• În timp ce vorbim, programul trebuie să transmită vocea ta către Gemini în timp real.
• În timp ce Gemini răspunde, programul trebuie să redea bucățile audio prin difuzoare pe măsură ce sosesc.
• Cel mai important, programul trebuie să continue să asculte chiar și în timp ce Gemini vorbește, ca să putem întrerupe (barge in).

Pentru a obține asta fără blocaje, folosim Python asyncio pentru a rula 4 taskuri ușoare în fundal ("workeri") care comunică folosind două buffere asyncio.Queue (gândește-te la ele ca la benzi transportoare):

1. Banda transportoare de intrare (input_queue):

• audio_recorder(): Ascultă continuu microfonul și pune felii de audio pe bandă.

• send_audio_loop(): Ia feliile audio de pe bandă și le transmite către Gemini.

2. Banda transportoare de ieșire (audio_queue):

• receive_loop(): Ascultă Gemini. Când sosește text, îl afișează. Când sosește vorbire, pune bucățile audio pe bandă.

• audio_player(): Ia bucăți audio de pe bandă și le redă în difuzoare sau căști.

Pentru că fiecare worker se concentrează doar pe mica lui sarcină, toți patru pot rula concurent pe event loop-ul Python fără să se încurce între ei.

Codul complet folosit în acest tutorial este disponibil în acest repo GitHub.

Cum să generezi și să configurezi o cheie API Gemini

Pentru a folosi API-ul Gemini, trebuie să creăm și să configurăm o cheie API, astfel încât codul nostru să poată comunica cu API-ul.

Cel mai simplu mod este:

• Vizitează pagina de chei API Google’s AI Studio și autentifică-te.

• Dă clic pe butonul Create API key din colțul din dreapta sus.

• Copiază cheia API într-un fișier numit .env în același folder cu codul Python, în următorul format:

GEMINI_API_KEY=replace_with_api_key

Reține că folosirea API-ului implică de obicei costuri. Nivelul gratuit acoperă acces limitat la ambele modele Gemini 3.8 Live, dar datele din free tier sunt folosite pentru a îmbunătăți produsele Google. Pentru producție sau limite de rată mai mari, trebuie să ne asigurăm că avem o metodă de plată configurată pe pagina de facturare Google’s AI Studio.

Cum să implementezi arhitectura asistentului vocal cu Gemini 3.8 Live

Acești pași au fost concepuți pentru a rula într-un notebook Jupyter local, fiecare fragment de cod corespunzând unei celule de cod. Pentru că avem nevoie de acces la microfon și difuzoare, nu va funcționa direct pe un notebook online precum Google Colab.

Pasul 1: Configurarea mediului și importuri

Mai întâi, ne asigurăm că pachetele necesare sunt instalate:

pip install google-genai sounddevice python-dotenv

Iată pe scurt ce fac aceste pachete:

• google-genai: Pachetul oficial Google folosit pentru a interacționa cu modelele Gemini.

• sounddevice: Folosit pentru a gestiona hardware-ul audio, înregistrarea de la microfon și redarea în difuzoare.

• python-dotenv: Pachet utilitar pentru a încărca cheia noastră API Gemini dintr-un fișier .env.

Acum putem încărca variabilele de mediu, verifica cheia API și inițializa 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!")

Pasul 2: Prima noastră cerere

Să începem prin a înțelege ciclul de viață al conexiunii Gemini Live trimițând o singură tură de text și primind vorbire și transcriere în streaming. Vom trimite un prompt text și vom primi răspunsul text și audio. Totuși, nu vom reda încă audio. Deocamdată, ne concentrăm doar pe colectarea bucăților audio.

API-ul Gemini Live folosește o conexiune WebSocket persistentă accesată prin client.aio.live.connect(). Pentru a configura ieșirea vocală și transcrierea în timp real, furnizăm un dicționar config:

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

• response_modalities: Folosește valoarea ["AUDIO"] pentru a spune lui Gemini să răspundă cu audio de vorbire.

• output_audio_transcription: Valoarea {} îi spune lui Gemini să transmită simultan transcriptul text al a ceea ce spune.

Acum putem testa trimiterea unui prompt text folosind session.send_client_content() și streamingul transcrierii de text care sosește.

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).")

Când rulăm acest cod, ar trebui să vedem ceva de genul:

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

Codul a capturat bucățile audio, dar nu aveam configurat un player audio, așa că nu le-am putut auzi. Hai să învățăm cum definim playerul audio în continuare.

Pasul 3: Redare audio în timp real

La Pasul 2, am primit mii de bytes de date audio, dar nu am auzit nimic. Dacă scriem direct către hardware-ul audio în bucla de recepție, orice întârziere de rețea va cauza sacadări, iar orice întârziere la redare va bloca recepția de la rețea.

Pentru a preveni ca redarea audio să blocheze receptorul de rețea, implementăm primul nostru worker: audio_player().

Nu e nevoie să te îngrijorezi de detaliile audio de nivel jos. Îți recomandăm să le tratezi ca pe niște „cutii negre”. 

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!")

Pentru a-l testa, conectăm audio_player() la cererea noastră. De data aceasta, îl vom auzi pe Gemini vorbind cu voce tare în timp real, în timp ce observăm transcrierea în streaming:

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!")

Rulând acest fragment, acum putem auzi răspunsul lui Gemini.

Pasul 4: Capturarea intrării audio a utilizatorului

Pentru a vorbi cu Gemini în timp real, trebuie să capturăm continuu vocea noastră de la microfon.

Al doilea nostru worker este audio_recorder(). Ascultă microfonul în fundal, feliază vorbirea în bucăți mici și le pune pe input_queue. Setăm rata de eșantionare la 16 kHz, formatul standard de vorbire pe care Gemini îl așteaptă.

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!")

Pasul 5: Scrierea unei funcții care transmite audio continuu

La Pasul 2, am folosit send_client_content() pentru a trimite o tură cu text static. Pentru streaming vocal continuu, API-ul Live oferă session.send_realtime_input().

Al treilea nostru worker este send_audio_loop(). El urmărește input_queue și, de îndată ce sosește o bucată audio de la microfon, o redirecționează către Gemini prin WebSocket-ul deschis.

Observă că nu trebuie să îi spunem manual lui Gemini când începem sau terminăm de vorbit: Gemini folosește detectarea activității vocale (VAD) încorporată pentru a detecta automat când începi și când termini.

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!")

La fel cum am testat redarea audio cu un prompt text în Pasul 3, acum putem testa capătul nostru de microfon end-to-end cu o singură întrebare rostită.

Când rulăm celula de mai jos, rostim o întrebare la microfon (de exemplu: „Care este capitala Franței?”). Gemini va procesa direct vocea noastră și va răspunde cu vorbire sintetică și transcriere în timp real:

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!")

Pasul 6: Mai multe ture și întreruperi

Observă ce s-a întâmplat în testul de mai sus: am pus o întrebare folosind microfonul, iar Gemini ne-a înțeles vocea direct și a răspuns cu voce tare. Totuși, dacă încercăm să punem o întrebare de continuare, sesiunea s-a încheiat deja. 

Pentru a depăși asta, trebuie să abordăm două aspecte esențiale în construirea unui asistent vocal real: persistența pe mai multe ture și întreruperea.

Persistența sesiunii multi-turn:

În SDK-ul google-genai, session.receive() este un generator async pentru o singură tură. Când Gemini termină de vorbit răspunsul, session.receive() se încheie. Fără să îl înfășurăm într-o buclă exterioară, asistentul se oprește după primul răspuns.

Pentru a susține conversații continue pe mai multe ture, înfășurăm session.receive() într-o buclă exterioară while not stop_event.is_set()::

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

Barge-in/întrerupere și golirea bufferului:

Gemini 3.8 Live are detecție nativă a activității vocale și suport pentru barge-in. Dacă Gemini vorbește și tu începi să vorbești, Gemini oprește imediat generarea de audio și trimite un flag: server_content.interrupted == True.

Chiar dacă Gemini încetează să trimită audio nou, audio_queue locală poate încă să rețină câteva bucăți audio care așteaptă să fie redate. Dacă nu golim această coadă, difuzoarele vor continua să redea răspunsul anterior.

De aceea, imediat ce primim server_content.interrupted, golim coada astfel încât redarea să se oprească instant:

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

Punând totul cap la cap

Iată al patrulea și ultimul nostru worker: receive_loop(). El combină persistența pe mai multe ture, transcrierea în timp real și întreruperea instantanee:

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!")

Pasul 7: Asamblarea asistentului vocal complet

Acum orchestrăm cei patru workeri concurenți în run_voice_assistant:

• audio_player(): Consumă din audio_queue și scrie în difuzoare.

• audio_recorder(): Citește din microfon și împinge audio în input_queue.

• send_audio_loop(): Consumă din input_queue și transmite către Gemini folosind session.send_realtime_input().

• receive_loop(): Consumă ieșirea lui Gemini folosind session.receive(), afișează transcrierea și împinge audio în audio_queue pentru redare.

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!")

Pasul 8: Rularea asistentului live

Iată cum rulezi asistentul vocal în notebook-ul tău:

await run_voice_assistant()

Note:

• Căștile sunt recomandate cu tărie. Dacă vocea lui Gemini se redă prin difuzoarele laptopului, microfonul o va prelua, iar Gemini va crede că încerci să îl întrerupi.
• Pentru a opri asistentul, dă clic pe butonul de întrerupere al notebook-ului (■).
• Dacă conectăm sau deconectăm căști în timp ce notebook-ul rulează, setările dispozitivului audio se pot schimba și putem întâmpina o eroare audio. În acest caz, trebuie să repornim kernelul notebook-ului și să rulăm celulele în ordine.

Utilizare avansată cu Gemini 3.8 Live Extended Thinking

Gemini 3.8 Live are două versiuni:

• Standard (gemini-3.8-live): Optimizat pentru conversații speech-to-speech cu latență ultra-redusă. La apelarea uneltelor, așteaptă în tăcere răspunsul uneltei înainte de a răspunde.

• Extended Thinking (gemini-3.8-live-extended-thinking): Include raționament în fundal și umpluturi conversaționale paralele. Poate rosti actualizări naturale (de ex. „Let me look that up for you...”) în timp ce execută unelte în fundal.

Iată o defalcare a diferențelor dintre cele două:

Când să folosești Gemini 3.8 Live vs 3.8 Live Extended Thinking

Dacă nu ești sigur pe care dintre cele două versiuni să o folosești, acesta este cadrul meu de decizie. Când construiești agenți conversaționali:

• Folosește gemini-3.8-live pentru întrebări-răspuns directe și comenzi vocale rapide unde minimizarea latenței este prioritară.

• Folosește gemini-3.8-live-extended-thinking pentru asistenți conversaționali bogați și agenți care fac raționament multi-pas, recuperare de date externe sau apeluri API menținând un dialog activ, natural cu utilizatorul.

Cum să implementezi apelarea uneltelor cu Gemini 3.8 Live

Unul dintre punctele forte ale versiunii extended-thinking este că poate raționa și executa unelte în fundal, menținând conversația. 

Înainte să intrăm în cod, hai să vedem asta în acțiune. Am echipat modelul de bază cu o unealtă pentru verificarea vremii. Iată un video în care întreb vremea în New York; observă cum modelul rămâne tăcut în timp ce calculează răspunsul:

Iată aceeași interacțiune, dar cu extended thinking:

A doua interacțiune este mai animată și se simte mai aproape de o conversație normală pentru că modelul poate menține dialogul în timp ce procesează informații în fundal.

Construirea uneltei de folosit în asistent

Modelul nu execută de fapt uneltele pentru noi. Configurația uneltei doar îi spune modelului că uneltele există, când și cum să le folosească. Când Gemini decide că sunt necesare date externe, populează response.tool_call cu numele și argumentele funcției.

Pentru a integra o unealtă personalizată în Gemini 3.8 Live, trebuie să facem legătura între codul nostru local și motorul de raționare al modelului. Asta cere următoarele:

• Logica de execuție: Definește o funcție Python standard care face efectiv munca și returnează rezultatul.

• Mapare unelte: Creează un dicționar (tool_map) care leagă numele funcției, ca șir, de obiectul Python executabil. 

• Declarație de funcție: Construiește un FunctionDeclaration care acționează ca manualul uneltei. Definind clar numele, descrierea și Schema parametrilor (inclusiv tipuri și câmpuri obligatorii), îl învățăm pe Gemini exact când să folosească unealta și cum să-și formateze cererea. Setăm și behavior="NON_BLOCKING", cerut de Extended Thinking, ca să poată continua să vorbească în timp ce unealta rulează.

• Configurarea sesiunii: Injectează declarația în payload-ul tools_config al sesiunii. 

Pentru a ilustra, creăm o unealtă de căutare a vremii:

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!")

Gestionarea apelurilor de unelte asincron

A vorbi în timp ce o unealtă rulează implică două lucruri. Pe server, declarația NON_BLOCKING îi permite lui Extended Thinking să continue să vorbească în loc să aștepte rezultatul. Pe client, nici codul nostru nu trebuie să se blocheze. Dacă am rula unealta direct în bucla de recepție, un apel API de 1,5 secunde ne-ar opri din a citi vorbirea de umplutură a lui Gemini și semnalele de întrerupere până când unealta s-ar termina.

Pentru a permite cu adevărat „vorbește în timp ce execuți”, actualizăm receive_loop_with_tools() cu două alegeri de proiectare cheie:

1. Execuție non-blocantă: Lansăm handle_tool_call ca task de fundal concurent prin asyncio.create_task(). Astfel, bucla de recepție continuă să proceseze și să redea vorbirea lui Gemini fără întreruperi, în timp ce Python recuperează vremea în paralel.

2. Urmărirea stării interacțiunii: În Extended Thinking, Gemini emite turn_complete: True când termină de rostit fraze intermediare de umplere (de ex. „Checking the weather for you...”). Dacă codul ar verifica doar turn_complete, asistentul ar afișa prematur [Listening... Speak now] în timp ce unealta încă rulează! Verificând server_content.interaction_status == "IDLE", clientul așteaptă până când tot raționamentul de fundal, apelurile de unelte și vorbirea finală sunt într-adevăr terminate înainte de a deschide microfonul.

Iată receive_loop_with_tools(). Este identică cu receive_loop() cu excepția noii funcții ajutătoare handle_tool_call() și a blocului 1, care dispecerizează apelurile de unelte:

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!")

În final, implementăm run_voice_assistant_with_tools(). Pe lângă furnizarea tools_config, această funcție ne permite să selectăm între modelul standard și cel extended thinking. Pentru că modelul Extended Thinking cere un dicționar thinking_config care specifică thinking_level ("low", "medium" sau "high"), îl injectăm condițional în configurația sesiunii:

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!")

Rularea asistentului cu unelte

Acum putem rula asistentul vocal cu unelte și compara comportamentul live al celor două modele.

Mai întâi, testează asistentul cu Extended Thinking:

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

După ce asistentul ascultă, pune o întrebare care necesită date live, de exemplu: 

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

Pentru că interogarea API-ului Open-Meteo pe internet durează ~1,5 secunde, vom observa raționamentul de fundal în acțiune:

1. Gemini vorbește imediat cu voce tare pentru a confirma întrebarea noastră: „Let me check the current weather in Tokyo for you...”
2. În timp ce Gemini vorbește, taskul nostru de fundal preia în paralel datele meteo live.
3. După ce ajunge răspunsul uneltei, Gemini trece la anunțarea temperaturii live.

Apoi, rulăm același asistent folosind modelul standard Gemini 3.8 Live:

await run_voice_assistant_with_tools("gemini-3.8-live")

Când punem aceeași întrebare modelului standard. În acest caz, modelul rămâne complet tăcut aproximativ 1,5 secunde, așteptând răspunsul uneltei de pe rețea, apoi anunță direct temperatura fără să rostească vreo frază de umplere.

Pentru a vedea proiectul integral, consultă repo-ul GitHub aferent.

Concluzie

În acest tutorial, am construit un asistent vocal full-duplex complet cu Python și Gemini 3.8 Live. Trei funcții îl fac deosebit de util pentru munca în timp real:

• Arhitectură audio concurentă: Patru workeri asyncio ușori comunică prin două cozi, permițând înregistrare simultană, streaming audio în timp real, redare a vorbirii și întreruperi instantanee (barge-in).

• Apelarea uneltelor în fundal: Lansarea execuției uneltelor ca taskuri non-blocante (asyncio.create_task) permite lui Gemini 3.8 Live Extended Thinking să vorbească în timp ce raționează și execută funcții externe.

• Managementul stării: Urmărirea interaction_status == "IDLE" asigură că asistentul reîncepe să asculte doar după ce s-au încheiat tot raționamentul de fundal, apelurile de unelte și turele finale de vorbire.

Dacă vrei să îți începi cariera în AI engineering, îți recomand să începi cu AI Engineer for Developers, traseu de carieră care te învață să lucrezi cu OpenAI API, Hugging Face, MCP și multe altele!