Live API

Live API обеспечивает двунаправленное голосовое и видео взаимодействие с Gemini с низкой задержкой, позволяя вам общаться с Gemini в прямом эфире, одновременно транслируя видео или делясь своим экраном. Используя Live API, вы можете предоставить конечным пользователям опыт естественного, похожего на человеческое голосовое общение.

Вы можете попробовать Live API в Google AI Studio . Чтобы использовать Live API в Google AI Studio, выберите Stream .

Как работает Live API

Потоковое вещание

Live API использует потоковую модель через соединение WebSocket . При взаимодействии с API создается постоянное соединение. Ваши входные данные (аудио, видео или текст) непрерывно передаются в модель, а ответ модели (текст или аудио) передается обратно в режиме реального времени через то же соединение.

Двунаправленная потоковая передача обеспечивает низкую задержку и поддерживает такие функции, как обнаружение голосовой активности, использование инструментов и генерация речи.

Обзор API в реальном времени

Более подробную информацию о базовом API WebSockets см. в справочнике по API WebSockets .

Создание с помощью Live API {:#building-with-live-api}ß

Установление связи

В следующем примере показано, как создать соединение с помощью ключа API:

Питон 

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:
        print("Session started")

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

JavaScript 

import { GoogleGenAI } from '@google/genai';

const ai = new GoogleGenAI({ apiKey: "GOOGLE_API_KEY" });
const model = 'gemini-2.0-flash-live-001';
const config = { responseModalities: [Modality.TEXT] };

async function main() {

    const session = await ai.live.connect({
        model: model,
        callbacks: {
            onopen: function () {
                console.debug('Opened');
            },
            onmessage: function (message) {
                console.debug(message);
            },
            onerror: function (e) {
                console.debug('Error:', e.message);
            },
            onclose: function (e) {
                console.debug('Close:', e.reason);
            },
        },
        config: config,
    });

    // Send content...

    session.close();
}

main();

Отправка и получение текста

Вот как можно отправлять и получать текстовые сообщения:

Питон 

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:
        message = "Hello, how are you?"
        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())

JavaScript 

import { GoogleGenAI, Modality } from '@google/genai';

const ai = new GoogleGenAI({ apiKey: "GOOGLE_API_KEY" });
const model = 'gemini-2.0-flash-live-001';
const config = { responseModalities: [Modality.TEXT] };

async function live() {
    const responseQueue = [];

    async function waitMessage() {
        let done = false;
        let message = undefined;
        while (!done) {
            message = responseQueue.shift();
            if (message) {
                done = true;
            } else {
                await new Promise((resolve) => setTimeout(resolve, 100));
            }
        }
        return message;
    }

    async function handleTurn() {
        const turns = [];
        let done = false;
        while (!done) {
            const message = await waitMessage();
            turns.push(message);
            if (message.serverContent && message.serverContent.turnComplete) {
                done = true;
            }
        }
        return turns;
    }

    const session = await ai.live.connect({
        model: model,
        callbacks: {
            onopen: function () {
                console.debug('Opened');
            },
            onmessage: function (message) {
                responseQueue.push(message);
            },
            onerror: function (e) {
                console.debug('Error:', e.message);
            },
            onclose: function (e) {
                console.debug('Close:', e.reason);
            },
        },
        config: config,
    });

    const simple = 'Hello how are you?';
    session.sendClientContent({ turns: simple });

    const turns = await handleTurn();
    for (const turn of turns) {
        if (turn.text) {
            console.debug('Received text: %s\n', turn.text);
        }
        else if (turn.data) {
            console.debug('Received inline data: %s\n', turn.data);
        }
    }

    session.close();
}

async function main() {
    await live().catch((e) => console.error('got error', e));
}

main();

Отправка и получение аудио

Вы можете отправить аудио, преобразовав его в формат 16 бит PCM, 16 кГц, моно. Этот пример считывает файл WAV и отправляет его в правильном формате:

Питон 

