In breve

• Gemini 3.8 Live fa streaming audio bidirezionale su un unico WebSocket, così puoi creare un assistente vocale che ascolta mentre parla e gestisce le interruzioni.

• Il tutorial lo realizza in Python con quattro worker asyncio (registratore mic, mittente audio, ricevitore, riproduttore) collegati da due code.

• Il barge‑in funziona svuotando la coda di riproduzione locale quando Gemini invia interrupted.

• Aggiungere un tool (una ricerca meteo live) mostra la differenza tra i due modelli: il modello standard sta zitto mentre i tool sono in esecuzione, mentre Extended Thinking continua a parlare.

• Con Extended Thinking, traccia interaction_status == "IDLE” invece di turn_complete e esegui le chiamate ai tool come task in background in modo che il ciclo di ricezione non si blocchi mai.

Cosa rende speciale Gemini 3.8 Live?

Gemini 3.8 Live di Google è un modello nativo speech‑to‑speech pensato specificamente per lo streaming in tempo reale e le applicazioni audio interattive. Gemini 3.8 Live elabora input multimodali direttamente su una connessione WebSocket persistente. 

Questa capacità di streaming bidirezionale permette di creare agenti conversazionali full‑duplex che possono ascoltare e parlare simultaneamente, supportando funzionalità come le interruzioni naturali da parte dell’utente e la trascrizione audio in tempo reale.

Per lo sviluppo di applicazioni, Gemini 3.8 Live introduce chiamate a tool asincrone e ragionamento in background, consentendo agli agenti di eseguire chiamate a funzioni esterne o recuperare dati mentre mantengono un dialogo attivo con l’utente.

Per una panoramica completa di funzionalità, benchmark e prezzi, fai riferimento alla nostra guida a Gemini 3.8 Live.

Come funziona un assistente vocale Live: 4 worker e 2 code

Prima di passare al codice, capiamo come funziona sotto il cofano un assistente vocale in tempo reale.

Nei normali script Python, il codice gira riga per riga: la funzione A termina, poi parte la funzione B. Ma in una conversazione vocale live, aspettare non funziona:

• Mentre parliamo, il programma deve streammare la tua voce a Gemini in tempo reale.
• Mentre Gemini risponde, il programma deve riprodurre i chunk audio sugli altoparlanti man mano che arrivano.
• Soprattutto, il programma deve continuare ad ascoltare anche mentre Gemini parla, così possiamo interrompere (barge‑in).

Per ottenere questo senza blocchi, usiamo Python asyncio per eseguire 4 task leggeri in background ("worker") che comunicano attraverso due buffer asyncio.Queue (pensali come nastri trasportatori):

1. Il nastro trasportatore in ingresso (input_queue):

• audio_recorder(): Ascolta continuamente il microfono e lascia delle porzioni di audio sul nastro.

• send_audio_loop(): Prende le porzioni dal nastro e le invia in streaming a Gemini.

2. Il nastro trasportatore in uscita (audio_queue):

• receive_loop(): Ascolta Gemini. Quando arriva testo, lo stampa. Quando arriva parlato, lascia i chunk audio sul nastro.

• audio_player(): Prende i chunk audio dal nastro e li riproduce su altoparlanti o cuffie.

Poiché ogni worker si concentra solo sul proprio piccolo compito, tutti e quattro possono girare in concorrenza sull’event loop di Python senza intralciarsi.

Il codice completo usato in questo tutorial è disponibile in questa repo GitHub.

Come generare e configurare una chiave API di Gemini

Per usare la Gemini API, dobbiamo creare e configurare una chiave API così che il nostro codice possa comunicare con l’API.

Il modo più semplice è:

• Visita la pagina delle chiavi API di Google AI Studio ed effettua l’accesso.

• Clicca il pulsante Create API key in alto a destra.

• Copia la chiave API in un file chiamato .env nella stessa cartella del codice Python, con il seguente formato:

