Live API

L'API Live consente interazioni vocali e video bidirezionali a bassa latenza con Gemini. Con l'API Live, puoi offrire agli utenti finali l'esperienza di conversazioni vocali naturali e simili a quelle umane, nonché la possibilità di interrompere le risposte del modello utilizzando i comandi vocali. Il modello può elaborare input di testo, audio e video e fornire output di testo e audio.

Puoi provare l'API Live in Google AI Studio.

Novità

Consulta il log delle modifiche per scoprire le ultime funzionalità e le nuove funzionalità dell'API Live.

Utilizzare l'API Live

Questa sezione descrive come utilizzare l'API Live con uno dei nostri SDK. Per ulteriori informazioni sull'API WebSockets sottostante, consulta il riferimento all'API WebSockets.

Per utilizzare tutte le funzionalità, assicurati di installare la versione più recente dell'SDK, ad esempio pip install -U google-genai.

Inviare e ricevere messaggi

import asyncio
from google import genai

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

config = {"response_modalities": ["TEXT"]}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        while True:
            message = input("User> ")
            if message.lower() == "exit":
                break
            await session.send_client_content(
                turns={"role": "user", "parts": [{"text": message}]}, turn_complete=True
            )

            async for response in session.receive():
                if response.text is not None:
                    print(response.text, end="")

if __name__ == "__main__":
    asyncio.run(main())

Ricevere audio

L'esempio seguente mostra come ricevere i dati audio e scriverli in un file .wav.

import asyncio
import wave
from google import genai

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

config = {"response_modalities": ["AUDIO"]}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        wf = wave.open("audio.wav", "wb")
        wf.setnchannels(1)
        wf.setsampwidth(2)
        wf.setframerate(24000)

        message = "Hello? Gemini are you there?"
        await session.send_client_content(
            turns={"role": "user", "parts": [{"text": message}]}, turn_complete=True
        )

        async for idx,response in async_enumerate(session.receive()):
            if response.data is not None:
                wf.writeframes(response.data)

            # Un-comment this code to print audio data info
            # if response.server_content.model_turn is not None:
            #      print(response.server_content.model_turn.parts[0].inline_data.mime_type)

        wf.close()

if __name__ == "__main__":
    asyncio.run(main())

Formati audio

L'API Live supporta i seguenti formati audio:

  • Formato audio di input: audio PCM non compresso a 16 bit a 16 kHz little-endian
  • Formato audio di output: audio PCM non compresso a 16 bit a 24 kHz little-endian

Riprodurre audio e video in streaming

Istruzioni di sistema

Le istruzioni di sistema ti consentono di indirizzare il comportamento di un modello in base alle tue esigenze e ai tuoi casi d'uso specifici. Le istruzioni di sistema possono essere impostate nella configurazione di setup e rimarranno in vigore per l'intera sessione.

from google.genai import types

config = {
    "system_instruction": types.Content(
        parts=[
            types.Part(
                text="You are a helpful assistant and answer in a friendly tone."
            )
        ]
    ),
    "response_modalities": ["TEXT"],
}

Aggiornamenti incrementali dei contenuti

Utilizza gli aggiornamenti incrementali per inviare input di testo, stabilire il contesto della sessione o ripristinare il contesto della sessione. Per contesti brevi, puoi inviare interazioni passo passo per rappresentare la sequenza esatta di eventi:

PythonJSON
turns = [
    {"role": "user", "parts": [{"text": "What is the capital of France?"}]},
    {"role": "model", "parts": [{"text": "Paris"}]},
]

await session.send_client_content(turns=turns, turn_complete=False)

turns = [{"role": "user", "parts": [{"text": "What is the capital of Germany?"}]}]

await session.send_client_content(turns=turns, turn_complete=True)
{
  "clientContent": {
    "turns": [
      {
        "parts":[
          {
            "text": ""
          }
        ],
        "role":"user"
      },
      {
        "parts":[
          {
            "text": ""
          }
        ],
        "role":"model"
      }
    ],
    "turnComplete": true
  }
}

