راه بازگشت به خانه - سطح 1: مکان دقیق را مشخص کنید

ماموریت

مدت زمان: ۲ دقیقه

سربرگ

شما خودتان را به هوش مصنوعی اورژانس معرفی کرده‌اید و چراغ راهنمای شما اکنون روی نقشه سیاره می‌درخشد - اما این چشمک زدن به سختی در میان امواج ثابت گم شده است. تیم‌های نجات که از مدار در حال اسکن هستند، می‌توانند چیزی در مختصات شما ببینند، اما نمی‌توانند آن را قفل کنند. سیگنال خیلی ضعیف است.

برای تقویت کامل قدرت چراغ‌هایتان، باید مکان دقیق خود را تأیید کنید . سیستم ناوبری کپسول از کار افتاده است، اما سقوط، شواهد قابل بازیابی را در محل فرود پراکنده کرده است. نمونه‌های خاک. گیاهان عجیب. منظره‌ای واضح از آسمان شب بیگانه.

اگر بتوانید این شواهد را تجزیه و تحلیل کنید و مشخص کنید که در کدام منطقه از سیاره هستید، هوش مصنوعی می‌تواند موقعیت شما را مثلث‌بندی کرده و سیگنال چراغ راهنما را تقویت کند. آنگاه—شاید—کسی شما را پیدا کند.

وقتشه که تکه‌ها رو کنار هم بذاریم.

پیش‌نیازها

⚠️ این سطح نیاز به تکمیل سطح 0 دارد.

قبل از شروع، مطمئن شوید که موارد زیر را دارید:
- [ ] config.json در ریشه پروژه به همراه شناسه و مختصات شرکت‌کننده شما
- [ ] آواتار شما روی نقشه جهان قابل مشاهده است
- [ ] چراغ راهنمای شما در مختصات شما (کم نور) نشان داده می‌شود

اگر سطح ۰ را تمام نکرده‌اید، ابتدا از آنجا شروع کنید .

آنچه خواهید ساخت

در این سطح، شما یک سیستم هوش مصنوعی چندعامله خواهید ساخت که شواهد محل سقوط را با استفاده از پردازش موازی تجزیه و تحلیل می‌کند:

معماری

اهداف یادگیری

مفهوم آنچه یاد خواهید گرفت
سیستم‌های چندعاملی ایجاد نمایندگان تخصصی با مسئولیت‌های واحد
عامل موازی ایجاد عامل‌های مستقل برای اجرای همزمان
قبل از تماس عامل پیکربندی را دریافت کنید و قبل از اجرای عامل، وضعیت را تنظیم کنید
ابزار زمینه دسترسی به مقادیر حالت در توابع ابزار
سرورهای سفارشی MCP ساخت ابزارها با الگوی دستوری (کد پایتون در Cloud Run)
OneMCP بیگ‌کوئری برای دسترسی به BigQuery به MCP مدیریت‌شده‌ی گوگل متصل شوید
هوش مصنوعی چندوجهی تصاویر و ویدیو + صدا را با Gemini تجزیه و تحلیل کنید
هماهنگ‌سازی عامل هماهنگ‌سازی چندین عامل با یک هماهنگ‌کننده ریشه
استقرار ابری سرور و عامل MCP را در Cloud Run مستقر کنید
آماده‌سازی A2A ساختاردهی عامل‌ها برای ارتباطات عامل به عامل در آینده

زیست‌بوم‌های سیاره

سطح سیاره به چهار زیست‌بوم مجزا تقسیم شده است که هر کدام ویژگی‌های منحصر به فردی دارند:

زیست‌بوم سیاره

مختصات شما مشخص می‌کند که در کدام زیست‌بوم سقوط کرده‌اید. شواهد موجود در محل سقوط شما، ویژگی‌های آن زیست‌بوم را نشان می‌دهد:

زیست‌بوم ربع شواهد زمین‌شناسی شواهد گیاه‌شناسی شواهد نجومی
🧊 کرایو شمال غربی (x<50، y≥50) متان منجمد، بلورهای یخ سرخس‌های یخ‌زده، گیاهان سرمازا ستاره غول آبی
🌋 آتشفشانی NE (x≥50، y≥50) ذخایر ابسیدین، سنگ آهک شکوفه‌های آتش، گیاهان مقاوم در برابر گرما دودویی کوتوله قرمز
💜 زیست‌تابی SW (x<50، y<50) خاک فسفردار قارچ‌های درخشان، گیاهان شب‌تاب تپ‌اختر سبز
🦴 فسیل شده SE (x≥50، y<50) ذخایر کهربا، کانی‌های ite درختان سنگ‌شده، گیاهان باستانی خورشید زرد

وظیفه شما: ساخت عوامل هوش مصنوعی که بتوانند شواهد را تجزیه و تحلیل کرده و استنباط کنند که در کدام زیست‌بوم (زیست‌بوم) هستید.

محیط خود را آماده کنید

مدت زمان: ۳ دقیقه

قبل از تولید شواهد، باید 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

ایجاد محیط مجازی

👉💻 محیط مجازی پایتون را برای سطح ۱ ایجاد و فعال کنید:

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!

ایجاد شواهد محل سقوط

مدت زمان: ۲ دقیقه

اکنون بر اساس مختصات خود، شواهد شخصی‌سازی‌شده‌ای از محل سقوط ایجاد کنید.

مولد شواهد را اجرا کنید

👉💻 از دایرکتوری level_1 (با فعال بودن venv)، دستور زیر را اجرا کنید:

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

مدت زمان: ۸ دقیقه

سیستم‌های تجزیه و تحلیل داخلی کپسول فرار شما از کار افتاده‌اند، اما داده‌های خام حسگر از تصادف جان سالم به در برده‌اند. شما یک سرور 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

سرور FastMCP اکنون با انتقال HTTP در حال اجرا است. برای توقف، Ctrl+C را فشار دهید.

سرور MCP را روی Cloud Run مستقر کنید

👉💻 استقرار: 

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"

آدرس اینترنتی سرویس را ذخیره کنید

👉💻 آدرس اینترنتی سرویس را ذخیره کنید: 

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

نمایندگان متخصص بسازید

مدت زمان: ۸ دقیقه

حالا شما سه مامور متخصص ایجاد خواهید کرد که هر کدام یک مسئولیت واحد دارند.

ایجاد عامل تحلیلگر زمین شناسی

👉✏️ 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()]
)

ایجاد عامل تحلیلگر نجومی

این عامل از رویکرد متفاوتی با دو الگوی ابزار استفاده می‌کند:

  1. ابزار عملکرد محلی : Gemini Vision برای استخراج ویژگی‌های ستاره
  2. OneMCP BigQuery : از طریق 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 را بسازید

مدت زمان: ۸ دقیقه

بسته‌های ابزاری را ایجاد کنید که به سرور 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 گوگل متصل می‌شویم.

👉✏️ 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)

ساخت ارکستراتور

مدت زمان: ۸ دقیقه

حالا گروه موازی و هماهنگ‌کننده‌ی اصلی را ایجاد کنید که همه چیز را هماهنگ می‌کند.

تیم تحلیل موازی را تشکیل دهید

ابتدا، بیایید تابع فراخوانی و 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]
)

ایجاد ارکستراتور ریشه

حالا عامل ریشه‌ای (root agent) را ایجاد کنید که همه چیز را هماهنگ می‌کند و از callback استفاده می‌کند.

