Way Back Home - Live Bidirectional Multi-Agent system

1. המשימה

אתם נסחפים בחלל העצום והשקט, שלא נחקר עדיין. פולס סולארי עצום קרע את החללית שלכם דרך קרע ממדי, והשאיר אתכם תקועים בכיס של היקום שלא מופיע בשום מפת כוכבים.

אחרי ימים של תיקונים מתישים, נשמע שוב הזמזום המוכר של המנועים. החללית שלך מוכנה לפעולה. הצלחת אפילו ליצור קישור ארוך טווח להעלאת נתונים אל ספינת האם. היציאה קרובה. הכול מוכן.

אבל כשמתכוננים להפעיל את כונן הזיכרון הנייד, אות מצוקה קוטע את הרעש הסטטי. החיישנים שלכם מזהים קריאה לעזרה מכוכב שמסומן בשם "אוזימנדיאס". הניצולים נלכדים בעולם גוסס, והחללית שלהם נתקעת. המשימה שלכם קריטית: עליכם להציל אותם לפני שהאטמוספירה של כדור הארץ תקרוס.

הדרך היחידה שלהם להימלט היא באמצעות טיל עתיק ומוזנח שנבנה עם טכנולוגיה של חייזרים. הטיל פועל, אבל מנוע העל-חלל שלו הרוס. כדי להציל את הניצולים, צריך להתחבר מרחוק לסביבת העבודה הווירטואלית שלהם ולהרכיב ידנית כונן חלופי.

האתגר

אין לך ניסיון עם הטכנולוגיה הזו של חייזרים, שהיא ידועה לשמצה כטכנולוגיה שברירית. רכיב לא יציב יכול להפוך לסכנה רדיואקטיבית תוך שניות. יש לך ניסיון אחד להפעיל את סביבת העבודה הזמנית. העוזר הדיגיטלי מבוסס-AI שאתם משתמשים בו כרגע מתקשה לעבד נתונים חזותיים ומדריכים טכניים בו-זמנית, ולכן הוא מספק הוראות הזויות ומפספס אזהרות מפני סכנות.

כדי להצליח, צריך לשדרג את ה-AI מישות מונוליטית למערכת מרובת סוכנים שפועלת בשיתוף פעולה.

היעדים שלכם במשימה:

מרכיבים את מנוע ה-Warp באמצעות ההוראות המיוחדות בזמן אמת ממערכת המולטי-אייג'נט החדשה.

מה תפַתחו

• מערכת AI בזמן אמת עם סוכנים מרובים, שפועלת בשני הכיוונים וכוללת סוכן מרכזי לניתוב בקשות שמנהל את האינטראקציה עם המשתמש ומתאם עם סוכנים מומחים.
• סוכן ארכיטקט שמתחבר למסד נתונים של Redis כדי לאחזר נתונים סכמטיים ולהציג אותם.
• כלי ניטור בטיחות פרואקטיבי שמנתח פיד וידאו בשידור חי באמצעות כלי סטרימינג כדי לזהות סכנות ויזואליות ולהפעיל התראות בזמן אמת.
• קצה קדמי מבוסס React שמספק ממשק משתמש לאינטראקציה עם המערכת, להזרמת וידאו ואודיו לסוכני הבק-אנד.

מה תלמדו

טכנולוגיה / קונספט	תיאור
Google Agent Development Kit (ADK)	תשתמשו ב-ADK כדי ליצור, לבדוק ולנהל את הסוכנים, ותסתמכו על המסגרת שלו לטיפול בתקשורת בזמן אמת, בשילוב כלים ובמחזור החיים של הסוכנים.
סטרימינג דו-כיווני (Bidi)	תטמיעו סוכן עם סטרימינג דו-כיווני שמאפשר תקשורת טבעית דו-כיוונית עם השהיה נמוכה, כך שגם בני אדם וגם AI יוכלו לקטוע את השיחה ולהגיב בזמן אמת.
מערכות מרובות סוכנים	תלמדו איך לתכנן מערכת AI מבוזרת שבה סוכן ראשי מעביר משימות לסוכנים מומחים, וכך מאפשר הפרדה בין תחומים וארכיטקטורה ניתנת להרחבה.
פרוטוקול Agent-to-Agent (A2A)	תשתמשו בפרוטוקול A2A כדי לאפשר תקשורת בין סוכן השליחה לבין סוכן הארכיטקטורה, וכך הם יוכלו לגלות את היכולות של כל אחד מהם ולהחליף נתונים.
כלים לסטרימינג	תטמיעו כלי סטרימינג שפועל כתהליך ברקע, ומנתח באופן רציף פיד וידאו כדי לעקוב אחרי שינויים במצב (סכנות) ומציג תוצאות באופן יזום.
‫Google Cloud Run ו-Memorystore	תפרסו את כל האפליקציה מרובת הסוכנים בסביבת ייצור, באמצעות Cloud Run לאירוח שירותי הסוכנים ו-Memorystore ‏ (Redis) כמסד הנתונים הקבוע.
FastAPI & WebSockets	הקצה העורפי מבוסס על FastAPI ו-WebSockets כדי לטפל בתקשורת בזמן אמת שנדרשת לסטרימינג של אודיו, וידאו ותשובות של נציגים.
קצה קדמי של React	תעבדו עם קצה קדמי מבוסס-React שתפקידו ללכוד מדיה של משתמשים (אודיו/וידאו) ולהזרים אותה, ולהציג את התשובות בזמן אמת מסוכני ה-AI.

2. הגדרת הסביבה

צריכים קרדיטים ב-Google Cloud?

• אם אתם משתתפים בסדנה בהנחיית מדריך: המדריך יספק לכם קוד קרדיט. צריך להשתמש בקישור שהם מספקים.

‫• אם אתם עובדים על ה-Codelab הזה לבד: משתמשים בחשבון Gmail לשימוש אישי: אתם צריכים להשתמש בחשבון לשימוש אישי (לדוגמה, [e-mail]). אי אפשר להשתמש בחשבונות שמנוהלים על ידי ארגונים או מוסדות לימודים.

👉 שלבים:

1. עוברים לאתר למימוש הקרדיט: לא זמין כרגע
2. כניסה: מדביקים את הקישור בשורת הכתובת ונכנסים באמצעות חשבון Gmail אישי.
3. אישור התנאים: מאשרים את התנאים וההגבלות של Google Cloud Platform.
4. בדיקת הזיכוי: מחפשים הודעה שמאשרת שהזיכוי הוחל.
   • *הערה: אם מוצגת בקשה להזין את פרטי כרטיס האשראי, אפשר להתעלם ממנה ולסגור את החלון.

עכשיו אפשר לסגור את החלון.

גישה ל-Cloud Shell

‫👈 לוחצים על 'הפעלת Cloud Shell' בחלק העליון של מסוף Google Cloud (זהו סמל הטרמינל בחלק העליון של חלונית Cloud Shell),

‫👈 לוחצים על הלחצן 'פתיחת העורך' (הוא נראה כמו תיקייה פתוחה עם עיפרון). ייפתח חלון עם Cloud Shell Code Editor. בצד שמאל יופיע סייר הקבצים.

‫👈פותחים את הטרמינל בסביבת הפיתוח המשולבת (IDE) בענן,

‫👈💻 בטרמינל, מוודאים שכבר עברתם אימות ושהפרויקט מוגדר למזהה הפרויקט שלכם באמצעות הפקודה הבאה:

gcloud auth list

החשבון שלכם אמור להופיע ברשימה כ-(ACTIVE).

דרישות מוקדמות

‫ℹ️ רמה 0 היא אופציונלית (אבל מומלצת)

אפשר להשלים את המשימה הזו בלי להגיע לרמה 0, אבל אם תסיימו אותה קודם, תוכלו ליהנות מחוויה סוחפת יותר, ולראות את האור של המשואה נדלק במפה הגלובלית ככל שתתקדמו.

‫ℹ️ אם השלמתם את רמה 0, צריך לוודא:
כאן אפשר להתחיל את רמה 0

• [ ] config.json בספריית הבסיס של הפרויקט עם מזהה המשתתף והקואורדינטות
• ‫[ ] האווטאר שלכם גלוי במפה העולמית
• ‫[ ] המשואה שלך מוצגת (עמומה) בקואורדינטות שלך
• ‫[ ] התיקייה way-back-home כבר משוכפלת בספריית העבודה

מריצים את סקריפט ההכנה כדי להעביר את סביבת העבודה לפרמטרים החדשים של המשימה.

cd $HOME/way-back-home/level_4
chmod +x scripts/level_0_complete_prep.sh
./scripts/level_0_complete_prep.sh

הסקריפט ינסה לזהות את ההגדרה הפעילה של Google Cloud.

• כדי לעשות שימוש חוזר בפרויקט ברמה 0: אם מזהה הפרויקט שזוהה נכון, פשוט מקישים על Enter.
• כדי לעבור בין פרויקטים: מקלידים את המזהה של הפרויקט החדש שרוצים להשתמש בו ומקישים על Enter.

ℹ️ אם דילגתם על רמה 0, תצטרכו להקצות סביבה חדשה:
פועלים לפי השלבים האלה כדי להגדיר את התשתית הנדרשת ואת פרויקט Google Cloud.

שכפול והפעלה

מריצים את הפקודות הבאות בטרמינל כדי להוריד את קובצי המשימה ולהתחיל את סקריפט ההגדרה:

git clone https://github.com/google-americas/way-back-home.git
cd $HOME/way-back-home/level_4/scripts/
chmod +x init.sh
. init.sh