GEMINI_API_KEY=replace_with_api_key

Nota che usare l’API comporta di solito dei costi. Il piano gratuito copre un accesso limitato a entrambi i modelli Gemini 3.8 Live, ma i dati del free tier vengono usati per migliorare i prodotti Google. Per l’uso in produzione o limiti più alti, dobbiamo assicurarci di avere un metodo di pagamento configurato sulla pagina di fatturazione di Google AI Studio.

Come implementare l’architettura dell’assistente vocale con Gemini 3.8 Live

Questi passaggi sono pensati per essere eseguiti in un notebook Jupyter locale, con ciascun frammento di codice che corrisponde a una cella del notebook. Poiché richiediamo accesso a microfono e altoparlanti, non funzionerà subito su un notebook online come Google Colab.

Passo 1: Setup dell’ambiente e import

Per prima cosa, ci assicuriamo che i pacchetti richiesti siano installati:

pip install google-genai sounddevice python-dotenv

Ecco a cosa servono questi pacchetti:

• google-genai: Il pacchetto ufficiale Google per interagire con i modelli Gemini.

• sounddevice: Gestisce l’hardware audio, registra dal microfono e riproduce sugli altoparlanti.

• python-dotenv: Pacchetto utility per caricare la chiave API di Gemini da un file .env.

Ora possiamo caricare le variabili d’ambiente, verificare la chiave API e inizializzare 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!")

Passo 2: La nostra prima richiesta

Iniziamo capendo il ciclo di vita della connessione Gemini Live inviando un singolo turno di testo e ricevendo parlato e trascrizione in streaming. Invieremo un prompt testuale e riceveremo la risposta in testo e audio. Tuttavia, non riprodurremo ancora l’audio. Per ora concentriamoci sul raccogliere i chunk audio.

La Gemini Live API usa una connessione WebSocket persistente accessibile tramite client.aio.live.connect(). Per configurare l’output vocale e la trascrizione in tempo reale, forniamo un dizionario config:

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

• response_modalities: Usa il valore ["AUDIO"] per dire a Gemini di rispondere con audio parlato.

• output_audio_transcription: Il valore {} dice a Gemini di streammare in simultanea la trascrizione testuale di ciò che sta dicendo.

Ora possiamo testare l’invio di un prompt testuale usando session.send_client_content() e streammare la trascrizione in arrivo.

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

Eseguendo questo codice, dovremmo vedere qualcosa del tipo:

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

Il codice ha catturato i chunk audio, ma non avevamo un riproduttore audio configurato, quindi non potevamo sentirli. Vediamo ora come definire il riproduttore audio.

Passo 3: Riproduzione audio in tempo reale

Nel Passo 2 abbiamo ricevuto migliaia di byte di dati audio, ma non abbiamo sentito nulla. Se scriviamo direttamente sull’hardware audio dentro il ciclo di ricezione, qualsiasi ritardo di rete causerà balbettii, e ogni ritardo di riproduzione bloccherà la ricezione di rete.

Per evitare che la riproduzione blocchi il ricevitore di rete, implementiamo il nostro primo worker: audio_player().

Non serve preoccuparti dei dettagli audio a basso livello. Ti consigliamo di trattarli come black box. 

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

Per testarlo, colleghiamo audio_player() alla nostra richiesta. Questa volta sentiremo Gemini parlare ad alta voce in tempo reale mentre osserviamo la trascrizione in 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!")

Eseguendo questo snippet, ora possiamo sentire la risposta di Gemini.

Passo 4: Acquisire l’input audio dell’utente

Per parlare con Gemini in tempo reale, dobbiamo catturare continuamente la nostra voce dal microfono.

Il nostro secondo worker è audio_recorder(). Ascolta il microfono in background, suddivide il parlato in piccoli chunk e li piazza su input_queue. Impostiamo il sample rate a 16 kHz, il formato standard di parlato atteso da 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!")