Per contesti più lunghi, ti consigliamo di fornire un riepilogo di un singolo messaggio per liberare la finestra di contesto per le interazioni successive.

Cambiare le voci

L'API Live supporta le seguenti voci: Puck, Charon, Kore, Fenrir, Aoede, Leda, Orus e Zephyr.

Per specificare una voce, imposta il nome della voce all'interno dell'oggetto speechConfig come parte della configurazione della sessione:

PythonJSON
from google.genai import types

config = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    speech_config=types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Kore")
        )
    )
)
{
  "voiceConfig": {
    "prebuiltVoiceConfig": {
      "voiceName": "Kore"
    }
  }
}

Cambia lingua

L'API Live supporta più lingue.

Per cambiare la lingua, imposta il codice lingua all'interno dell'oggetto speechConfig come parte della configurazione della sessione:

from google.genai import types

config = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    speech_config=types.SpeechConfig(
        language_code="de-DE",
    )
)

Utilizzare gli strumenti

Con l'API Live puoi definire strumenti come chiamate di funzioni, esecuzione di codice e Ricerca Google.

Utilizzare le chiamate di funzione

Puoi definire le dichiarazioni di funzione come parte della configurazione della sessione. Per saperne di più, consulta il tutorial sulle chiamate di funzione.

Dopo aver ricevuto le chiamate dello strumento, il client deve rispondere con un elenco di oggetti FunctionResponse utilizzando il metodo session.send_tool_response.

import asyncio
from google import genai
from google.genai import types

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

# Simple function definitions
turn_on_the_lights = {"name": "turn_on_the_lights"}
turn_off_the_lights = {"name": "turn_off_the_lights"}

tools = [{"function_declarations": [turn_on_the_lights, turn_off_the_lights]}]
config = {"response_modalities": ["TEXT"], "tools": tools}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        prompt = "Turn on the lights please"
        await session.send_client_content(turns={"parts": [{"text": prompt}]})

        async for chunk in session.receive():
            if chunk.server_content:
                if chunk.text is not None:
                    print(chunk.text)
            elif chunk.tool_call:
                function_responses = []
                for fc in tool_call.function_calls:
                    function_response = types.FunctionResponse(
                        id=fc.id,
                        name=fc.name,
                        response={ "result": "ok" } # simple, hard-coded function response
                    )
                    function_responses.append(function_response)

                await session.send_tool_response(function_responses=function_responses)


if __name__ == "__main__":
    asyncio.run(main())

Da un singolo prompt, il modello può generare più chiamate di funzione e il codice necessario per collegarne gli output. Questo codice viene eseguito in un ambiente sandbox, generando messaggi BidiGenerateContentToolCall successivi. L'esecuzione viene messa in pausa fino a quando non sono disponibili i risultati di ogni chiamata di funzione, il che garantisce l'elaborazione sequenziale.

Gli input e gli output audio influiscono negativamente sulla capacità del modello di utilizzare la chiamata di funzioni.

Utilizza l'esecuzione di codice

Puoi definire l'esecuzione del codice nell'ambito della configurazione della sessione. Per scoprire di più, consulta il tutorial sull'esecuzione del codice.

import asyncio
from google import genai
from google.genai import types

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

tools = [{'code_execution': {}}]
config = {"response_modalities": ["TEXT"], "tools": tools}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        prompt = "Compute the largest prime palindrome under 100000."
        await session.send_client_content(turns={"parts": [{"text": prompt}]})

        async for chunk in session.receive():
            if chunk.server_content:
                if chunk.text is not None:
                    print(chunk.text)
            
                model_turn = chunk.server_content.model_turn
                if model_turn:
                    for part in model_turn.parts:
                      if part.executable_code is not None:
                        print(part.executable_code.code)

                      if part.code_execution_result is not None:
                        print(part.code_execution_result.output)

if __name__ == "__main__":
    asyncio.run(main())

Puoi attivare il grounding con la Ricerca Google nell'ambito della configurazione della sessione. Per saperne di più, consulta il tutorial sull'isolamento.

import asyncio
from google import genai
from google.genai import types

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