# Test file: https://storage.googleapis.com/generativeai-downloads/data/16000.wav
# Install helpers for converting files: pip install librosa soundfile
import asyncio
import io
from pathlib import Path
from google import genai
from google.genai import types
import soundfile as sf
import librosa

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:

        buffer = io.BytesIO()
        y, sr = librosa.load("sample.wav", sr=16000)
        sf.write(buffer, y, sr, format='RAW', subtype='PCM_16')
        buffer.seek(0)
        audio_bytes = buffer.read()

        # If already in correct format, you can use this:
        # audio_bytes = Path("sample.pcm").read_bytes()

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

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

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

JavaScript 

// Test file: https://storage.googleapis.com/generativeai-downloads/data/16000.wav
// Install helpers for converting files: npm install wavefile
import { GoogleGenAI, Modality } from '@google/genai';
import * as fs from "node:fs";
import pkg from 'wavefile';
const { WaveFile } = pkg;

const ai = new GoogleGenAI({ apiKey: "GOOGLE_API_KEY" });
const model = 'gemini-2.0-flash-live-001';
const config = { responseModalities: [Modality.TEXT] };

async function live() {
    const responseQueue = [];

    async function waitMessage() {
        let done = false;
        let message = undefined;
        while (!done) {
            message = responseQueue.shift();
            if (message) {
                done = true;
            } else {
                await new Promise((resolve) => setTimeout(resolve, 100));
            }
        }
        return message;
    }

    async function handleTurn() {
        const turns = [];
        let done = false;
        while (!done) {
            const message = await waitMessage();
            turns.push(message);
            if (message.serverContent && message.serverContent.turnComplete) {
                done = true;
            }
        }
        return turns;
    }

    const session = await ai.live.connect({
        model: model,
        callbacks: {
            onopen: function () {
                console.debug('Opened');
            },
            onmessage: function (message) {
                responseQueue.push(message);
            },
            onerror: function (e) {
                console.debug('Error:', e.message);
            },
            onclose: function (e) {
                console.debug('Close:', e.reason);
            },
        },
        config: config,
    });

    // Send Audio Chunk
    const fileBuffer = fs.readFileSync("sample.wav");

    // Ensure audio conforms to API requirements (16-bit PCM, 16kHz, mono)
    const wav = new WaveFile();
    wav.fromBuffer(fileBuffer);
    wav.toSampleRate(16000);
    wav.toBitDepth("16");
    const base64Audio = wav.toBase64();

    // If already in correct format, you can use this:
    // const fileBuffer = fs.readFileSync("sample.pcm");
    // const base64Audio = Buffer.from(fileBuffer).toString('base64');

    session.sendRealtimeInput(
        {
            audio: {
                data: base64Audio,
                mimeType: "audio/pcm;rate=16000"
            }
        }

    );

    const turns = await handleTurn();
    for (const turn of turns) {
        if (turn.text) {
            console.debug('Received text: %s\n', turn.text);
        }
        else if (turn.data) {
            console.debug('Received inline data: %s\n', turn.data);
        }
    }

    session.close();
}

async function main() {
    await live().catch((e) => console.error('got error', e));
}

main();

Вы можете получать аудио, установив AUDIO в качестве модальности ответа. Этот пример сохраняет полученные данные в виде 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 how are you?"
        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())

JavaScript 

import { GoogleGenAI, Modality } from '@google/genai';

const ai = new GoogleGenAI({ apiKey: "GOOGLE_API_KEY" });
const model = 'gemini-2.0-flash-live-001';
const config = { responseModalities: [Modality.AUDIO] };

async function live() {
    const responseQueue = [];

    async function waitMessage() {
        let done = false;
        let message = undefined;
        while (!done) {
            message = responseQueue.shift();
            if (message) {
                done = true;
            } else {
                await new Promise((resolve) => setTimeout(resolve, 100));
            }
        }
        return message;
    }

    async function handleTurn() {
        const turns = [];
        let done = false;
        while (!done) {
            const message = await waitMessage();
            turns.push(message);
            if (message.serverContent && message.serverContent.turnComplete) {
                done = true;
            }
        }
        return turns;
    }

    const session = await ai.live.connect({
        model: model,
        callbacks: {
            onopen: function () {
                console.debug('Opened');
            },
            onmessage: function (message) {
                responseQueue.push(message);
            },
            onerror: function (e) {
                console.debug('Error:', e.message);
            },
            onclose: function (e) {
                console.debug('Close:', e.reason);
            },
        },
        config: config,
    });

    const simple = 'Hello how are you?';
    session.sendClientContent({ turns: simple });

    const turns = await handleTurn();

    // Combine audio data strings and save as wave file
    const combinedAudio = turns.reduce((acc, turn) => {
        if (turn.data) {
            const buffer = Buffer.from(turn.data, 'base64');
            const intArray = new Int16Array(buffer.buffer, buffer.byteOffset, buffer.byteLength / Int16Array.BYTES_PER_ELEMENT);
            return acc.concat(Array.from(intArray));
        }
        return acc;
    }, []);

    const audioBuffer = new Int16Array(combinedAudio);

    const wf = new WaveFile();
    wf.fromScratch(1, 24000, '16', audioBuffer);
    fs.writeFileSync('output.wav', wf.toBuffer());

    session.close();
}