👉✏️ در همان فایل ( 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 تست کنید

مدت زمان: ۵ دقیقه

حالا بیایید کل سیستم چندعامله را به صورت محلی آزمایش کنیم.

وب سرور 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 (بالا سمت راست)، گزینه Change port را انتخاب کنید.

آیکون پیش‌نمایش وب

👉 پورت را روی ۸۰۰۰ تنظیم کنید و روی «تغییر و پیش‌نمایش» کلیک کنید.

پنجره تغییر پورت

👉 رابط کاربری وب ADK باز خواهد شد. از منوی کشویی، agent را انتخاب کنید.

انتخاب عامل

تحلیل را اجرا کنید

👉 در رابط چت، تایپ کنید: 

Analyze the evidence from my crash site and confirm my location to activate the beacon.

سیستم چندعاملی را در عمل تماشا کنید:

نسخه آزمایشی وب adk

  1. before_agent_callback ابتدا اجرا می‌شود و داده‌های شرکت‌کننده شما را دریافت می‌کند.
  2. هماهنگ‌کننده ریشه درخواست شما را با وضعیت ثبت‌شده دریافت می‌کند.
  3. EvidenceAnalysisCrew فعال می‌شود (ParallelAgent)
  4. سه متخصص به صورت موازی با استفاده از قالب‌بندی {key} اجرا می‌شوند:
    • GeologicalAnalyst → {soil_url} را از حالت حل شده می‌بیند
    • BotanicalAnalyst → وضعیت {flora_url} را حل‌شده می‌بیند
    • AstronomicalAnalyst → {stars_url} و {project_id} را حل‌شده می‌بیند
  5. هماهنگ‌کننده ریشه (توافق ۲ از ۳) سنتز می‌کند
  6. confirm_location با ToolContext فراخوانی شد → "🔦 بیکن فعال شد!"

پنل ردیابی در سمت راست، تمام تعاملات عامل و فراخوانی‌های ابزار را نشان می‌دهد.

👉 برای متوقف کردن سرور پس از انجام تست، Ctrl+C را در ترمینال فشار دهید.

استقرار در Cloud Run

مدت زمان: ۵ دقیقه

اکنون سیستم چندعاملی خود را برای آمادگی A2A در Cloud Run مستقر کنید.

عامل را مستقر کنید

👉💻 با استفاده از ADK CLI، روی Cloud Run مستقر شوید: 

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) منتشر شده را دریافت کنید: 

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"

شما باید پاسخی را ببینید که نماینده شما را فهرست می‌کند.

نتیجه‌گیری

مدت زمان: ۱ دقیقه

چک لیست تأیید

✅ سرور MCP
- [] به Cloud Run منتقل شد
- [] ابزار analyze_geological کار می‌کند
- [] ابزار analyze_botanical کار می‌کند

✅ نمایندگان متخصص
- [] GeologicalAnalyst از {soil_url} از state استفاده می‌کند.
- [ ] BotanicalAnalyst از {flora_url} از state استفاده می‌کند.
- [ ] AstronomicalAnalyst از {stars_url} و {project_id} از state استفاده می‌کند

✅ قبل از تماس مجدد با اپراتور
- [] داده‌های شرکت‌کننده را از API بک‌اند دریافت می‌کند
- [ ] مقادیر حالت را برای همه زیرعامل‌ها تنظیم می‌کند
- [] با PARTICIPANT_ID از محیط کار می‌کند

✅ عامل موازی
- [ ] هر سه متخصص به طور همزمان فعالیت می‌کنند
- [] وضعیت از طریق InvocationContext به اشتراک گذاشته می‌شود

✅ ارکستراسیون ریشه
- [ ] با توافق ۲ از ۳ ترکیب می‌شود
- [ ] 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 روی عامل ریشه تنظیم شده است؟ - آیا callback مقادیر state را به درستی تنظیم می‌کند؟ - آیا عامل‌های فرعی از {soil_url} استفاده می‌کنند (نه f-strings)؟

«هر سه تحلیل با هم مخالفند» شواهد را بازسازی کنید: python generate_evidence.py

«اپراتور در adk web پاسخ نمی‌دهد» - بررسی کنید که پورت ۸۰۰۰ صحیح باشد - تأیید کنید که MCP_SERVER_URL و PARTICIPANT_ID تنظیم شده‌اند - ترمینال را برای پیام‌های خطا بررسی کنید

خلاصه معماری

کامپوننت نوع الگو هدف
setup_participant_context تماس برگشتی قبل از تماس عامل دریافت پیکربندی، تنظیم وضعیت
تحلیلگر زمین شناسی عامل قالب‌بندی {soil_url} طبقه‌بندی خاک
تحلیلگر گیاه‌شناسی عامل قالب‌بندی {flora_url} طبقه‌بندی گیاهان
تحلیلگر نجوم عامل {stars_url}، {project_id} مثلث‌بندی ستاره‌ای
تایید_موقعیت ابزار دسترسی به وضعیت ToolContext فعال کردن چراغ راهنما
تیم تحلیل شواهد عامل موازی ترکیب زیرعامل متخصصان را همزمان اجرا کنید
تحلیل ماموریت هوش مصنوعی عامل (ریشه) تنظیم کننده ارکستر + تماس مجدد هماهنگ‌سازی + ترکیب
تحلیلگر مکان سرور FastMCP MCP سفارشی تجزیه و تحلیل زمین شناسی + گیاه شناسی
bigquery.googleapis.com/mcp وان‌ام‌سی‌پی MCP مدیریت‌شده دسترسی به بیگ‌کوئری

