راه بازگشت به خانه - سیستم چندعامله دوطرفه زنده

۱. ماموریت

داستان

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

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

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

تنها راه فرار آنها یک موشک قدیمی و متروکه است که با فناوری بیگانه ساخته شده است. در حالی که این موشک کار می‌کند، اما Warp Drive آن خرد شده است. برای نجات بازماندگان، باید از راه دور به Volatile Workbench آنها متصل شوید و به صورت دستی یک درایو جایگزین بسازید.

چالش

شما هیچ تجربه‌ای با این فناوری بیگانه که به شکننده بودن معروف است، ندارید. یک قطعه بی‌ثبات می‌تواند در عرض چند ثانیه به یک خطر رادیواکتیو تبدیل شود. شما فقط یک بار فرصت دارید تا با میز کار فرار (Volatile Workbench) کار کنید. دستیار هوش مصنوعی فعلی شما در پردازش همزمان داده‌های بصری و دستورالعمل‌های فنی مشکل دارد که منجر به دستورالعمل‌های توهم‌زا و عدم توجه به هشدارهای خطر می‌شود.

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

اهداف ماموریت شما:

با دنبال کردن دستورالعمل‌های تخصصی و بلادرنگ از سیستم چندعاملی جدید خود، Warp Drive را مونتاژ کنید.

ماموریت آلفا

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

نمای کلی

  • یک سیستم هوش مصنوعی چندعاملی دوطرفه و بلادرنگ که دارای یک عامل اعزام مرکزی است که تعامل کاربر را مدیریت کرده و با عوامل تخصصی هماهنگ می‌کند.
  • یک عامل معمار که به پایگاه داده Redis متصل می‌شود تا داده‌های شماتیک را بازیابی و ارائه دهد.
  • یک مانیتور ایمنی پیشگیرانه که از ابزارهای پخش جریانی برای تجزیه و تحلیل فید ویدیویی زنده برای خطرات بصری و ایجاد هشدارهای بلادرنگ استفاده می‌کند.
  • یک رابط کاربری مبتنی بر React که رابط کاربری برای تعامل با سیستم، پخش ویدئو و صدا به عوامل backend فراهم می‌کند.

آنچه یاد خواهید گرفت

فناوری / مفهوم

توضیحات

کیت توسعه عامل گوگل (ADK)

شما از ADK برای ساخت، آزمایش و مدیریت عامل‌ها استفاده خواهید کرد و از چارچوب آن برای مدیریت ارتباطات بلادرنگ، یکپارچه‌سازی ابزار و چرخه حیات عامل بهره خواهید برد.

جریان دو طرفه (Bidi)

شما یک عامل پخش بیدی (bidi-streaming agent) پیاده‌سازی خواهید کرد که امکان ارتباط طبیعی، کم‌تاخیر و دوطرفه را فراهم می‌کند و به انسان و هوش مصنوعی این امکان را می‌دهد که در لحظه (real time) وقفه ایجاد کرده و پاسخ دهند.

سیستم‌های چندعاملی

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

پروتکل عامل به عامل (A2A)

شما از پروتکل A2A برای فعال کردن ارتباط بین Dispatch Agent و Architect Agent استفاده خواهید کرد و به آنها اجازه می‌دهید تا قابلیت‌های یکدیگر را کشف کرده و داده‌ها را تبادل کنند.

ابزارهای استریمینگ

شما یک ابزار استریمینگ پیاده‌سازی خواهید کرد که به عنوان یک فرآیند پس‌زمینه عمل می‌کند و به طور مداوم یک فید ویدیویی را برای نظارت بر تغییرات وضعیت (خطرات) تجزیه و تحلیل می‌کند و به صورت پیشگیرانه نتایج را ارائه می‌دهد.

گوگل کلود ران و مموری استور

شما کل برنامه چندعاملی را در یک محیط عملیاتی مستقر خواهید کرد و از Cloud Run برای میزبانی سرویس‌های عامل و Memorystore (Redis) به عنوان پایگاه داده پایدار استفاده خواهید کرد.

FastAPI و وب‌سوکت‌ها

بخش پشتی (backend) با استفاده از FastAPI و WebSockets ساخته شده است تا ارتباطات با کارایی بالا و بلادرنگ مورد نیاز برای پخش صدا، تصویر و پاسخ‌های عامل (agent) را مدیریت کند.

فرانت‌اند واکنش نشان دهید

شما با یک رابط کاربری مبتنی بر React کار خواهید کرد که رسانه‌های کاربر (صوتی/تصویری) را ضبط و پخش می‌کند و پاسخ‌های بلادرنگ از عوامل هوش مصنوعی را نمایش می‌دهد.

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

دسترسی به پوسته ابری

👉 روی فعال کردن پوسته ابری (Activate Cloud Shell) در بالای کنسول گوگل کلود کلیک کنید (این آیکون به شکل ترمینال در بالای پنل پوسته ابری قرار دارد)، پوسته ابری.png