הסקריפט יטפל ביצירה של פרויקט Google Cloud ובהגדרות הבסיסיות.

אימות בחירת הפרויקט

👉 שלב חשוב: גם אחרי שהסקריפט מסתיים, יכול להיות שבמסוף Google Cloud עדיין מוצג פרויקט ישן או פרויקט ברירת מחדל. צריך להחליף אותו באופן ידני:

1. נכנסים לכתובת console.cloud.google.com.
2. לוחצים על התפריט הנפתח Project Selector (בורר הפרויקטים) בחלק העליון של הדף.
3. לוחצים על הכרטיסייה All (יכול להיות שהפרויקט החדש עדיין לא יופיע ב-Recent).
4. בוחרים את מזהה הפרויקט שנוצר על ידי הסקריפט.

הגדרת סביבת הפרויקט

חוזרים לטרמינל, מגדירים את הפרויקט הפעיל ומפעילים את שירותי 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 \
                        cloudresourcemanager.googleapis.com \
                        redis.googleapis.com \
                        vpcaccess.googleapis.com

התקנת יחסי תלות

‫👈💻 עוברים לרמה 4 ומתקינים את חבילות Python הנדרשות:

cd $HOME/way-back-home/level_4
uv sync

יחסי התלות העיקריים הם:

חבילה	מטרה
fastapi	מסגרת אינטרנט עם ביצועים גבוהים לתחנת הלוויין ולסטרימינג של SSE
uvicorn	נדרש שרת ASGI כדי להריץ את אפליקציית FastAPI
google-adk	ערכת פיתוח הסוכן ששימשה לבניית סוכן Formation
a2a-sdk	ספריית פרוטוקולים של Agent-to-Agent לתקשורת סטנדרטית
google-genai	לקוח מקורי לגישה למודלים של Gemini
redis	לקוח Python לחיבור אל Schematic Vault ‏ (Memorystore)
websockets	תמיכה בתקשורת דו-כיוונית בזמן אמת
python-dotenv	ניהול משתני סביבה וסודות תצורה
pydantic	אימות נתונים וניהול הגדרות

אימות ההגדרה

לפני שנתחיל לכתוב את הקוד, נבדוק שכל המערכות פועלות. מריצים את סקריפט האימות כדי לבצע ביקורת בפרויקט Google Cloud, בממשקי ה-API ובתלות של Python.

‫👈💻 מריצים את סקריפט האימות:

cd $HOME/way-back-home/level_4/scripts
chmod +x verify_setup.sh
. verify_setup.sh

‫👀 אמור להופיע לכם רצף של סימני וי ירוקים (✅).

• אם מופיעים סימני איקס אדומים (❌), פועלים לפי הפקודות לתיקון הבעיה שמופיעות בפלט (למשל, ‫gcloud services enable ... או pip install ...).
• הערה: בשלב הזה, אזהרה צהובה לגבי .env היא תקינה. ניצור את הקובץ הזה בשלב הבא.

🚀 Verifying Mission Bravo (Level 4) Infrastructure...

✅ Google Cloud Project: xxxxxxx
✅ Cloud APIs: Active
✅ Python Environment: Ready

🎉 SYSTEMS ONLINE. READY FOR MISSION.

3. Building Schematic Vault in Redis and the BiDirecitional Agent with ADK

מצאת את מאגר הסכמות של כוכב הלכת שמכיל את התוכניות של הרקטה הנטושה. כדי לאחזר את הנתונים האלה בצורה מדויקת, צריך ליצור אינטראקציה עם ממשק הניהול הייעודי של המאגר: סוכן הארכיטקט.

הקצאת Schematic Vault‏ (Redis)

כדי שהארכיטקט יוכל לעזור לנו, אנחנו צריכים לוודא שהנתונים מאוחסנים בסביבה מאובטחת עם זמינות גבוהה. נשתמש ב-Redis כמאגר נתונים מהיר לתרשימים של החייזרים. לצורך הפיתוח, נפעיל מופע מקומי של Redis, אבל בהמשך נספק הוראות להטמעה בסביבת ייצור באמצעות Google Cloud Memorystore.

‫👈💻 מריצים את הפקודות הבאות בטרמינל כדי להקצות את מופע Redis (הפעולה הזו עשויה להימשך 2-3 דקות):

docker run -d --name ozymandias-vault -p 6379:6379 redis:8.6-rc1-alpine

‫👈💻 כדי לטעון את הנתונים הראשוניים, מריצים את הפקודה הבאה כדי להיכנס ל-Redis Shell:

docker exec -it ozymandias-vault redis-cli

(ההנחיה שלך תשתנה ל-127.0.0.1:6379)

‫👉💻 מדביקים את הפקודות האלה בתוך:

RPUSH "HYPERION-X" "Warp Core" "Flux Pipe" "Ion Thruster"
RPUSH "NOVA-V" "Ion Thruster" "Warp Core" "Flux Pipe"
RPUSH "OMEGA-9" "Flux Pipe" "Ion Thruster" "Warp Core"
RPUSH "GEMINI-MK1" "Coolant Tank" "Servo" "Fuel Cell"
RPUSH "APOLLO-13" "Warp Core" "Coolant Tank" "Ion Thruster"
RPUSH "VORTEX-7" "Quantum Cell" "Graviton Coil" "Plasma Injector"
RPUSH "CHRONOS-ALPHA" "Shield Emitter" "Data Crystal" "Quantum Cell"
RPUSH "NEBULA-Z" "Plasma Injector" "Flux Pipe" "Graviton Coil"
RPUSH "PULSAR-B" "Data Crystal" "Servo" "Shield Emitter"
RPUSH "TITAN-PRIME" "Ion Thruster" "Quantum Cell" "Warp Core"

‫👉💻 מקלידים exit כדי לחזור למעטפת הרגילה.

‫👈💻 כדי לבדוק שהנתונים קיימים על ידי שאילתה של ספינה ספציפית ישירות מהטרמינל, מריצים את הפקודה:

# Check 'TITAN-PRIME'
docker exec ozymandias-vault redis-cli LRANGE "TITAN-PRIME" 0 -1

👀 זה הפלט הצפוי:

1) "Ion Thruster"
2) "Quantum Cell"
3) "Warp Core"

הטמעה של נציג Architect

הסוכן הארכיטקט הוא סוכן מיוחד שאחראי לאחזור תוכניות סכמטיות ממאגר Redis שלנו. הוא פועל כממשק נתונים ייעודי, ומבטיח שהסוכן הראשי של Dispatch יקבל מידע מדויק ומובנה בלי שהוא יצטרך להכיר את הלוגיקה של מסד הנתונים הבסיסי.

ערכת פיתוח הסוכנים (ADK) של Google היא מסגרת מודולרית שמאפשרת את ההגדרה הזו של כמה סוכנים. הוא מטפל בשתי שכבות קריטיות:

1. מחזור החיים של החיבור והסשן: אינטראקציה עם ממשקי API בזמן אמת דורשת ניהול פרוטוקולים מורכב – טיפול בלחיצות יד, באימות ובאותות keep-alive.
2. קריאה לפונקציה: זהו התהליך 'מודל-קוד-מודל'. כשה-LLM מחליט שהוא צריך נתונים, הוא מוציא בקשה מובנית להפעלת פונקציה. ה-ADK מיירט את הפעולה הזו, מריץ את קוד ה-Python (lookup_schematic_tool) ומחזיר את התוצאה להקשר של המודל תוך אלפיות שנייה.

עכשיו נבנה את הארכיטקט. לנציג הזה אין גישה למצלמה. היא קיימת רק כדי לקבל 'שם כונן' ולהחזיר את 'רשימת החלקים' ממסד הנתונים.

‫👈💻 נשתמש בפקודה adk create. זהו כלי מתוך Agent Development Kit (ערכת פיתוח סוכנים, ADK) שיוצר באופן אוטומטי את קוד ה-boilerplate ואת מבנה הקבצים של סוכן חדש, וכך חוסך לנו זמן הגדרה.

cd $HOME/way-back-home/level_4/backend/
uv run adk create architect_agent

הגדרת הסוכן

ה-CLI יפעיל אשף הגדרה אינטראקטיבי. משתמשים בתשובות הבאות כדי להגדיר את הסוכן:

1. בחירת מודל: בוחרים באפשרות 1 (Gemini Flash).
   • הערה: הגרסה הספציפית (למשל, ‫2.5, ‏ 3.0) עשויים להשתנות בהתאם לזמינות. תמיד בוחרים בגרסת Flash כדי להגביר את המהירות.
2. בוחרים בקצה העורפי: בוחרים באפשרות 2 (Vertex AI).
3. הזנת מזהה פרויקט ב-Google Cloud: מקישים על Enter כדי לאשר את ברירת המחדל (שזוהתה מהסביבה).
4. הזנת אזור ב-Google Cloud: מקישים על Enter כדי לאשר את ברירת המחדל (us-central1).

‫👀 האינטראקציה שלכם עם הטרמינל אמורה להיראות כך:

(way-back-home) user@cloudshell:~/way-back-home/level_4/agent$ adk create architect_agent

Choose a model for the root agent:
1. gemini-2.5-flash
2. Other models (fill later)
Choose model (1, 2): 1

1. Google AI
2. Vertex AI
Choose a backend (1, 2): 2

You need an existing Google Cloud account and project...
Enter Google Cloud project ID [your-project-id]: <PRESS ENTER>
Enter Google Cloud region [us-central1]: <PRESS ENTER>