async function main() {
    await live().catch((e) => console.error('got error', e));
}

main();

Форматы аудио

Аудиоданные в Live API всегда являются необработанными, little-endian, 16-битными PCM. Аудиовыход всегда использует частоту дискретизации 24 кГц. Входной звук изначально имеет частоту 16 кГц, но Live API при необходимости будет выполнять повторную выборку, чтобы можно было отправить любую частоту дискретизации. Чтобы передать частоту дискретизации входного звука, установите MIME-тип каждого Blob , содержащего аудио, на значение, например audio/pcm;rate=16000 .

Получение аудиотранскрипций

Вы можете включить транскрипцию аудиовыхода модели, отправив output_audio_transcription в конфигурации настройки. Язык транскрипции выводится из ответа модели.

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

Вы можете включить транскрипцию аудиовхода, отправив input_audio_transcription в конфигурации настройки. 

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": ["TEXT"],
    "realtime_input_config": {
        "automatic_activity_detection": {"disabled": True},
        "activity_handling": "NO_INTERRUPTION",
    },
    "input_audio_transcription": {},
}

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

        await session.send_realtime_input(activity_start=types.ActivityStart())
        await session.send_realtime_input(
            audio=types.Blob(data=audio_data, mime_type='audio/pcm;rate=16000')
        )
        await session.send_realtime_input(activity_end=types.ActivityEnd())

        async for msg in session.receive():
            if msg.server_content.input_transcription:
                print('Transcript:', msg.server_content.input_transcription.text)

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

Потоковое аудио и видео

Системные инструкции

Системные инструкции позволяют вам управлять поведением модели на основе ваших конкретных потребностей и вариантов использования. Системные инструкции могут быть установлены в конфигурации настройки и будут действовать в течение всего сеанса. 

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"],
}

Постепенные обновления контента

Используйте инкрементные обновления для отправки текстового ввода, установления контекста сеанса или восстановления контекста сеанса. Для коротких контекстов вы можете отправлять пошаговые взаимодействия, чтобы представить точную последовательность событий:

Питон 

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)

JSON 

{
  "clientContent": {
    "turns": [
      {
        "parts":[
          {
            "text": ""
          }
        ],
        "role":"user"
      },
      {
        "parts":[
          {
            "text": ""
          }
        ],
        "role":"model"
      }
    ],
    "turnComplete": true
  }
}

Для более длинных контекстов рекомендуется предоставлять единое резюме сообщения, чтобы освободить окно контекста для последующих взаимодействий.

Изменение голоса и языка

Live API поддерживает следующие голоса: Puck, Charon, Kore, Fenrir, Aoede, Leda, Orus и Zephyr.

Чтобы указать голос, задайте имя голоса в объекте speechConfig как часть конфигурации сеанса:

Питон 

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

JSON 

{
  "voiceConfig": {
    "prebuiltVoiceConfig": {
      "voiceName": "Kore"
    }
  }
}

API Live поддерживает несколько языков .

Чтобы изменить язык, установите код языка в объекте speechConfig как часть конфигурации сеанса: 

from google.genai import types

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

Использование инструмента с Live API

С помощью Live API можно определить такие инструменты, как вызов функций , выполнение кода и поиск Google .

Вызов функции

Вы можете определить объявления функций как часть конфигурации сеанса. См. руководство по вызову функций, чтобы узнать больше.

После получения вызовов инструментов клиент должен ответить списком объектов FunctionResponse , используя метод 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())

Из одного приглашения модель может генерировать несколько вызовов функций и код, необходимый для цепочки их выходов. Этот код выполняется в среде песочницы, генерируя последующие сообщения BidiGenerateContentToolCall .

Асинхронный вызов функции

По умолчанию выполнение приостанавливается до тех пор, пока не станут доступны результаты каждого вызова функции, что обеспечивает последовательную обработку. Это означает, что вы не сможете продолжать взаимодействовать с моделью, пока функции выполняются.

Если вы не хотите блокировать диалог, вы можете указать модели запускать функции асинхронно.

Для этого сначала необходимо добавить behavior к определениям функций: 

  # Non-blocking function definitions
  turn_on_the_lights = {"name": "turn_on_the_lights", "behavior": "NON_BLOCKING"} # turn_on_the_lights will run asynchronously
  turn_off_the_lights = {"name": "turn_off_the_lights"} # turn_off_the_lights will still pause all interactions with the model

NON-BLOCKING гарантирует, что функция будет работать асинхронно, в то время как вы сможете продолжать взаимодействовать с моделью.

Затем вам нужно указать модели, как себя вести, когда она получает FunctionResponse , используя параметр scheduling . Она может:

  • Прервите его работу и сразу же сообщите о полученном ответе ( scheduling="INTERRUPT" ),
  • Подождите, пока он закончит то, что он в данный момент делает ( scheduling="WHEN_IDLE" ),
  • Или ничего не делать и использовать эти знания позже в обсуждении ( scheduling="SILENT" ) 
# Non-blocking function definitions
  function_response = types.FunctionResponse(
      id=fc.id,
      name=fc.name,
      response={
          "result": "ok",
          "scheduling": "INTERRUPT" # Can also be WHEN_IDLE or SILENT
      }
  )

Выполнение кода

Вы можете определить выполнение кода как часть конфигурации сеанса. См. руководство по выполнению кода, чтобы узнать больше. 

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

Вы можете включить Grounding with Google Search как часть конфигурации сеанса. См. учебник Grounding , чтобы узнать больше. 

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

Объединение нескольких инструментов

В рамках Live API можно комбинировать несколько инструментов: 

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}

Обработка прерываний

Пользователи могут прервать вывод модели в любое время. Когда обнаружение голосовой активности (VAD) обнаруживает прерывание, текущая генерация отменяется и удаляется. В истории сеанса сохраняется только информация, уже отправленная клиенту. Затем сервер отправляет сообщение BidiGenerateContentServerContent , чтобы сообщить о прерывании.

Кроме того, сервер Gemini отменяет все ожидающие вызовы функций и отправляет сообщение BidiGenerateContentServerContent с идентификаторами отмененных вызовов. 

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

Обнаружение голосовой активности (VAD)

Вы можете настроить или отключить функцию обнаружения голосовой активности (VAD).

Использование автоматического VAD

По умолчанию модель автоматически выполняет VAD на непрерывном входном аудиопотоке. VAD можно настроить с помощью поля realtimeInputConfig.automaticActivityDetection конфигурации настройки .

Когда аудиопоток останавливается более чем на секунду (например, потому что пользователь выключил микрофон), должно быть отправлено событие audioStreamEnd для очистки кэшированного аудио. Клиент может возобновить отправку аудиоданных в любое время. 

# 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
from google.genai import types

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

С send_realtime_input API будет автоматически реагировать на аудио на основе VAD. В то время как send_client_content добавляет сообщения в контекст модели по порядку, send_realtime_input оптимизирован для реагирования за счет детерминированного упорядочивания.

Настройка автоматического VAD

Для большего контроля над активностью VAD вы можете настроить следующие параметры. См. справочник 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,
        }
    }
}

Отключение автоматического VAD

В качестве альтернативы можно отключить автоматический VAD, установив realtimeInputConfig.automaticActivityDetection.disabled в true в сообщении настройки. В этой конфигурации клиент отвечает за обнаружение речи пользователя и отправку сообщений activityStart и activityEnd в соответствующее время. В этой конфигурации audioStreamEnd не отправляется. Вместо этого любое прерывание потока отмечается сообщением 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())
    # ...

Количество токенов

Общее количество использованных токенов можно найти в поле usageMetadata возвращаемого сервером сообщения. 

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