👉 روی دکمه‌ی «باز کردن ویرایشگر» کلیک کنید (شبیه یک پوشه‌ی باز شده با مداد است). با این کار ویرایشگر کد Cloud Shell در پنجره باز می‌شود. یک فایل اکسپلورر در سمت چپ خواهید دید. ویرایشگر باز.png

👉 ترمینال را در محیط توسعه ابری (cloud IDE) باز کنید،

۰۳-۰۵-new-terminal.png

👉💻 در ترمینال، با استفاده از دستور زیر تأیید کنید که از قبل احراز هویت شده‌اید و پروژه روی شناسه پروژه شما تنظیم شده است:

gcloud auth list

باید حساب خود را به عنوان (ACTIVE) مشاهده کنید.

پیش‌نیازها

ℹ️ سطح ۰ اختیاری است (اما توصیه می‌شود)

شما می‌توانید این ماموریت را بدون سطح ۰ انجام دهید، اما تمام کردن آن در ابتدا تجربه‌ای فراگیرتر ارائه می‌دهد و به شما این امکان را می‌دهد که با پیشرفت خود، چراغ راهنمای خود را روی نقشه جهانی ببینید.

راه‌اندازی محیط پروژه

به ترمینال خود برگردید، با تنظیم پروژه فعال و فعال کردن سرویس‌های مورد نیاز 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

نصب وابستگی‌ها

👉💻 به سطح ۴ بروید و بسته‌های پایتون مورد نیاز را نصب کنید:

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

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

بسته

هدف

fastapi

چارچوب وب با کارایی بالا برای ایستگاه ماهواره‌ای و استریم SSE

uvicorn

سرور ASGI برای اجرای برنامه FastAPI مورد نیاز است

google-adk

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

a2a-sdk

کتابخانه پروتکل عامل به عامل برای ارتباط استاندارد

google-genai

کلاینت بومی برای دسترسی به مدل‌های Gemini

redis

کلاینت پایتون برای اتصال به Schematic Vault (Memorystore)

websockets

پشتیبانی از ارتباط دو طرفه بلادرنگ

python-dotenv

متغیرهای محیطی و اسرار پیکربندی را مدیریت می‌کند

pydantic

اعتبارسنجی داده‌ها و مدیریت تنظیمات

تأیید تنظیمات

قبل از اینکه وارد کد شویم، بیایید مطمئن شویم که همه سیستم‌ها سبز هستند. اسکریپت تأیید را اجرا کنید تا پروژه Google Cloud، APIها و وابستگی‌های پایتون خود را بررسی کنید.

👉💻 اسکریپت تأیید را اجرا کنید:

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.

۳. ساخت Schematic Vault در Redis و BiDirectional Agent با ADK

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

نمای کلی

آماده‌سازی مخزن شماتیک (Redis)

قبل از اینکه معمار بتواند به ما کمک کند، باید مطمئن شویم که داده‌ها در یک محیط امن و با دسترسی بالا میزبانی می‌شوند. ما از Redis به عنوان یک مخزن داده سریع برای طرح‌های بیگانه خود استفاده خواهیم کرد. برای راحتی توسعه، یک نمونه محلی Redis راه‌اندازی خواهیم کرد، اما دستورالعمل‌های نحوه استقرار در یک محیط عملیاتی با Google Cloud Memorystore بعداً ارائه خواهد شد.

👉💻 دستورات زیر را در ترمینال خود اجرا کنید تا نمونه Redis را آماده کنید (این کار ممکن است ۲-۳ دقیقه طول بکشد):

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 Agent ) یک عامل تخصصی است که مسئول بازیابی طرح‌های اولیه شماتیک از مخزن Redis ما است. این عامل به عنوان یک رابط داده اختصاصی عمل می‌کند و تضمین می‌کند که عامل اعزام اصلی، اطلاعات دقیق و ساختاریافته را بدون نیاز به دانستن منطق پایگاه داده اصلی دریافت کند.

نمای کلی

کیت توسعه عامل گوگل (ADK) چارچوب ماژولاری است که این راه‌اندازی چندعاملی را امکان‌پذیر می‌کند. این کیت دو لایه حیاتی را مدیریت می‌کند:

  1. چرخه عمر اتصال و جلسه: تعامل با APIهای بلادرنگ نیازمند مدیریت پیچیده پروتکل است - مدیریت handshakeها، احراز هویت و سیگنال‌های keep-alive.
  2. فراخوانی تابع: این «مسیر رفت و برگشت مدل-کد-مدل» است. وقتی LLM تصمیم می‌گیرد که به داده نیاز دارد، یک فراخوانی تابع ساختاریافته را خروجی می‌دهد. ADK این را رهگیری می‌کند، کد پایتون شما ( lookup_schematic_tool ) را اجرا می‌کند و نتیجه را در عرض چند میلی‌ثانیه به متن مدل برمی‌گرداند.

