Миссия
Продолжительность: 2 мин.
Вы идентифицировали себя для системы искусственного интеллекта экстренной помощи, и ваш маяк теперь пульсирует на карте планеты — но это едва заметное мерцание, затерянное среди помех. Спасательные команды, сканирующие местность с орбиты, видят что-то по вашим координатам, но не могут зафиксировать точку захвата. Сигнал слишком слабый.
Чтобы усилить свой маяк до максимума, вам нужно подтвердить своё точное местоположение . Навигационная система капсулы вышла из строя, но место крушения разбросало по нему улики, которые ещё можно спасти. Образцы почвы. Странная флора. Чёткий вид на инопланетное ночное небо.
Если вы сможете проанализировать эти данные и определить, в каком регионе планеты вы находитесь, искусственный интеллект сможет определить ваше местоположение методом триангуляции и усилить сигнал маяка. Тогда — возможно — кто-нибудь вас найдет.
Пора собрать все кусочки воедино.
Предварительные требования
⚠️ Для прохождения этого уровня необходимо завершить Уровень 0.
Перед началом убедитесь, что у вас есть:
- [ ] config.json в корне проекта, содержащий ваш ID участника и координаты.
- [ ] Ваш аватар виден на карте мира
- [ ] Ваш маяк отображается (тускло) по вашим координатам
Если вы еще не прошли нулевой уровень, начните с него .
Что вы построите
На этом уровне вы создадите многоагентную систему искусственного интеллекта , которая будет анализировать улики с места крушения с помощью параллельной обработки:
Цели обучения
| Концепция | Что вы узнаете |
|---|---|
| Многоагентные системы | Создавайте специализированных агентов с одной единственной обязанностью. |
| ParallelAgent | Создавайте независимые агенты для параллельного выполнения. |
| before_agent_callback | Перед запуском агента необходимо получить конфигурацию и установить состояние. |
| ToolContext | Доступ к значениям состояния в функциях инструмента |
| Пользовательские MCP-серверы | Создавайте инструменты, используя императивный шаблон проектирования (код на Python в Cloud Run). |
| OneMCP BigQuery | Подключитесь к управляемой Google MCP для доступа к BigQuery. |
| Мультимодальный ИИ | Анализируйте изображения, видео и аудио с помощью Gemini. |
| Организация агентов | Координация действий нескольких агентов с помощью корневого оркестратора. |
| Развертывание в облаке | Разверните сервер и агент MCP в Cloud Run. |
| Подготовка A2A | Структурируйте агентов для будущей коммуникации между ними. |
Биомы планеты
Поверхность планеты разделена на четыре различных биома, каждый из которых обладает уникальными характеристиками:
Ваши координаты определяют, в каком биоме вы потерпели крушение. Улики на месте крушения отражают характеристики этого биома:
| Биом | Квадрант | Геологические доказательства | Ботанические доказательства | Астрономические доказательства |
|---|---|---|---|---|
| 🧊 Криотерапия | СЗ (x<50, y≥50) | Замороженный метан, кристаллы льда | Морозные папоротники, криофлора | Голубая гигантская звезда |
| 🌋 ВУЛКАНИЧЕСКИЙ | NE (x≥50, y≥50) | месторождения обсидиана | Огненные цветения, жароустойчивая флора | двойная система красных карликов |
| 💜 БИОЛЮМИНЕСЦЕНТНЫЙ | SW (x<50, y<50) | Фосфоресцентная почва | Светящиеся грибы, светящиеся растения | Зеленый пульсар |
| 🦴 ОКАМЕНЕЛЫЙ | SE (x≥50, y<50) | Месторождения янтаря, минералы ит. | Окаменевшие деревья, древняя флора | Жёлтое солнце |
Ваша задача: создать агентов искусственного интеллекта, способных анализировать улики и определять, в каком биоме вы находитесь.
Настройте свою среду
Продолжительность: 3 мин.
Перед созданием доказательств необходимо включить требуемые API Google Cloud, включая OneMCP для BigQuery, который обеспечивает управляемый доступ MCP к BigQuery.
Запустите скрипт настройки среды.
👉💻 Запустите скрипт настройки среды:
cd ~/way-back-home/level_1
chmod +x setup/setup_env.sh
./setup/setup_env.sh
Вы должны увидеть примерно такой вывод:
================================================================
Level 1: Environment Setup
================================================================
Project: your-project-id
[1/6] Enabling core Google Cloud APIs...
✓ Vertex AI API enabled
✓ Cloud Run API enabled
✓ Cloud Build API enabled
✓ BigQuery API enabled
✓ Artifact Registry API enabled
✓ IAM API enabled
[2/6] Enabling OneMCP BigQuery (Managed MCP)...
✓ OneMCP BigQuery enabled
[3/6] Setting up service account and IAM permissions...
✓ Service account 'way-back-home-sa' created
✓ Vertex AI User role granted
✓ Cloud Run Invoker role granted
✓ BigQuery User role granted
✓ BigQuery Data Viewer role granted
✓ Storage Object Viewer role granted
[4/6] Configuring Cloud Build IAM for deployments...
✓ Cloud Build can now deploy services as way-back-home-sa
✓ Cloud Run Admin role granted to Compute SA
[5/6] Creating Artifact Registry repository...
✓ Repository 'way-back-home' created
[6/6] Creating environment variables file...
Found PARTICIPANT_ID in config.json: abc123...
✓ Created ../set_env.sh
================================================================
✅ Environment Setup Complete!
================================================================
Исходные переменные среды
👉💻 Укажите переменные окружения:
source ~/way-back-home/set_env.sh
Создание виртуальной среды
👉💻 Создайте и активируйте виртуальную среду Python для уровня 1:
cd ~/way-back-home/level_1
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
Настройте каталог звёзд
👉💻 Настройте звездный каталог в BigQuery:
python setup/setup_star_catalog.py
Вам следует увидеть:
Setting up star catalog in project: your-project-id
==================================================
✓ Dataset way_back_home already exists
✓ Created table star_catalog
✓ Inserted 12 rows into star_catalog
📊 Star Catalog Summary:
----------------------------------------
NE (VOLCANIC): 3 stellar patterns
NW (CRYO): 3 stellar patterns
SE (FOSSILIZED): 3 stellar patterns
SW (BIOLUMINESCENT): 3 stellar patterns
----------------------------------------
✓ Star catalog is ready for triangulation queries
==================================================
✅ Star catalog setup complete!
Соберите доказательства с места крушения.
Продолжительность: 2 мин.
Теперь вы можете создать персонализированные данные о месте крушения на основе ваших координат.
Запустите генератор доказательств
👉💻 Из каталога level_1 (с активированным виртуальным окружением) выполните:
cd ~/way-back-home/level_1
python generate_evidence.py
Вы должны увидеть примерно такой вывод:
✓ Welcome back, Explorer_Aria!
Coordinates: (23, 67)
Ready to analyze your crash site.
📍 Crash site analysis initiated...
Generating evidence for your location...
🔬 Generating soil sample...
✓ Soil sample captured: outputs/soil_sample.png
✨ Capturing star field...
✓ Star field captured: outputs/star_field.png
🌿 Recording flora activity...
(This may take 1-2 minutes for video generation)
Generating video...
Generating video...
Generating video...
✓ Flora recorded: outputs/flora_recording.mp4
📤 Uploading evidence to Mission Control...
✓ Config updated with evidence URLs
==================================================
✅ Evidence generation complete!
==================================================
Проанализируйте свои доказательства
👉 Уделите немного времени, чтобы просмотреть сгенерированные файлы с доказательствами в папке outputs/ . Каждый из них отражает характеристики биома места крушения — хотя вы не узнаете, какой именно биом, пока ваши агенты ИИ не проанализируют их!
В зависимости от вашего местоположения, сгенерированные вами доказательства могут выглядеть примерно так:
Создайте собственный MCP-сервер.
Продолжительность: 8 мин.
Системы анализа на борту вашей спасательной капсулы вышли из строя, но исходные данные с датчиков сохранились после крушения. Вам предстоит создать MCP-сервер с помощью FastMCP , предоставляющий инструменты для геологического и ботанического анализа.
Создайте инструмент геологического анализа.
Этот инструмент анализирует изображения образцов почвы для определения минерального состава.
👉✏️ Откройте файл mcp-server/main.py и найдите #REPLACE-GEOLOGICAL-TOOL . Замените её на:
GEOLOGICAL_PROMPT = """Analyze this alien soil sample image.
Classify the PRIMARY characteristic (choose exactly one):
1. CRYO - Frozen/icy minerals, crystalline structures, frost patterns,
blue-white coloration, permafrost indicators
2. VOLCANIC - Volcanic rock, basalt, obsidian, sulfur deposits,
red-orange minerals, heat-formed crystite structures
3. BIOLUMINESCENT - Glowing particles, phosphorescent minerals,
organic-mineral hybrids, purple-green luminescence
4. FOSSILIZED - Ancient compressed minerals, amber deposits,
petrified organic matter, golden-brown stratification
Respond ONLY with valid JSON (no markdown, no explanation):
{
"biome": "CRYO|VOLCANIC|BIOLUMINESCENT|FOSSILIZED",
"confidence": 0.0-1.0,
"minerals_detected": ["mineral1", "mineral2"],
"description": "Brief description of what you observe"
}
"""
@mcp.tool()
def analyze_geological(
image_url: Annotated[
str,
Field(description="Cloud Storage URL (gs://...) of the soil sample image")
]
) -> dict:
"""
Analyzes a soil sample image to identify mineral composition and classify the planetary biome.
Args:
image_url: Cloud Storage URL of the soil sample image (gs://bucket/path/image.png)
Returns:
dict with biome, confidence, minerals_detected, and description
"""
logger.info(f">>> 🔬 Tool: 'analyze_geological' called for '{image_url}'")
try:
response = client.models.generate_content(
model="gemini-2.5-flash",
contents=[
GEOLOGICAL_PROMPT,
genai_types.Part.from_uri(file_uri=image_url, mime_type="image/png")
]
)
result = parse_json_response(response.text)
logger.info(f" ✓ Geological analysis complete: {result.get('biome', 'UNKNOWN')}")
return result
except Exception as e:
logger.error(f" ✗ Geological analysis failed: {str(e)}")
return {"error": str(e), "biome": "UNKNOWN", "confidence": 0.0}
Создайте инструмент для ботанического анализа.
Этот инструмент анализирует видеозаписи растений, включая звуковую дорожку.
👉✏️ Найдите #REPLACE-BOTANICAL-TOOL и замените его на:
BOTANICAL_PROMPT = """Analyze this alien flora video recording.
Pay attention to BOTH:
1. VISUAL elements: Plant appearance, movement patterns, colors, bioluminescence
2. AUDIO elements: Ambient sounds, rustling, organic noises, frequencies
Classify the PRIMARY biome (choose exactly one):
1. CRYO - Crystalline ice-plants, frost-covered vegetation,
crackling/tinkling sounds, slow brittle movements, blue-white flora
2. VOLCANIC - Heat-resistant plants, sulfur-adapted species,
hissing/bubbling sounds, smoke-filtering vegetation, red-orange flora
3. BIOLUMINESCENT - Glowing plants, pulsing light patterns,
humming/resonating sounds, reactive to stimuli, purple-green flora
4. FOSSILIZED - Ancient petrified plants, amber-preserved specimens,
deep resonant sounds, minimal movement, golden-brown flora
Respond ONLY with valid JSON (no markdown, no explanation):
{
"biome": "CRYO|VOLCANIC|BIOLUMINESCENT|FOSSILIZED",
"confidence": 0.0-1.0,
"species_detected": ["species1", "species2"],
"audio_signatures": ["sound1", "sound2"],
"description": "Brief description of visual and audio observations"
}
"""
@mcp.tool()
def analyze_botanical(
video_url: Annotated[
str,
Field(description="Cloud Storage URL (gs://...) of the flora video recording")
]
) -> dict:
"""
Analyzes a flora video recording (visual + audio) to identify plant species and classify the biome.
Args:
video_url: Cloud Storage URL of the flora video (gs://bucket/path/video.mp4)
Returns:
dict with biome, confidence, species_detected, audio_signatures, and description
"""
logger.info(f">>> 🌿 Tool: 'analyze_botanical' called for '{video_url}'")
try:
response = client.models.generate_content(
model="gemini-2.5-flash",
contents=[
BOTANICAL_PROMPT,
genai_types.Part.from_uri(file_uri=video_url, mime_type="video/mp4")
]
)
result = parse_json_response(response.text)
logger.info(f" ✓ Botanical analysis complete: {result.get('biome', 'UNKNOWN')}")
return result
except Exception as e:
logger.error(f" ✗ Botanical analysis failed: {str(e)}")
return {"error": str(e), "biome": "UNKNOWN", "confidence": 0.0}
Протестируйте MCP-сервер локально.
👉💻 Протестируйте сервер MCP:
cd ~/way-back-home/level_1/mcp-server
pip install -r requirements.txt
python main.py
Вам следует увидеть:
[INFO] Initialized Gemini client for project: your-project-id
[INFO] 🚀 Location Analyzer MCP Server starting on port 8080
[INFO] 📍 MCP endpoint: http://0.0.0.0:8080/mcp
[INFO] 🔧 Tools: analyze_geological, analyze_botanical
Сервер FastMCP запущен с использованием протокола HTTP. Нажмите Ctrl+C для остановки.
Разверните сервер MCP в облаке.
👉💻 Развернуть:
cd ~/way-back-home/level_1/mcp-server
source ~/way-back-home/set_env.sh
gcloud builds submit . \
--config=cloudbuild.yaml \
--substitutions=_REGION="$REGION",_REPO_NAME="$REPO_NAME",_SERVICE_ACCOUNT="$SERVICE_ACCOUNT"
Сохраните URL-адрес сервиса
👉💻 Сохраните URL-адрес сервиса:
export MCP_SERVER_URL=$(gcloud run services describe location-analyzer \
--region=$REGION --format='value(status.url)')
echo "MCP Server URL: $MCP_SERVER_URL"
# Add to set_env.sh for later use
echo "export MCP_SERVER_URL=\"$MCP_SERVER_URL\"" >> ~/way-back-home/set_env.sh
Создайте команду специалистов-агентов.
Продолжительность: 8 мин.
Теперь вам нужно создать трех агентов-специалистов, каждый из которых будет выполнять одну единственную задачу.
Создайте агента геологического анализа.
👉✏️ Откройте agent/agents/geological_analyst.py и найдите #REPLACE-GEOLOGICAL-AGENT . Замените её на:
from google.adk.agents import Agent
from agent.tools.mcp_tools import get_geological_tool
geological_analyst = Agent(
name="GeologicalAnalyst",
model="gemini-2.5-flash",
description="Analyzes soil samples to classify planetary biome based on mineral composition.",
instruction="""You are a geological specialist analyzing alien soil samples.
## YOUR EVIDENCE TO ANALYZE
Soil sample URL: {soil_url}
## YOUR TASK
1. Call the analyze_geological tool with the soil sample URL above
2. Examine the results for mineral composition and biome indicators
3. Report your findings clearly
The four possible biomes are:
- CRYO: Frozen, icy minerals, blue/white coloring
- VOLCANIC: Magma, obsidian, volcanic rock, red/orange coloring
- BIOLUMINESCENT: Glowing, phosphorescent minerals, purple/green
- FOSSILIZED: Amber, ancient preserved matter, golden/brown
## REPORTING FORMAT
Always report your classification clearly:
"GEOLOGICAL ANALYSIS: [BIOME] (confidence: X%)"
Include a brief description of what you observed in the sample.
## IMPORTANT
- You do NOT synthesize with other evidence
- You do NOT confirm locations
- Just analyze the soil sample and report what you find
- Call the tool immediately with the URL provided above""",
tools=[get_geological_tool()]
)
Создайте агента «Ботанический аналитик».
👉✏️ Откройте agent/agents/botanical_analyst.py и найдите #REPLACE-BOTANICAL-AGENT . Замените её на:
from google.adk.agents import Agent
from agent.tools.mcp_tools import get_botanical_tool
botanical_analyst = Agent(
name="BotanicalAnalyst",
model="gemini-2.5-flash",
description="Analyzes flora recordings to classify planetary biome based on plant life and ambient sounds.",
instruction="""You are a botanical specialist analyzing alien flora recordings.
## YOUR EVIDENCE TO ANALYZE
Flora recording URL: {flora_url}
## YOUR TASK
1. Call the analyze_botanical tool with the flora recording URL above
2. Pay attention to BOTH visual AND audio elements in the recording
3. Report your findings clearly
The four possible biomes are:
- CRYO: Frost ferns, crystalline plants, cold wind sounds, crackling ice
- VOLCANIC: Fire blooms, heat-resistant flora, crackling/hissing sounds
- BIOLUMINESCENT: Glowing fungi, luminescent plants, ethereal hum, chiming
- FOSSILIZED: Petrified trees, ancient formations, deep resonant sounds
## REPORTING FORMAT
Always report your classification clearly:
"BOTANICAL ANALYSIS: [BIOME] (confidence: X%)"
Include descriptions of what you SAW and what you HEARD.
## IMPORTANT
- You do NOT synthesize with other evidence
- You do NOT confirm locations
- Just analyze the flora recording and report what you find
- Call the tool immediately with the URL provided above""",
tools=[get_botanical_tool()]
)
Создайте агента астрономического аналитика.
Этот агент использует другой подход с двумя вариантами инструментов :
- Local FunctionTool : Gemini Vision для извлечения характеристик звезд
- OneMCP BigQuery : Запрос к каталогу звезд через управляемый Google MCP.
👉✏️ Откройте agent/agents/astronomical_analyst.py и найдите строку #REPLACE-ASTRONOMICAL-AGENT . Замените её на:
from google.adk.agents import Agent
from agent.tools.star_tools import (
extract_star_features_tool,
get_bigquery_mcp_toolset,
)
# Get the BigQuery MCP toolset
bigquery_toolset = get_bigquery_mcp_toolset()
astronomical_analyst = Agent(
name="AstronomicalAnalyst",
model="gemini-2.5-flash",
description="Analyzes star field images and queries the star catalog via OneMCP BigQuery.",
instruction="""You are an astronomical specialist analyzing alien night skies.
## YOUR EVIDENCE TO ANALYZE
Star field URL: {stars_url}
## YOUR TWO TOOLS
### TOOL 1: extract_star_features (Local Gemini Vision)
Call this FIRST with the star field URL above.
Returns: "primary_star": "...", "nebula_type": "...", "stellar_color": "..."
### TOOL 2: BigQuery MCP (execute_query)
Call this SECOND with the results from Tool 1.
Use this exact SQL query (replace the placeholders with values from Step 1):
SELECT quadrant, biome, primary_star, nebula_type
FROM `{project_id}.way_back_home.star_catalog`
WHERE LOWER(primary_star) = LOWER('PRIMARY_STAR_FROM_STEP_1')
AND LOWER(nebula_type) = LOWER('NEBULA_TYPE_FROM_STEP_1')
LIMIT 1
## YOUR WORKFLOW
1. Call extract_star_features with: {stars_url}
2. Get the primary_star and nebula_type from the result
3. Call execute_query with the SQL above (replacing placeholders)
4. Report the biome and quadrant from the query result
## BIOME REFERENCE
| Biome | Quadrant | Primary Star | Nebula Type |
|-------|----------|--------------|-------------|
| CRYO | NW | blue_giant | ice_blue |
| VOLCANIC | NE | red_dwarf_binary | fire |
| BIOLUMINESCENT | SW | green_pulsar | purple_magenta |
| FOSSILIZED | SE | yellow_sun | golden |
## REPORTING FORMAT
"ASTRONOMICAL ANALYSIS: [BIOME] in [QUADRANT] quadrant (confidence: X%)"
Include a description of the stellar features you observed.
## IMPORTANT
- You do NOT synthesize with other evidence
- You do NOT confirm locations
- Just analyze the stars and report what you find
- Start by calling extract_star_features with the URL above""",
tools=[extract_star_features_tool, bigquery_toolset]
)
Создайте подключения к инструменту MCP.
Продолжительность: 8 мин.
Создайте оболочки инструментов, которые будут подключаться к развернутому серверу MCP.
Создание подключения инструмента MCP (пользовательский MCP)
Это соединение устанавливается с вашим пользовательским сервером FastMCP, развернутым в Cloud Run.
👉✏️ Откройте agent/tools/mcp_tools.py и найдите строку #REPLACE-MCP-TOOL-CONNECTION . Замените её на:
import os
import logging
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
logger = logging.getLogger(__name__)
MCP_SERVER_URL = os.environ.get("MCP_SERVER_URL")
_mcp_toolset = None
def get_mcp_toolset():
"""Get the MCPToolset connected to the location-analyzer server."""
global _mcp_toolset
if _mcp_toolset is not None:
return _mcp_toolset
if not MCP_SERVER_URL:
raise ValueError(
"MCP_SERVER_URL not set. Please run:\n"
" export MCP_SERVER_URL='https://location-analyzer-xxx.a.run.app'"
)
# FastMCP exposes MCP protocol at /mcp endpoint
mcp_endpoint = f"{MCP_SERVER_URL}/mcp"
logger.info(f"[MCP Tools] Connecting to: {mcp_endpoint}")
_mcp_toolset = MCPToolset(
connection_params=StreamableHTTPConnectionParams(
url=mcp_endpoint,
timeout=120, # 2 minutes for Gemini analysis
)
)
return _mcp_toolset
def get_geological_tool():
"""Get the geological analysis tool from the MCP server."""
return get_mcp_toolset()
def get_botanical_tool():
"""Get the botanical analysis tool from the MCP server."""
return get_mcp_toolset()
Создание инструментов анализа звезд (OneMCP BigQuery)
В этом разделе демонстрируется шаблон управляемого MCP . Вместо написания собственного клиентского кода для BigQuery мы подключаемся к серверу OneMCP BigQuery от Google.
👉✏️ Откройте agent/tools/star_tools.py и найдите строку #REPLACE-STAR-TOOLS . Замените её на:
import os
import json
import logging
from google import genai
from google.genai import types as genai_types
from google.adk.tools import FunctionTool
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
import google.auth
import google.auth.transport.requests
logger = logging.getLogger(__name__)
# =============================================================================
# CONFIGURATION - Environment variables only
# =============================================================================
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "")
if not PROJECT_ID:
logger.warning("[Star Tools] GOOGLE_CLOUD_PROJECT not set")
# Initialize Gemini client for star feature extraction
genai_client = genai.Client(
vertexai=True,
project=PROJECT_ID or "placeholder",
location=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
)
logger.info(f"[Star Tools] Initialized for project: {PROJECT_ID}")
# =============================================================================
# OneMCP BigQuery Connection
# =============================================================================
BIGQUERY_MCP_URL = "https://bigquery.googleapis.com/mcp"
_bigquery_toolset = None
def get_bigquery_mcp_toolset():
"""
Get the MCPToolset connected to Google's BigQuery MCP server.
This uses OAuth 2.0 authentication with Application Default Credentials.
The toolset provides access to BigQuery's pre-built MCP tools like:
- execute_query: Run SQL queries
- list_datasets: List available datasets
- get_table_schema: Get table structure
"""
global _bigquery_toolset
if _bigquery_toolset is not None:
return _bigquery_toolset
logger.info("[Star Tools] Connecting to OneMCP BigQuery...")
# Get OAuth credentials
credentials, project_id = google.auth.default(
scopes=["https://www.googleapis.com/auth/bigquery"]
)
# Refresh to get a valid token
credentials.refresh(google.auth.transport.requests.Request())
oauth_token = credentials.token
# Configure headers for BigQuery MCP
headers = {
"Authorization": f"Bearer {oauth_token}",
"x-goog-user-project": project_id or PROJECT_ID
}
# Create MCPToolset with StreamableHTTP connection
_bigquery_toolset = MCPToolset(
connection_params=StreamableHTTPConnectionParams(
url=BIGQUERY_MCP_URL,
headers=headers
)
)
logger.info("[Star Tools] Connected to BigQuery MCP")
return _bigquery_toolset
# =============================================================================
# Local FunctionTool: Star Feature Extraction
# =============================================================================
# This is a LOCAL tool that calls Gemini directly - demonstrating that
# you can mix local FunctionTools with MCP tools in the same agent.
STAR_EXTRACTION_PROMPT = """Analyze this alien night sky image and extract stellar features.
Identify:
1. PRIMARY STAR TYPE: blue_giant, red_dwarf, red_dwarf_binary, green_pulsar, yellow_sun, etc.
2. NEBULA TYPE: ice_blue, fire, purple_magenta, golden, etc.
3. STELLAR COLOR: blue_white, red_orange, green_purple, yellow_gold, etc.
Respond ONLY with valid JSON:
{"primary_star": "...", "nebula_type": "...", "stellar_color": "...", "description": "..."}
"""
def _parse_json_response(text: str) -> dict:
"""Parse JSON from Gemini response, handling markdown formatting."""
cleaned = text.strip()
if cleaned.startswith("```json"):
cleaned = cleaned[7:]
elif cleaned.startswith("```"):
cleaned = cleaned[3:]
if cleaned.endswith("```"):
cleaned = cleaned[:-3]
cleaned = cleaned.strip()
try:
return json.loads(cleaned)
except json.JSONDecodeError as e:
logger.error(f"Failed to parse JSON: {e}")
return {"error": f"Failed to parse response: {str(e)}"}
def extract_star_features(image_url: str) -> dict:
"""
Extract stellar features from a star field image using Gemini Vision.
This is a LOCAL FunctionTool - we call Gemini directly, not through MCP.
The agent will use this alongside the BigQuery MCP tools.
"""
logger.info(f"[Stars] Extracting features from: {image_url}")
response = genai_client.models.generate_content(
model="gemini-2.5-flash",
contents=[
STAR_EXTRACTION_PROMPT,
genai_types.Part.from_uri(file_uri=image_url, mime_type="image/png")
]
)
result = _parse_json_response(response.text)
logger.info(f"[Stars] Extracted: primary_star={result.get('primary_star')}")
return result
# Create the local FunctionTool
extract_star_features_tool = FunctionTool(extract_star_features)
Создайте оркестратор
Продолжительность: 8 мин.
Теперь создайте параллельную группу и корневой оркестратор, который будет координировать все процессы.
Создайте группу параллельного анализа.
Для начала создадим функцию обратного вызова и ParallelAgent, которые будут запускать специалистов одновременно.
👉✏️ Откройте agent/agent.py и найдите #REPLACE-PARALLEL-CREW . Замените её на:
import os
import logging
import httpx
from google.adk.agents import Agent, ParallelAgent
from google.adk.agents.callback_context import CallbackContext
# Import specialist agents
from agent.agents.geological_analyst import geological_analyst
from agent.agents.botanical_analyst import botanical_analyst
from agent.agents.astronomical_analyst import astronomical_analyst
# Import confirmation tool
from agent.tools.confirm_tools import confirm_location_tool
logger = logging.getLogger(__name__)
# =============================================================================
# BEFORE AGENT CALLBACK - Fetches config and sets state
# =============================================================================
async def setup_participant_context(callback_context: CallbackContext) -> None:
"""
Fetch participant configuration and populate state for all agents.
This callback:
1. Reads PARTICIPANT_ID and BACKEND_URL from environment
2. Fetches participant data from the backend API
3. Sets state values: soil_url, flora_url, stars_url, username, x, y, etc.
4. Returns None to continue normal agent execution
"""
participant_id = os.environ.get("PARTICIPANT_ID", "")
backend_url = os.environ.get("BACKEND_URL", "https://api.waybackhome.dev")
project_id = os.environ.get("GOOGLE_CLOUD_PROJECT", "")
logger.info(f"[Callback] Setting up context for participant: {participant_id}")
# Set project_id and backend_url in state immediately
callback_context.state["project_id"] = project_id
callback_context.state["backend_url"] = backend_url
callback_context.state["participant_id"] = participant_id
if not participant_id:
logger.warning("[Callback] No PARTICIPANT_ID set - using placeholder values")
callback_context.state["username"] = "Explorer"
callback_context.state["x"] = 0
callback_context.state["y"] = 0
callback_context.state["soil_url"] = "Not available - set PARTICIPANT_ID"
callback_context.state["flora_url"] = "Not available - set PARTICIPANT_ID"
callback_context.state["stars_url"] = "Not available - set PARTICIPANT_ID"
return None
# Fetch participant data from backend API
try:
url = f"{backend_url}/participants/{participant_id}"
logger.info(f"[Callback] Fetching from: {url}")
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(url)
response.raise_for_status()
data = response.json()
# Extract evidence URLs
evidence_urls = data.get("evidence_urls", {})
# Set all state values for sub-agents to access
callback_context.state["username"] = data.get("username", "Explorer")
callback_context.state["x"] = data.get("x", 0)
callback_context.state["y"] = data.get("y", 0)
callback_context.state["soil_url"] = evidence_urls.get("soil", "Not available")
callback_context.state["flora_url"] = evidence_urls.get("flora", "Not available")
callback_context.state["stars_url"] = evidence_urls.get("stars", "Not available")
logger.info(f"[Callback] State populated for {data.get('username')}")
except Exception as e:
logger.error(f"[Callback] Error fetching participant config: {e}")
callback_context.state["username"] = "Explorer"
callback_context.state["x"] = 0
callback_context.state["y"] = 0
callback_context.state["soil_url"] = f"Error: {e}"
callback_context.state["flora_url"] = f"Error: {e}"
callback_context.state["stars_url"] = f"Error: {e}"
return None
# =============================================================================
# PARALLEL ANALYSIS CREW
# =============================================================================
evidence_analysis_crew = ParallelAgent(
name="EvidenceAnalysisCrew",
description="Runs geological, botanical, and astronomical analysis in parallel.",
sub_agents=[geological_analyst, botanical_analyst, astronomical_analyst]
)
Создайте корневой оркестратор.
Теперь создайте корневой агент, который координирует все процессы и использует функцию обратного вызова.
👉✏️ В том же файле ( agent/agent.py ) найдите строку #REPLACE-ROOT-ORCHESTRATOR . Замените её на:
# =============================================================================
# ROOT ORCHESTRATOR
# =============================================================================
root_agent = Agent(
name="MissionAnalysisAI",
model="gemini-2.5-flash",
description="Coordinates crash site analysis to confirm explorer location.",
instruction="""You are the Mission Analysis AI coordinating a rescue operation.
## Explorer Information
- Name: {username}
- Coordinates: ({x}, {y})
## Evidence URLs (automatically provided to specialists via state)
- Soil sample: {soil_url}
- Flora recording: {flora_url}
- Star field: {stars_url}
## Your Workflow
### STEP 1: DELEGATE TO ANALYSIS CREW
Tell the EvidenceAnalysisCrew to analyze all the evidence.
The evidence URLs are already available to the specialists.
### STEP 2: COLLECT RESULTS
Each specialist will report:
- "GEOLOGICAL ANALYSIS: [BIOME] (confidence: X%)"
- "BOTANICAL ANALYSIS: [BIOME] (confidence: X%)"
- "ASTRONOMICAL ANALYSIS: [BIOME] in [QUADRANT] quadrant (confidence: X%)"
### STEP 3: APPLY 2-OF-3 AGREEMENT RULE
- If 2 or 3 specialists agree → that's the answer
- If all 3 disagree → use judgment based on confidence
### STEP 4: CONFIRM LOCATION
Call confirm_location with the determined biome.
## Biome Reference
| Biome | Quadrant | Key Characteristics |
|-------|----------|---------------------|
| CRYO | NW | Frozen, blue, ice crystals |
| VOLCANIC | NE | Magma, red/orange, obsidian |
| BIOLUMINESCENT | SW | Glowing, purple/green |
| FOSSILIZED | SE | Amber, golden, ancient |
## Response Style
Be encouraging and narrative! Celebrate when the beacon activates!
""",
sub_agents=[evidence_analysis_crew],
tools=[confirm_location_tool],
before_agent_callback=setup_participant_context
)
Создайте инструмент подтверждения местоположения.
Этот инструмент использует ToolContext для чтения значений состояния, установленных функцией обратного вызова.
👉✏️ В agent/tools/confirm_tools.py найдите #REPLACE-CONFIRM-TOOL . Замените её на:
import os
import logging
import requests
from google.adk.tools import FunctionTool
from google.adk.tools.tool_context import ToolContext
logger = logging.getLogger(__name__)
BIOME_TO_QUADRANT = {
"CRYO": "NW",
"VOLCANIC": "NE",
"BIOLUMINESCENT": "SW",
"FOSSILIZED": "SE"
}
def _get_actual_biome(x: int, y: int) -> tuple[str, str]:
"""Determine actual biome and quadrant from coordinates."""
if x < 50 and y >= 50:
return "NW", "CRYO"
elif x >= 50 and y >= 50:
return "NE", "VOLCANIC"
elif x < 50 and y < 50:
return "SW", "BIOLUMINESCENT"
else:
return "SE", "FOSSILIZED"
def confirm_location(biome: str, tool_context: ToolContext) -> dict:
"""
Confirm the explorer's location and activate the rescue beacon.
Uses ToolContext to read state values set by before_agent_callback.
"""
# Read from state (set by before_agent_callback)
participant_id = tool_context.state.get("participant_id", "")
x = tool_context.state.get("x", 0)
y = tool_context.state.get("y", 0)
backend_url = tool_context.state.get("backend_url", "https://api.waybackhome.dev")
# Fallback to environment variables
if not participant_id:
participant_id = os.environ.get("PARTICIPANT_ID", "")
if not backend_url:
backend_url = os.environ.get("BACKEND_URL", "https://api.waybackhome.dev")
if not participant_id:
return {"success": False, "message": "❌ No participant ID available."}
biome_upper = biome.upper().strip()
if biome_upper not in BIOME_TO_QUADRANT:
return {"success": False, "message": f"❌ Unknown biome: {biome}"}
# Get actual biome from coordinates
actual_quadrant, actual_biome = _get_actual_biome(x, y)
if biome_upper != actual_biome:
return {
"success": False,
"message": f"❌ Mismatch! Analysis: {biome_upper}, Actual: {actual_biome}"
}
quadrant = BIOME_TO_QUADRANT[biome_upper]
try:
response = requests.patch(
f"{backend_url}/participants/{participant_id}/location",
params={"x": x, "y": y},
timeout=10
)
response.raise_for_status()
return {
"success": True,
"message": f"🔦 BEACON ACTIVATED!\n\nLocation: {biome_upper} in {quadrant}\nCoordinates: ({x}, {y})"
}
except requests.exceptions.ConnectionError:
return {
"success": True,
"message": f"🔦 BEACON ACTIVATED! (Local)\n\nLocation: {biome_upper} in {quadrant}",
"simulated": True
}
except Exception as e:
return {"success": False, "message": f"❌ Failed: {str(e)}"}
confirm_location_tool = FunctionTool(confirm_location)
Тестирование с помощью веб-интерфейса ADK.
Продолжительность: 5 мин.
Теперь давайте протестируем всю многоагентную систему локально.
Запустите веб-сервер ADK.
👉💻 Установите переменные окружения и запустите веб-сервер ADK:
cd ~/way-back-home/level_1
source ~/way-back-home/set_env.sh
# Verify environment is set
echo "PARTICIPANT_ID: $PARTICIPANT_ID"
echo "MCP Server: $MCP_SERVER_URL"
# Start ADK web server
adk web
Вам следует увидеть:
+-----------------------------------------------------------------------------+
| ADK Web Server started |
| |
| For local testing, access at http://localhost:8000. |
+-----------------------------------------------------------------------------+
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
Получите доступ к веб-интерфейсу.
👉 На панели инструментов Cloud Shell (вверху справа) в меню « Предварительный просмотр веб-страниц» выберите «Изменить порт» .
👉 Установите порт на 8000 и нажмите «Изменить и просмотреть» .
👉 Откроется веб-интерфейс ADK. Выберите агента из выпадающего меню.
Запустить анализ
👉 В интерфейсе чата введите:
Analyze the evidence from my crash site and confirm my location to activate the beacon.
Посмотрите, как работает многоагентная система:
- Функция before_agent_callback выполняется первой, получая данные об участниках.
- Корневой оркестратор получает ваш запрос с заполненным состоянием.
- Команда анализа доказательств активирует (Параллельный агент)
- Три специалиста работают параллельно, используя шаблонизацию
{key}:- GeologicalAnalyst → видит, что
{soil_url}получен из состояния - BotanicalAnalyst → видит, что
{flora_url}получен из состояния - AstronomicalAnalyst → видит,
{stars_url}и{project_id}разрешены.
- GeologicalAnalyst → видит, что
- Основной оркестратор синтезирует (согласование 2 из 3)
- Вызов функции confirm_location с ToolContext → "🔦 МАЯК АКТИВИРОВАН!"
На панели трассировки справа отображаются все взаимодействия агентов и вызовы инструментов.
👉 Нажмите Ctrl+C в терминале, чтобы остановить сервер после завершения тестирования.
Развертывание в облаке. Запуск.
Продолжительность: 5 мин.
Теперь разверните свою многоагентную систему в Cloud Run для обеспечения готовности к взаимодействию между агентами (A2A) .
Разверните агента
👉💻 Развертывание в Cloud Run с помощью ADK CLI:
cd ~/way-back-home/level_1
source ~/way-back-home/set_env.sh
adk deploy cloud_run \
--project=$GOOGLE_CLOUD_PROJECT \
--region=$REGION \
--service_name=mission-analysis-ai \
--with_ui \
--a2a \
./agent
Когда появится запрос « Allow unauthenticated invocations to [mission-analysis-ai] (y/N)? , введите y , чтобы разрешить публичный доступ.
Вы должны увидеть примерно такой вывод:
Building and deploying agent to Cloud Run...
✓ Container built successfully
✓ Deploying to Cloud Run...
✓ Service deployed: https://mission-analysis-ai-abc123-uc.a.run.app
Your agent is now live!
Настройка переменных среды в Cloud Run
Развернутому агенту необходим доступ к переменным среды. Обновите службу:
👉💻 Установите необходимые переменные среды:
gcloud run services update mission-analysis-ai \
--region=$REGION \
--set-env-vars="GOOGLE_CLOUD_PROJECT=$GOOGLE_CLOUD_PROJECT,GOOGLE_CLOUD_LOCATION=$REGION,MCP_SERVER_URL=$MCP_SERVER_URL,BACKEND_URL=$BACKEND_URL,PARTICIPANT_ID=$PARTICIPANT_ID,GOOGLE_GENAI_USE_VERTEXAI=True"
Сохраните URL-адрес агента
👉💻 Получите развернутый URL-адрес:
export AGENT_URL=$(gcloud run services describe mission-analysis-ai \
--region=$REGION --format='value(status.url)')
echo "Agent URL: $AGENT_URL"
# Add to set_env.sh
echo "export LEVEL1_AGENT_URL=\"$AGENT_URL\"" >> ~/way-back-home/set_env.sh
Проверка развертывания
👉💻 Проверьте развернутый агент, открыв URL-адрес в браузере (флаг --with_ui развертывает веб-интерфейс ADK), или проверьте с помощью curl:
curl -X GET "$AGENT_URL/list-apps"
Вы должны увидеть ответ со списком ваших агентов.
Заключение
Продолжительность: 1 мин.
Контрольный список проверки
✅ Сервер MCP
- [ ] Развернуто в Cloud Run
- [ ] Инструмент analyze_geological работает
- [ ] Инструмент analyze_botanical работает
✅ Специализированные агенты
- [ ] GeologicalAnalyst использует {soil_url} из штата
- [ ] BotanicalAnalyst использует {flora_url} из штата
- [ ] AstronomicalAnalyst использует {stars_url} и {project_id} из состояния
✅ before_agent_callback
- [ ] Получает данные участников из бэкэнд API
- [ ] Устанавливает значения состояния для всех субагентов
- [ ] Работает с PARTICIPANT_ID из среды
✅ ParallelAgent
- [ ] Все три специалиста работают одновременно
- [ ] Состояние передается через InvocationContext
✅ Root Orchestrator
- [ ] Синтезирует с согласованием 2 из 3
- [ ] confirm_location использует ToolContext для состояния
- [ ] Маяк активирован!
✅ Развертывание
- [ ] Агент развернут в Cloud Run
- [ ] Доступна конечная точка A2A
✅ Карта мира
- [ ] Маяк теперь ЯРКИЙ (а не тусклый)
- [ ] Биом отображается при наведении курсора
🎉 Первый уровень пройден!
Ваш спасательный маяк теперь работает на полную мощность. Треугольный сигнал пробивается сквозь атмосферные помехи, представляя собой устойчивый импульс, говорящий: «Я здесь. Я выжил. Приходите меня найти».
Но вы не единственный на этой планете. Когда ваш маяк активируется, вы замечаете, как на горизонте загораются другие огни — другие выжившие, другие места крушения, другие исследователи, которым удалось выжить.
На втором уровне вы научитесь обрабатывать поступающие сигналы SOS и координировать свои действия с другими выжившими. Спасение – это не просто поиск себя, это поиск друг друга.
Поиск неисправностей
"MCP_SERVER_URL не установлен" bash export MCP_SERVER_URL=$(gcloud run services describe location-analyzer \ --region=$REGION --format='value(status.url)')
"PARTICIPANT_ID не установлен" bash source ~/way-back-home/set_env.sh echo $PARTICIPANT_ID
"Таблица BigQuery не найдена" bash python setup/setup_star_catalog.py
"Специалисты запрашивают URL-адреса". Это означает, что шаблонизация {key} не работает. Проверьте: - Установлен ли параметр before_agent_callback на корневом агенте? - Правильно ли устанавливается значение состояния в функции обратного вызова? - Используют ли субагенты {soil_url} (а не f-строки)?
«Все три анализа расходятся во мнениях». Восстановить данные: python generate_evidence.py
"Агент не отвечает в веб-интерфейсе adk" - Убедитесь, что порт 8000 указан правильно - Проверьте, установлены ли параметры MCP_SERVER_URL и PARTICIPANT_ID - Проверьте терминал на наличие сообщений об ошибках
Архитектурное резюме
| Компонент | Тип | Шаблон | Цель |
|---|---|---|---|
| setup_participant_context | Перезвонить | before_agent_callback | Получить конфигурацию, установить состояние |
| Геологический аналитик | Агент | {soil_url} шаблонирование | Классификация почв |
| Ботанический аналитик | Агент | {flora_url} шаблонирование | Классификация флоры |
| Астрономический аналитик | Агент | {stars_url}, {project_id} | Звездная триангуляция |
| подтвердить_местоположение | Инструмент | Доступ к состоянию ToolContext | Активировать маяк |
| Команда анализа доказательств | ParallelAgent | Состав суб-агентов | Запускайте специалистов одновременно. |
| MissionAnalysisAI | Агент (Корень) | Оркестратор + обратный вызов | Координировать + синтезировать |
| анализатор местоположения | Сервер FastMCP | Пользовательский MCP | Геологический и ботанический анализ |
| bigquery.googleapis.com/mcp | OneMCP | Управляемый MCP | доступ к BigQuery |
Освоенные ключевые понятия
✅ before_agent_callback : Получение конфигурации перед запуском агента
✅ {ключ} Шаблонизация состояний : Доступ к значениям состояний в инструкциях агента
✅ ToolContext : Доступ к значениям состояния в функциях инструмента
✅ Совместное использование состояния : состояние родительского объекта автоматически становится доступно дочерним агентам через InvocationContext.
✅ Многоагентная архитектура : специализированные агенты с одной единственной обязанностью
✅ ParallelAgent : Параллельное выполнение независимых задач
✅ Пользовательский MCP-сервер : Ваш собственный MCP-сервер в облаке
✅ OneMCP BigQuery : управляемый шаблон MCP для доступа к базе данных
✅ Облачное развертывание : развертывание без сохранения состояния с использованием переменных среды.
✅ Подготовка к A2A : Агент готов к межагентскому общению
Для тех, кто не увлекается видеоиграми: практическое применение.
«Определение вашего местоположения» представляет собой параллельный экспертный анализ с учетом консенсуса — одновременное выполнение нескольких специализированных анализов с использованием ИИ и синтез результатов.
Корпоративные приложения
| Вариант использования | Параллельные эксперты | Правило синтеза |
|---|---|---|
| Медицинский диагноз | аналитик изображений, аналитик симптомов, лаборант | Порог доверия 2 из 3 |
| Выявление мошенничества | Аналитик транзакций, поведенческий аналитик, сетевой аналитик | Любой 1 флаг = проверка |
| Обработка документов | Агент оптического распознавания символов, агент классификации, агент экстракции | Все должны согласиться |
| Контроль качества | Визуальный инспектор, аналитик датчиков, специалист по проверке технических характеристик. | пас 2 из 3 |
Ключевые архитектурные идеи
before_agent_callback для конфигурации : Получает конфигурацию один раз в начале, заполняет состояние для всех субагентов. Чтение файла конфигурации субагентами не выполняется.
{key} Шаблонизация состояния : декларативная, чистая, идиоматическая. Без f-строк, без импортов, без манипуляций с sys.path.
Механизмы достижения консенсуса : метод «2 из 3» позволяет надежно обрабатывать неоднозначность, не требуя единогласного согласия.
ParallelAgent для независимых задач : когда анализы не зависят друг от друга, запускайте их параллельно для повышения скорости.
Два варианта MCP : собственный (создание на заказ) и OneMCP (размещение на серверах Google). Оба используют StreamableHTTP.
Развертывание без сохранения состояния : один и тот же код работает локально и в развернутом режиме. Переменные среды + бэкэнд API = отсутствие конфигурационных файлов в контейнерах.
Что дальше?
Уровень 2: Обработка сигнала SOS →
Научитесь обрабатывать поступающие сигналы бедствия от других выживших, используя закономерности, основанные на событиях, и более совершенную координацию действий агентов.