Passo 5: Scrivere una funzione per streammare audio in continuo

Nel Passo 2 abbiamo usato send_client_content() per inviare un turno con testo statico. Per lo streaming vocale continuo, la Live API fornisce session.send_realtime_input().

Il nostro terzo worker è send_audio_loop(). Osserva input_queue e, non appena arriva un chunk audio dal microfono, lo inoltra a Gemini sul WebSocket aperto.

Nota che non dobbiamo dire manualmente a Gemini quando iniziamo o smettiamo di parlare: Gemini usa il suo Voice Activity Detection (VAD) integrato per rilevare automaticamente inizio e fine dell’intervento.

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

Così come abbiamo testato la riproduzione audio con un prompt testuale nel Passo 3, ora possiamo testare il nostro streaming dal microfono end‑to‑end con una singola domanda parlata.

Quando eseguiamo la cella seguente, pronunciamo ad alta voce una domanda nel microfono (ad esempio: "Qual è la capitale della Francia?"). Gemini elaborerà direttamente la nostra voce e risponderà con parlato sintetizzato e trascrizione in tempo reale:

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

Passo 6: Multi‑turn e interruzioni

Nota cosa è successo nel test sopra: abbiamo fatto una domanda col microfono, e Gemini ha capito direttamente la nostra voce e ha risposto ad alta voce. Tuttavia, se proviamo a fare una domanda di follow‑up, la sessione è già terminata. 

Per superare questo, dobbiamo affrontare due aspetti cruciali nella creazione di un assistente vocale reale: persistenza multi‑turn e interruzioni.

Persistenza della sessione multi‑turn:

Nel SDK google-genai, session.receive() è un generatore async per un turno. Quando Gemini finisce di parlare la risposta, session.receive() termina. Senza incapsularlo in un ciclo esterno, l’assistente si chiude dopo la prima risposta.

Per supportare conversazioni multi‑turn continue, avvolgiamo session.receive() in un ciclo esterno while not stop_event.is_set()::

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

Barge‑in/interruzione e svuotamento del buffer:

Gemini 3.8 Live ha rilevamento dell’attività vocale e barge‑in nativi. Se Gemini sta parlando e tu inizi a parlare, Gemini interrompe immediatamente la generazione audio e invia un flag: server_content.interrupted == True.

Anche se Gemini smette di inviare nuovo audio, la nostra audio_queue locale potrebbe contenere ancora alcuni chunk in attesa di essere riprodotti. Se non svuotiamo questa coda, gli altoparlanti continueranno a riprodurre la risposta precedente.

Quindi, non appena riceviamo server_content.interrupted, svuotiamo la coda così la riproduzione si ferma all’istante:

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

Mettere tutto insieme

Ecco il nostro quarto e ultimo worker: receive_loop(). Combina persistenza multi‑turn, trascrizione in tempo reale e interruzione istantanea:

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

Passo 7: Assemblare l’assistente vocale completo

Ora orchestriamo i nostri quattro worker concorrenti in run_voice_assistant:

• audio_player(): Consuma da audio_queue e scrive sugli altoparlanti.

• audio_recorder(): Legge dal microfono e inserisce l’audio in input_queue.

• send_audio_loop(): Consuma da input_queue e invia a Gemini usando session.send_realtime_input().

• receive_loop(): Consuma l’output di Gemini con session.receive(), stampa la trascrizione e inserisce l’audio in audio_queue per la riproduzione.

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

Passo 8: Eseguire l’assistente live

Ecco come eseguire l’assistente vocale nel tuo notebook:

await run_voice_assistant()

Note:

• Si raccomandano fortemente le cuffie. Se la voce di Gemini viene riprodotta dagli altoparlanti del laptop, il microfono la capterà e Gemini penserà che tu stia cercando di interromperlo.
• Per fermare l’assistente, clicca semplicemente il pulsante di interruzione del notebook (■).
• Se colleghiamo o scolleghiamo le cuffie mentre il notebook è in esecuzione, le impostazioni del dispositivo audio potrebbero cambiare e potremmo incorrere in un errore audio. In tal caso, dobbiamo riavviare il kernel del notebook e rieseguire le celle in ordine.