اکنون معمار را خواهیم ساخت. این عامل به دوربین دسترسی ندارد. صرفاً برای دریافت "نام درایو" و بازگرداندن "لیست قطعات" از پایگاه داده وجود دارد.

👉💻 ما از دستور adk create استفاده خواهیم کرد. این ابزاری از مجموعه توسعه عامل (ADK) است که به طور خودکار کد و ساختار فایل استاندارد را برای یک عامل جدید تولید می‌کند و در زمان راه‌اندازی ما صرفه‌جویی می‌کند. 

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

پیکربندی عامل

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

  1. یک مدل انتخاب کنید : گزینه ۱ (Gemini Flash) را انتخاب کنید.
    • توجه: نسخه خاص (مثلاً ۲.۵، ۳.۰) ممکن است بسته به موجود بودن متفاوت باشد. برای سرعت، همیشه نوع "فلش" را انتخاب کنید.
  2. انتخاب یک backend : گزینه ۲ (Vertex AI) را انتخاب کنید.
  3. شناسه پروژه گوگل کلود را وارد کنید : برای پذیرش پیش‌فرض (شناسایی شده از محیط شما) ، Enter را فشار دهید.
  4. منطقه ابری گوگل را وارد کنید : برای پذیرش پیش‌فرض ( us-central1 )، Enter را فشار دهید.

👀 تعامل ترمینال شما باید مشابه این باشد: 

(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 success) را ببینید. این کد، اسکلت کدی را که در مرحله بعدی اصلاح خواهیم کرد، ایجاد می‌کند.

👉✏️ به فایل $HOME/way-back-home/level_4/backend/architect_agent/agent.py که به تازگی ایجاد شده است، بروید و آن را در ویرایشگر خود باز کنید. قطعه کد ابزار را بعد از اولین خط import به فایل اضافه کنید: 

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

با معمار آنلاین، اکنون یک منبع حقیقت داریم. قبل از اینکه این را به عامل اصلی متصل کنیم، کیت توسعه عامل (ADK) با ساده‌سازی پیچیدگی‌های ساخت و آزمایش عامل‌های هوش مصنوعی، مزیت قابل توجهی را ارائه می‌دهد. با کنسول توسعه‌دهنده adk web داخلی آن، می‌توانیم عملکرد Architect Agent خود، به ویژه قابلیت‌های فراخوانی ابزار آن را، قبل از ادغام آن در سیستم چندعاملی بزرگتر، جداسازی و تأیید کنیم. این رویکرد مدولار برای توسعه و آزمایش برای ساخت برنامه‌های هوش مصنوعی قوی و قابل اعتماد بسیار مهم است.

👉💻 در ترمینال خود، دستور زیر را اجرا کنید: 

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 کلیک کنید. گزینه Change port را انتخاب کنید، آن را روی ۸۰۰۰ تنظیم کنید و روی Change and Preview کلیک کنید. *پیش‌نمایش وب
  • architect_agent را انتخاب کنید.
  • فعال کردن ابزار: در رابط چت، عبارت CHRONOS-ALPHA (یا هر شناسه درایو از پایگاه داده شماتیک) را تایپ کنید.
  • رفتار را مشاهده کنید:
    • The Architect should immediately trigger the 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]

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

👉💻 برای خروج، Ctrl+C را فشار دهید.

مقداردهی اولیه سرور A2A

برای اتصال عامل اعزام به معمار، از پروتکل عامل به عامل (A2A) استفاده می‌کنیم.

در حالی که پروتکل‌هایی مانند MCP (پروتکل زمینه مدل) بر اتصال عامل‌ها به ابزارها تمرکز دارند، A2A بر اتصال عامل‌ها به سایر عامل‌ها تمرکز دارد. این استانداردی است که به Dispatcher ما اجازه می‌دهد تا معمار را "کشف" کند و قابلیت آن را در جستجوی شماتیک‌ها درک کند.

A2A

جریان A2A: در این ماموریت، ما از یک مدل کلاینت-سرور استفاده می‌کنیم:

  1. سرور (معمار): میزبان ابزارهای پایگاه داده است و مهارت‌های خود را از طریق کارت عامل «تبلیغ» می‌کند.
  2. کلاینت (Dispatch): کارت معمار را می‌خواند، API آن را می‌فهمد و یک درخواست شماتیک ارسال می‌کند.

کارت نمایندگی چیست؟

کارت عامل را به عنوان یک کارت ویزیت دیجیتال یا "گواهینامه رانندگی" برای هوش مصنوعی در نظر بگیرید. وقتی یک سرور A2A شروع به کار می‌کند، این شیء JSON را منتشر می‌کند که شامل موارد زیر است:

  • هویت: نام عامل ( architect_agent ) و شناسه آن.
  • شرح: خلاصه‌ای قابل فهم برای انسان و ماشین از کاری که انجام می‌دهد ("نقش سیستم: رابط برنامه‌نویسی کاربردی پایگاه داده...").
  • رابط: کلیدهای ورودی خاص ( drive_name ) و فرمت‌های خروجی مورد انتظار آن.