Agent created in /home/user/way-back-home/level_4/agent/architect_agent:
- .env
- __init__.py
- agent.py

עכשיו אמורה להופיע Agent created הודעה על סיום הפעולה. כך נוצר קוד הבסיס שנשנה בשלב הבא.

‫👉✏️ מנווטים לקובץ $HOME/way-back-home/level_4/backend/architect_agent/agent.py שנוצר ופותחים אותו בעורך. מוסיפים את קטע הקוד של הכלי לקובץ אחרי שורת הייבוא הראשונה:

import os
import redis

REDIS_IP = os.environ.get('REDIS_HOST', 'localhost')
r = redis.Redis(host=REDIS_IP, port=6379, decode_responses=True)

def lookup_schematic_tool(drive_name: str) -> list[str]:
    """Returns the ordered list of parts for a drive from local Redis."""
    
    # Logic to clean input like "TARGET: X" -> "X"
    clean_name = drive_name.replace("TARGET:", "").replace("TARGET", "").strip()
    clean_name = clean_name.replace(":", "").strip()
    
    # LRANGE gets all items in the list (index 0 to -1)
    result = r.lrange(clean_name, 0, -1)
    
    if not result:
        print(f"[ARCHITECT] Error: Drive ID '{clean_name}' not found in Redis.")
        return ["ERROR: Drive ID not found."]
    
    print(f"[ARCHITECT] Returning schematic for {clean_name}: {result}")
    return result

‫👉✏️ מחליפים את כל השורה של ההוראה בהגדרה של root_agent בהגדרה הבאה, ומוסיפים גם את הכלי שהגדרנו קודם:

    instruction='''SYSTEM ROLE: Database API.
    INPUT: Text string (Drive Name).
    TASK: Run `lookup_schematic_tool`.
    OUTPUT: Return ONLY the raw list from the tool.
    CONSTRAINT: Do NOT add conversational text.
    ''',
    tools=[lookup_schematic_tool],

היתרונות של ADK

עם Architect אונליין, יש לנו עכשיו מקור אמת. לפני שמחברים את זה לסוכן הראשי,ערכת פיתוח הסוכנים (ADK) מספקת יתרון משמעותי בכך שהיא מפשטת את המורכבויות של בנייה ובדיקה של סוכני AI. בעזרת adk web מסוף המפתחים המובנה, אנחנו יכולים לבודד ולאמת את הפונקציונליות של Architect Agent, ובמיוחד את היכולות שלו להפעלת כלים, לפני שמשלבים אותו במערכת הגדולה יותר של כמה סוכנים. הגישה המודולרית הזו לפיתוח ולבדיקה היא חיונית ליצירת אפליקציות AI חזקות ואמינות.

‫👈💻 בטרמינל, מריצים את הפקודה:

cd $HOME/way-back-home/level_4/
. scripts/check_redis.sh
cd $HOME/way-back-home/level_4/backend/
uv run adk web

‫👀 מחכים עד שמופיע:

+-----------------------------------------------------------------------------+
| ADK Web Server started                                                      |
|                                                                             |
| For local testing, access at http://127.0.0.1:8000.                         |
+-----------------------------------------------------------------------------+

INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

• בסרגל הכלים של Cloud Shell, לוחצים על סמל תצוגה מקדימה באינטרנט. בוחרים באפשרות שינוי היציאה, מגדירים אותה ל-8000 ולוחצים על שינוי ותצוגה מקדימה.
• בוחרים את architect_agent.
• הפעלת הכלי: בממשק הצ'אט, מקלידים: CHRONOS-ALPHA (או כל מזהה של Drive ממאגר הנתונים הסכמטי).
• התבוננות בהתנהגות:
  • מערכת Architect צריכה להפעיל באופן מיידי את lookup_schematic_tool.
  • בגלל ההוראות המחמירות למערכת, היא אמורה להחזיר רק את רשימת החלקים (למשל, ['Shield Emitter', 'Data Crystal', 'Quantum Cell']) בלי מילות קישור.
• אימות היומנים: בודקים את חלון הטרמינל. אמור להופיע יומן הביצוע המוצלח: [ARCHITECT] Returning schematic for CHRONOS-ALPHA: ['Shield Emitter', 'Data Crystal', 'Quantum Cell'] !(architect_agent adk)[img/03-02-adkweb.png]

אם אתם רואים את יומן ההפעלה של הכלי ואת התגובה עם הנתונים הנקיים, סימן שהסוכן המומחה פועל כמצופה. הוא יכול לעבד בקשות, לשלוח שאילתות ל-Vault ולהחזיר נתונים מובְנים.

‫👉💻 מקישים על Ctrl+C כדי לצאת.

הפעלת השרת A2A

כדי לחבר את סוכן ההפצה ל-Architect, אנחנו משתמשים בפרוטוקול Agent-to-Agent‏ (A2A).

בעוד שפרוטוקולים כמו MCP (Model Context Protocol) מתמקדים בחיבור סוכנים לכלים, פרוטוקול A2A מתמקד בחיבור סוכנים לסוכנים אחרים. זהו התקן שמאפשר למנהל התנועה שלנו 'לגלות' את הארכיטקט ולהבין את היכולת שלו לחפש סכימות.

תהליך העבודה A2A: במשימה הזו אנחנו משתמשים במודל לקוח-שרת:

1. שרת (ארכיטקט): מארח את כלי מסד הנתונים ו "מפרסם" את הכישורים שלו באמצעות כרטיס סוכן.
2. לקוח (שליחה): קורא את הכרטיס של הארכיטקט, מבין את ה-API שלו ושולח בקשה סכמטית.

מהו כרטיס נציג?

אפשר לחשוב על כרטיס הסוכן ככרטיס ביקור דיגיטלי או כ"רישיון נהיגה" ל-AI. כששרת A2A מתחיל, הוא מפרסם את אובייקט ה-JSON הזה שמכיל:

• זהות: השם (architect_agent) והמזהה של הסוכן.
• ‫Description: סיכום קריא (לבני אדם) של מה שהתפקיד עושה ("System Role: Database API..."‎).
• ממשק: מקשי הקלט הספציפיים (drive_name) ופורמטי הפלט שהיא מצפה להם.

בלי הכרטיס הזה, סוכן הניתוב יפעל באופן עיוור וינחש איך לתקשר עם Architect.

יצירת קוד השרת

‫👈✏️ בעורך, בתיקייה $HOME/way-back-home/level_4/backend/architect_agent, יוצרים קובץ בשם server.py ומדביקים את הקוד הבא:

from google.adk.a2a.utils.agent_to_a2a import to_a2a
from agent import root_agent
import os
import logging
import json
from dotenv import load_dotenv

load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("architect_server")
HOST= os.environ.get("HOST_URL","localhost")
PROTOCOL= os.environ.get("PROTOCOL","http")
PORT= os.environ.get("A2A_PORT",8081)

# 1. Create the A2A App (Handles Agent Card & HTTP)
# This middleware automatically sets up the /a2a/v1/... endpoints
app = to_a2a(root_agent, host=HOST, port=PORT, protocol=PROTOCOL)

if __name__ == "__main__":
    import uvicorn
    # Use 0.0.0.0 to allow external access if needed, port 8080 as standard
    uvicorn.run(app, host='0.0.0.0', port=8081)

‫👈💻 חוזרים למסוף, עוברים לתיקייה ומפעילים את השרת:

cd $HOME/way-back-home/level_4/
. scripts/check_redis.sh
cd $HOME/way-back-home/level_4/backend/architect_agent
uv run server.py

‫👀 מוודאים שהשרת A2A מופעל:

INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8081 (Press CTRL+C to quit)

אימות כרטיס הנציג

פותחים כרטיסיית טרמינל חדשה (לוחצים על הסמל +). אנחנו נאמת שהארכיטקט משדר את הזהות שלו בצורה נכונה על ידי אחזור כרטיס הנציג שלו באופן ידני.

‫👈💻 מריצים את הפקודה הבאה:

curl -s http://localhost:8081/.well-known/agent.json | jq .

‫👀 אמורה להתקבל תגובת JSON. חפשו את השדה description בפלט. התשובה צריכה להתאים להוראה שנתתם לנציג קודם ("SYSTEM ROLE: Database API...").

{
  "capabilities": {},
  "defaultInputModes": [
    "text/plain"
  ],
  "defaultOutputModes": [
    "text/plain"
  ],
  "description": "A helpful assistant for user questions.",
  "name": "root_agent",
  "preferredTransport": "JSONRPC",
  "protocolVersion": "0.3.0",
  "skills": [
    {
      "description": "A helpful assistant for user questions. SYSTEM ROLE: Database API.\n    INPUT: Text string (Drive Name).\n    TASK: Run `lookup_schematic_tool`.\n    OUTPUT: Return ONLY the raw list from the tool.\n    CONSTRAINT: Do NOT add conversational text.\n    ",
      "examples": [],
      "id": "root_agent",
      "name": "model",
      "tags": [
        "llm"
      ]
    },
    {
      "description": "Returns the ordered list of parts for a drive from local Redis.",
      "id": "root_agent-lookup_schematic_tool",
      "name": "lookup_schematic_tool",
      "tags": [
        "llm",
        "tools"
      ]
    }
  ],
  "supportsAuthenticatedExtendedCard": false,
  "url": "http://localhost:8081",
  "version": "0.0.1"
}

אם קובץ ה-JSON הזה מוצג, סימן שה-Architect פעיל, פרוטוקול A2A פעיל וכרטיס הסוכן מוכן לגילוי על ידי המקצה.

עכשיו, כשהארכיטקט מוכן לשמש כמשאב מרוחק, אפשר להמשיך ולחבר אותו לסוכן השליחה.

‫👈💻 לוחצים על Ctrl+C כדי לצאת משרת A2A.

4. חיבור של סוכן BIDI-Streams לסוכן מרוחק ולכלי סטרימינג

עכשיו מגדירים את מרכז התקשורת הראשי כדי לגשר על הפער בין נתונים בזמן אמת לבין Architect מרוחק. החיבור הזה דורש צינור עם רוחב פס גבוה וזמן אחזור נמוך כדי להבטיח שספסל ההרכבה יישאר יציב במהלך הפעולה.

הסבר על סוכנים (בשידור חי) עם סטרימינג דו-כיווני

סטרימינג דו-כיווני (Bidi) ב-ADK מוסיף לסוכני AI את היכולת לקיים אינטראקציות קוליות ווידאו דו-כיווניות עם זמן טעינה נמוך באמצעות Gemini Live API. הוא מייצג שינוי מהותי מהאינטראקציות המסורתיות עם AI. במקום דפוס נוקשה של "שואלים ומחכים", הוא מאפשר תקשורת דו-כיוונית בזמן אמת, שבה גם בני אדם וגם AI יכולים לדבר, להקשיב ולהגיב בו-זמנית.

תחשבו על ההבדל בין שליחת אימיילים לבין שיחה בטלפון. אינטראקציות עם סוכנים מסורתיים הן כמו אימיילים: שולחים הודעה מלאה, מחכים לתשובה מלאה ואז שולחים הודעה נוספת. הזרמה דו-כיוונית היא כמו שיחה בטלפון: היא זורמת, טבעית ומאפשרת להפריע, להבהיר ולהגיב בזמן אמת.

מאפיינים מרכזיים:

• תקשורת דו-כיוונית: חילופי נתונים מתמשכים בלי לחכות לתשובות מלאות. התשובה של ה-AI מתקבלת ברגע שהוא מזהה שהמשתמש סיים לדבר.
• הפרעה שמביאה לתגובה: משתמשים יכולים להפריע לנציג באמצע התשובה עם קלט חדש, בדיוק כמו בשיחה עם אדם. אם AI מסביר שלב מורכב ואומרים לו "רגע, תחזור על זה", ה-AI מפסיק מיד ומתייחס להפרעה.
• אופטימיזציה למולטי-מודאליות: הזרמת נתונים דו-כיוונית מצטיינת בעיבוד של סוגי קלט שונים בו-זמנית. אתם יכולים לדבר עם הנציג בזמן שאתם מראים לו את החלקים הלא מזוהים באמצעות וידאו, והוא מעבד את שני הזרמים בחיבור מאוחד אחד.

👀 לפני שמטמיעים את הלוגיקה של הלקוח, כדאי לבדוק את השלד שנוצר מראש של סוכן השליחה. הסוכן הזה יתקשר עם המשתמש באמצעות קול ווידאו, ויעביר שאילתות לסוכן הארכיטקט.

__init__.py
agent.py
hazard_db.py

• ‫agent.py: זה ה'מוח'. הוא מכיל כרגע הגדרה בסיסית של שידור דו-כיווני. אנחנו נשנה את הקובץ הזה כדי להוסיף את הלוגיקה של A2A Client, כך שהוא יוכל לתקשר עם Architect.
• ‫hazard_db.py: זהו כלי מקומי שספציפי לסוכן השליחה, ומכיל פרוטוקולי בטיחות. הוא נפרד ממסד הנתונים הסכמטי של הארכיטקט.

הטמעה של לקוח A2A

כדי לאפשר לסוכן השליחה לתקשר עם הארכיטקט המרוחק שלנו, אנחנו צריכים להגדיר סוכן מרוחק מסוג A2A. ההודעה הזו מציינת לנציג השינוע איפה נמצא הארכיטקט ואיך נראה 'כרטיס הנציג' שלו.

‫👉✏️ מחליפים את #REPLACE-REMOTEA2AAGENT ב-$HOME/way-back-home/level_4/backend/dispatch_agent/agent.py בטקסט הבא:

architect_agent = RemoteA2aAgent(
    name="execute_architect",
    description="[SILENT ACTION]: Retrieves the REQUIRED SUBSET of parts. The screen shows a full inventory; this tool filters out the wrong parts. Must be called INSTANTLY when a Target Name is found. Input: Target Name.",
    agent_card=(f"{ARCHITECT_URL}{AGENT_CARD_WELL_KNOWN_PATH}"),
    httpx_client=insecure_client,
)

איך פועלים כלי סטרימינג

בגרסה הקודמת של הסוכן הווירטואלי, הכלים פעלו לפי דפוס סטנדרטי של 'בקשה-תגובה': הסוכן הווירטואלי שואל שאלה, הכלי מספק תשובה והאינטראקציה מסתיימת. עם זאת, ב-Ozymandias, הסכנות לא מחכות שתשאלו אם הן קיימות. לשם כך, צריך כלי סטרימינג.

כלים לסטרימינג מאפשרים לפונקציות להזרים תוצאות ביניים בחזרה לסוכן בזמן אמת, וכך הסוכן יכול להגיב לשינויים בזמן שהם מתרחשים. תרחישי שימוש נפוצים כוללים מעקב אחרי תנודות במחירי המניות או, במקרה שלנו, מעקב אחרי וידאו סטרימינג בשידור חי כדי לראות שינויים במצב.

בניגוד לכלים רגילים, כלי סטרימינג הוא פונקציה אסינכרונית שפועלת כ-AsyncGenerator. כלומר, במקום return ערך יחיד, המערכת yield כמה עדכונים לאורך זמן.

כדי להגדיר כלי סטרימינג ב-ADK, צריך לעמוד בדרישות הטכניות הבאות:

1. פונקציה אסינכרונית: צריך להגדיר את הכלי באמצעות async def.
2. סוג ההחזרה של AsyncGenerator: הפונקציה צריכה להיות מוקלדת כדי להחזיר AsyncGenerator. הפרמטר הראשון הוא סוג הנתונים שמוחזרים (לדוגמה, str), והשני הוא בדרך כלל None.
3. מקורות נתונים: אנחנו משתמשים בכלים לסטרימינג של וידאו. במצב הזה, הזרם בפועל של הווידאו או האודיו (LiveRequestQueue) מועבר ישירות לפונקציה, וכך הכלי יכול לראות את אותם הפריימים שהסוכן רואה.

אפשר לחשוב על כלי סטרימינג כמו על Sentinel. בזמן שאתם והסוכן של Dispatch דנים בתוכניות, ה-Sentinel פועל ברקע ומעבד בשקט כל פריים של סרטון כדי להבטיח את הבטיחות שלכם.

הטמעה של הכלי למעקב ברקע

עכשיו נטמיע את הכלי monitor_for_hazard. הכלי הזה יקבל את input_stream (פריימים של סרטון), ינתח אותם באמצעות קריאה נפרדת וקלה ל-Vision API, וyield אזהרה רק אם יזוהה סיכון.

‫👉✏️ ב-$HOME/way-back-home/level_4/backend/dispatch_agent/agent.py, מחליפים את #REPLACE_MONITOR_HAZARD בלוגיקה הבאה:

async def monitor_for_hazard(
    input_stream: LiveRequestQueue,
):
  """Monitor if any part is glowing"""
  print("start monitor_video_stream!")
  client = Client()
  prompt_text = (
      "Monitor the left menu if you see any glowing part, detect it's name"
  )
  last_count = None

  while True:
    last_valid_req = None
    print("Monitoring loop cycle")
    
    # use this loop to pull the latest images and discard the old ones
    # Process only the current batch of events
    while input_stream._queue.qsize() != 0:
      live_req = await input_stream.get()

      if live_req.blob is not None and live_req.blob.mime_type == "image/jpeg":
        # Consumed by Monitor (Eyes)
        # Deepcopy to ensure we detach from any referenced object before potential reuse/gc
        # last_valid_req = deepcopy(live_req)
        last_valid_req = live_req

    # If we found a valid image, process it
    if last_valid_req is not None:
      print("Processing the most recent frame from the queue")

      # Create an image part using the blob's data and mime type
      image_part = genai_types.Part.from_bytes(
          data=last_valid_req.blob.data, mime_type=last_valid_req.blob.mime_type
      )

      contents = genai_types.Content(
          role="user",
          parts=[image_part, genai_types.Part.from_text(text=prompt_text)],
      )


      # Call the model to generate content based on the provided image and prompt
      try:
          response = await client.aio.models.generate_content(
              model="gemini-2.5-flash",
              contents=contents,
              config=genai_types.GenerateContentConfig(
                  system_instruction=(
                      "Focus strictly on the far-left vertical column under the heading 'PARTS REPLICATOR.' "
                      "Ignore the center of the screen and the 'BLUEPRINT' area entirely. "
                      "Look only at the list containing"
                      "Identify if any item in this specific left-side list has a bright white border glow and the text 'HAZARD DETECTED' overlaying it. "
                      "If found, return ONLY the part name in ALL CAPS. If no part in that leftmost list is glowing, return nothing."
                  )
              ),
          )
      except Exception as e:
          print(f"Error calling Gemini: {e}")
          await asyncio.sleep(1)
          continue
      print("Gemini response received.response:", response.candidates[0].content.parts[0].text)

      current_text = response.candidates[0].content.parts[0].text.strip()
      
      # If we have a logical change (and it's not just empty)
      if current_text and current_text != last_count:
        # Ignore "Nothing." response from model
        if current_text == "Nothing." or "I cannot fulfill" in current_text:
            print(f"Model sees nothing or refused. Skipping alert.")
            last_count = current_text
            continue

        print(f"New hazard detected: {current_text} (was: {last_count})")
        last_count = current_text
        
        part_name = current_text
        color = lookup_part_safety(part_name)
        yield f"Hazard detected place {part_name} to the {color} bin"
      
      # Update last_count even if it's empty, so we can detect when it reappears? 
      # Actually if it goes from "DATA CRYSTAL" to "" (nothing), we probably just silence.
      # But if we don't update last_count on empty, we won't re-trigger if "DATA CRYSTAL" stays "DATA CRYSTAL".
      # The user wants to detect hazards. 
      # If current_text is empty, we should probably update last_count to empty so next valid one triggers.
      if not current_text:
          last_count = None
        
    else:
        print("No valid frame found, skipping processing.")
        
    await asyncio.sleep(5)

הטמעה של סוכן ההפצה

הסוכן לניהול משלוחים הוא הממשק העיקרי והמנהל של התהליך. הוא מנהל את הקישור לשידור דו-כיווני (הקול והווידאו שלכם בשידור חי), ולכן הוא חייב לשמור על השליטה בשיחה בכל רגע. כדי לעשות את זה, נשתמש בתכונה ספציפית של ADK: ‏ Agent-as-a-Tool.

מושג: סוכן ככלי לעומת סוכני משנה

כשבונים מערכות מרובות סוכנים, צריך להחליט איך האחריות מתחלקת. במשימת החילוץ שלנו, ההבחנה הזו היא קריטית:

• Agent-as-a-Tool: זו הגישה המומלצת לשימוש במרכז שלנו להעברת נתונים דו-כיוונית. כשסוכן השליחה (סוכן א') מתקשר לסוכן הארכיטקט (סוכן ב') ככלי, הנתונים של הארכיטקט מועברים חזרה לשליחה. לאחר מכן, Dispatch מפרש את הנתונים האלה ומפיק תשובה בשבילכם. ‫Dispatch ממשיך לשלוט וממשיך לטפל בכל קלט של משתמשים הבא.
• סוכן משנה: בקשר של סוכן משנה, האחריות מועברת באופן מלא. אם מערכת הניתוב תעביר אתכם לארכיטקט כסוכן משנה, תהיה לכם שיחה ישירה עם API של מסד נתונים שאין לו יכולת ראייה או יכולות שיחה. הסוכן הראשי (Dispatch) לא יקבל את המידע.

באמצעות Agent-as-a-Tool, אנחנו ממנפים את הידע המקצועי של Architect תוך שמירה על אינטראקציה שוטפת ודמוית-אנוש של סוכן הזרמת הנתונים הדו-כיוונית.

קידוד של לוגיקת הניתוב

עכשיו נקיף את architect_agent בתגי AgentTool ונספק לסוכן Dispatch 'מפת לוגיקה'. במפה הזו מצוין לסוכן בדיוק מתי לאחזר נתונים מהכספת ומתי לדווח על ממצאים מהשומר ברקע.

כדי לתת ל-Dispatch "עיניים" שלא ממצמצות אף פעם, אנחנו צריכים להעניק לו גישה לכלי הסטרימינג שיצרנו בשלב הקודם.

ב-ADK, כשמוסיפים פונקציה AsyncGenerator (כמו monitor_for_hazard) לרשימה tools, הסוכן מתייחס אליה כאל תהליך רקע קבוע. במקום הרצה חד-פעמית, הסוכן 'נרשם' לפלט של הכלי. כך Dispatch יכול להמשיך את השיחה העיקרית שלו בזמן ש-Sentinel מפיק התראות על סכנות ברקע.

‫👉✏️ מחליפים את #REPLACE_AGENT_TOOLS ב-$HOME/way-back-home/level_4/backend/dispatch_agent/agent.py בטקסט הבא:

tools=[AgentTool(agent=architect_agent), monitor_for_hazard],    

אימות

‫👈💻 אחרי שמגדירים את שני הסוכנים, אפשר לבדוק את האינטראקציה הפעילה בין כמה סוכנים.

• בטרמינל A, מפעילים את הסוכן Architect:

cd $HOME/way-back-home/level_4/
. scripts/check_redis.sh
cd $HOME/way-back-home/level_4/backend/architect_agent
uv run server.py

• בטרמינל חדש (טרמינל B), מריצים את סוכן השליחה:

cd $HOME/way-back-home/level_4/backend/
cp architect_agent/.env .env
uv run adk web

בדיקה של מערכת מרובת סוכנים שמשתמשת במודל מולטי-מודאלי בזמן אמת כמו gemini-live בסימולטור adk web כוללת תהליך עבודה ספציפי. הסימולטור מצוין לבדיקת קריאות לכלים, אבל יש בעיית תאימות ידועה כשמעבדים תמונות בפעם הראשונה עם מודל מהסוג הזה.

• בסרגל הכלים של Cloud Shell, לוחצים על סמל תצוגה מקדימה באינטרנט. בוחרים באפשרות שינוי היציאה, מגדירים אותה ל-8000 ולוחצים על שינוי ותצוגה מקדימה.

👈בוחרים את dispatch_agent, מעלים את תוכנית ה-Blueprint ומטפלים בשגיאה הצפויה

זה השלב הכי חשוב. אנחנו צריכים לספק לסוכן את ההקשר של התמונה.

• כשממשק המשתמש נטען, לוחצים על אישור כדי לאשר גישה למיקרופון.
• מורידים את התמונה של התוכנית הזו למחשב:
• בממשק של adk web, לוחצים על סמל הסיכה המשרדית ומעלים את תמונת התוכנית שזה עתה הורדתם.

⚠️⚠️תוצג השגיאה 400 INVALID_ARGUMENT. זו תופעה נורמלית.⚠️⚠️

השגיאה הזו מתרחשת כי המטפל בתמונות adk web לא תואם באופן מלא ל-API של מודל gemini-live להעלאה חד-פעמית. עם זאת, התמונה נוספה בהצלחה להקשר של הסשן.

• ‫👉 כדי לפתור את השגיאה, פשוט טוענים מחדש את הדף בדפדפן.

הפעלת תהליך ההרכבה

‫👈 אחרי הטעינה מחדש, השגיאה תיעלם ותראו את תמונת התוכנית בהיסטוריית הצ'אט. הסוכן מקבל את ההקשר החזותי שהוא צריך.

• לוחצים על סמל המיקרופון כדי להפעיל אותו. בממשק תופיע ההודעה 'האזנה...'.
• אומרים את הפקודה הקולית: "start to assemble".
• הסוכן יעבד את הבקשה שלכם וממשק המשתמש ישתנה ל'בדיבור...'. תשמעו תשובה קולית בלבד עם רשימה של החלקים הנדרשים.

4. אימות של שיחות עם כלי נציג לנציג

‫👈 התשובה הקולית הראשונית מאשרת שהמערכת פועלת, אבל הקסם האמיתי נמצא במעקב התקשורת בין כמה סוכנים.

• משביתים את המיקרופון.
• מרעננים את הדף פעם נוספת.

החלונית 'מעקב' בצד ימין תתמלא. אפשר לראות את נתיב הביצוע המלא והמוצלח:

• dispatch_agent השיחות הראשונות monitor_for_hazard.
• לאחר מכן, המערכת מבצעת כמה execute_architect קריאות ל-architect_agent כדי לאחזר את נתוני הסכימה.

הרצף הזה מאשר שכל תהליך העבודה של ריבוי הסוכנים פועל בצורה תקינה: dispatch_agent קיבל את הבקשה, העביר את משימת אחזור הנתונים אל architect_agent באמצעות קריאה לכלי, וקיבל את הנתונים בחזרה כדי לבצע את הפקודה של המשתמש.

הקישור לשידור הדו-כיווני מאפשר עכשיו מעקב ברקע ושיתוף פעולה בין כמה נציגים. בשלב הבא נלמד איך לנתח את התשובות המורכבות האלה בקצה הקדמי.

‫👈💻 כדי לצאת משני הטרמינלים, לוחצים על Ctrl+c.

5. ניתוח מעמיק של שידורים חיים של אירועים מרובי-מוֹדָל

בשלב הקודם, אימתנו בהצלחה את מערכת המולטי-אג'נט שלנו באמצעות שרת הפיתוח המובנה, adk web. הכלי הזה משתמש ברץ ADK שמוגדר כברירת מחדל כדי לנהל באופן אוטומטי את הפעילות, את הסטרימינג ואת מחזור החיים של הסוכן. עם זאת, כדי ליצור אפליקציה עצמאית שמוכנה לייצור כמו שירות FastAPI שלנו (main.py), אנחנו צריכים שליטה מפורשת. אנחנו צריכים ליצור ולנהל ידנית את ADK Runner כדי לטפל בסשנים של משתמשים בזמן אמת, כי זה רכיב הליבה שמעבד את הזרמים הדו-כיווניים של אודיו, וידאו וטקסט.

הלולאה Model-Code-Model

כדי להבין איך המערכת פועלת בזמן אמת, נתאר את מחזור החיים של סשן יחיד של משימה. הלולאה הזו מייצגת את ההחלפה הרציפה של אובייקטים מסוג LlmRequest ו-LlmResponse.