Uso avanzato con Gemini 3.8 Live Extended Thinking

Gemini 3.8 Live è disponibile in due versioni:

• Standard (gemini-3.8-live): Ottimizzata per conversazioni speech‑to‑speech a latenza ultra‑bassa. Quando chiama tool, attende in silenzio la risposta del tool prima di rispondere.

• Extended Thinking (gemini-3.8-live-extended-thinking): Offre ragionamento in background e filler conversazionali in parallelo. Può fornire aggiornamenti naturali (per es. "Lascia che lo controlli per te...") mentre esegue tool in background.

Ecco un riepilogo delle differenze tra le due:

Quando usare Gemini 3.8 Live vs 3.8 Live Extended Thinking

Se non sei sicuro di quale delle due versioni usare, ecco il mio schema decisionale. Quando costruisci agenti conversazionali:

• Usa gemini-3.8-live per domande‑risposte dirette e comandi vocali veloci, dove la massima priorità è minimizzare la latenza.

• Usa gemini-3.8-live-extended-thinking per assistenti conversazionali ricchi e agenti che eseguono ragionamento multi‑step, recupero di dati esterni o chiamate API mentre mantengono un dialogo attivo e naturale con l’utente.

Come implementare le chiamate a tool con Gemini 3.8 Live

Uno dei punti di forza della versione extended‑thinking del modello è che può ragionare ed eseguire tool in background mentre mantiene la conversazione. 

Prima di tuffarci nel codice, vediamolo in azione. Ho dotato il modello base di un tool per controllare il meteo. Ecco un video in cui chiedo il meteo a New York; nota come il modello resta in silenzio mentre calcola la risposta:

Ecco la stessa interazione ma con extended thinking:

La seconda interazione è più vivace e sembra più una conversazione normale perché il modello può mantenere la conversazione mentre elabora informazioni in background.

Costruire il tool da usare nell’assistente

Il modello in realtà non esegue i tool per noi. Quello che fa la configurazione del tool è informare il modello che i tool esistono, quando e come usarli. Quando Gemini decide che servono dati esterni, popola response.tool_call con il nome e gli argomenti della funzione.

Per integrare un tool personalizzato in Gemini 3.8 Live, dobbiamo fare da ponte tra il nostro codice locale e il motore di ragionamento del modello. Questo richiede quanto segue:

• Logica di esecuzione: Definisci una normale funzione Python che svolge il lavoro effettivo e restituisce il risultato.

• Mapping del tool: Crea un dizionario (tool_map) che colleghi il nome stringa della funzione all’oggetto Python eseguibile. 

• Dichiarazione della funzione: Crea una FunctionDeclaration che funga da manuale d’uso del tool. Definendo chiaramente nome, descrizione e schema dei parametri (inclusi tipi e campi obbligatori), insegniamo a Gemini esattamente quando usare il tool e come formattare la richiesta. Impostiamo anche behavior="NON_BLOCKING", richiesto da Extended Thinking, così può continuare a parlare mentre il tool gira.

• Configurazione della sessione: Inserisci la dichiarazione nel payload tools_config della sessione. 

Per illustrarlo, creiamo un tool di consultazione meteo:

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

Gestire le chiamate ai tool in modo asincrono

Parlare mentre un tool è in esecuzione richiede due cose. Lato server, la dichiarazione NON_BLOCKING permette a Extended Thinking di continuare a parlare invece di aspettare il risultato. Lato client, neppure il nostro codice deve bloccarsi. Se eseguissimo il tool direttamente dentro il ciclo di ricezione, una chiamata API da 1,5 secondi ci impedirebbe di leggere il parlato filler e i segnali di interruzione di Gemini fino a fine tool.