مفاهیم کلیدی تسلط یافته

✅ before_agent_callback : دریافت پیکربندی قبل از اجرای عامل
✅ {key} قالب‌بندی حالت : دسترسی به مقادیر حالت در دستورالعمل‌های عامل
✅ ToolContext : دسترسی به مقادیر حالت در توابع ابزار
✅ اشتراک‌گذاری وضعیت : وضعیت والد به طور خودکار از طریق InvocationContext برای زیرعامل‌ها در دسترس است
✅ معماری چندعاملی : عامل‌های تخصصی با مسئولیت‌های واحد
✅ ParallelAgent : اجرای همزمان وظایف مستقل
✅ سرور MCP سفارشی : سرور MCP خودتان در Cloud Run
✅ OneMCP BigQuery : الگوی MCP مدیریت‌شده برای دسترسی به پایگاه داده
✅ استقرار ابری : استقرار بدون تابعیت با استفاده از متغیرهای محیطی
✅ آمادگی A2A : نماینده آماده برای ارتباط بین نمایندگان

برای غیر گیمرها: کاربردهای دنیای واقعی

«مشخص کردن موقعیت مکانی شما» نشان‌دهنده‌ی تحلیل تخصصی موازی با اجماع است - اجرای همزمان چندین تحلیل تخصصی هوش مصنوعی و ترکیب نتایج.

برنامه‌های سازمانی

مورد استفاده کارشناسان موازی قانون سنتز
تشخیص پزشکی تحلیلگر تصویر، تحلیلگر علائم، تحلیلگر آزمایشگاه آستانه اطمینان ۲ از ۳
تشخیص کلاهبرداری تحلیلگر تراکنش، تحلیلگر رفتار، تحلیلگر شبکه هر ۱ پرچم = بررسی
پردازش اسناد عامل OCR، عامل طبقه‌بندی، عامل استخراج همه باید موافق باشند
کنترل کیفیت بازرس بصری، تحلیلگر حسگر، بررسی‌کننده مشخصات پاس ۲ از ۳

بینش‌های کلیدی معماری

  1. before_agent_callback برای پیکربندی : فایل پیکربندی را یک بار در ابتدا دریافت کنید، وضعیت را برای همه زیرعامل‌ها پر کنید. هیچ فایل پیکربندی در زیرعامل‌ها خوانده نمی‌شود.

  2. {key} قالب‌بندی حالت : اعلانی، تمیز، اصطلاحی. بدون f-string، بدون import، بدون دستکاری sys.path.

  3. مکانیسم‌های اجماع : توافق ۲ از ۳، ابهام را به طور قوی و بدون نیاز به توافق کامل، مدیریت می‌کند.

  4. ParallelAgent برای وظایف مستقل : وقتی تحلیل‌ها به یکدیگر وابسته نیستند، برای سرعت بیشتر، آنها را همزمان اجرا کنید.

  5. دو الگوی MCP : سفارشی (خودتان بسازید) در مقابل OneMCP (میزبانی شده توسط گوگل). هر دو از StreamableHTTP استفاده می‌کنند.

  6. استقرار بدون وضعیت : همان کد به صورت محلی و مستقر کار می‌کند. متغیرهای محیطی + API بک‌اند = بدون فایل‌های پیکربندی در کانتینرها.

قدم بعدی چیست؟

سطح ۲: پردازش سیگنال SOS →

یاد بگیرید که سیگنال‌های اضطراری دریافتی از سایر بازماندگان را با استفاده از الگوهای مبتنی بر رویداد و هماهنگی پیشرفته‌تر عامل‌ها پردازش کنید.