tools = [{'google_search': {}}]
config = {"response_modalities": ["TEXT"], "tools": tools}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        prompt = "When did the last Brazil vs. Argentina soccer match happen?"
        await session.send_client_content(turns={"parts": [{"text": prompt}]})

        async for chunk in session.receive():
            if chunk.server_content:
                if chunk.text is not None:
                    print(chunk.text)

                # The model might generate and execute Python code to use Search
                model_turn = chunk.server_content.model_turn
                if model_turn:
                    for part in model_turn.parts:
                      if part.executable_code is not None:
                        print(part.executable_code.code)

                      if part.code_execution_result is not None:
                        print(part.code_execution_result.output)

if __name__ == "__main__":
    asyncio.run(main())

Combinare più strumenti

Puoi combinare più strumenti all'interno dell'API Live:

prompt = """
Hey, I need you to do three things for me.

1. Compute the largest prime palindrome under 100000.
2. Then use Google Search to look up information about the largest earthquake in California the week of Dec 5 2024?
3. Turn on the lights

Thanks!
"""

tools = [
    {"google_search": {}},
    {"code_execution": {}},
    {"function_declarations": [turn_on_the_lights, turn_off_the_lights]},
]

config = {"response_modalities": ["TEXT"], "tools": tools}

Gestire le interruzioni

Gli utenti possono interrompere l'output del modello in qualsiasi momento. Quando il rilevamento dell'attività vocale (VAD) rileva un'interruzione, la generazione in corso viene annullata e ignorata. Nella cronologia della sessione vengono conservate solo le informazioni già inviate al client. Il server invia quindi un messaggio BidiGenerateContentServerContent per segnalare l'interruzione.

Inoltre, il server Gemini ignora le chiamate di funzione in attesa e invia un messaggio BidiGenerateContentServerContent con gli ID delle chiamate annullate.

async for response in session.receive():
    if response.server_content.interrupted is True:
        # The generation was interrupted

Configurare il rilevamento dell'attività vocale (VAD)

Puoi configurare o disattivare il rilevamento dell'attività vocale (VAD).

Utilizzare la VAD automatica

Per impostazione predefinita, il modello esegue automaticamente la VAD su un flusso di input audio continuo. La VAD può essere configurata con il campo realtimeInputConfig.automaticActivityDetection della configurazione di configurazione.