Per abilitare un vero "parla mentre esegui", aggiorniamo receive_loop_with_tools() con due scelte chiave di design:

1. Esecuzione non bloccante: Avviamo handle_tool_call come task in background concorrente tramite asyncio.create_task(). Così il ciclo di ricezione continua a processare e riprodurre la voce di Gemini senza interruzioni mentre Python recupera il meteo in parallelo.

2. Tracciamento dello stato di interazione: In Extended Thinking, Gemini emette turn_complete: True quando finisce di pronunciare frasi filler intermedie (per es. "Sto controllando il meteo per te..."). Se il codice verificasse solo turn_complete, l’assistente segnalerebbe prematuramente [Listening... Speak now] mentre il tool è ancora in esecuzione! Controllando server_content.interaction_status == "IDLE", il client attende che tutto il ragionamento in background, le chiamate ai tool e il parlato finale siano davvero conclusi prima di riaprire il microfono.

Ecco receive_loop_with_tools(). È identico a receive_loop() a eccezione del nuovo helper handle_tool_call() e del blocco 1, che smista le chiamate ai tool:

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

Infine, implementiamo run_voice_assistant_with_tools(). Oltre a fornire tools_config, questa funzione permette di selezionare tra il modello standard e quello extended thinking. Poiché il modello Extended Thinking richiede un dizionario thinking_config con thinking_level ("low", "medium" o "high"), lo inseriamo nella configurazione della sessione in modo condizionale:

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

Eseguire l’assistente con tool

Ora possiamo eseguire il nostro assistente vocale con tool e confrontare il comportamento live dei due modelli.

Per prima cosa, testa l’assistente con Extended Thinking:

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

Una volta che l’assistente sta ascoltando, fai una domanda che richiede dati live, ad esempio: 

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

Poiché interrogare l’API Open‑Meteo via internet richiede ~1,5 secondi, osserveremo il ragionamento in background in azione:

1. Gemini parla subito ad alta voce per riconoscere la domanda: "Let me check the current weather in Tokyo for you..."
2. Mentre Gemini sta parlando, il nostro task in background recupera in parallelo i dati meteo live.
3. Quando arriva la risposta del tool, Gemini passa a leggere la temperatura live.

Poi eseguiamo lo stesso assistente usando il modello standard Gemini 3.8 Live:

await run_voice_assistant_with_tools("gemini-3.8-live")

Quando facciamo la stessa domanda al modello standard, in questo caso il modello resta completamente in silenzio per ~1,5 secondi in attesa della risposta del tool sulla rete, e poi annuncia direttamente la temperatura senza pronunciare alcun filler.

Per vedere il progetto per intero, consulta la repo GitHub di accompagnamento.

Conclusione

In questo tutorial, abbiamo costruito un assistente vocale full‑duplex completo con Python e Gemini 3.8 Live. Le tre caratteristiche che lo rendono particolarmente utile per il lavoro in tempo reale:

• Architettura audio concorrente: Quattro worker asyncio leggeri comunicano su due code, abilitando registrazione simultanea, streaming audio in tempo reale, riproduzione del parlato e interruzioni istantanee (barge‑in).

• Chiamate a tool in background: Avviare l’esecuzione dei tool come task non bloccanti (asyncio.create_task) consente a Gemini 3.8 Live Extended Thinking di parlare mentre ragiona ed esegue funzioni esterne.

• Gestione dello stato: Tracciare interaction_status == "IDLE" assicura che l’assistente torni ad ascoltare solo dopo che tutto il ragionamento in background, le chiamate ai tool e i turni di parlato finali sono terminati.

Se vuoi iniziare la tua carriera nell’AI engineering, ti raccomando di partire dal nostro AI Engineer for Developers career track, che ti insegna a lavorare con la OpenAI API, Hugging Face, MCP e molto altro!