Увеличение продолжительности сеанса

Максимальную продолжительность сеанса можно увеличить до неограниченного значения с помощью двух механизмов:

Кроме того, перед завершением сеанса вы получите сообщение GoAway , что позволит вам предпринять дальнейшие действия.

Сжатие контекстного окна

Чтобы включить более длительные сеансы и избежать внезапного завершения соединения, можно включить сжатие контекстного окна, установив поле contextWindowCompression как часть конфигурации сеанса.

В ContextWindowCompressionConfig можно настроить механизм скользящего окна и количество токенов , запускающих сжатие. 

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(),
        )
    ),
)

Возобновление сеанса

Чтобы предотвратить завершение сеанса, когда сервер периодически сбрасывает соединение WebSocket, настройте поле sessionResumption в конфигурации установки .

Передача этой конфигурации заставляет сервер отправлять сообщения SessionResumptionUpdate , которые можно использовать для возобновления сеанса, передавая последний токен возобновления как SessionResumptionConfig.handle последующего соединения. 

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

Получение сообщения до отключения сеанса

Сервер отправляет сообщение GoAway , которое сигнализирует о том, что текущее соединение скоро будет разорвано. Это сообщение включает timeLeft , указывающее оставшееся время, и позволяет вам предпринять дальнейшие действия, прежде чем соединение будет разорвано как 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)

Получение сообщения о завершении генерации

Сервер отправляет сообщение generationComplete , которое сигнализирует о том, что модель завершила генерацию ответа. 

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

Разрешение СМИ

Вы можете указать разрешение для входного носителя, установив поле mediaResolution как часть конфигурации сеанса: 

from google.genai import types

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

Ограничения

При планировании своего проекта учитывайте следующие ограничения Live API.

Методы реагирования

В конфигурации сеанса можно задать только одну модальность ответа ( TEXT или AUDIO ) на сеанс. Установка обоих приводит к сообщению об ошибке конфигурации. Это означает, что можно настроить модель на ответ либо текстом, либо аудио, но не обоими способами в одном сеансе.

Аутентификация клиента

Live API обеспечивает только аутентификацию сервера к серверу и не рекомендуется для прямого использования клиентом. Клиентский ввод должен быть направлен через промежуточный сервер приложений для безопасной аутентификации с Live API.

Продолжительность сеанса

Продолжительность сеанса может быть увеличена до неограниченной, если включить сжатие сеанса. Без сжатия сеансы только с аудио ограничены 15 минутами, а сеансы с аудио и видео ограничены 2 минутами. Превышение этих ограничений без сжатия приведет к разрыву соединения.

Кроме того, вы можете настроить возобновление сеанса , чтобы позволить клиенту возобновить прерванный сеанс.

Контекстное окно

Сеанс имеет ограничение на окно контекста в 32 тыс. токенов.

Поддерживаемые языки

Live API поддерживает следующие языки.

Язык Код BCP-47
Немецкий (Германия) де-ДЕ
Английский (Австралия) en-AU
Английский (Великобритания) ru-GB
Английский (Индия) ru-IN
Английский (США) en-US
Испанский (США) es-US
Французский (Франция) fr-FR
Хинди (Индия) привет-IN
Португальский (Бразилия) пт-BR
Арабский (общий) ar-XA
Испанский (Испания) es-ES
Французский (Канада) fr-CA
Индонезийский (Индонезия) я сделал
Итальянский (Италия) это-ИТ
Японский (Япония) ja-JP
Турецкий (Турция) тр-ТР
Вьетнамский (Вьетнам) ви-ВН
Бенгальский (Индия) bn-IN
Гуджарати (Индия) гу-ИН
Каннада (Индия) кн-IN
Малаялам (Индия) мл-IN
Маратхи (Индия) г-н-IN
Тамильский (Индия) та-ИН
Телугу (Индия) те-ИН
Голландский (Нидерланды) nl-NL
Корейский (Южная Корея) ко-КР
Мандаринский китайский (Китай) cmn-CN
Польский (Польша) pl-PL
Русский (Россия) ру-RU
Тайский (Таиланд) th-TH

Интеграции третьих сторон

Для развертывания веб- и мобильных приложений вы можете изучить следующие варианты:

Что дальше?