بدون این کارت، مأمور اعزام کورکورانه عمل می‌کرد و حدس می‌زد که چگونه با معمار ارتباط برقرار کند.

کد سرور را ایجاد کنید

👉✏️ در ویرایشگر خود، در زیر پوشه $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 را مشاهده کردید، معمار فعال است، پروتکل A2A فعال است و کارت عامل آماده است تا توسط Dispatcher کشف شود.

اکنون که معمار آماده است تا به عنوان یک منبع از راه دور عمل کند، می‌توانیم آن را به Dispatch Agent متصل کنیم.

👉💻 برای خروج از سرور A2A، Ctrl+C را فشار دهید.

۴. اتصال عامل BIDI-Streams به عامل از راه دور و ابزارهای استریمینگ

اکنون مرکز ارتباط اصلی را پیکربندی خواهید کرد تا شکاف بین داده‌های زنده و معمار از راه دور را پر کند. این اتصال به یک خط لوله با پهنای باند بالا و تأخیر کم نیاز دارد تا از پایداری میز مونتاژ در حین کار اطمینان حاصل شود.

درک عوامل پخش دو طرفه (زنده)

استریمینگ دوطرفه (Bidi) در ADK، قابلیت تعامل صوتی و تصویری دوطرفه و با تأخیر کمِ Gemini Live API را به عوامل هوش مصنوعی اضافه می‌کند. این قابلیت، یک تغییر اساسی در تعاملات سنتی هوش مصنوعی را نشان می‌دهد. به جای الگوی سفت و سخت «پرسیدن و انتظار»، ارتباط دوطرفه و بلادرنگ را امکان‌پذیر می‌کند که در آن هم انسان و هم هوش مصنوعی می‌توانند همزمان صحبت کنند، گوش دهند و پاسخ دهند.

به تفاوت بین ارسال ایمیل و مکالمه تلفنی فکر کنید. تعاملات سنتی اپراتورها مانند ایمیل است: شما یک پیام کامل ارسال می‌کنید، منتظر پاسخ کامل هستید و سپس پیام دیگری ارسال می‌کنید. اما Bidi-streaming مانند یک مکالمه تلفنی است: روان، طبیعی، با قابلیت قطع کردن، شفاف‌سازی و پاسخ دادن در لحظه.

ویژگی‌های کلیدی:

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

چرخه حیات

👀 قبل از پیاده‌سازی منطق کلاینت، بیایید اسکلت از پیش تولید شده برای Dispatch Agent را بررسی کنیم. این Agent از طریق صدا و تصویر با کاربر ارتباط برقرار می‌کند و کوئری‌ها را به Architect Agent واگذار می‌کند. 

__init__.py
agent.py
hazard_db.py
  • agent.py : این "مغز" است. در حال حاضر شامل تنظیمات اولیه‌ی Bidi-streaming است. ما این فایل را برای اضافه کردن منطق کلاینت A2A تغییر خواهیم داد تا بتواند با معمار ارتباط برقرار کند.
  • hazard_db.py : این یک ابزار محلی مختص Dispatch Agent است که شامل پروتکل‌های ایمنی می‌باشد. این ابزار جدا از پایگاه داده شماتیک معمار است.

پیاده‌سازی کلاینت A2A

برای اینکه به مامور اعزام اجازه دهیم با معمار راه دور ما ارتباط برقرار کند، باید یک مامور راه دور 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، خطرات منتظر نمی‌مانند تا شما بپرسید که آیا آنها وجود دارند یا خیر. برای این کار، به یک ابزار Streaming نیاز دارید.

جریان ابزار جریان

ابزارهای استریمینگ به توابع اجازه می‌دهند تا نتایج میانی را به صورت بلادرنگ به عامل (agent) ارسال کنند و عامل را قادر می‌سازند تا به محض وقوع تغییرات، به آنها واکنش نشان دهد. موارد استفاده رایج شامل نظارت بر نوسانات قیمت سهام یا در مورد ما، نظارت بر یک استریم ویدیویی زنده برای تغییرات وضعیت است.

برخلاف ابزارهای استاندارد، یک ابزار استریمینگ یک تابع ناهمزمان است که به عنوان یک AsyncGenerator عمل می‌کند. این بدان معناست که به جای return یک مقدار واحد، چندین به‌روزرسانی را در طول زمان yield .

برای تعریف یک ابزار استریمینگ در ADK، باید این الزامات فنی را رعایت کنید:

  1. تابع ناهمزمان: این ابزار باید با async def تعریف شود.
  2. نوع بازگشتی AsyncGenerator: برای بازگرداندن یک AsyncGenerator ، باید نوع تابع را مشخص کرد. پارامتر اول نوع داده‌ای است که قرار است تولید شود (مثلاً str ) و پارامتر دوم معمولاً None است.
  3. جریان‌های ورودی: ما از ابزارهای پخش ویدئو استفاده می‌کنیم. در این حالت، جریان ویدئو/صوت واقعی ( LiveRequestQueue ) مستقیماً به تابع ارسال می‌شود و به ابزار اجازه می‌دهد همان فریم‌هایی را که عامل می‌بیند، "ببیند".

