La API de Live permite interacciones de voz y video bidireccionales de baja latencia con Gemini. Con la API de Live, puedes brindar a los usuarios finales la experiencia de conversaciones de voz naturales y similares a las humanas, y la capacidad de interrumpir las respuestas del modelo con comandos por voz. El modelo puede procesar entradas de texto, audio y video, y puede proporcionar salidas de texto y audio.
Puedes probar la API de Live en Google AI Studio.
Novedades
Consulta el Registro de cambios para conocer las funciones y capacidades nuevas más recientes de la API de Live.
Usa la API de Live
En esta sección, se describe cómo usar la API de Live con uno de nuestros SDKs. Para obtener más información sobre la API subyacente de WebSockets, consulta la referencia de la API de WebSockets.
Para usar todas las funciones, asegúrate de instalar la versión más reciente del SDK, p.ej.,
pip install -U google-genai.
Cómo enviar y recibir mensajes de texto
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())
Cómo recibir audio
En el siguiente ejemplo, se muestra cómo recibir datos de audio y escribirlos en un archivo .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())
Formatos de audio
Los datos de audio de la API de Live siempre son PCM sin procesar, de formato little-endian y de 16 bits. La salida de audio siempre usa una tasa de muestreo de 24 kHz. El audio de entrada es de 16 kHz de forma nativa, pero la API de Live volverá a muestrear si es necesario para que se pueda enviar cualquier tasa de muestreo. Para transmitir la tasa de muestreo del audio de entrada, establece el tipo MIME de cada BLOB que contiene audio en un valor como audio/pcm;rate=16000.
Cómo transmitir audio y video
Instrucciones del sistema
Las instrucciones del sistema te permiten dirigir el comportamiento de un modelo según tus necesidades y casos de uso específicos. Las instrucciones del sistema se pueden establecer en la configuración de configuración y permanecerán vigentes durante toda la sesión.
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"],
}
Actualizaciones de contenido incrementales
Usa actualizaciones incrementales para enviar entradas de texto, establecer el contexto de la sesión o restablecerlo. Para contextos cortos, puedes enviar interacciones paso a paso para representar la secuencia exacta de eventos:
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
}
}
Para contextos más largos, se recomienda proporcionar un solo resumen del mensaje para liberar la ventana de contexto para interacciones posteriores.
Cómo cambiar las voces
La API de Live admite las siguientes voces: Puck, Charon, Kore, Fenrir, Aoede, Leda, Orus y Zephyr.
Para especificar una voz, establece el nombre de la voz dentro del objeto speechConfig como parte de la configuración de la sesión:
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"
}
}
}
Cambiar idioma
La API de Live admite varios idiomas.
Para cambiar el idioma, configura el código de idioma dentro del objeto speechConfig como parte de la configuración de la sesión:
from google.genai import types
config = types.LiveConnectConfig(
response_modalities=["AUDIO"],
speech_config=types.SpeechConfig(
language_code="de-DE",
)
)
Usar herramientas
Puedes definir herramientas como Llamadas a funciones, Ejecución de código y Búsqueda de Google con la API en vivo.
Usa llamadas a función
Puedes definir declaraciones de funciones como parte de la configuración de la sesión. Consulta el instructivo de llamadas a funciones para obtener más información.
Después de recibir llamadas de herramientas, el cliente debe responder con una lista de objetos FunctionResponse con el método 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())
A partir de una sola instrucción, el modelo puede generar varias llamadas a funciones y el código necesario para encadenar sus resultados. Este código se ejecuta en un entorno de zona de pruebas y genera mensajes BidiGenerateContentToolCall posteriores. La ejecución se detiene hasta que los resultados de cada llamada a función están disponibles, lo que garantiza el procesamiento secuencial.
Las entradas y salidas de audio afectan de forma negativa la capacidad del modelo para usar llamadas a función.
Usa la ejecución de código
Puedes definir la ejecución de código como parte de la configuración de la sesión. Consulta el instructivo de ejecución de código para obtener más información.
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())
Usa la fundamentación con la Búsqueda de Google
Puedes habilitar la vinculación con la Búsqueda de Google como parte de la configuración de la sesión. Consulta el instructivo de puesta a tierra para obtener más información.
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())
Combina varias herramientas
Puedes combinar varias herramientas dentro de la API de 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}
Controla las interrupciones
Los usuarios pueden interrumpir la salida del modelo en cualquier momento. Cuando la detección de actividad de voz (VAD) detecta una interrupción, se cancela y descarta la generación en curso. Solo la información que ya se envió al cliente se conserva en el historial de la sesión. Luego, el servidor envía un mensaje BidiGenerateContentServerContent para informar la interrupción.
Además, el servidor de Gemini descarta las llamadas a funciones pendientes y envía
un mensaje BidiGenerateContentServerContent con los IDs de las llamadas canceladas.
async for response in session.receive():
if response.server_content.interrupted is True:
# The generation was interrupted
Configura la detección de actividad de voz (VAD)
Puedes configurar o inhabilitar la detección de actividad de voz (VAD).
Usa la VAD automática
De forma predeterminada, el modelo realiza automáticamente la VAD en un flujo de entrada de audio continuo. La VAD se puede configurar con el campo realtimeInputConfig.automaticActivityDetection de la configuración de configuración.
Cuando la transmisión de audio se pausa durante más de un segundo (por ejemplo, porque el usuario apagó el micrófono), se debe enviar un evento audioStreamEnd para borrar el audio almacenado en caché. El cliente puede reanudar el envío de datos de audio en cualquier 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, la API responderá al audio automáticamente según el VAD. Mientras que send_client_content agrega mensajes al contexto del modelo en orden, send_realtime_input está optimizado para la capacidad de respuesta a expensas del orden determinista.
Cómo configurar el VAD automático
Para tener más control sobre la actividad de VAD, puedes configurar los siguientes parámetros. Consulta la referencia de la API para obtener más información.
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,
}
}
}
Inhabilita la VAD automática
Como alternativa, puedes inhabilitar la VAD automática configurando realtimeInputConfig.automaticActivityDetection.disabled como true en el mensaje de configuración. En esta configuración, el cliente es responsable de detectar la voz del usuario y enviar mensajes activityStart y activityEnd en los momentos adecuados. No se envía un audioStreamEnd en esta configuración. En cambio, cualquier interrupción de la transmisión se marca con un mensaje 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())
# ...
Cómo obtener el recuento de tokens
Puedes encontrar la cantidad total de tokens consumidos en el campo usageMetadata del mensaje del servidor que se muestra.
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}")
Cómo extender la duración de la sesión
La duración máxima de la sesión se puede extender a ilimitada con dos mecanismos:
Además, recibirás un mensaje GoAway antes de que finalice la sesión, lo que te permitirá realizar más acciones.
Habilita la compresión de la ventana de contexto
Para habilitar sesiones más largas y evitar la terminación abrupta de la conexión, puedes habilitar la compresión de la ventana de contexto configurando el campo contextWindowCompression como parte de la configuración de la sesión.
En ContextWindowCompressionConfig, puedes configurar un mecanismo de ventana deslizante y la cantidad de tokens que activa la compresión.
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(),
)
),
)
Configura la reanudación de la sesión
Para evitar que se cierre la sesión cuando el servidor restablezca periódicamente la conexión de WebSocket, configura el campo sessionResumption dentro de la configuración de configuración.
Si pasas esta configuración, el servidor enviará mensajes SessionResumptionUpdate, que se pueden usar para reanudar la sesión pasando el último token de reanudación como SessionResumptionConfig.handle de la conexión posterior.
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())
Recibir un mensaje antes de que se desconecte la sesión
El servidor envía un mensaje GoAway que indica que la conexión actual se finalizará pronto. Este mensaje incluye timeLeft, que indica el tiempo restante y te permite realizar más acciones antes de que la conexión se cierre como 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)
Recibir un mensaje cuando se complete la generación
El servidor envía un mensaje generationComplete que indica que el modelo terminó de generar la respuesta.
async for response in session.receive():
if response.server_content.generation_complete is True:
# The generation is complete
Cambia la resolución del contenido multimedia
Para especificar la resolución de contenido multimedia de entrada, configura el campo mediaResolution como parte de la configuración de la sesión:
from google.genai import types
config = types.LiveConnectConfig(
response_modalities=["AUDIO"],
media_resolution=types.MediaResolution.MEDIA_RESOLUTION_LOW,
)
Cómo recibir transcripciones de audio
Puedes habilitar la transcripción de la salida de audio del modelo. El idioma de transcripción se infiere de la respuesta del modelo.
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())
Limitaciones
Ten en cuenta las siguientes limitaciones de la API de Live y Gemini 2.0 cuando planifiques tu proyecto.
Modalidades de respuesta
Solo puedes establecer una modalidad de respuesta (TEXT o AUDIO) por sesión en la configuración de la sesión. Si intentas configurar ambos, se mostrará un mensaje de error de configuración. Esto significa que puedes configurar el modelo para que responda con texto o audio, pero no con ambos en la misma sesión.
Autenticación de clientes
La API de Live solo proporciona autenticación de servidor a servidor y no se recomienda para el uso directo de clientes. La entrada del cliente debe enrutarse a través de un servidor de aplicaciones intermedio para obtener una autenticación segura con la API de Live.
Duración de la sesión
Si habilitas la compresión de la sesión, puedes extender su duración hasta un tiempo ilimitado. Sin compresión, las sesiones de solo audio se limitan a 15 minutos, y las sesiones de audio y video se limitan a 2 minutos. Si superas estos límites sin compresión, se finalizará la conexión.
Además, puedes configurar la reanudación de la sesión para permitir que el cliente reanude una sesión que se cerró.
Ventana de contexto
Una sesión tiene un límite de ventana de contexto de 32,000 tokens.
Idiomas admitidos
La API de Live admite los siguientes idiomas:
| Idioma | Código BCP-47 |
|---|---|
| Alemán (Alemania) | de-DE |
| inglés (Australia) | en-AU |
| Inglés (Reino Unido) | en-GB |
| Inglés (India) | en-IN |
| Inglés (EE.UU.) | en-US |
| Español (Estados Unidos) | es-US |
| Francés (Francia) | fr-FR |
| Hindi (India) | hi-IN |
| Portugués (Brasil) | pt-BR |
| Árabe (genérico) | ar-XA |
| Español (España) | es-ES |
| Francés (Canadá) | fr-CA |
| Indonesio (Indonesia) | id-ID |
| Italiano (Italia) | it-IT |
| Japonés (Japón) | ja-JP |
| Turco (Turquía) | tr-TR |
| Vietnamita (Vietnam) | vi-VN |
| Bengalí (India) | bn-IN |
| Guyaratí (India) | gu-IN |
| Canarés (India) | kn-IN |
| Malabar (India) | ml-IN |
| Maratí (India) | mr-IN |
| Tamil (India) | ta-IN |
| Telugu (India) | te-IN |
| Holandés (Países Bajos) | nl-NL |
| Coreano (Corea del Sur) | ko-KR |
| Chino mandarín (China) | cmn-CN |
| Polaco (Polonia) | pl-PL |
| Ruso (Rusia) | ru-RU |
| Tailandés (Tailandia) | th-TH |
Integraciones de terceros
Para implementaciones de aplicaciones web y para dispositivos móviles, puedes explorar las siguientes opciones: