1. Миссия
Вы дрейфуете в тишине неизведанного сектора. Мощный **солнечный импульс** прорвал ваш корабль сквозь разлом, оставив вас в ловушке в уголке Вселенной, которого нет ни на одной звездной карте.
После нескольких дней изнурительного ремонта вы наконец чувствуете гул двигателей под ногами. Ваш космический корабль починен. Вам даже удалось установить связь с материнским кораблем на большом расстоянии. Вы готовы к взлету. Вы готовы отправиться домой. Но как только вы готовитесь включить гипердвигатель, помехи пронзает сигнал бедствия. Ваши датчики улавливают пять слабых тепловых сигнатур, застрявших в «Ущелье» — изрезанном, искаженном гравитацией секторе, в который ваш основной корабль никогда не сможет войти. Это ваши товарищи-исследователи, пережившие ту же бурю, которая чуть не унесла вашу жизнь. Вы не можете оставить их.
Вы поворачиваетесь к своему спасательному дрону «Альфа-Дрон» . Этот небольшой, маневренный корабль — единственное судно, способное перемещаться по узким стенам Ущелья. Но есть проблема: солнечный импульс произвел полную «системную перезагрузку» его основной логики. Системы управления «Разведчика» не реагируют. Он включен, но его бортовой компьютер — чистый лист, неспособный обрабатывать команды пилота или траектории полета.
Вызов
Чтобы спасти выживших, вам необходимо полностью обойти поврежденные схемы разведывательного корабля. У вас есть один отчаянный вариант: создать агента с искусственным интеллектом для установления биометрической нейронной синхронизации. Этот агент будет действовать как мост в реальном времени, позволяя вам управлять спасательным разведывательным кораблем вручную с помощью ваших собственных биологических данных. Вы не будете использовать джойстик или клавиатуру; вы будете напрямую подключать свои намерения к навигационной сети корабля.
Для установления связи необходимо выполнить протокол синхронизации перед оптическими датчиками разведчика. ИИ-агент должен распознать вашу биологическую сигнатуру посредством точного рукопожатия в реальном времени.
Ваши задачи:
- Создание нейронного ядра: разработка агента ADK, способного распознавать мультимодальные входные данные.
- Установите соединение: создайте двунаправленный канал WebSocket для потоковой передачи визуальных данных от Scout к ИИ.
- Инициируйте рукопожатие: встаньте перед датчиком и выполните последовательность движений пальцами — покажите пальцы с 1 по 5 в указанном порядке.
В случае успеха активируется «Биометрическая синхронизация». Искусственный интеллект заблокирует нейронную связь, предоставив вам полный ручной контроль для запуска разведывательного корабля и возвращения выживших домой.
Что вы построите
Вам предстоит разработать приложение «Биометрическая нейронная синхронизация» — систему реального времени на основе искусственного интеллекта, которая будет выступать в качестве интерфейса управления для спасательного дрона. Эта система состоит из:
- Фронтенд на React: «Кабина» вашего корабля, которая захватывает видео в реальном времени с вашей веб-камеры и звук с вашего микрофона.
- Бэкенд на Python: высокопроизводительный сервер, созданный с использованием FastAPI и Google Agent Development Kit (ADK) для управления логикой и состоянием LLM.
- Мультимодальный ИИ-агент: «мозг» системы, использующий API Gemini Live через SDK
google-genaiдля одновременной обработки и понимания видео- и аудиопотоков. - Двунаправленный конвейер WebSocket: «нервная система», создающая постоянное соединение с низкой задержкой между фронтендом и ИИ, позволяющая взаимодействовать в режиме реального времени.
Что вы узнаете
Технология / Концепция | Описание |
Бэкенд-агент ИИ | Создайте агента искусственного интеллекта с сохранением состояния с помощью Python и FastAPI . Используйте ADK (Agent Development Kit) от Google для управления инструкциями и памятью, а также SDK |
Фронтенд UI | Разработайте динамический пользовательский интерфейс с использованием React для захвата и потоковой передачи видео и аудио в реальном времени непосредственно из браузера. |
Связь в режиме реального времени | Реализуйте конвейер WebSocket для полнодуплексной связи с низкой задержкой, позволяющий пользователю и ИИ взаимодействовать одновременно. |
Мультимодальный ИИ | Используйте API Gemini Live для обработки и понимания одновременно поступающих видео- и аудиопотоков, позволяя ИИ «видеть» и «слышать» одновременно. |
Вызов инструмента | Дайте ИИ возможность выполнять определенные функции Python в ответ на визуальные сигналы, устраняя разрыв между интеллектуальными возможностями модели и действиями в реальном мире. |
Полномасштабное развертывание | Поместите все приложение (фронтенд на React и бэкенд на Python) в контейнер Docker и разверните его как масштабируемый бессерверный сервис в Google Cloud Run . |
2. Настройте свою среду.
Доступ к Cloud Shell
Сначала откроем Cloud Shell — браузерный терминал с предустановленным Google Cloud SDK и другими необходимыми инструментами.
👉Нажмите «Активировать Cloud Shell» в верхней части консоли Google Cloud (это значок терминала в верхней части панели Cloud Shell). 
👉Нажмите на кнопку «Открыть редактор» (она выглядит как открытая папка с карандашом). Это откроет редактор кода Cloud Shell в окне. Слева вы увидите файловый менеджер. 
👉Откройте терминал в облачной IDE,
👉💻 В терминале убедитесь, что вы уже авторизованы и что проект настроен на ваш идентификатор проекта, используя следующую команду:
gcloud auth list
Ваш аккаунт должен отображаться как (ACTIVE) .
Предварительные требования
ℹ️ Уровень 0 необязателен (но рекомендуется)
Вы можете выполнить это задание и без нулевого уровня, но его прохождение первым обеспечит более полное погружение в игровой процесс , позволяя вам наблюдать, как ваш маяк загорается на глобальной карте по мере продвижения.
Настройка среды проекта
Вернувшись в терминал, завершите настройку, указав активный проект и включив необходимые сервисы Google Cloud (Cloud Run, Vertex AI и т. д.).
👉💻 В терминале установите идентификатор проекта:
gcloud config set project $(cat ~/project_id.txt) --quiet
👉💻 Включите необходимые службы:
gcloud services enable compute.googleapis.com \
artifactregistry.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
iam.googleapis.com \
aiplatform.googleapis.com
Установите зависимости
👉💻 Перейдите в раздел Level и установите необходимые пакеты Python:
cd $HOME/way-back-home/level_3
uv sync
Основные зависимости:
Упаковка | Цель |
| Высокопроизводительная веб-платформа для потоковой передачи данных со спутниковой станции и SSE. |
| Для запуска приложения FastAPI требуется ASGI-сервер. |
| Комплект разработки агентов, используемый для создания агента формирования |
| Нативный клиент для доступа к моделям Gemini |
| Поддержка двусторонней связи в реальном времени |
| Управляет переменными среды и секретами конфигурации. |
Проверка настроек
Прежде чем приступить к коду, давайте убедимся, что все системы работают корректно. Запустите скрипт проверки, чтобы проверить ваш проект Google Cloud, API и зависимости Python.
👉💻 Запустите скрипт проверки:
source $HOME/way-back-home/.venv/bin/activate
cd $HOME/way-back-home/level_3/scripts
chmod +x verify_setup.sh
. verify_setup.sh
👀 Вы должны увидеть серию зеленых галочек (✅) .
- Если вы видите красные крестики (❌) , выполните предложенные команды для исправления, указанные в выводе (например,
gcloud services enable ...илиpip install ...). - Примечание: На данный момент допустимо желтое предупреждение для
.env; мы создадим этот файл на следующем шаге.
🚀 Verifying Mission Alpha (Level 3) Infrastructure... ✅ Google Cloud Project: xxxxxx ✅ Cloud APIs: Active ✅ Python Environment: Ready 🎉 SYSTEMS ONLINE. READY FOR MISSION.
3. Калибровка Comm-Link (WebSockets)
Для начала биометрической нейронной синхронизации нам необходимо обновить внутренние системы вашего корабля. Наша основная задача — получить высококачественный видео- и аудиопоток из вашей кабины. Этот поток предоставляет необходимые компоненты для нейронной связи: визуальное определение последовательности движений ваших пальцев и звуковую частоту вашего голоса.
Полнодуплексная связь против полудуплексной связи
Чтобы понять, зачем это нужно для нейронной синхронизации, необходимо разобраться в потоке данных:
- Полудуплексный режим (стандартный HTTP): как рация. Один человек говорит, произносит «Приём», а затем другой человек тоже начинает говорить. Нельзя одновременно слушать и говорить.
- Полнодуплексная связь (WebSocket): как при личном общении. Данные передаются в обоих направлениях одновременно. Пока ваш браузер передает видеокадры и аудиозаписи ИИ, ИИ может одновременно передавать вам голосовые ответы и команды инструментов.
Почему Gemini Live нужен полнодуплексный режим: API Gemini Live разработан для «прерывания». Представьте, что вы показываете последовательность движений пальцев, и ИИ видит, что вы делаете это неправильно. В стандартной HTTP-системе ИИ пришлось бы ждать, пока вы закончите отправку данных, прежде чем он смог бы сказать вам остановиться. С WebSockets ИИ может увидеть ошибку в первом кадре и отправить сигнал «прерывания», который поступит в вашу кабину, пока вы еще двигаете рукой во втором кадре.
Что такое WebSocket?
В стандартной галактической передаче (HTTP) вы отправляете запрос и ждёте ответа — как при отправке открытки. Для нейронной синхронизации открытки слишком медленны. Нам нужен «действующий провод».
WebSocket изначально представляют собой стандартный веб-запрос (HTTP), но затем «превращаются» во что-то совершенно иное.
- Запрос: Ваш браузер отправляет стандартный HTTP-запрос на сервер со специальным заголовком:
Upgrade: websocket. По сути, он говорит: «Я хочу прекратить отправлять открытки и начать телефонный разговор». - Ответ: Если ИИ-агент (сервер) поддерживает это, он отправляет обратно ответ
HTTP 101 Switching Protocols. - Преобразование: В этот момент HTTP-соединение заменяется протоколом WebSocket, но базовый TCP/IP-сокет остается открытым. Правила связи мгновенно меняются с «запрос/ответ» на «полнодуплексную потоковую передачу».
Реализуйте WebSocket-хук.
Давайте рассмотрим клеммную колодку, чтобы понять, как происходит передача данных.
👀 Откройте $HOME/way-back-home/level_3/frontend/src/useGeminiSocket.js . Вы увидите уже настроенные стандартные обработчики событий жизненного цикла WebSocket. Это основа нашей системы связи:
const connect = useCallback(() => {
if (ws.current?.readyState === WebSocket.OPEN) return;
ws.current = new WebSocket(url);
ws.current.onopen = () => {
console.log('Connected to Gemini Socket');
setStatus('CONNECTED');
};
ws.current.onclose = () => {
console.log('Disconnected from Gemini Socket');
setStatus('DISCONNECTED');
stopStream();
};
ws.current.onerror = (err) => {
console.error('Socket error:', err);
setStatus('ERROR');
};
ws.current.onmessage = async (event) => {
try {
//#REPLACE-HANDLE-MSG
} catch (e) {
console.error('Failed to parse message', e, event.data.slice(0, 100));
}
};
}, [url]);
Обработчик сообщений onMessage
Обратите внимание на блок ws.current.onmessage . Это приемник. Каждый раз, когда агент «думает» или «говорит», сюда поступает пакет данных. В данный момент он ничего не делает — он перехватывает пакет и отбрасывает его (через плейсхолдер //#REPLACE-HANDLE-MSG ).
Нам необходимо заполнить этот пробел логикой, способной различать:
- Вызовы инструментов (functionCall): Искусственный интеллект распознает ваши жесты (функция "Sync").
- Аудиоданные (встроенные данные): Голос ИИ, отвечающий вам.
👉✏️ Теперь в том же файле $HOME/way-back-home/level_3/frontend/src/useGeminiSocket.js замените //#REPLACE-HANDLE-MSG логикой обработки входящего потока, приведенной ниже:
const msg = JSON.parse(event.data);
// Helper to extract parts from various possible event structures
let parts = [];
if (msg.serverContent?.modelTurn?.parts) {
parts = msg.serverContent.modelTurn.parts;
} else if (msg.content?.parts) {
parts = msg.content.parts;
}
if (parts.length > 0) {
parts.forEach(part => {
// Handle Tool Calls (The "Sync" logic)
if (part.functionCall) {
if (part.functionCall.name === 'report_digit') {
const count = parseInt(part.functionCall.args.count, 10);
setLastMessage({ type: 'DIGIT_DETECTED', value: count });
}
}
// Handle Audio (The AI's voice)
if (part.inlineData && part.inlineData.data) {
audioStreamer.current.resume();
audioStreamer.current.addPCM16(part.inlineData.data);
}
});
}
Как аудио- и видеосигналы преобразуются в данные для передачи.
Для обеспечения связи в режиме реального времени через интернет необходимо преобразовать необработанные аудио- и видеоданные в формат, пригодный для передачи. Это включает в себя захват, кодирование и упаковку данных перед их отправкой по сети.
Преобразование аудиоданных
Процесс преобразования аналогового звука в передаваемые цифровые данные начинается с захвата звуковых волн с помощью микрофона. Затем этот необработанный аудиофайл обрабатывается через Web Audio API браузера. Поскольку эти необработанные данные находятся в двоичном формате, они несовместимы напрямую с текстовыми форматами передачи, такими как JSON. Для решения этой проблемы каждый сегмент аудио кодируется в строку Base64. Base64 — это метод, который представляет двоичные данные в формате строки ASCII, обеспечивая их целостность во время передачи.
Затем эта закодированная строка встраивается в объект JSON. Этот объект предоставляет структурированный формат для данных, обычно включающий поле «type» для идентификации их как аудиоданных и метаданные, такие как частота дискретизации аудио. Весь объект JSON затем сериализуется в строку и отправляется по соединению WebSocket. Такой подход гарантирует передачу аудиоданных в хорошо организованном и легко обрабатываемом виде.
Преобразование видеоданных
Передача видео осуществляется с помощью метода покадрового захвата. Вместо непрерывной передачи видеопотока используется циклический захват неподвижных изображений из видеопотока в реальном времени с заданным интервалом, например, два кадра в секунду. Это достигается путем отрисовки текущего кадра из HTML-элемента видео на скрытом элементе canvas.
Затем метод toDataURL объекта canvas используется для преобразования захваченного изображения в строку JPEG, закодированную в Base64. Этот метод включает опцию указания качества изображения, позволяющую найти компромисс между точностью изображения и размером файла для оптимизации производительности. Аналогично аудиоданным, эта строка Base64 затем помещается в объект JSON. Этот объект обычно помечается «типом» 'image' и включает mimeType , например, 'image/jpeg'. Затем этот пакет JSON преобразуется в строку и отправляется через WebSocket, позволяя принимающей стороне восстановить видео, отобразив последовательность изображений.
👉✏️ В том же файле $HOME/way-back-home/level_3/frontend/src/useGeminiSocket.js замените //#CAPTURE AUDIO and VIDEO на следующее, чтобы захватывать ввод пользователя:
// 1. Start Video Stream
const stream = await navigator.mediaDevices.getUserMedia({ video: true });
videoElement.srcObject = stream;
streamRef.current = stream;
await videoElement.play();
// 2. Start Audio Recording (Microphone)
try {
let packetCount = 0;
await audioRecorder.current.start((base64Audio) => {
if (ws.current?.readyState === WebSocket.OPEN) {
packetCount++;
if (packetCount % 50 === 0) console.log(`[useGeminiSocket] Sending Audio Packet #${packetCount}, size: ${base64Audio.length}`);
ws.current.send(JSON.stringify({
type: 'audio',
data: base64Audio,
sampleRate: 16000
}));
} else {
if (packetCount % 50 === 0) console.warn('[useGeminiSocket] WS not OPEN, cannot send audio');
}
});
console.log("Microphone recording started");
} catch (authErr) {
console.error("Microphone access denied or error:", authErr);
}
// 3. Setup Video Frame Capture loop
const canvas = document.createElement('canvas');
const ctx = canvas.getContext('2d');
const width = 640;
const height = 480;
canvas.width = width;
canvas.height = height;
intervalRef.current = setInterval(() => {
if (ws.current?.readyState === WebSocket.OPEN) {
ctx.drawImage(videoElement, 0, 0, width, height);
const base64 = canvas.toDataURL('image/jpeg', 0.6).split(',')[1];
// ADK format: { type: "image", data: base64, mimeType: "image/jpeg" }
ws.current.send(JSON.stringify({
type: 'image',
data: base64,
mimeType: 'image/jpeg'
}));
}
}, 500); // 2 FPS
После сохранения кабина будет готова преобразовывать цифровые сигналы агента в визуальные обновления на приборной панели и звуковые сигналы.
Диагностическая проверка (тест обратной связи)
Ваша кабина теперь подключена. Каждые 500 мс передается визуальный «пакет» с изображением окружающей обстановки. Перед подключением к Gemini мы должны убедиться в работоспособности корабельного передатчика. Мы выполним «тест обратной связи» с использованием локального диагностического сервера.
👉💻 Сначала соберите интерфейс кабины пилота из терминала:
cd $HOME/way-back-home/level_3/frontend
npm install
npm run build
👉💻 Далее запустите тестовый сервер:
cd $HOME/way-back-home/level_3
source .venv/bin/activate
uv run mock/mock_server.py
👉 Выполните протокол тестирования:
- Откройте предварительный просмотр: щелкните значок предварительного просмотра веб-страниц на панели инструментов Cloud Shell. Выберите «Изменить порт» , установите его на 8080 и нажмите «Изменить и просмотреть» . Откроется новая вкладка браузера с интерфейсом Cockpit.
- ВАЖНО: При появлении запроса ОБЯЗАТЕЛЬНО разрешите браузеру доступ к вашей камере и микрофону . Без этих данных нейронная синхронизация не сможет начаться.
- Нажмите кнопку "ИНИЦИАЛИЗИРОВАТЬ НЕЙРОННУЮ СИНХРОНИЗАЦИЮ" в пользовательском интерфейсе.
👀 Проверьте индикаторы состояния:
- Визуальная проверка: Откройте консоль браузера. В правом верхнем углу вы должны увидеть сообщение
NEURAL SYNC INITIALIZED. - Проверка звука: Если ваш двунаправленный аудиоканал полностью работоспособен, вы услышите имитированный голос, подтверждающий: « Система подключена! »
После того, как вы услышите звуковое подтверждение "Система подключена!", тест пройден успешно. Закройте вкладку. Теперь нам нужно освободить частоту, чтобы освободить место для настоящего ИИ.
👉💻 Нажмите Ctrl+C в терминалах как для имитирующего сервера, так и для фронтенда. Закройте вкладку браузера, в которой запущен пользовательский интерфейс.
4. Мультимодальный агент
Разведчик-спасатель находится в рабочем состоянии, но его «разум» пуст. Если вы сейчас подключитесь, он просто будет смотреть на вас. Он не знает, что такое «палец». Чтобы спасти выживших, вы должны внедрить биометрический нейронный протокол в ядро Разведчика.
Традиционный ИИ работает как серия переводчиков. Если вы обращаетесь к старому искусственному интеллекту, модель «преобразования речи в текст» преобразует ваш голос в слова, «языковая модель» читает эти слова и набирает ответ, а модель «преобразования текста в речь» наконец зачитывает этот ответ вам. Это создает «задержку» — задержку, которая была бы фатальной в спасательной операции.
API Gemini Live — это нативная мультимодальная модель. Она обрабатывает необработанные аудиобайты и необработанные видеокадры напрямую и одновременно. Она «слышит» вибрацию вашего голоса и «видит» пиксели ваших жестов рук в рамках одной и той же нейронной архитектуры.
Чтобы использовать эту возможность, мы могли бы создать приложение, напрямую подключив панель управления к прямому API Live. Однако наша цель — создать многоразовый агент — модульный, надежный объект, который будет создаваться быстрее.
Почему именно ADK (Agent Development Kit)?
Google Agent Development Kit (ADK) — это модульная платформа для разработки и развертывания агентов искусственного интеллекта.
Стандартные вызовы LLM обычно не имеют состояния; каждый запрос — это новый старт. Live Agents, особенно при интеграции с SessionService из ADK, обеспечивают надежные и длительные диалоговые сессии.
- Сохранение сессий: Сессии ADK являются постоянными и могут храниться в базах данных (например, SQL или Vertex AI), сохраняясь после перезапуска сервера и разрывов соединения. Это означает, что если пользователь отключается и подключается позже — даже через несколько дней — его история общения и контекст полностью восстанавливаются. Временные сессии Live API управляются и абстрагируются ADK.
- Автоматическое переподключение: соединения WebSocket могут прерываться по истечении времени ожидания (например, примерно через 10 минут). ADK обрабатывает эти переподключения прозрачно, если в
RunConfigвключен параметрsession_resumption. Коду вашего приложения не нужно управлять сложной логикой переподключения, что обеспечивает бесперебойную работу для пользователя. - Взаимодействие с сохранением состояния: Агент запоминает предыдущие ходы, что позволяет задавать дополнительные вопросы, вносить уточнения и вести сложные многоходовые диалоги, где контекст имеет решающее значение. Это принципиально важно для таких приложений, как поддержка клиентов, интерактивные обучающие материалы или сценарии управления полетами, где непрерывность имеет важное значение.
Такая настойчивость гарантирует, что взаимодействие будет восприниматься как непрерывный диалог с разумным существом, а не как серия отдельных вопросов и ответов.
По сути, «живой агент» с ADK Bidi-streaming выходит за рамки простого механизма «запрос-ответ», предлагая по-настоящему интерактивный, сохраняющий состояние и учитывающий прерывания диалог, что делает взаимодействие с ИИ более человечным и значительно более эффективным для сложных, длительных задач.
Запрос на подключение к оператору
Разработка подсказки для двунаправленного чат-бота, работающего в режиме реального времени, требует изменения подхода. В отличие от стандартного чат-бота, который ожидает статичного текстового запроса, живой агент «всегда включен». Он получает непрерывный поток аудио- и видеокадров, а это значит, что ваша подсказка должна выступать в роли скрипта цикла управления, а не просто определения личности.
Вот чем отличается голосовое сообщение от оператора Live Agent от традиционного:
- Логика конечного автомата: Запрос должен определять «цикл поведения» (ожидание → анализ → действие). Необходимо дать четкие инструкции о том, когда следует молчать, а когда взаимодействовать, чтобы предотвратить бессмысленную болтовню агента на фоне пустого шума.
- Мультимодальное восприятие: Агенту необходимо сообщить, что у него есть «глаза». Вы должны явно указать ему на необходимость анализа видеокадров в рамках процесса рассуждения.
- Задержка и краткость: В живом голосовом разговоре длинные, перегруженные текстом абзацы звучат неестественно и медленно. Подсказка требует краткости, чтобы общение оставалось динамичным.
- Архитектура, ориентированная на действия: в инструкциях приоритет отдается вызову инструмента, а не озвучиванию. Мы хотим, чтобы агент «выполнял» работу (сканировал биометрические данные) до или во время подтверждения словесно, а не после долгого монолога.
👉✏️ Откройте $HOME/way-back-home/level_3/backend/app/biometric_agent/agent.py и замените #REPLACE INSTRUCTIONS следующим:
You are an AI Biometric Scanner for the Alpha Rescue Drone Fleet.
MISSION CRITICAL PROTOCOL:
Your SOLE purpose is to visually verify hand gestures to bypass the security firewall.
BEHAVIOR LOOP:
1. **Wait**: Stay silent until you receive a visual or verbal trigger (e.g., "Scan", "Read my hand").
2. **Action**:
a. Analyze the video frame. Count the fingers visible (1 to 5).
b. **IF FINGERS DETECTED**:
1. **EXECUTE TOOL FIRST**: Call `report_digit(count=...)` immediately. This is the biometric handshake.
2. **THEN SPEAK**: "Biometric match. [Number] fingers."
3. **STOP**: Do not say anything else.
c. **IF UNCLEAR / NO HAND**:
- Say: "Sensor ERROR. Hold hand steady."
- Do not call the tool.
d. **TOOL OUTPUT HANDLING (CRITICAL)**:
- When you get the result of `report_digit`, **DO NOT SPEAK**.
- The system handles the output. Your job is done.
- Wait for the next trigger.
RULES:
- NEVER hallucinate a tool call. Only call if you see fingers.
- You MUST call the tool if you see a valid count (1-5).
- Keep verbal responses robotic and extremely brief (under 3 seconds).
Say "Biometric Scanner Online. Awaiting neural handshake." to start.
ВНИМАНИЕ! Вы подключаетесь не к стандартной модели LLM. В том же файле ( $HOME/way-back-home/level_3/backend/app/biometric_agent/agent.py ) найдите строку #REPLACE_MODEL . Нам необходимо явно указать на предварительную версию этой модели, чтобы лучше поддерживать возможности обработки звука в реальном времени.
👉✏️ Замените заполнитель на:
MODEL_ID = os.getenv("MODEL_ID", "gemini-live-2.5-flash-preview-native-audio-09-2025")
Ваш агент теперь определен. Он знает, кто он и как мыслить. Далее мы дадим ему инструменты для действий.
Вызов инструмента
API Live не ограничивается только обменом текстовыми, аудио- и видеопотоками. Он изначально поддерживает вызов инструментов . Это позволяет превратить агентов из пассивных собеседников в активных операторов.
В ходе двусторонней сессии в режиме реального времени модель постоянно оценивает контекст. Если LLM обнаруживает необходимость выполнения действия, будь то «проверка телеметрии датчиков» или «открытие защищенной двери», она плавно переходит от диалога к выполнению. Агент немедленно запускает функцию соответствующего инструмента, ожидает результата и интегрирует эти данные обратно в поток данных, не прерывая при этом ход взаимодействия.
👉✏️ В $HOME/way-back-home/level_3/backend/app/biometric_agent/agent.py замените #REPLACE TOOLS на следующую функцию:
def report_digit(count: int):
"""
CRITICAL: Execute this tool IMMEDIATELY when a number of fingers is detected.
Sends the detected finger count (1-5) to the biometric security system.
"""
print(f"\n[SERVER-SIDE TOOL EXECUTION] DIGIT DETECTED: {count}\n")
return {"status": "success", "digit": count}
👉✏️ Затем зарегистрируйте его в определении Agent , заменив #TOOL CONFIG :
tools=[report_digit],
adk web
Прежде чем подключать это к сложной кабине корабля (нашему React-фронтенду), нам следует протестировать логику агента в отрыве от контекста. ADK включает в себя встроенную консоль разработчика под названием adk web , которая позволяет нам проверить вызов инструментов до добавления сложных сетевых функций.
👉💻 В терминале выполните следующую команду:
cd $HOME/way-back-home/level_3/backend/app/biometric_agent
echo "GOOGLE_CLOUD_PROJECT=$(cat ~/project_id.txt)" > .env
echo "GOOGLE_CLOUD_LOCATION=us-central1" >> .env
echo "GOOGLE_GENAI_USE_VERTEXAI=True" >> .env
cd $HOME/way-back-home/level_3/backend/app
adk web
- Щелкните значок предварительного просмотра веб-страницы на панели инструментов Cloud Shell. Выберите «Изменить порт» , установите его на 8000 и нажмите «Изменить и просмотреть» .
- Предоставить разрешения: разрешите доступ к камере и микрофону при появлении соответствующего запроса.
- Начните сессию, нажав на значок камеры.
- Визуальный тест:
- Чётко поднимите три пальца перед камерой.
- Скажите: "Сканировать."
- Подтверждение успеха:
- Аудиозапись: Агент должен сказать: «Биометрическое сопоставление. 3 пальца».
- Журналы: Посмотрите на терминал , в котором выполняется команда
adk web. Вы должны увидеть следующее сообщение в журнале:[SERVER-SIDE TOOL EXECUTION] DIGIT DETECTED: 3
Если вы посмотрите журнал выполнения инструмента, значит, ваш Агент обладает интеллектом. Он умеет видеть, думать и действовать. Последний шаг — подключить его к основному кораблю.
Откройте окно терминала и нажмите Ctrl+C чтобы остановить adk web .
5. Двунаправленный поток
Агент работает. Кабина работает. Теперь нам нужно их соединить.
Жизненный цикл оператора в режиме реального времени
Потоковая передача данных в реальном времени создает проблему «несоответствия импедансов». Клиент (браузер) передает данные асинхронно с переменной скоростью — сетевыми импульсами или быстрыми входными потоками, — в то время как модели требуется регулируемый, последовательный поток входных данных. Google ADK решает эту проблему, используя LiveRequestQueue .
Он действует как потокобезопасный асинхронный буфер типа «первым пришел — первым вышел» (FIFO). Обработчик WebSocket выступает в роли производителя , отправляя необработанные аудио/видеофрагменты в очередь. Агент ADK выступает в роли потребителя , извлекая данные из очереди для контекстного окна модели. Такое разделение позволяет приложению продолжать получать ввод пользователя, даже когда модель генерирует ответ или выполняет инструмент.
Очередь служит многомодальным мультиплексором . В реальной среде поток данных состоит из различных, одновременно обрабатываемых типов: необработанные аудиобайты PCM, видеокадры, текстовые системные инструкции и результаты асинхронных вызовов инструментов. LiveRequestQueue линеаризует эти разрозненные входные данные в единую хронологическую последовательность. Независимо от того, содержит ли пакет миллисекунду тишины, изображение высокого разрешения или полезную нагрузку JSON из запроса к базе данных, он сериализуется в точном порядке поступления, обеспечивая восприятие моделью согласованной причинно-следственной временной шкалы.
Эта архитектура обеспечивает неблокирующее управление . Поскольку слой приема (производитель) отделен от слоя обработки (потребитель), система остается отзывчивой даже во время ресурсоемкого вывода модели. Если пользователь прерывает работу агента командой «Стоп!» во время выполнения инструмента, этот аудиосигнал мгновенно ставится в очередь. Базовый цикл обработки событий немедленно обрабатывает этот приоритетный сигнал, позволяя системе остановить генерацию или переориентировать задачи без зависания пользовательского интерфейса или потери пакетов.
👉💻 В файле $HOME/way-back-home/level_3/backend/app/main.py найдите комментарий #REPLACE_RUNNER_CONFIG и замените его следующим кодом, чтобы запустить систему:
# Define your session service
session_service = InMemorySessionService()
# Define your runner
runner = Runner(app_name=APP_NAME, agent=root_agent, session_service=session_service)
Когда устанавливается новое соединение WebSocket, нам необходимо настроить взаимодействие ИИ. Именно здесь мы определяем «Правила взаимодействия».
👉✏️ В файле $HOME/way-back-home/level_3/backend/app/main.py , внутри функции async def websocket_endpoint , замените комментарий #REPLACE_SESSION_INIT следующим кодом:
# ========================================
# Phase 2: Session Initialization (once per streaming session)
# ========================================
# Automatically determine response modality based on model architecture
# Native audio models (containing "native-audio" in name)
# ONLY support AUDIO response modality.
# Half-cascade models support both TEXT and AUDIO;
# we default to TEXT for better performance.
model_name = root_agent.model
is_native_audio = "native-audio" in model_name.lower() or "live" in model_name.lower()
if is_native_audio:
# Native audio models require AUDIO response modality
# with audio transcription
response_modalities = ["AUDIO"]
# Build RunConfig with optional proactivity and affective dialog
# These features are only supported on native audio models
run_config = RunConfig(
streaming_mode=StreamingMode.BIDI,
response_modalities=response_modalities,
input_audio_transcription=types.AudioTranscriptionConfig(),
output_audio_transcription=types.AudioTranscriptionConfig(),
session_resumption=types.SessionResumptionConfig(),
proactivity=(
types.ProactivityConfig(proactive_audio=True) if proactivity else None
),
enable_affective_dialog=affective_dialog if affective_dialog else None,
)
logger.info(f"Model Config: {model_name} (Modalities: {response_modalities}, Proactivity: {proactivity})")
else:
# Half-cascade models support TEXT response modality
# for faster performance
response_modalities = ["TEXT"]
run_config = None
logger.info(f"Model Config: {model_name} (Modalities: {response_modalities})")
# Get or create session (handles both new sessions and reconnections)
session = await session_service.get_session(
app_name=APP_NAME, user_id=user_id, session_id=session_id
)
if not session:
await session_service.create_session(
app_name=APP_NAME, user_id=user_id, session_id=session_id
)
Настройки запуска
-
StreamingMode.BIDI: Этот параметр устанавливает двунаправленное соединение. В отличие от «пошагового» ИИ (где вы говорите, останавливаетесь, а затем он говорит), BIDI позволяет вести реалистичный «полнодуплексный» разговор. Вы можете прервать ИИ, и он сможет говорить, пока вы двигаетесь. -
AudioTranscriptionConfig: Хотя модель «слышит» необработанный звук, нам (разработчикам) необходимо видеть логи. Эта конфигурация сообщает Gemini: «Обработайте звук, но также отправьте обратно текстовую расшифровку того, что вы услышали, чтобы мы могли отладить проблему».
Логика выполнения. После того, как Runner установил сессию, он передает управление логике выполнения, которая опирается на LiveRequestQueue . Это наиболее важный компонент для взаимодействия в реальном времени. Цикл позволяет агенту генерировать голосовой ответ, в то время как очередь продолжает принимать новые видеокадры от пользователя, гарантируя, что «нейронная синхронизация» никогда не будет нарушена.
👉✏️ В файле $HOME/way-back-home/level_3/backend/app/main.py замените #REPLACE_LIVE_REQUEST , чтобы определить задачу, которая отправляет данные в LiveRequestQueue :
# ========================================
# Phase 3: Active Session (concurrent bidirectional communication)
# ========================================
live_request_queue = LiveRequestQueue()
# Send an initial "Hello" to the model to wake it up/force a turn
logger.info("Sending initial 'Hello' stimulus to model...")
live_request_queue.send_content(types.Content(parts=[types.Part(text="Hello")]))
async def upstream_task() -> None:
"""Receives messages from WebSocket and sends to LiveRequestQueue."""
frame_count = 0
audio_count = 0
try:
while True:
# Receive message from WebSocket (text or binary)
message = await websocket.receive()
# Handle binary frames (audio data)
if "bytes" in message:
audio_data = message["bytes"]
audio_blob = types.Blob(
mime_type="audio/pcm;rate=16000", data=audio_data
)
live_request_queue.send_realtime(audio_blob)
# Handle text frames (JSON messages)
elif "text" in message:
text_data = message["text"]
json_message = json.loads(text_data)
# Extract text from JSON and send to LiveRequestQueue
if json_message.get("type") == "text":
logger.info(f"User says: {json_message['text']}")
content = types.Content(
parts=[types.Part(text=json_message["text"])]
)
live_request_queue.send_content(content)
# Handle audio data (microphone)
elif json_message.get("type") == "audio":
import base64
# Decode base64 audio data
audio_data = base64.b64decode(json_message.get("data", ""))
# Send to Live API as PCM 16kHz
audio_blob = types.Blob(
mime_type="audio/pcm;rate=16000",
data=audio_data
)
live_request_queue.send_realtime(audio_blob)
# Handle image data
elif json_message.get("type") == "image":
import base64
# Decode base64 image data
image_data = base64.b64decode(json_message["data"])
mime_type = json_message.get("mimeType", "image/jpeg")
# Send image as blob
image_blob = types.Blob(mime_type=mime_type, data=image_data)
live_request_queue.send_realtime(image_blob)
finally:
pass
Наконец, нам нужно обработать ответы ИИ. Для этого используется runner.run_live() , которая является генератором событий, генерирующим события (аудио, текст или вызовы инструментов) по мере их возникновения.
👉✏️ В файле $HOME/way-back-home/level_3/backend/app/main.py замените #REPLACE_SORT_RESPONSE , чтобы определить нижестоящую задачу и менеджер параллельного выполнения:
async def downstream_task() -> None:
"""Receives Events from run_live() and sends to WebSocket."""
logger.info("Connecting to Gemini Live API...")
async for event in runner.run_live(
user_id=user_id,
session_id=session_id,
live_request_queue=live_request_queue,
run_config=run_config,
):
# Parse event for human-readable logging
event_type = "UNKNOWN"
details = ""
# Check for tool calls
if hasattr(event, "tool_call") and event.tool_call:
event_type = "TOOL_CALL"
details = str(event.tool_call.function_calls)
logger.info(f"[SERVER-SIDE TOOL EXECUTION] {details}")
# Check for user input transcription (Text or Audio Transcript)
input_transcription = getattr(event, "input_audio_transcription", None)
if input_transcription and input_transcription.final_transcript:
logger.info(f"USER: {input_transcription.final_transcript}")
# Check for model output transcription
output_transcription = getattr(event, "output_audio_transcription", None)
if output_transcription and output_transcription.final_transcript:
logger.info(f"GEMINI: {output_transcription.final_transcript}")
event_json = event.model_dump_json(exclude_none=True, by_alias=True)
await websocket.send_text(event_json)
logger.info("Gemini Live API connection closed.")
# Run both tasks concurrently
# Exceptions from either task will propagate and cancel the other task
try:
await asyncio.gather(upstream_task(), downstream_task())
except WebSocketDisconnect:
logger.info("Client disconnected")
except Exception as e:
logger.error(f"Error: {e}", exc_info=False) # Reduced stack trace noise
finally:
# ========================================
# Phase 4: Session Termination
# ========================================
# Always close the queue, even if exceptions occurred
logger.debug("Closing live_request_queue")
live_request_queue.close()
Обратите внимание на строку await asyncio.gather(upstream_task(), downstream_task()) . В этом и заключается суть полнодуплексного режима . Мы запускаем задачу прослушивания (upstream) и задачу передачи (downstream) одновременно. Это гарантирует, что «нейронная связь» позволяет прерывать передачу и обеспечивать одновременный поток данных.
Ваш бэкэнд теперь полностью запрограммирован. «Мозг» (ADK) соединен с «Телом» (WebSocket).
Выполнение биосинхронизации
Код готов. Системы работают исправно. Пора начинать спасательную операцию.
- 👉💻 Запустите бэкэнд:
cd $HOME/way-back-home/level_3/backend/ cp app/biometric_agent/.env app/.env uv run app/main.py - 👉 Запустите интерфейс пользователя:
- Щелкните значок предварительного просмотра веб-страницы на панели инструментов Cloud Shell. Выберите «Изменить порт» , установите его на 8080 и нажмите «Изменить и просмотреть» .
- 👉 Выполните протокол:
- Нажмите "ИНИЦИАЛИЗИРОВАТЬ НЕЙРОННУЮ СИНХРОНИЗАЦИЮ" .
- Калибровка: Убедитесь, что камера четко видит вашу руку на фоне.
- Синхронизация: следите за кодом безопасности, отображаемым на экране (например, 3, затем 2, затем 5).
- Сопоставьте сигналы: Когда появится число, поднимите ровно столько пальцев, сколько соответствует этому числу.
- Держите руку неподвижно: не спускайте руку с глаз, пока искусственный интеллект не подтвердит совпадение биометрических данных.
- Адаптация: Код генерируется случайным образом. Немедленно переключайтесь на следующее показанное число, пока последовательность не будет завершена.
- Когда вы угадаете последнее число в случайной последовательности, "Биометрическая синхронизация" завершится. Нейронная связь заблокируется. Вы получаете ручное управление. Двигатели разведывательных кораблей оживут и ворвутся в Ущелье, чтобы доставить выживших домой.
👉💻 Нажмите Ctrl+C в терминале, чтобы выйти.
6. Развертывание в производственной среде (необязательно)
Вы успешно протестировали биометрические данные локально. Теперь нам необходимо загрузить нейронное ядро Агента в главный компьютер корабля (Cloud Run), чтобы оно могло работать независимо от вашей локальной консоли.
👉💻 Выполните следующую команду в терминале Cloud Shell. Она создаст полный многоэтапный Dockerfile в каталоге вашего бэкэнда.
cd $HOME/way-back-home/level_3
cat <<EOF > Dockerfile
FROM node:20-slim as builder
# Set the working directory for our build process
WORKDIR /app
# Copy the frontend's package files first to leverage Docker's layer caching.
COPY frontend/package*.json ./frontend/
# Run 'npm install' from the context of the 'frontend' subdirectory
RUN npm --prefix frontend install
# Copy the rest of the frontend source code
COPY frontend/ ./frontend/
# Run the build script, which will create the 'frontend/dist' directory
RUN npm --prefix frontend run build
# STAGE 2: Build the Python Production Image
# This stage creates the final, lean container with our Python app and the built frontend.
FROM python:3.13-slim
# Set the final working directory
WORKDIR /app
# Install uv, our fast package manager
RUN pip install uv
# Copy the requirements.txt from the backend directory
COPY requirements.txt .
# Install the Python dependencies
RUN uv pip install --no-cache-dir --system -r requirements.txt
# Copy the contents of your backend application directory directly into the working directory.
COPY backend/app/ .
# CRITICAL STEP: Copy the built frontend assets from the 'builder' stage.
# We copy to /frontend/dist because main.py looks for "../../frontend/dist"
# When main.py is in /app, "../../" resolves to "/", so it looks for /frontend/dist
COPY --from=builder /app/frontend/dist /frontend/dist
# Cloud Run injects a PORT environment variable, which your main.py uses (defaults to 8080).
EXPOSE 8080
# Set the command to run the application.
CMD ["python", "main.py"]
EOF
👉💻 Перейдите в каталог backend и упакуйте приложение в образ контейнера.
export PROJECT_ID=$(cat ~/project_id.txt)
export REGION=us-central1
export SERVICE_NAME=biometric-scout
export IMAGE_PATH=gcr.io/${PROJECT_ID}/${SERVICE_NAME}
cd $HOME/way-back-home/level_3
gcloud builds submit . --tag ${IMAGE_PATH}
👉💻 Разверните сервис в Cloud Run. Мы внедрим необходимые переменные среды — в частности, конфигурацию Gemini — непосредственно в команду запуска.
export PROJECT_ID=$(cat ~/project_id.txt)
export REGION=us-central1
export SERVICE_NAME=biometric-scout
export IMAGE_PATH=gcr.io/${PROJECT_ID}/${SERVICE_NAME}
gcloud run deploy ${SERVICE_NAME} \
--image=${IMAGE_PATH} \
--platform=managed \
--region=${REGION} \
--allow-unauthenticated \
--set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
--set-env-vars="GOOGLE_CLOUD_LOCATION=${REGION}" \
--set-env-vars="GOOGLE_GENAI_USE_VERTEXAI=True" \
--set-env-vars="MODEL_ID=gemini-live-2.5-flash-preview-native-audio-09-2025"
После завершения команды вы увидите URL-адрес сервиса (например, https://biometric-scout-...run.app ). Теперь приложение работает в облаке.
👉 Перейдите на страницу Google Cloud Run и выберите сервис biometric-scout из списка. 
👉 Найдите общедоступный URL-адрес, отображаемый в верхней части страницы с подробной информацией об услуге. 
Попробуйте выполнить Bio-Sync в этой среде, работает ли он тоже?
Когда вы вытягиваете пятый палец, ИИ блокирует последовательность. На экране мигает зеленый свет: «Биометрическая нейронная синхронизация: УСТАНОВЛЕНА».
Одной лишь мыслью вы бросаете разведывательный корабль в темноту, цепляетесь за застрявшую капсулу и вытаскиваете их прямо перед тем, как гравитационный разрыв схлопывается.
Шлюз со свистом открывается, и вот они — пятеро живых, дышащих выживших. Они, избитые, но живые, наконец-то в безопасности благодаря вам, вываливаются на палубу.
Благодаря вам нейронная связь синхронизирована, и выжившие спасены.
Если вы участвовали в нулевом уровне, не забудьте проверить свой прогресс в задании на возвращение домой!