یک ابزار استریمینگ را به عنوان یک نگهبان (Sentinel) در نظر بگیرید. در حالی که شما و مامور اعزام در حال بحث در مورد طرح‌ها هستید، نگهبان در پس‌زمینه در حال اجرا است و بی‌صدا هر فریم ویدیو را پردازش می‌کند تا امنیت شما را تضمین کند.

ابزار استریمینگ

پیاده‌سازی ابزار نظارت بر پس‌زمینه

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

مفهوم: عامل به عنوان ابزار در مقابل زیرعامل‌ها

هنگام ساخت سیستم‌های چندعاملی، باید تصمیم بگیرید که مسئولیت چگونه به اشتراک گذاشته شود. در ماموریت نجات ما، تمایز بین این موارد بسیار مهم است:

  • عامل به عنوان ابزار: این رویکرد پیشنهادی برای هاب استریمینگ bidi ما است. وقتی عامل اعزام (عامل A) عامل معمار (عامل B) را به عنوان ابزار فراخوانی می‌کند، داده‌های معمار به Dispatch بازگردانده می‌شود. سپس Dispatch آن داده‌ها را تفسیر کرده و پاسخی برای شما تولید می‌کند. Dispatch کنترل را در دست می‌گیرد و به مدیریت تمام ورودی‌های بعدی کاربر ادامه می‌دهد.
  • زیر-عامل: در یک رابطه زیر-عامل، مسئولیت کاملاً منتقل می‌شود. اگر Dispatch شما را به عنوان زیر-عامل به معمار تحویل دهد، شما مستقیماً با یک API پایگاه داده صحبت خواهید کرد که هیچ "بینش" و مهارت مکالمه‌ای ندارد. عامل اصلی (Dispatch) عملاً از حلقه خارج خواهد شد.

کنترل

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

کدنویسی منطق مسیریابی

اکنون architect_agent خود را در یک AgentTool قرار می‌دهیم و یک "نقشه منطقی" به عامل اعزام ارائه می‌دهیم. این نقشه دقیقاً به عامل می‌گوید چه زمانی داده‌ها را از گاوصندوق دریافت کند و چه زمانی یافته‌ها را از نگهبان پس‌زمینه گزارش دهد.

برای اینکه به Dispatch «چشم‌هایی» بدهیم که هرگز پلک نزنند، باید به آن دسترسی به Streaming Tool که در مرحله قبل ساختیم را بدهیم.

In ADK, when you add an AsyncGenerator function (like monitor_for_hazard ) to the tools list, the agent treats it as a persistent background process. Instead of a one-time execution, the agent "subscribes" to the tool's output. This allows Dispatch to continue its primary conversation while the Sentinel silently yields hazard alerts in the background.

👉✏️ #REPLACE_AGENT_TOOLS در $HOME/way-back-home/level_4/backend/dispatch_agent/agent.py با موارد زیر جایگزین کنید: 

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

تأیید

👉💻 با پیکربندی هر دو عامل، می‌توانیم تعامل چندعاملی را به صورت زنده آزمایش کنیم.

  • در ترمینال A، Architect Agent را اجرا کنید: 
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)، Dispatch Agent را اجرا کنید: 
cd $HOME/way-back-home/level_4/backend/
cp architect_agent/.env .env
uv run adk web

آزمایش یک سیستم چندعامله که از یک مدل چندوجهی و بلادرنگ مانند gemini-live در شبیه‌ساز adk web استفاده می‌کند، شامل یک گردش کار خاص است. این شبیه‌ساز برای بررسی فراخوانی‌های ابزار عالی است، اما هنگام پردازش اولیه تصاویر با این نوع مدل، ناسازگاری شناخته‌شده‌ای دارد.

  • روی نماد پیش‌نمایش وب در نوار ابزار Cloud Shell کلیک کنید. گزینه Change port را انتخاب کنید، آن را روی ۸۰۰۰ تنظیم کنید و روی Change and Preview کلیک کنید.

👉 dispatch_agent را انتخاب کنید و طرح اولیه را آپلود کنید و خطای مورد انتظار را مدیریت کنید

این مهم‌ترین مرحله است. ما باید زمینه تصویر را برای عامل فراهم کنیم.

  • وقتی رابط کاربری بارگذاری شد، در صورت درخواست، به آن اجازه دسترسی به میکروفون خود را بدهید .
  • این تصویر طرح اولیه را روی رایانه خود دانلود کنید: نمونه طرح اولیه
  • در رابط adk web ، روی آیکون گیره کاغذ کلیک کنید و تصویر طرح اولیه‌ای که دانلود کرده‌اید را آپلود کنید. اضافه کردن فایل

⚠️⚠️شما خطای ۴۰۰ INVALID_ARGUMENT را مشاهده خواهید کرد. این مورد قابل پیش‌بینی بود.⚠️⚠️