Quando lo stream audio viene messo in pausa per più di un secondo (ad esempio perché l'utente ha disattivato il microfono), deve essere inviato un evento audioStreamEnd per svuotare la cache dell'audio. Il client può riprendere l'invio di dati audio in qualsiasi momento.

# example audio file to try:
# URL = "https://storage.googleapis.com/generativeai-downloads/data/hello_are_you_there.pcm"
# !wget -q $URL -O sample.pcm
import asyncio
from pathlib import Path
from google import genai

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

config = {"response_modalities": ["TEXT"]}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        audio_bytes = Path("sample.pcm").read_bytes()

        await session.send_realtime_input(
            audio=types.Blob(data=audio_bytes, mime_type="audio/pcm;rate=16000")
        )

        # if stream gets paused, send:
        # await session.send_realtime_input(audio_stream_end=True)

        async for response in session.receive():
            if response.text is not None:
                print(response.text)

if __name__ == "__main__":
    asyncio.run(main())

Con send_realtime_input, l'API risponderà automaticamente all'audio in base al VAD. Sebbene send_client_content aggiunga i messaggi al contesto del modello in ordine, send_realtime_input è ottimizzato per la reattività a spese dell'ordinamento deterministico.

Configurare la VAD automatica

Per un maggiore controllo sull'attività VAD, puoi configurare i seguenti parametri. Per ulteriori informazioni, consulta il riferimento API.

from google.genai import types

config = {
    "response_modalities": ["TEXT"],
    "realtime_input_config": {
        "automatic_activity_detection": {
            "disabled": False, # default
            "start_of_speech_sensitivity": types.StartSensitivity.START_SENSITIVITY_LOW,
            "end_of_speech_sensitivity": types.EndSensitivity.END_SENSITIVITY_LOW,
            "prefix_padding_ms": 20,
            "silence_duration_ms": 100,
        }
    }
}

Disattivare la VAD automatica

In alternativa, la VAD automatica può essere disattivata impostando realtimeInputConfig.automaticActivityDetection.disabled su true nel messaggio di configurazione. In questa configurazione, il client è responsabile del rilevamento della voce dell'utente e dell'invio di messaggi activityStart e activityEnd al momento opportuno. In questa configurazione non viene inviato un audioStreamEnd. Invece, qualsiasi interruzione dello stream viene contrassegnata da un messaggio activityEnd.

config = {
    "response_modalities": ["TEXT"],
    "realtime_input_config": {"automatic_activity_detection": {"disabled": True}},
}

async with client.aio.live.connect(model=model, config=config) as session:
    # ...
    await session.send_realtime_input(activity_start=types.ActivityStart())
    await session.send_realtime_input(
        audio=types.Blob(data=audio_bytes, mime_type="audio/pcm;rate=16000")
    )
    await session.send_realtime_input(activity_end=types.ActivityEnd())
    # ...

Ottieni il conteggio dei token

Puoi trovare il numero totale di token consumati nel campo usageMetadata del messaggio del server restituito.

async for message in session.receive():
    # The server will periodically send messages that include UsageMetadata.
    if message.usage_metadata:
        usage = message.usage_metadata
        print(
            f"Used {usage.total_token_count} tokens in total. Response token breakdown:"
        )
        for detail in usage.response_tokens_details:
            match detail:
                case types.ModalityTokenCount(modality=modality, token_count=count):
                    print(f"{modality}: {count}")

Estendere la durata della sessione

La durata massima della sessione può essere estesa a illimitata con due meccanismi:

Inoltre, prima del termine della sessione riceverai un messaggio GoAway che ti consentirà di intraprendere ulteriori azioni.

Attiva la compressione della finestra di contesto

Per consentire sessioni più lunghe ed evitare interruzioni improvvise della connessione, puoi attivare la compressione della finestra di contesto impostando il campo contextWindowCompression nell'ambito della configurazione della sessione.

In ContextWindowCompressionConfig, puoi configurare un meccanismo a finestra scorrevole e il numero di token che attivano la compressione.

from google.genai import types

config = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    context_window_compression=(
        # Configures compression with default parameters.
        types.ContextWindowCompressionConfig(
            sliding_window=types.SlidingWindow(),
        )
    ),
)

Configurare la ripresa della sessione

Per impedire la terminazione della sessione quando il server reimposta periodicamente la connessione WebSocket, configura il campo sessionResumption all'interno della configurazione di configurazione.

Se passi questa configurazione, il server invia messaggi SessionResumptionUpdate, che possono essere utilizzati per riprendere la sessione passando l'ultimo token di ripresa come SessionResumptionConfig.handle della connessione successiva.

import asyncio
from google import genai
from google.genai import types

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

async def main():
    print(f"Connecting to the service with handle {previous_session_handle}...")
    async with client.aio.live.connect(
        model=model,
        config=types.LiveConnectConfig(
            response_modalities=["AUDIO"],
            session_resumption=types.SessionResumptionConfig(
                # The handle of the session to resume is passed here,
                # or else None to start a new session.
                handle=previous_session_handle
            ),
        ),
    ) as session:
        while True:
            await session.send_client_content(
                turns=types.Content(
                    role="user", parts=[types.Part(text="Hello world!")]
                )
            )
            async for message in session.receive():
                # Periodically, the server will send update messages that may
                # contain a handle for the current state of the session.
                if message.session_resumption_update:
                    update = message.session_resumption_update
                    if update.resumable and update.new_handle:
                        # The handle should be retained and linked to the session.
                        return update.new_handle

                # For the purposes of this example, placeholder input is continually fed
                # to the model. In non-sample code, the model inputs would come from
                # the user.
                if message.server_content and message.server_content.turn_complete:
                    break

if __name__ == "__main__":
    asyncio.run(main())

Ricevere un messaggio prima che la sessione si disconnetta

