La API de Gemini Live permite la interacción bidireccional en tiempo real con los modelos de Gemini, admite entradas de audio, video y texto, y salidas de audio nativas. En esta guía, se explica cómo realizar la integración directamente con la API a través de WebSockets sin procesar.
Descripción general
La API de Gemini Live usa WebSockets para la comunicación en tiempo real. A diferencia del uso de un SDK, este enfoque implica administrar directamente la conexión de WebSocket y enviar o recibir mensajes en un formato JSON específico definido por la API.
Conceptos clave:
- Extremo de WebSocket: Es la URL específica a la que se debe conectar.
- Formato de mensaje: Toda la comunicación se realiza a través de mensajes JSON que cumplen con las estructuras
LiveSessionRequestyLiveSessionResponse. - Administración de sesiones: Eres responsable de mantener la conexión de WebSocket.
Autenticación
La autenticación se controla incluyendo tu clave de API como parámetro de consulta en la URL de WebSocket.
El formato del extremo es el siguiente:
wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent?key=YOUR_API_KEY
Reemplaza YOUR_API_KEY por tu clave de API real.
Autenticación con tokens efímeros
Si usas tokens efímeros, debes conectarte al extremo v1alpha.
El token efímero se debe pasar como un parámetro de búsqueda access_token.
El formato del extremo para las claves efímeras es el siguiente:
wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1alpha.GenerativeService.BidiGenerateContentConstrained?access_token={short-lived-token}
Reemplaza {short-lived-token} por el token efímero real.
Cómo conectarse a la API de Live
Para iniciar una sesión en vivo, establece una conexión WebSocket con el extremo autenticado.
El primer mensaje que se envía a través de WebSocket debe ser un LiveSessionRequest que contenga el config.
Para conocer todas las opciones de configuración, consulta la referencia de la API de Live - WebSockets.
Python
import asyncio
import websockets
import json
API_KEY = "YOUR_API_KEY"
MODEL_NAME = "gemini-2.5-flash-native-audio-preview-12-2025"
WS_URL = f"wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent?key={API_KEY}"
async def connect_and_configure():
async with websockets.connect(WS_URL) as websocket:
print("WebSocket Connected")
# 1. Send the initial configuration
config_message = {
"config": {
"model": f"models/{MODEL_NAME}",
"responseModalities": ["AUDIO"],
"systemInstruction": {
"parts": [{"text": "You are a helpful assistant."}]
}
}
}
await websocket.send(json.dumps(config_message))
print("Configuration sent")
# Keep the session alive for further interactions
await asyncio.sleep(3600) # Example: keep open for an hour
async def main():
await connect_and_configure()
if __name__ == "__main__":
asyncio.run(main())
JavaScript
const API_KEY = "YOUR_API_KEY";
const MODEL_NAME = "gemini-2.5-flash-native-audio-preview-12-2025";
const WS_URL = `wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent?key=${API_KEY}`;
const websocket = new WebSocket(WS_URL);
websocket.onopen = () => {
console.log('WebSocket Connected');
// 1. Send the initial configuration
const configMessage = {
config: {
model: `models/${MODEL_NAME}`,
responseModalities: ['AUDIO'],
systemInstruction: {
parts: [{ text: 'You are a helpful assistant.' }]
}
}
};
websocket.send(JSON.stringify(configMessage));
console.log('Configuration sent');
};
websocket.onmessage = (event) => {
const response = JSON.parse(event.data);
console.log('Received:', response);
// Handle different types of responses here
};
websocket.onerror = (error) => {
console.error('WebSocket Error:', error);
};
websocket.onclose = () => {
console.log('WebSocket Closed');
};
Enviando mensaje de texto
Para enviar texto como entrada, crea un objeto LiveSessionRequest con el campo realtimeInput completado con texto.
Python
# Inside the websocket context
async def send_text(websocket, text):
text_message = {
"realtimeInput": {
"text": text
}
}
await websocket.send(json.dumps(text_message))
print(f"Sent text: {text}")
# Example usage: await send_text(websocket, "Hello, how are you?")
JavaScript
function sendTextMessage(text) {
if (websocket.readyState === WebSocket.OPEN) {
const textMessage = {
realtimeInput: {
text: text
}
};
websocket.send(JSON.stringify(textMessage));
console.log('Text message sent:', text);
} else {
console.warn('WebSocket not open.');
}
}
// Example usage:
sendTextMessage("Hello, how are you?");
Cómo enviar audio
El audio debe enviarse como datos PCM sin procesar (audio PCM sin procesar de 16 bits, 16 kHz, little-endian). Construye un objeto LiveSessionRequest con el campo realtimeInput, que contiene un objeto Blob con los datos de audio. El mimeType es fundamental.
Python
# Inside the websocket context
async def send_audio_chunk(websocket, chunk_bytes):
import base64
encoded_data = base64.b64encode(chunk_bytes).decode('utf-8')
audio_message = {
"realtimeInput": {
"audio": {
"data": encoded_data,
"mimeType": "audio/pcm;rate=16000"
}
}
}
await websocket.send(json.dumps(audio_message))
# print("Sent audio chunk") # Avoid excessive logging
# Assuming 'chunk' is your raw PCM audio bytes
# await send_audio_chunk(websocket, chunk)
JavaScript
// Assuming 'chunk' is a Buffer of raw PCM audio
function sendAudioChunk(chunk) {
if (websocket.readyState === WebSocket.OPEN) {
const audioMessage = {
realtimeInput: {
audio: {
data: chunk.toString('base64'),
mimeType: 'audio/pcm;rate=16000'
}
}
};
websocket.send(JSON.stringify(audioMessage));
// console.log('Sent audio chunk');
}
}
// Example usage: sendAudioChunk(audioBuffer);
Para ver un ejemplo de cómo obtener el audio del dispositivo cliente (p.ej., el navegador), consulta el ejemplo de extremo a extremo en GitHub.
Enviando video
Los fotogramas de video se envían como imágenes individuales (p.ej., JPEG o PNG). Al igual que con el audio, usa realtimeInput con un Blob y especifica el mimeType correcto.
Python
# Inside the websocket context
async def send_video_frame(websocket, frame_bytes, mime_type="image/jpeg"):
import base64
encoded_data = base64.b64encode(frame_bytes).decode('utf-8')
video_message = {
"realtimeInput": {
"video": {
"data": encoded_data,
"mimeType": mime_type
}
}
}
await websocket.send(json.dumps(video_message))
# print("Sent video frame")
# Assuming 'frame' is your JPEG-encoded image bytes
# await send_video_frame(websocket, frame)
JavaScript
// Assuming 'frame' is a Buffer of JPEG-encoded image data
function sendVideoFrame(frame, mimeType = 'image/jpeg') {
if (websocket.readyState === WebSocket.OPEN) {
const videoMessage = {
realtimeInput: {
video: {
data: frame.toString('base64'),
mimeType: mimeType
}
}
};
websocket.send(JSON.stringify(videoMessage));
// console.log('Sent video frame');
}
}
// Example usage: sendVideoFrame(jpegBuffer);
Para ver un ejemplo de cómo obtener el video del dispositivo cliente (p.ej., el navegador), consulta el ejemplo de extremo a extremo en GitHub.
Cómo recibir respuestas
El WebSocket enviará mensajes LiveSessionResponse. Debes analizar estos mensajes JSON y controlar diferentes tipos de contenido.
Python
# Inside the websocket context, in a receive loop
async def receive_loop(websocket):
async for message in websocket:
response = json.loads(message)
print("Received:", response)
if "serverContent" in response:
server_content = response["serverContent"]
# Receiving Audio
if "modelTurn" in server_content and "parts" in server_content["modelTurn"]:
for part in server_content["modelTurn"]["parts"]:
if "inlineData" in part:
audio_data_b64 = part["inlineData"]["data"]
# Process or play the base64 encoded audio data
# audio_data = base64.b64decode(audio_data_b64)
print(f"Received audio data (base64 len: {len(audio_data_b64)})")
# Receiving Text Transcriptions
if "inputTranscription" in server_content:
print(f"User: {server_content['inputTranscription']['text']}")
if "outputTranscription" in server_content:
print(f"Gemini: {server_content['outputTranscription']['text']}")
# Handling Tool Calls
if "toolCall" in response:
await handle_tool_call(websocket, response["toolCall"])
# Example usage: await receive_loop(websocket)
Para ver un ejemplo de cómo controlar la respuesta, consulta el ejemplo de extremo a extremo en GitHub.
JavaScript
websocket.onmessage = (event) => {
const response = JSON.parse(event.data);
console.log('Received:', response);
if (response.serverContent) {
const serverContent = response.serverContent;
// Receiving Audio
if (serverContent.modelTurn?.parts) {
for (const part of serverContent.modelTurn.parts) {
if (part.inlineData) {
const audioData = part.inlineData.data; // Base64 encoded string
// Process or play audioData
console.log(`Received audio data (base64 len: ${audioData.length})`);
}
}
}
// Receiving Text Transcriptions
if (serverContent.inputTranscription) {
console.log('User:', serverContent.inputTranscription.text);
}
if (serverContent.outputTranscription) {
console.log('Gemini:', serverContent.outputTranscription.text);
}
}
// Handling Tool Calls
if (response.toolCall) {
handleToolCall(response.toolCall);
}
};
Cómo controlar las llamadas a herramientas
Cuando el modelo solicita una llamada a una herramienta, el objeto LiveSessionResponse contendrá un campo toolCall. Debes ejecutar la función de forma local y enviar el resultado de vuelta al WebSocket con un LiveSessionRequest que incluya el campo toolResponse.
Python
# Placeholder for your tool function
def my_tool_function(args):
print(f"Executing tool with args: {args}")
# Implement your tool logic here
return {"status": "success", "data": "some result"}
async def handle_tool_call(websocket, tool_call):
function_responses = []
for fc in tool_call["functionCalls"]:
# 1. Execute the function locally
try:
result = my_tool_function(fc.get("args", {}))
response_data = {"result": result}
except Exception as e:
print(f"Error executing tool {fc['name']}: {e}")
response_data = {"error": str(e)}
# 2. Prepare the response
function_responses.append({
"name": fc["name"],
"id": fc["id"],
"response": response_data
})
# 3. Send the tool response back to the session
tool_response_message = {
"toolResponse": {
"functionResponses": function_responses
}
}
await websocket.send(json.dumps(tool_response_message))
print("Sent tool response")
# This function is called within the receive_loop when a toolCall is detected.
JavaScript
// Placeholder for your tool function
function myToolFunction(args) {
console.log(`Executing tool with args:`, args);
// Implement your tool logic here
return { status: 'success', data: 'some result' };
}
function handleToolCall(toolCall) {
const functionResponses = [];
for (const fc of toolCall.functionCalls) {
// 1. Execute the function locally
let result;
try {
result = myToolFunction(fc.args || {});
} catch (e) {
console.error(`Error executing tool ${fc.name}:`, e);
result = { error: e.message };
}
// 2. Prepare the response
functionResponses.push({
name: fc.name,
id: fc.id,
response: { result }
});
}
// 3. Send the tool response back to the session
if (websocket.readyState === WebSocket.OPEN) {
const toolResponseMessage = {
toolResponse: {
functionResponses: functionResponses
}
};
websocket.send(JSON.stringify(toolResponseMessage));
console.log('Sent tool response');
} else {
console.warn('WebSocket not open to send tool response.');
}
}
// This function is called within websocket.onmessage when a toolCall is detected.
¿Qué sigue?
- Lee la guía completa de Capacidades de la API de Live para conocer las capacidades y configuraciones clave, incluidas la detección de actividad de voz y las funciones de audio nativas.
- Lee la guía de Uso de herramientas para aprender a integrar la API en vivo con herramientas y llamadas a funciones.
- Lee la guía de administración de sesiones para administrar conversaciones de larga duración.
- Lee la guía sobre tokens efímeros para obtener información sobre la autenticación segura en aplicaciones cliente-servidor.
- Para obtener más información sobre la API de WebSockets subyacente, consulta la referencia de la API de WebSockets.