پیام خطای مورد انتظار

این خطا به این دلیل رخ می‌دهد که کنترل‌کننده تصویر adk web به طور کامل با API مدل gemini-live برای آپلود یک‌باره سازگار نیست. با این حال، تصویر با موفقیت به متن جلسه اضافه شده است .

  • 👉 برای پاک کردن خطا، کافیست صفحه مرورگر را مجدداً بارگذاری کنید .

فرآیند مونتاژ را آغاز کنید

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

  • برای روشن کردن میکروفون، روی آیکون آن کلیک کنید. رابط کاربری عبارت «در حال گوش دادن...» را نشان می‌دهد.
  • دستور صوتی را بگویید: «شروع به مونتاژ کنید» .
  • اپراتور درخواست شما را پردازش می‌کند و رابط کاربری به «در حال صحبت...» تغییر می‌کند. شما باید یک پاسخ صوتی بشنوید که بخش‌های مورد نیاز را فهرست می‌کند.

پاسخ سخنگوی مامور

۴. فراخوانی‌های ابزار بین عامل‌ها را تأیید کنید

👉 پاسخ صوتی اولیه تأیید می‌کند که سیستم کار می‌کند، اما جادوی واقعی در ردیابی ارتباط چندعاملی است.

  • میکروفون را خاموش کنید.
  • یه بار دیگه صفحه رو رفرش کن.

اکنون پنل "Trace" در سمت چپ پر خواهد شد. می‌توانید جریان اجرای کامل و موفق را مشاهده کنید:

  • dispatch_agent ابتدا monitor_for_hazard را فراخوانی می‌کند.
  • سپس، چندین فراخوانی execute_architect به architect_agent انجام می‌دهد تا داده‌های شماتیک را بازیابی کند.

تأیید تماس ابزار

این توالی تأیید می‌کند که کل گردش کار چندعاملی به درستی کار می‌کند: dispatch_agent درخواست را دریافت کرد، وظیفه بازیابی داده‌ها را از طریق یک فراخوانی ابزار به architect_agent واگذار کرد و داده‌ها را برای انجام دستور کاربر دریافت کرد.

لینک bidi-streaming شما اکنون قادر به نظارت بر پس‌زمینه و همکاری چندعاملی است. در ادامه، یاد خواهیم گرفت که چگونه این پاسخ‌های پیچیده را در سمت کاربر تجزیه کنیم.

👉💻 برای خروج، در هر دو ترمینال Ctrl+c را فشار دهید.

۵. نگاهی عمیق به پخش زنده رویدادهای چندوجهی

در مرحله قبل، ما با موفقیت سیستم چندعاملی خود را با استفاده از سرور توسعه داخلی، adk web تأیید کردیم. این ابزار از یک اجراکننده ADK پیش‌فرض برای مدیریت خودکار جلسه، جریان‌ها و چرخه عمر عامل استفاده می‌کند. با این حال، برای ایجاد یک برنامه مستقل و آماده برای تولید مانند سرویس FastAPI ما ( main.py )، به کنترل صریح نیاز داریم. ما باید ADK Runner را به صورت دستی ایجاد و مدیریت کنیم تا جلسات زنده کاربر را مدیریت کند، زیرا این مؤلفه اصلی است که جریان‌های دو طرفه صدا، ویدئو و متن را پردازش می‌کند.

حلقه مدل-کد-مدل

برای درک نحوه عملکرد سیستم در حالت بلادرنگ، بیایید چرخه حیات یک جلسه ماموریت واحد را دنبال کنیم. این حلقه نشان‌دهنده تبادل مداوم اشیاء LlmRequest و LlmResponse است.

  1. پیوند بصری: شما اتصال را آغاز می‌کنید و وب‌کم/صفحه نمایش خود را به اشتراک می‌گذارید. فریم‌های JPEG با کیفیت بالا از طریق realtimeInput (با استفاده از LiveRequestQueue ) به بالادست جریان می‌یابند.
  2. فعال‌سازی نگهبان: سیستم یک محرک اولیه "سلام" ارسال می‌کند. طبق دستورالعمل‌هایش، مامور اعزام بلافاصله ابزار پخش monitor_for_hazard را فعال می‌کند. این یک حلقه پس‌زمینه را آغاز می‌کند که بی‌صدا هر فریم ورودی را تماشا می‌کند.
  3. فرمان خلبان: شما در ارتباطات می‌گویید: «شروع به جمع‌آوری کنید.»
  4. Vocal Upstream: صدای شما به صورت صوتی ۱۶ کیلوهرتز ضبط شده و در کنار فریم‌های ویدیویی به صورت Upstream ارسال می‌شود.
  5. واگذاری (A2A): دیسپچ نیت شما را «می‌شنود». متوجه می‌شود که طرحواره‌ها را ندارد، بنابراین با استفاده از پروتکل AgentTool (عامل به عنوان ابزار) با عامل معمار تماس می‌گیرد.
  6. بازیابی اطلاعات: معمار از پایگاه داده Redis پرس و جو می‌کند و لیست قطعات را به Dispatch برمی‌گرداند. Dispatch همچنان "مدیر جلسه" باقی می‌ماند و داده‌ها را بدون تحویل دادن به شما دریافت می‌کند.
  7. اطلاعات پایین‌دستی: دیسپچ یک modelTurn (پایین‌دستی) حاوی متن و صدای اصلی ارسال می‌کند: "معمار تأیید شد. زیرمجموعه مورد نیاز: Warp Core، Flux Pipe، Ion Thruster است."
  8. بحران: ناگهان، بخشی از میز کار بی‌ثبات می‌شود و شروع به درخشش سفید می‌کند.
  9. تشخیص خودکار: حلقه monitor_for_hazard در پس‌زمینه (Sentinel) فریم JPEG خاص حاوی تابش را دریافت می‌کند. این فریم را با فراخوانی Gemini پردازش کرده و خطر را شناسایی می‌کند.
  10. ایمنی در پایین‌دست: ابزار استریمینگ نتیجه‌ای را yields . از آنجا که این یک عامل Bidi-Streaming است، Dispatch می‌تواند وضعیت فعلی آن را قطع کند تا فوراً یک هشدار ایمنی حیاتی در پایین‌دست ارسال کند: "خطر شناسایی شد! اکنون کریستال داده را خنثی کنید. آن را به سطل قرمز منتقل کنید."

جریان

تنظیم پیکربندی زمان اجرای عامل

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

پیاده‌سازی درخواست به نماینده

در مرحله بعد، ما آپ‌لینک ارتباطی اصلی را پیاده‌سازی خواهیم کرد که داده‌های چندوجهی و بلادرنگ را از Volatile Workbench کاربر از طریق یک WebSocket به Dispatch Agent ارسال می‌کند. این کار باعث می‌شود که Agent به طور مداوم "ببیند" (فریم‌های ویدیویی) و "بشنود" (دستورات صوتی) باشد. منطق به طور مداوم جریان داده‌ها را دریافت می‌کند، بین تکه‌های صوتی باینری ورودی و بسته‌های متنی/تصویری پیچیده شده در JSON تمایز قائل می‌شود و آن را در اشیاء Blob (برای چندرسانه‌ای) یا Content (برای متن) کپسوله می‌کند و آن را به LiveRequestQueue ارسال می‌کند تا جلسه دو طرفه Agent را فعال کند.

بیدی

عبارت #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 اجرا می‌کنید، داده‌هایی که از عامل برمی‌گردند در نوع خاصی از Event بسته‌بندی می‌شوند که از ساختارهای اصلی GenAI SDK ارث‌بری می‌کند. شیء Event که در حلقه async for event in runner.run_live(...) دریافت می‌کنید، یک شیء واحد است که شامل چندین فیلد اختیاری است که هر کدام برای نوع متفاوتی از اطلاعات هستند:

رویداد

نحوه ساختار محتوا:

  • وقتی عامل صحبت می‌کند (از طریق .server_content ): این فیلد فقط متن ساده نیست. بلکه شامل لیستی از Parts است. هر Part ، ظرفی برای یک نوع داده است - یا یک رشته متنی (مانند "The part is stable." ) یا یک حباب صوتی خام (صدا).
  • وقتی عامل عمل می‌کند (از طریق .tool_call ): این فیلد شامل لیستی از اشیاء FunctionCall است. هر FunctionCall یک شیء ساده و ساختاریافته است که نام ابزار و آرگومان‌های ورودی را در قالبی تمیز مشخص می‌کند که کد backend شما به راحتی می‌تواند آن را بخواند و اجرا کند.

👀 If you were to look at a single Event yielded by the run_live loop, the JSON (produced by event.model_dump(by_alias=True) ) would look like this, strictly following the GenAI SDK shapes: 

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

👉✏️ We will now update the downstream_task in main.py to forward the complete event data. This logic ensures that every "thought" the AI has is logged in the ship's diagnostic terminal and sent as a single JSON object to the frontend UI.

Locate the #PROCESS_AGENT_RESPONSE placeholder in your $HOME/way-back-home/level_4/backend/main.py file and replace it with the following dissection logic: 

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

Mission Execution

With the backend vault connected and both agents configured, all systems are now mission-ready. The following steps will launch the full application, allowing you to interact with the two-agent system you just built.

Objective: Assemble the randomly assigned warp drive that appears on your workbench. Protocol: You must follow the vocal guidance of the Dispatch Agent, especially the hazard warnings for specific components.

Activate the Specialist (The Architect)

👉💻 In your first terminal window , launch the Architect agent. This backend service will connect to the Redis vault and wait for schematic requests from the 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

(Leave this terminal running. It is now your active "database agent.")

Launch the Cockpit (The Dispatcher)

👉💻 In a new terminal window (Terminal B), we will build the frontend UI and start the main Dispatch agent, which serves the user interface and handles all live communication. 

# 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

(This starts the primary server on Port 8080.)

Run the Test Scenario

The system is now live. Your goal is to follow the agent's instructions to complete the assembly.

  1. 👉 Access the Workbench:
    • Click the Web preview icon in the Cloud Shell toolbar.
    • Select Change port , set it to 8080 , and click Change and Preview .
  2. 👉 Start the Mission:
    • When the interface loads, make sure you allow it to access your screen and microphone. پنجره
    • You will be ask to select a tab or a window to share, if you are sharing the window, to avoid problem, make sure this is the ONLY tab in the window.
    • A drive with a random name (eg, "NOVA-V", "OMEGA-9") will be assigned to you.
  3. 👉 The Assembly Loop:
    • Request: To start assembling the drive say: "Start assembling." Assemble
    • Architect Respond: The agent will provide the correct parts to assemble the drive.
    • Hazard Check: When a part appears to be hazardous on the workbench:
      • The Dispatch agent's monitor_for_hazard tool will visually identify it.
      • It will yield a "VISUAL HAZARD ALERT". (This will take about 30 sec)
      • It will check which bin to use to disengage the hazard. خطر
    • Action: The Dispatch Agent will give you a direct command: "Hazard Confirmed. Place XXX in the Red bin immediately." You must follow this instruction to proceed.

Mission Accomplished. You have successfully built an interactive, multi-agent system. The survivors are safe, the rocket has cleared the atmosphere, and your "Way Back Home" continues.

👉💻 Press Ctrl+c in both terminal to exit.

6. Deploy to Production (Optional)

You have successfully tested the agent locally. Now, we must upload the Architect's neural core to the ship's mainframes (Cloud Run). This will allow it to operate as a permanent, independent service that the Dispatch agent can query from anywhere.

نمای کلی

Provision the Secure Vault (Infrastructure)

Before deploying the agent, we must create its persistent memory (Memorystore) and the secure channel to access it (VPC Connector).

👉💻 Create the Memorystore Instance (Redis Vault): 

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

👉💻 Retrieve the Vault's Network Address: Execute this command and copy the host IP address. This is the private address of your new Redis instance. 

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

👉💻 Create the VPC Access Connector (Secure Bridge): This connector acts as a private bridge, allowing Cloud Run to access the Redis instance inside your 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

👉💻 Load the data: 

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

Deploy the Agent Application

Compile and Build Agent Image

👉💻 Navigate to the backend directory and create the 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

👉💻 Package the application into a container image. 

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}

Deploy to Cloud Run

👉💻 Deploy the agent to Cloud Run. We will inject the Redis IP and link the VPC Connector directly into the launch command. This ensures the agent starts with a secure, private connection to its database. 

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"

👉💻 Verify if the A2A server is running. 

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

Once the command finishes, you will see a Service URL . The Architect Agent is now live in the cloud, permanently connected to its vault and ready to serve schematic data to other agents.

Deploy Dispatch Hub to Production Mainframe

With the Architect Agent operational in the cloud, we must now deploy the Dispatch Hub. This agent will serve as the primary user interface, handling live voice/video streams and delegating database queries to the Architect's secure endpoint.

👉💻 Run the following command in your Cloud Shell terminal. It will create the complete, multi-stage Dockerfile in your backend directory. 

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

Compile and Build Agent/Frontend Image

👉💻 Navigate to the backend directory containing the Dispatch agent's code ( main.py ) and package it into a container image. 

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}

Deploy to Cloud Run

👉💻 Deploy the Dispatch Hub to Cloud Run. We will inject the Architect's URL as an environment variable, creating the critical link between our two cloud-native agents. 

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

Once the command finishes, you will see a Service URL (eg, https://mission-bravo-...run.app ). The application is now live in the cloud.

👉 Go to the Google Cloud Run page and select the biometric-scout service from the list. CloudRun

👉 Locate the Public URL displayed at the top of the Service details page. CloudRun

Final Systems Check (End-to-End Test)

👉 Now you will interact with the live system.

  1. Get the URL: Copy the Service URL from the output of the last deployment command (it should end with run.app ).
  2. Open the Cockpit: Paste the URL into your web browser.
  3. Initiate Contact: When the interface loads, make sure you allow it to access your screen and microphone.
  4. Request Data: When a drive is assigned, ask to start assembling. For example: "Start to assemble"

CloudRun

You are now interacting with a fully deployed, multi-agent system running entirely on Google Cloud.

The Multi-agent system locks the final containment ring into place, and the erratic radiation flatlines into a steady hum.

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

Ending

On your monitor, the alien ship streaks upward, narrowly escaping the crumbling surface of Ozymandias as the atmosphere collapses. It settles into a safe orbit alongside your vessel, and the comms fill with the voices of the survivors—shaken but alive. With the rescue complete and your path home clear, the remote link severs.

Thanks to you, the survivors are rescued.

If you participated in Level 0, don't forget to check where your progress is on the way back home mission!

نهایی