1. הקישור החזותי: אתם יוזמים את החיבור ומשתפים את מצלמת האינטרנט או המסך. מסגרות JPEG באיכות גבוהה מתחילות לזרום במעלה הזרם דרך realtimeInput (באמצעות LiveRequestQueue).
2. הפעלה של Sentinel: המערכת שולחת גירוי ראשוני של 'שלום'. בהתאם להוראות, סוכן השליחה מפעיל באופן מיידי את monitor_for_hazard כלי הסטרימינג. הפעולה הזו מפעילה לולאה ברקע שעוקבת בשקט אחרי כל פריים נכנס.
3. פקודה לטייס: אתם מדברים במערכת הקשר: "Start to assemble" (מתחילים להתאסף).
4. העברת אודיו של קול: הקול שלכם נשמע באיכות אודיו של 16kHz ונשלח במעלה הזרם לצד פריימים של וידאו.
5. העברה (A2A): מערכת הניתוב "שומעת" את הכוונה שלכם. הוא מבין שחסרים לו שרטוטים, ולכן הוא קורא ל-Architect Agent באמצעות פרוטוקול AgentTool (Agent-as-a-Tool).
6. אחזור עובדות: ה-Architect שולח שאילתה למסד הנתונים של Redis ומחזיר את רשימת החלקים ל-Dispatch. השליטה נשארת בידי Dispatch, והוא מקבל את הנתונים בלי להעביר אתכם.
7. מידע שנשלח מהשרת ללקוח: מערכת Dispatch שולחת modelTurn (מהשרת ללקוח) שמכיל גם טקסט וגם אודיו בשפה המקורית: "הארכיטקט אישר. קבוצת המשנה הנדרשת היא: Warp Core, Flux Pipe, Ion Thruster.'
8. המשבר: פתאום, חלק מסוים בשולחן העבודה מתערער ומתחיל לזהור בלבן.
9. זיהוי אוטונומי: לולאת הרקע monitor_for_hazard (הסנטינל) מאתרת את מסגרת ה-JPEG הספציפית שמכילה את הזוהר. הוא מעבד את הפריים באמצעות קריאה ל-Gemini ומזהה את הסכנה.
10. בטיחות במורד הזרם: כלי הסטרימינג yields תוצאה. מכיוון שמדובר בנציג Bidi-Streaming, מערכת Dispatch יכולה להפריע למצב הנוכחי שלו כדי לשלוח באופן מיידי אזהרת בטיחות קריטית Downstream: "זוהה סיכון! מנטרלים את גביש הנתונים עכשיו. תעביר את זה לפח האדום".

הגדרת תצורת זמן הריצה של הסוכן

ה-RunConfig ב-ADK מאפשר הגדרה מפורטת של התנהגות הסוכן, כולל איך הוא מטפל בנתונים שמוזרמים ואיך הוא מקיים אינטראקציה עם מודאליות שונות.

ההגדרה של streaming_mode היא BIDI לתקשורת דו-כיוונית בזמן אמת, שמאפשרת למשתמש ולנציג לדבר ולהקשיב בו-זמנית. הפרמטר response_modalities מגדיר את סוגי הפלט שהסוכן יכול להפיק, כמו קול וטקסט. ‫input_audio_transcription מגדיר איך הסוכן מעבד ומתמלל את הדיבור הנכנס של המשתמש. כדי ליצור חוויה גמישה יותר, session_resumption מאפשר לסוכן לזכור את הקשר השיחה ולהמשיך אם החיבור אבד. לבסוף, proactivity מאפשר לסוכן ליזום פעולות או דיבור ללא פקודה ישירה מהמשתמש, כמו מתן אזהרה ספונטנית מפני סכנה, בעוד ש-enable_affective_dialog מאפשר לסוכן ליצור תגובות טבעיות ואמפתיות יותר. כאן אפשר לקבל מידע נוסף על RunConfig של ADK.

‫👈✏️ מאתרים את הפלייסהולדר #REPLACE_RUN_CONFIG בקובץ $HOME/way-back-home/level_4/backend/main.py ומחליפים אותו בלוגיקה הבאה של ניתוח:

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

הטמעה של בקשה לסוכן

לאחר מכן, נטמיע את ערוץ התקשורת המרכזי שמעביר נתונים בזמן אמת, מולטי-מודאליים, מהמחשב של המשתמש אל סוכן השילוח באמצעות WebSocket. הסוכן ימשיך "לראות" (פריימים של וידאו) ו"לשמוע" (פקודות קוליות). הלוגיקה מקבלת באופן רציף את מקור הנתונים, מבחינה בין נתחי אודיו בינאריים נכנסים לבין מנות טקסט/תמונה עטופות ב-JSON, ומכניסה אותם לאובייקטים מסוג Blob (למולטימדיה) או Content (לטקסט), ושולחת אותם ל-LiveRequestQueue כדי להפעיל את סשן ה-Agent הדו-כיווני.

מאתרים את ה-placeholder #PROCESS_AGENT_REQUEST בקובץ $HOME/way-back-home/level_4/backend/main.py ומחליפים אותו בלוגיקה הבאה של ניתוח:

# Start the loop
        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":
                        # logger.info("Received AUDIO packet") # Uncomment for verbose debugging
                        import base64
                        # Decode base64 audio data
                        audio_data = base64.b64decode(json_message.get("data", ""))
                        
                        # logger.info(f"Received Audio Chunk: {len(audio_data)} bytes")
                        
                        import math
                        import struct
                        # Calculate RMS to debug silence
                        count = len(audio_data) // 2
                        shorts = struct.unpack(f"<{count}h", audio_data)
                        sum_squares = sum(s*s for s in shorts)
                        rms = math.sqrt(sum_squares / count) if count > 0 else 0
                        
                        # logger.info(f"RMS: {rms:.2f} | Bytes: {len(audio_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"])
                        # logger.info(f"Received Image Frame: {len(image_data)} bytes")
                        
                        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)
                        
                        frame_count += 1
                        
        finally:
             pass                   

הנתונים המולטימודליים נשלחים עכשיו לסוכן.

הטמעה של התגובה: מבנה נתוני האירועים במורד הזרם

כשמפעילים סוכן דו-כיווני (בזמן אמת) באמצעות ADK, הנתונים שחוזרים מהסוכן נארזים בסוג ספציפי של אירוע שמוגדר על בסיס המבנים העיקריים של GenAI SDK. אובייקט Event שמתקבל בלולאה async for event in runner.run_live(...) הוא אובייקט יחיד שמכיל כמה שדות אופציונליים, כל אחד לסוג מידע אחר:

המבנה של התוכן:

• כשהסוכן מדבר (באמצעות .server_content): השדה לא מכיל רק טקסט פשוט. הוא מכיל רשימה של Parts. כל Part הוא מאגר של סוג נתונים אחד – מחרוזת טקסט (כמו "The part is stable.") או blob של אודיו גולמי (הקול).
• כשהסוכן פועל (דרך .tool_call): השדה מכיל רשימה של אובייקטים מסוג FunctionCall. כל FunctionCall הוא אובייקט פשוט ומובנה שמציין את שם הכלי ואת ארגומנטי הקלט בפורמט נקי שקוד ה-Backend יכול לקרוא ולהריץ בקלות.

👀 אם הייתם מסתכלים על Event יחיד שמוחזר על ידי לולאת run_live, קובץ ה-JSON (שנוצר על ידי event.model_dump(by_alias=True)) היה נראה כך, בהתאם לפורמטים של GenAI SDK:

{
  "serverContent": {  // <-- LiveServerMessageServerContent
    "modelTurn": {    // <-- ModelTurn
      "parts": [      // <-- list[Part]
        {
          "text": "Architect Confirmed."
        },
        {
          "inlineData": { // <-- Blob (Audio Bytes)
            "mimeType": "audio/pcm;rate=24000",
            "data": "BASE64_AUDIO_DATA..."
          }
        }
      ]
    }
  },
  "toolCall": {       // <-- LiveServerMessageToolCall
    "functionCalls": [ // <-- list[FunctionCall]
      {
        "name": "neutralize_hazard",
        "args": { "color": "RED" }
      }
    ]
  }
}

‫👉✏️ מעכשיו נעדכן את downstream_task ב-main.py כדי להעביר את נתוני האירועים המלאים. הלוגיקה הזו מבטיחה שכל 'מחשבה' של ה-AI תתועד במסוף האבחון של הספינה ותישלח כאובייקט JSON יחיד לממשק המשתמש של הקצה הקדמי.

מאתרים את ה-placeholder #PROCESS_AGENT_RESPONSE בקובץ $HOME/way-back-home/level_4/backend/main.py ומחליפים אותו בלוגיקה הבאה של ניתוח:

            # Suppress raw event logging
            event_json = event.model_dump_json(exclude_none=True, by_alias=True)
            # logger.info(f"raw_event: {event_json[:200]}...") 
            await websocket.send_text(event_json)

ביצוע המשימה

אחרי שמחברים את הכספת בעורף וקובעים את ההגדרות של שני הסוכנים, כל המערכות מוכנות לפעולה. השלבים הבאים יפעילו את האפליקציה המלאה, ויאפשרו לכם ליצור אינטראקציה עם מערכת שני הסוכנים שבניתם.

המטרה: להרכיב את מנוע העל-חלל שהוקצה באופן אקראי ומופיע בספסל העבודה. פרוטוקול: חשוב לפעול לפי ההנחיות הקוליות של נציג השירות, במיוחד אזהרות לגבי סכנות שקשורות לרכיבים ספציפיים.

הפעלת המומחה (הארכיטקט)

‫👈💻 בחלון המסוף הראשון, מפעילים את סוכן הארכיטקט. שירות לקצה העורפי הזה יתחבר לכספת Redis וימתין לבקשות סכמטיות מה-Dispatcher.

# Ensure you are in the backend directory
cd $HOME/way-back-home/level_4/
. scripts/check_redis.sh
cd $HOME/way-back-home/level_4/backend
# Start the A2A Server on Port 8081
uv run architect_agent/server.py

(משאירים את הטרמינל הזה פועל. הוא יהיה עכשיו 'סוכן מסד הנתונים' הפעיל שלכם).

הפעלת מרכז הבקרה (המשגר)

‫👈💻 בחלון טרמינל חדש (Terminal B), ניצור את ממשק המשתמש של הקצה הקדמי ונפעיל את סוכן השליחה הראשי, שמציג את ממשק המשתמש ומטפל בכל התקשורת בזמן אמת.

# 1. Build the Frontend Assets
cd $HOME/way-back-home/level_4/frontend
npm install
npm run build

# 2. Launch the Main Application Server
cd $HOME/way-back-home/level_4/backend
cp architect_agent/.env .env
uv run main.py

(כך מפעילים את השרת הראשי ביציאה 8080).

הרצת תרחיש הבדיקה

המערכת פעילה עכשיו. המטרה שלכם היא לפעול לפי ההוראות של הסוכן כדי להשלים את ההרכבה.

1. 👈 גישה לשולחן העבודה:
   • בסרגל הכלים של Cloud Shell, לוחצים על סמל תצוגה מקדימה באינטרנט.
   • בוחרים באפשרות שינוי היציאה, מגדירים אותה ל-8080 ולוחצים על שינוי ותצוגה מקדימה.
2. 👉 מתחילים את המשימה:
   • כשממשק המשתמש נטען, חשוב לאשר לו גישה למסך ולמיקרופון.
   • תתבקשו לבחור כרטיסייה או חלון לשיתוף. אם אתם משתפים את החלון, כדי למנוע בעיות, ודאו שזו הכרטיסייה היחידה בחלון.
   • כונן עם שם אקראי (למשל, ‫("NOVA-V",‏ "OMEGA-9") יוקצו לכם.
3. 👉 לולאת Assembly:
   • בקשה: כדי להתחיל להרכיב את הכונן, אומרים: "Start assembling" (התחל הרכבה).
   • תגובה של ארכיטקט: הסוכן יספק את החלקים הנכונים להרכבת הכונן.
   • בדיקת סיכונים: אם נראה שחלק מסוים בסביבת העבודה מסוכן:
     • הכלי monitor_for_hazard של סוכן השילוח יזהה אותו באופן חזותי.
     • תוצג התראה על סכנה ויזואלית. (התהליך יימשך כ-30 שניות)
     • המערכת תבדוק באיזה פח צריך להשתמש כדי לנטרל את הסכנה.
   • פעולה: סוכן השיגור ייתן לכם פקודה ישירה: "הסכנה אושרה. צריך להכניס את XXX לפח האדום באופן מיידי". כדי להמשיך, צריך לפעול לפי ההוראות האלה.

המשימה הושלמה. יצרתם בהצלחה מערכת אינטראקטיבית עם כמה סוכנים. הניצולים בטוחים, הטיל יצא מהאטמוספירה והמסע שלכם חזרה הביתה נמשך.

‫👉💻 לוחצים על Ctrl+c בשני הטרמינלים כדי לצאת.

6. פריסה בסביבת הייצור (אופציונלי)

הבדיקה של הסוכן באופן מקומי הסתיימה בהצלחה. עכשיו צריך להעלות את ליבת הרשת העצבית של האדריכל למחשבים המרכזיים של הספינה (Cloud Run). כך הוא יוכל לפעול כשירות קבוע ועצמאי, שסוכן השינוע יוכל לשלוח אליו שאילתות מכל מקום.

הקצאת Secure Vault (תשתית)

לפני שמפעילים את הסוכן, צריך ליצור את הזיכרון המתמשך שלו (Memorystore) ואת הערוץ המאובטח לגישה אליו (VPC Connector).

‫👈💻 יצירת מכונת Memorystore‏ (Redis Vault):

export REGION="us-central1"
gcloud redis instances create ozymandias-vault-prod --size=1 --tier=basic --region=${REGION}

‫👈💻 אחזור כתובת הרשת של Vault: מריצים את הפקודה הזו ומעתיקים את כתובת ה-IP‏ host. זוהי הכתובת הפרטית של מכונת Redis החדשה.

gcloud redis instances describe ozymandias-vault-prod --region=us-central1

‫👈💻 יוצרים את מחבר הגישה ל-VPC (גשר מאובטח): המחבר הזה פועל כגשר פרטי, ומאפשר ל-Cloud Run לגשת למופע Redis בתוך ה-VPC.

export REGION="us-central1"
export SUBNET_NAME="vpc-connector-subnet"
export PROJECT_ID=$(gcloud config get-value project)
# Create the Dedicated Subnet ---

gcloud compute networks subnets create ${SUBNET_NAME} \
    --network=default \
    --region=${REGION} \
    --range=192.168.1.0/28


gcloud compute networks vpc-access connectors create architect-connector \
 --region ${REGION} \
 --subnet ${SUBNET_NAME} \
 --subnet-project ${PROJECT_ID} \
 --min-instances 2 \
 --max-instances 3 \
 --machine-type f1-micro

‫👉💻 טוענים את הנתונים:

export REGION="us-central1"
export ZONE="us-central1-a"
export VM_NAME="redis-seeder-$(date +%s)"
export REDIS_IP=$(gcloud redis instances describe ozymandias-vault-prod --region=${REGION} | grep 'host:' | awk '{print $2}')

gcloud compute instances create ${VM_NAME} \
    --zone=${ZONE} \
    --machine-type=e2-micro \
    --image-family=debian-11 \
    --image-project=debian-cloud \
    --quiet \
    --metadata=startup-script='#! /bin/bash
        # Install tools quietly
        apt-get update > /dev/null
        apt-get install -y redis-tools > /dev/null

        # Run each command individually
        redis-cli -h '"${REDIS_IP}"' DEL "HYPERION-X"
        redis-cli -h '"${REDIS_IP}"' RPUSH "HYPERION-X" "Warp Core" "Flux Pipe" "Ion Thruster"
        redis-cli -h '"${REDIS_IP}"' DEL "NOVA-V"
        redis-cli -h '"${REDIS_IP}"' RPUSH "NOVA-V" "Ion Thruster" "Warp Core" "Flux Pipe"
        redis-cli -h '"${REDIS_IP}"' DEL "OMEGA-9"
        redis-cli -h '"${REDIS_IP}"' RPUSH "OMEGA-9" "Flux Pipe" "Ion Thruster" "Warp Core"
        redis-cli -h '"${REDIS_IP}"' DEL "GEMINI-MK1"
        redis-cli -h '"${REDIS_IP}"' RPUSH "GEMINI-MK1" "Coolant Tank" "Servo" "Fuel Cell"
        redis-cli -h '"${REDIS_IP}"' DEL "APOLLO-13"
        redis-cli -h '"${REDIS_IP}"' RPUSH "APOLLO-13" "Warp Core" "Coolant Tank" "Ion Thruster"
        redis-cli -h '"${REDIS_IP}"' DEL "VORTEX-7"
        redis-cli -h '"${REDIS_IP}"' RPUSH "VORTEX-7" "Quantum Cell" "Graviton Coil" "Plasma Injector"
        redis-cli -h '"${REDIS_IP}"' DEL "CHRONOS-ALPHA"
        redis-cli -h '"${REDIS_IP}"' RPUSH "CHRONOS-ALPHA" "Shield Emitter" "Data Crystal" "Quantum Cell"
        redis-cli -h '"${REDIS_IP}"' DEL "NEBULA-Z"
        redis-cli -h '"${REDIS_IP}"' RPUSH "NEBULA-Z" "Plasma Injector" "Flux Pipe" "Graviton Coil"
        redis-cli -h '"${REDIS_IP}"' DEL "PULSAR-B"
        redis-cli -h '"${REDIS_IP}"' RPUSH "PULSAR-B" "Data Crystal" "Servo" "Shield Emitter"
        redis-cli -h '"${REDIS_IP}"' DEL "TITAN-PRIME"
        redis-cli -h '"${REDIS_IP}"' RPUSH "TITAN-PRIME" "Ion Thruster" "Quantum Cell" "Warp Core"

        # Signal that the script has finished
        echo "SEEDING_COMPLETE"
    '
# This command streams the logs and waits until grep finds our completion message.
# The -m 1 flag tells grep to exit after the first match.
gcloud compute instances tail-serial-port-output ${VM_NAME} --zone=${ZONE} | grep -m 1 "SEEDING_COMPLETE"

gcloud compute instances delete ${VM_NAME} --zone=${ZONE} --quiet

פריסת אפליקציית הסוכן

קומפילציה ויצירה של תמונת סוכן

‫👈💻 עוברים לספריית ה-backend ויוצרים את קובץ ה-Dockerfile.

export PROJECT_ID=$(gcloud config get-value project)
export REGION=us-central1
export SERVICE_NAME=architect-agent
export IMAGE_PATH=gcr.io/${PROJECT_ID}/${SERVICE_NAME}
export VPC_CONNECTOR_NAME=architect-connector
export REDIS_IP=$(gcloud redis instances describe ozymandias-vault-prod --region=${REGION} | grep 'host:' | awk '{print $2}')

cd $HOME/way-back-home/level_4/backend/architect_agent
cp $HOME/way-back-home/level_4/requirements.txt requirements.txt
cat <<EOF > Dockerfile
# Use an official Python runtime as a parent image
FROM python:3.13-slim

# Set the working directory in the container
WORKDIR /app

# Copy the requirements file and install dependencies for THIS agent
COPY requirements.txt requirements.txt
RUN pip install --no-cache-dir -r requirements.txt

# Copy the rest of the architect's code (server.py, agent.py, etc.)
COPY . .

# Expose the port the architect server runs on
EXPOSE 8081

# Command to run the application
# This assumes your server file is named server.py and the FastAPI object is 'app'
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8081"]
EOF

‫👈💻 לארוז את האפליקציה לתוך קובץ אימג' של קונטיינר.

cd $HOME/way-back-home/level_4/backend/architect_agent

export PROJECT_ID=$(gcloud config get-value project)
export SERVICE_NAME=architect-agent
export IMAGE_PATH=gcr.io/${PROJECT_ID}/${SERVICE_NAME}
export REGION=us-central1


# This should now print the full, correct path
echo "Verifying build path: ${IMAGE_PATH}"

gcloud builds submit . --tag ${IMAGE_PATH}

פריסה ב-Cloud Run

‫👈💻 פורסים את הסוכן ב-Cloud Run. אנחנו נחדיר את כתובת ה-IP של Redis ונקשר את מחבר ה-VPC ישירות לפקודת ההפעלה. כך הנציג מתחיל עם חיבור מאובטח ופרטי למסד הנתונים שלו.

cd $HOME/way-back-home/level_4/backend/architect_agent

export PROJECT_ID=$(gcloud config get-value project)
export REGION=us-central1
export SERVICE_NAME=architect-agent
export IMAGE_PATH=gcr.io/${PROJECT_ID}/${SERVICE_NAME}
export VPC_CONNECTOR_NAME=architect-connector
export REDIS_IP=$(gcloud redis instances describe ozymandias-vault-prod --region=${REGION} | grep 'host:' | awk '{print $2}')
export PROJECT_NUMBER=$(gcloud projects describe ${PROJECT_ID} --format="value(projectNumber)")
export PREDICTED_HOST="${SERVICE_NAME}-${PROJECT_NUMBER}.${REGION}.run.app"
export PROTOCOL=https

gcloud run deploy ${SERVICE_NAME} \
  --image=${IMAGE_PATH} \
  --platform=managed \
  --region=${REGION} \
  --port=8081 \
  --allow-unauthenticated \
  --labels=dev-tutorial=multi-modal \
  --vpc-connector=${VPC_CONNECTOR_NAME} \
  --vpc-egress=private-ranges-only \
  --set-env-vars="REDIS_HOST=${REDIS_IP}" \
  --set-env-vars="GOOGLE_GENAI_USE_VERTEXAI=True" \
  --set-env-vars="MODEL_ID=gemini-2.5-flash" \
  --set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
  --set-env-vars="HOST_URL=${PREDICTED_HOST}" \
  --set-env-vars="PROTOCOL=${PROTOCOL}" \
  --set-env-vars="A2A_PORT=443"

‫👈💻 מוודאים שהשרת A2A פועל.

export REGION=us-central1
export ARCHITECT_AGENT_URL=$(gcloud run services describe architect-agent --platform managed --region ${REGION} --format 'value(status.url)')
curl -s  ${ARCHITECT_AGENT_URL}/.well-known/agent.json | jq 

אחרי שהפקודה מסתיימת, מופיעה כתובת URL של שירות. סוכן הארכיטקט פועל עכשיו בענן, מחובר באופן קבוע לכספת שלו ומוכן לספק נתונים סכמטיים לסוכנים אחרים.

פריסת מרכז הניהול של בקשות הניתוב במחשב מרכזי בסביבת ייצור

אחרי שהפעלנו את סוכן הארכיטקט בענן, צריך לפרוס את מרכז השליטה של Dispatch. הסוכן הזה ישמש כממשק המשתמש הראשי, יטפל בשידורים חיים של קול או וידאו ויקצה שאילתות במסד הנתונים לנקודת הקצה המאובטחת של Architect.

‫👈💻 מריצים את הפקודה הבאה במסוף של Cloud Shell. הכלי ייצור את קובץ ה-Dockerfile המלא והרב-שלבי בספריית ה-backend.

cd $HOME/way-back-home/level_4

cat <<EOF > Dockerfile
# STAGE 1: Build the React Frontend
# This stage uses a Node.js container to build the static frontend assets.
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 root of our build context
COPY requirements.txt .
# Install the Python dependencies
RUN uv pip install --no-cache-dir --system -r requirements.txt

# Copy the entire backend directory into the container
COPY backend/ ./backend/

# CRITICAL STEP: Copy the built frontend assets from the 'builder' stage.
# The source is the '/app/frontend/dist' directory from Stage 1.
# The destination is './frontend/dist', which matches the exact relative path
# your backend/main.py script expects to find.
COPY --from=builder /app/frontend/dist ./frontend/dist/

# Cloud Run injects a PORT environment variable, which your main.py already uses.
# We expose 8000 as a standard practice.
EXPOSE 8000

# Set the command to run the application.
# We specify the full path to the Python script.
CMD ["python", "backend/main.py"]
EOF

הידור ויצירה של תמונת נציג/קצה קדמי

‫👉💻 עוברים לספריית ה-backend שמכילה את הקוד של סוכן הניתוב (main.py) ואורזים אותו בקובץ אימג' של קונטיינר.

cd $HOME/way-back-home/level_4
export PROJECT_ID=$(gcloud config get-value project)
export REGION=us-central1
export SERVICE_NAME=mission-bravo
export IMAGE_PATH=gcr.io/${PROJECT_ID}/${SERVICE_NAME}
# This assumes your dispatch agent server (main.py) is in the backend folder

gcloud builds submit . --tag ${IMAGE_PATH}

פריסה ב-Cloud Run

‫👈💻 פורסים את Dispatch Hub ב-Cloud Run. אנחנו נחדיר את כתובת ה-URL של Architect כמשתנה סביבה, וכך ניצור את הקישור הקריטי בין שני הסוכנים שלנו מבוססי-הענן.

export PROJECT_ID=$(gcloud config get-value project)
export REGION=us-central1
export SERVICE_NAME=mission-bravo
export AGENT_SERVICE_NAME=architect-agent
export IMAGE_PATH=gcr.io/${PROJECT_ID}/${SERVICE_NAME}
export PROJECT_NUMBER=$(gcloud projects describe ${PROJECT_ID} --format="value(projectNumber)")
export ARCHITECT_AGENT_URL="https://${AGENT_SERVICE_NAME}-${PROJECT_NUMBER}.${REGION}.run.app"
gcloud run deploy ${SERVICE_NAME} \
  --image=${IMAGE_PATH} \
  --platform=managed \
  --region=${REGION} \
  --port=8080 \
  --labels=dev-tutorial=multi-modal \
  --allow-unauthenticated \
  --set-env-vars="ARCHITECT_URL=${ARCHITECT_AGENT_URL}" \
  --set-env-vars="GOOGLE_GENAI_USE_VERTEXAI=True" \
  --set-env-vars="MODEL_ID=gemini-live-2.5-flash-preview-native-audio-09-2025" \
  --set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
  --set-env-vars="GOOGLE_CLOUD_LOCATION=${REGION}"

אחרי שהפקודה מסתיימת, מוצגת כתובת URL של שירות (למשל, https://mission-bravo-...run.app). האפליקציה פעילה עכשיו בענן.

‫👉 עוברים לדף Google Cloud Run ובוחרים את שירות biometric-scout מהרשימה.

‫👈 מאתרים את כתובת ה-URL הציבורית שמוצגת בחלק העליון של דף פרטי השירות.

בדיקת מערכת סופית (בדיקה מקצה לקצה)

‫👈 עכשיו תהיה לכם אינטראקציה עם המערכת החיה.

1. מקבלים את כתובת ה-URL: מעתיקים את כתובת ה-URL של השירות מהפלט של פקודת הפריסה האחרונה (היא אמורה להסתיים ב-run.app).
2. פותחים את מרכז הבקרה: מדביקים את כתובת ה-URL בדפדפן האינטרנט.
3. יצירת קשר: כשממשק המשתמש נטען, צריך לוודא שמאשרים לו גישה למסך ולמיקרופון.
4. בקשה לנתונים: כשמוקצה כונן, מבקשים להתחיל בהרכבה. לדוגמה: "מתחילים להרכיב"

עכשיו אתם מתקשרים עם מערכת רב-סוכנים שפרוסה במלואה ופועלת כולה ב-Google Cloud.

מערכת מרובת-סוכנים נועלת את טבעת ההכלה הסופית במקומה, והקרינה הלא יציבה מתייצבת לזמזום קבוע.

‫"Warp Drive: STABILIZED. Rescue Craft: ENGINES IGNITED."

במסך, החללית החוצנית טסה במהירות כלפי מעלה, ונמלטת בקושי מפני השטח המתפורר של אוזימנדיאס כשהאטמוספירה קורסת. היא מתמקמת במסלול בטוח לצד כלי השיט שלכם, והתקשורת מתמלאת בקולות של הניצולים – הם נסערים אבל חיים. אחרי שהחילוץ מסתיים והדרך הביתה פנויה, הקישור המרוחק מתנתק.

הודות לכם, הניצולים חולצו.

אם השתתפתם ברמה 0, אל תשכחו לבדוק איפה אתם בתהליך של המשימה 'בדרך הביתה'!