Il server invia un messaggio GoAway che indica che la connessione corrente verrà terminata a breve. Questo messaggio include timeLeft, che indica il tempo rimanente e ti consente di intraprendere ulteriori azioni prima che la connessione venga interrotta come ABORTED.

async for response in session.receive():
    if response.go_away is not None:
        # The connection will soon be terminated
        print(response.go_away.time_left)

Ricevere un messaggio al termine della generazione

Il server invia un messaggio generationComplete che indica che il modello ha completato la generazione della risposta.

async for response in session.receive():
    if response.server_content.generation_complete is True:
        # The generation is complete

Modificare la risoluzione dei contenuti multimediali

Puoi specificare la risoluzione dei contenuti multimediali di input impostando il campo mediaResolution nell'ambito della configurazione della sessione:

from google.genai import types

config = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    media_resolution=types.MediaResolution.MEDIA_RESOLUTION_LOW,
)

Ricevere trascrizioni audio

Puoi attivare la trascrizione dell'output audio del modello. La lingua della trascrizione viene dedotta dalla risposta del modello.

import asyncio
from google import genai
from google.genai import types

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

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

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        message = "Hello? Gemini are you there?"

        await session.send_client_content(
            turns={"role": "user", "parts": [{"text": message}]}, turn_complete=True
        )

        async for response in session.receive():
            if response.server_content.model_turn:
                print("Model turn:", response.server_content.model_turn)
            if response.server_content.output_transcription:
                print("Transcript:", response.server_content.output_transcription.text)


if __name__ == "__main__":
    asyncio.run(main())

Limitazioni

Tieni presente i seguenti limiti dell'API Live e di Gemini 2.0 quando pianifichi il tuo progetto.

Modalità di risposta

Nella configurazione della sessione puoi impostare una sola modalità di risposta (TEXT o AUDIO) per sessione. Se provi a impostare entrambi, verrà visualizzato un messaggio di errore di configurazione. Ciò significa che puoi configurare il modello in modo che risponda con testo o audio, ma non con entrambi nella stessa sessione.

Autenticazione client

L'API Live fornisce solo l'autenticazione da server a server e non è consigliata per l'utilizzo diretto del client. L'input del client deve essere instradato tramite un server di applicazioni intermedio per l'autenticazione sicura con l'API Live.

Durata della sessione

La durata della sessione può essere estesa a illimitata attivando la compressione della sessione. Senza compressione, le sessioni solo audio sono limitate a 15 minuti, mentre quelle con audio e video sono limitate a 2 minuti. Il superamento di questi limiti senza compressione interromperà la connessione.

Inoltre, puoi configurare la ripresa della sessione per consentire al client di riprendere una sessione interrotta.

Finestra contestuale

Una sessione ha un limite di finestra contestuale di 32.000 token.

Lingue supportate

L'API Live supporta le seguenti lingue:

Lingua Codice BCP-47
Tedesco (Germania) de-DE
Inglese (Australia) en-AU
Inglese (Regno Unito) en-GB
Inglese (India) en-IN
Inglese (USA) en-US
Spagnolo (Stati Uniti) es-US
Francese (Francia) fr-FR
Hindi (India) hi-IN
Portoghese (Brasile) pt-BR
Arabo (generico) ar-XA
Spagnolo (Spagna) es-ES
Francese (Canada) fr-CA
Indonesiano (Indonesia) id-ID
Italiano (Italia) it-IT
Giapponese (Giappone) ja-JP
Turco (Turchia) tr-TR
Vietnamita (Vietnam) vi-VN
Bengalese (India) bn-IN
Gujarati (India) gu-IN
Kannada (India) kn-IN
Malayalam (India) ml-IN
Marathi (India) mr-IN
Tamil (India) ta-IN
Telugu (India) te-IN
Olandese (Paesi Bassi) nl-NL
Coreano (Corea del Sud) ko-KR
Cinese mandarino (Cina) cmn-CN
Polacco (Polonia) pl-PL
Russo (Russia) ru-RU
Thailandese (Thailandia) th-TH

Integrazioni di terze parti

Per i deployment di app web e mobile, puoi esplorare le opzioni disponibili in: