Way Back Home - Live Bidirectional Multi-Agent system

1. المهمة

أنت تائه في الفضاء الشاسع الصامت غير المحدّد. تسبّبت نبضة شمسية هائلة في تمزيق سفينتك عبر صدع بُعدي، ما أدى إلى تقطُّع السُبل بك في جزء من الكون لا يتضمّن أي خريطة نجمية.

بعد أيام من أعمال الإصلاح الشاقة، عاد أخيرًا صوت المحركات المألوف. أصبحت مركبتك الفضائية جاهزة للعمل. لقد تمكّنت حتى من إنشاء وصلة صاعدة بعيدة المدى إلى السفينة الأم. موعد المغادرة وشيك. يمكنك الآن العودة إلى المنزل.

ولكن بينما تستعد لتشغيل محرك الأقراص، يقطع إشارة استغاثة التشويش. ترصد أجهزة الاستشعار لديك إشارة استغاثة من كوكب يُعرف باسم "أوزيماندياس". يجد الناجون أنفسهم عالقين في هذا العالم المحتضر، بعد أن تحطمت سفينتهم. مهمتك بالغة الأهمية: عليك إنقاذهم قبل أن ينهار الغلاف الجوي للكوكب.

وسيلتهم الوحيدة للهروب هي صاروخ قديم مهجور تم تصنيعه باستخدام تكنولوجيا فضائية، وهو يعمل ولكن محرك الانتقال الفضائي الخاص به مكسور. لإنقاذ الناجين، عليك الاتصال عن بُعد بـ Volatile Workbench وتجميع محرك بديل يدويًا.

التحدي

ليس لديك خبرة في هذه التكنولوجيا الفضائية، وهي معروفة بأنّها هشة للغاية. ويمكن أن يصبح المكوّن غير المستقر خطرًا إشعاعيًا في غضون ثوانٍ. لديك محاولة واحدة لتشغيل Volatile Workbench. يواجه مساعدك الحالي المستند إلى الذكاء الاصطناعي صعوبة في معالجة البيانات المرئية والأدلّة الفنية في الوقت نفسه، ما يؤدي إلى تقديم تعليمات غير صحيحة وتجاهل تحذيرات المخاطر.

لتحقيق النجاح، يجب ترقية الذكاء الاصطناعي من كيان واحد إلى نظام متعدد الوكلاء تعاوني.

أهداف مهمتك:

يمكنك تجميع Warp Drive باتّباع التعليمات المتخصّصة في الوقت الفعلي من نظامك الجديد المتعدّد الوكلاء.

ما ستنشئه

• نظام ذكاء اصطناعي متعدد الوكلاء وثنائي الاتجاه وفي الوقت الفعلي، ويضم وكيل إرسال مركزيًا يدير تفاعل المستخدم وينسّق مع الوكلاء المتخصصين.
• وكيل Architect يتصل بقاعدة بيانات Redis لاسترداد البيانات التخطيطية وعرضها
• ميزة "مراقبة السلامة" الاستباقية التي تستخدم أدوات البث لتحليل خلاصة فيديو مباشرة بحثًا عن المخاطر المرئية وإرسال تنبيهات في الوقت الفعلي
• واجهة مستخدم مستندة إلى React تتيح التفاعل مع النظام، وبث الفيديو والصوت إلى وكلاء الخلفية

ما ستتعلمه

التكنولوجيا / المفهوم	الوصف
Google Agent Development Kit (ADK)	ستستخدم "حزمة تطوير البرامج" لإنشاء الوكلاء واختبارهم وإدارتهم، والاستفادة من إطار عملها في التعامل مع الاتصالات في الوقت الفعلي ودمج الأدوات ومراحل نشاط الوكيل.
البث ثنائي الاتجاه	ستنفّذ وكيلاً لبث البيانات في اتجاهين يتيح التواصل الطبيعي في اتجاهين وبزمن انتقال منخفض، ما يتيح لكل من الإنسان والذكاء الاصطناعي مقاطعة المحادثة والرد في الوقت الفعلي.
الأنظمة المتعددة الوكلاء	ستتعلّم كيفية تصميم نظام ذكاء اصطناعي موزّع يفوّض فيه وكيل رئيسي المهام إلى وكلاء متخصصين، ما يتيح فصل الاهتمامات وبنية أكثر قابلية للتوسّع.
بروتوكول Agent-to-Agent (A2A)	ستستخدم بروتوكول A2A لتفعيل التواصل بين "وكيل الإرسال" و"وكيل المصمم"، ما يتيح لهما التعرّف على إمكانات بعضهما البعض وتبادل البيانات.
أدوات البث	عليك تنفيذ أداة بث تعمل كعملية في الخلفية، وتُحلّل باستمرار خلاصة فيديو لرصد التغييرات في الحالة (المخاطر) وتقديم النتائج بشكل استباقي.
‫Google Cloud Run وMemorystore	ستنشر تطبيق الوكلاء المتعددين بالكامل في بيئة إنتاج، باستخدام Cloud Run لاستضافة خدمات الوكيل وMemorystore (Redis) كقاعدة بيانات دائمة.
FastAPI وWebSockets	تم إنشاء الخلفية باستخدام FastAPI وWebSockets للتعامل مع الاتصالات العالية الأداء وفي الوقت الفعلي المطلوبة لبث الصوت والفيديو وردود الوكيل.
React Frontend	ستعمل على واجهة مستخدم أمامية مستندة إلى React تسجّل وسائط المستخدم (الصوت/الفيديو) وتبثّها وتعرض الردود في الوقت الفعلي من وكلاء الذكاء الاصطناعي.

2. إعداد البيئة

هل تحتاج إلى أرصدة Google Cloud؟

• إذا كنت تحضر ورشة عمل بإشراف معلّم: سيقدّم لك المعلّم رمز رصيد. يُرجى استخدام الرمز الذي يقدّمه لك.

• إذا كنت تعمل على هذا الدرس التطبيقي بنفسك: استخدام حساب Gmail شخصي: يجب استخدام حساب شخصي (مثل [e-mail]). لن تعمل الحسابات التي تديرها الشركات أو المؤسسات التعليمية .

👉 الخطوات:

1. انتقِل إلى الموقع الإلكتروني لاسترداد الرصيد: غير متاح في الوقت الحالي
2. تسجيل الدخول: الصِق الرابط في شريط العناوين وسجِّل الدخول باستخدام حسابك الشخصي على Gmail.
3. قبول البنود: وافِق على بنود خدمة Google Cloud Platform.
4. التحقّق من الرصيد: ابحث عن رسالة تؤكّد أنّه تم تطبيق الرصيد.
   • *ملاحظة: إذا طُلب منك إدخال معلومات بطاقة الائتمان، يمكنك تجاهل ذلك بأمان وإغلاق النافذة.

يمكنك الآن إغلاق النافذة.

الوصول إلى Cloud Shell

👉انقر على "تفعيل Cloud Shell" في أعلى "وحدة تحكّم Google Cloud" (رمز شكل الوحدة الطرفية في أعلى لوحة Cloud Shell)،

👉انقر على الزر "فتح المحرّر" (يبدو كملف مفتوح مع قلم رصاص). سيؤدي ذلك إلى فتح "محرِّر Cloud Shell للرموز" في النافذة. سيظهر لك مستكشف الملفات على الجانب الأيمن.

👉افتح المحطة الطرفية في بيئة التطوير المتكاملة المستندة إلى السحابة الإلكترونية.

👉💻 في نافذة الوحدة الطرفية، تأكَّد من أنّك قد أثبتّ هويتك وأنّ المشروع مضبوط على رقم تعريف مشروعك باستخدام الأمر التالي:

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. انقر على القائمة المنسدلة أداة اختيار المشروع في أعلى الصفحة.
3. انقر على علامة التبويب "الكل" (قد لا يظهر مشروعك الجديد في "الأخيرة" بعد).
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	إطار عمل ويب عالي الأداء لبث محتوى Satellite Station وSSE
uvicorn	مطلوب خادم ASGI لتشغيل تطبيق FastAPI
google-adk	مجموعة أدوات تطوير الوكيل المستخدَمة لإنشاء Formation Agent
a2a-sdk	مكتبة بروتوكولات للتواصل الموحّد بين العملاء
google-genai	برنامج أصلي للوصول إلى نماذج Gemini
redis	برنامج عميل Python للاتصال بخدمة Schematic Vault (Memorystore)
websockets	إتاحة التواصل الثنائي الاتجاه في الوقت الفعلي
python-dotenv	إدارة متغيرات البيئة وأسرار الإعدادات
pydantic	التحقّق من صحة البيانات وإدارة الإعدادات

التحقّق من الإعداد

قبل أن نبدأ في كتابة الرمز، لنحرص على أن تكون جميع الأنظمة جاهزة. نفِّذ نص التحقّق البرمجي لتدقيق مشروع Google Cloud وواجهات برمجة التطبيقات وتبعيات Python.

👉💻 تشغيل نص التحقّق البرمجي:

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

👀 ستظهر لك سلسلة من علامات الصح الخضراء (✅).

• إذا ظهرت لك علامات X حمراء (❌)، اتّبِع أوامر الإصلاح المقترَحة في الناتج (مثل ‫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- إنشاء مستودع المخططات التوضيحية في Redis ووكيل ثنائي الاتجاه باستخدام "حزمة تطوير الوكلاء"

لقد عثرت على مستودع المخططات الكوكبية الذي يحتوي على مخططات الصاروخ المهجور. لاسترداد هذه البيانات بدقة، يجب التفاعل مع واجهة الإدارة المخصّصة للمستودع: Architect agent.

توفير مستودع المخططات (Redis)

قبل أن يتمكّن Architect من مساعدتنا، يجب التأكّد من استضافة البيانات في بيئة آمنة وعالية التوفّر. سنستخدم Redis كمخزن بيانات سريع لمخططات الكائنات الفضائية. لتسهيل عملية التطوير، سننشئ مثيلاً محليًا من Redis، ولكن سنقدّم لاحقًا تعليمات حول كيفية التفعيل في بيئة دعم الإنتاج باستخدام Google Cloud Memorystore.

👉💻 نفِّذ الأوامر التالية في الوحدة الطرفية لتوفير مثيل Redis (قد يستغرق ذلك من دقيقتَين إلى 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 Agent

وكيل المهندس المعماري هو وكيل متخصص مسؤول عن استرداد المخططات التخطيطية من مستودع Redis. تعمل هذه الخدمة كواجهة مخصّصة للبيانات، ما يضمن حصول "وكيل الإرسال" الرئيسي على معلومات دقيقة ومنظَّمة بدون الحاجة إلى معرفة منطق قاعدة البيانات الأساسية.

حزمة تطوير وكلاء Google (ADK) هي إطار عمل نموذجي يتيح إعداد الوكلاء المتعدّدين. تتعامل هذه المنصة مع طبقتَين أساسيتَين:

1. دورة حياة الاتصال والجلسة: يتطلّب التفاعل مع واجهات برمجة التطبيقات في الوقت الفعلي إدارة معقّدة للبروتوكول، بما في ذلك التعامل مع المصافحات والمصادقة وإشارات إبقاء الاتصال نشطًا.
2. استدعاء الدوال: هذه هي "رحلة النموذج والرمز البرمجي والنموذج". عندما يقرّر النموذج اللغوي الكبير أنّه بحاجة إلى بيانات، يُصدر طلبًا منظَّمًا لاستدعاء دالة. يعترض ADK على ذلك، وينفّذ رمز Python (lookup_schematic_tool)، ويُعيد النتيجة إلى سياق النموذج في غضون أجزاء من الثانية.

سننشئ الآن Architect. لا يمكن لهذا الوكيل الوصول إلى الكاميرا. وهي موجودة فقط لتلقّي "اسم محرك أقراص" وعرض "قائمة الأجزاء" من قاعدة البيانات.

👉💻 سنستخدم الأمر adk create. هذه أداة من "حزمة تطوير الوكيل" (ADK) تنشئ تلقائيًا رمز النص النموذجي وبنية الملفات لوكيل جديد، ما يوفّر وقت الإعداد.

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

ضبط إعدادات "الوكيل"

ستطلق واجهة سطر الأوامر معالج إعداد تفاعليًا. استخدِم الردود التالية لإعداد برنامجك الآلي:

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

مزايا "حزمة تطوير التطبيقات"

مع توفّر Architect على الإنترنت، أصبح لدينا الآن مصدر موثوق للمعلومات. قبل ربط هذا النموذج بالوكيل الأساسي، يوفّر "حزمة تطوير الوكيل" (ADK) ميزة كبيرة من خلال تبسيط تعقيدات إنشاء وكلاء الذكاء الاصطناعي واختبارهم. باستخدام adk web وحدة تحكّم المطوّرين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. انقر على تغيير المنفذ، واضبطه على 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]

إذا ظهر لك سجلّ تنفيذ الأداة وردّ البيانات النظيفة، يعني ذلك أنّ الوكيل المتخصّص يعمل على النحو المطلوب. يمكنه معالجة الطلبات والاستعلام عن الخزنة وعرض البيانات المنظَّمة.

👉💻 اضغط على Ctrl+C للخروج.

إعداد خادم A2A

لربط "وكيل الإرسال" بـ "المصمّم"، نستخدم بروتوكول الوكيل إلى الوكيل (A2A).

في حين تركّز بروتوكولات مثل MCP (بروتوكول Model Context) على ربط النماذج الوكيلة بالأدوات، يركّز بروتوكول A2A على ربط النماذج الوكيلة بغيرها من النماذج الوكيلة. وهو المعيار الذي يتيح لبرنامج Dispatcher "اكتشاف" Architect وفهم قدرته على البحث عن المخططات.

مسار A2A: في هذه المهمة، نستخدم نموذج خادم-عميل:

1. الخادم (المصمّم): يستضيف أدوات قاعدة البيانات و "يعلن" عن مهاراته من خلال بطاقة وكيل.
2. العميل (الإرسال): يقرأ بطاقة Architect ويفهم واجهة برمجة التطبيقات الخاصة بها ويرسل طلبًا تخطيطيًا.

ما هي بطاقة الوكيل؟

يمكنك اعتبار بطاقة الوكيل بمثابة بطاقة عمل رقمية أو "رخصة قيادة" للذكاء الاصطناعي. عندما يبدأ خادم 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)

تأكيد بطاقة الوكيل

افتح علامة تبويب جديدة في الوحدة الطرفية (انقر على الرمز +). سنتأكّد من أنّ Architect يبث هويته بشكل صحيح من خلال استرداد بطاقة الوكيل يدويًا.

👉💻 شغِّل الأمر التالي:

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 نشط، وبطاقة Agent Card جاهزة ليتم العثور عليها من خلال Dispatcher.

بعد أن أصبح Architect جاهزًا للعمل كمورد بعيد، يمكننا المتابعة لربطه بـ وكيل الإرسال.

👉💻 اضغط على Ctrl+C للخروج من خادم A2A.

4. ربط BIDI-Streams Agent بـ Remote Agent وأدوات البث

عليك الآن ضبط مركز التواصل الأساسي لسدّ الفجوة بين البيانات المباشرة و"المصمّم" البعيد. يتطلّب هذا الاتصال مسارًا بمعدّل نقل بيانات عالٍ ووقت استجابة منخفض لضمان بقاء طاولة التجميع ثابتة أثناء التشغيل.

فهم وكلاء البث ثنائي الاتجاه (مباشر)

تضيف ميزة البث ثنائي الاتجاه في "حزمة تطوير التطبيقات" إمكانية التفاعل الصوتي والمرئي ثنائي الاتجاه مع زمن استجابة منخفض من خلال Gemini Live API إلى وكلاء الذكاء الاصطناعي. فهو يشكّل قفزة نوعية في طريقة تفاعلنا مع الذكاء الاصطناعي. وبدلاً من نمط "السؤال والانتظار" الصارم، يتيح هذا النموذج التواصل في الوقت الفعلي وباتجاهين، حيث يمكن لكل من الإنسان والذكاء الاصطناعي التحدث والاستماع والرد في الوقت نفسه.

فكِّر في الفرق بين إرسال رسائل إلكترونية وإجراء محادثة هاتفية. تتشابه تفاعلات "الوكيل التقليدي" مع رسائل البريد الإلكتروني: أنت ترسل رسالة كاملة، وتنتظر ردًا كاملاً، ثم ترسل رسالة أخرى. يشبه البث ثنائي الاتجاه محادثة هاتفية: فهو سلس وطبيعي ويتيح مقاطعة المحادثة وطلب التوضيح والرد في الوقت الفعلي.

الخصائص الرئيسية:

• التواصل في اتجاهَين: تبادل البيانات بشكل مستمر بدون انتظار الردود الكاملة يردّ الذكاء الاصطناعي فورًا عند رصد انتهاء المستخدم من التحدث.
• المقاطعة التفاعلية: يمكن للمستخدمين مقاطعة الوكيل أثناء الردّ من خلال تقديم إدخال جديد، تمامًا كما هو الحال في المحادثة مع شخص آخر. إذا كان الذكاء الاصطناعي يشرح خطوة معقّدة وقلت "لحظة، كرِّر ذلك"، سيتوقف الذكاء الاصطناعي على الفور وسيتعامل مع طلبك.
• مُحسَّن لتعدُّد الوسائط: تتفوّق ميزة "البث الثنائي الاتجاه" في معالجة أنواع مختلفة من الإدخالات في الوقت نفسه. يمكنك التحدث إلى الموظف أثناء عرض أجزاء الجهاز الغريب عليه عبر الفيديو، وهو يعالج كلا البثّين في اتصال واحد موحّد.

👀 قبل تنفيذ منطق العميل، لنلقِ نظرة على الهيكل الذي تم إنشاؤه مسبقًا لـ Dispatch Agent. سيتواصل هذا الوكيل مع المستخدم من خلال الصوت والفيديو وسيفوّض طلبات البحث إلى "الوكيل المعماري".

__init__.py
agent.py
hazard_db.py

• agent.py: هذا هو "الدماغ". يحتوي حاليًا على إعداد أساسي للبث ثنائي الاتجاه. سنعدّل هذا الملف لإضافة منطق عميل A2A حتى يتمكّن من التواصل مع "المصمّم".
• ‫hazard_db.py: هذه أداة محلية خاصة بـ "وكيل الإرسال"، وتحتوي على بروتوكولات الأمان. وهي منفصلة عن قاعدة بيانات المخططات الهندسية.

تنفيذ عميل A2A

للسماح لـ "وكيل الإرسال" بالتواصل مع "مهندس الحلول" البعيد، يجب تحديد وكيل A2A البعيد. يخبر هذا الملف وكيل الإرسال بمكان العثور على Architect وشكل "بطاقة الوكيل".

👉✏️ استبدِل #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) مباشرةً إلى الدالة، ما يسمح للأداة "برؤية" اللقطات نفسها التي يراها العميل.

يمكنك اعتبار أداة البث حارسًا. أثناء مناقشة المخططات مع وكيل Dispatch، يعمل نظام الحراسة في الخلفية، ويعالج كل إطار فيديو بصمت لضمان سلامتك.

تنفيذ "أداة المراقبة في الخلفية"

سننفّذ الآن أداة 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)

تنفيذ وكيل الإرسال

وكيل الإرسال هو واجهتك الأساسية ومنسّق العمليات. وبما أنّها تدير رابط البث الثنائي الاتجاه (الصوت والفيديو المباشر)، يجب أن تحتفظ بالتحكّم في المحادثة في جميع الأوقات. لتحقيق ذلك، سنستخدم ميزة معيّنة في "مجموعة أدوات المطوّرين في Android"، وهي Agent-as-a-Tool.

المفهوم: استخدام الوكيل كأداة مقابل استخدام الوكلاء الفرعيين

عند إنشاء أنظمة متعددة الوكلاء، يجب تحديد كيفية تقسيم المسؤولية. في مهمة الإنقاذ التي نضطلع بها، يكون التمييز بين هذين المفهومَين أمرًا بالغ الأهمية:

• Agent-as-a-Tool: هذا هو النهج المقترَح لمركز البث ثنائي الاتجاه. عندما يستدعي وكيل الإرسال (الوكيل أ) وكيل التصميم (الوكيل ب) كأداة، يتم إعادة بيانات وكيل التصميم إلى وكيل الإرسال. بعد ذلك، تفسّر Dispatch هذه البيانات وتُنشئ ردًا لك. يظلّ Dispatch في وضع التحكّم ويواصل التعامل مع جميع بيانات أدخلها المستخدم اللاحقة.
• الوكيل الفرعي: في علاقة الوكيل الفرعي، يتم نقل المسؤولية بالكامل. إذا أحالك Dispatch إلى Architect كوكيل فرعي، ستتحدث مباشرةً إلى واجهة برمجة تطبيقات لقاعدة بيانات لا تملك "رؤية" ولا مهارات محادثة. لن يكون الوكيل الأساسي (Dispatch) على علم بذلك.

من خلال استخدام Agent-as-a-Tool، نستفيد من المعرفة المتخصصة التي يقدّمها Architect مع الحفاظ على التفاعل السلس الذي يقدّمه وكيل البث الثنائي الاتجاه والمشابه لتفاعل البشر.

كتابة رمز منطق التوجيه

سنضع الآن architect_agent في AgentTool ونزوّد وكيل الإرسال بـ "خريطة منطقية". تحدّد هذه الخريطة الوقت الذي يجب أن يسترجع فيه الوكيل البيانات من الخزنة والوقت الذي يجب أن يبلغ فيه عن النتائج من نظام المراقبة في الخلفية.

لكي نمنح 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],    

التحقّق

👉💻 بعد إعداد كلا الوكيلَين، يمكننا اختبار التفاعل المباشر بين الوكلاء المتعدّدين.

• في نافذة Terminal 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. انقر على تغيير المنفذ، واضبطه على 8000، ثم انقر على تغيير ومعاينة.

👉اختيار dispatch_agent وتحميل Blueprint والتعامل مع الخطأ المتوقّع

هذه هي الخطوة الأكثر أهمية. يجب أن نقدّم سياق الصورة إلى الوكيل.

• عند تحميل الواجهة، اسمح لها بالوصول إلى الميكروفون عندما يُطلب منك ذلك.
• نزِّل صورة المخطط هذه على جهاز الكمبيوتر:
• في واجهة adk web، انقر على رمز مشبك الورق وحمِّل صورة التصميم التي نزّلتها للتو.

⚠️⚠️سيظهر لك الخطأ 400 INVALID_ARGUMENT. هذا متوقّع.⚠️⚠️

يحدث هذا الخطأ لأنّ معالج الصور adk web غير متوافق تمامًا مع واجهة برمجة التطبيقات الخاصة بنموذج gemini-live للتحميل لمرة واحدة. ومع ذلك، تمت إضافة الصورة إلى سياق الجلسة بنجاح.

• 👉 لإزالة الخطأ، ما عليك سوى إعادة تحميل صفحة المتصفح.

بدء عملية التجميع

👉 بعد إعادة التحميل، سيختفي الخطأ وستظهر صورة التصميم في سجلّ المحادثة. أصبح لدى الوكيل الآن السياق المرئي الذي يحتاجه.

• انقر على رمز الميكروفون لتفعيله. ستعرض الواجهة الرسالة "جارٍ الاستماع...".
• قُل الطلب الصوتي: "بدء التجميع".
• سيعالج الوكيل طلبك، وسيتغيّر واجهة المستخدم إلى "جارٍ التحدث...". يجب أن تسمع ردًا صوتيًا فقط يتضمّن الأجزاء المطلوبة.

4. التحقّق من مكالمات الأدوات بين الوكلاء

👈 يؤكّد الرد الصوتي الأوّلي أنّ النظام يعمل، ولكنّ السحر الحقيقي يكمن في مسار التواصل بين الوكلاء المتعدّدين.

• أوقِف الميكروفون.
• أعِد تحميل الصفحة مرة أخرى.

سيتم الآن ملء اللوحة "تتبُّع" (Trace) على يمين الشاشة. يمكنك الاطّلاع على مسار التنفيذ الكامل والناجح:

• يتم عرض dispatch_agent المكالمات الأولى monitor_for_hazard.
• بعد ذلك، يتم إجراء عدة طلبات execute_architect إلى architect_agent لاسترداد البيانات التخطيطية.

يؤكّد هذا التسلسل أنّ سير العمل الكامل المتضمّن عدة وكلاء يعمل بشكل صحيح: تلقّى dispatch_agent الطلب، وفوّض مهمة استرداد البيانات إلى architect_agent من خلال استدعاء أداة، وتلقّى البيانات مرة أخرى لتنفيذ أمر المستخدم.

أصبح بإمكان رابط البث ثنائي الاتجاه الآن إجراء المراقبة في الخلفية والتعاون بين عدة وكلاء. بعد ذلك، سنتعرّف على كيفية تحليل هذه الردود المعقّدة في الواجهة الأمامية.

👉💻 اضغط على Ctrl+c في كلتا نافذتَي Terminal للخروج.

5- نظرة تفصيلية على أحداث البث المباشر المتعدد الوسائط

في الخطوة السابقة، تحقّقنا بنجاح من نظامنا المتعدد الوكلاء باستخدام خادم التطوير المضمّن adk web. تستخدم هذه الأداة برنامج تشغيل تلقائيًا لحزمة تطوير التطبيقات (ADK) من أجل إدارة الجلسة وعمليات البث ودورة حياة الوكيل تلقائيًا. ومع ذلك، لإنشاء تطبيق مستقل وجاهز للإنتاج مثل خدمة FastAPI (main.py)، نحتاج إلى تحكّم صريح. يجب إنشاء ADK Runner وإدارته يدويًا للتعامل مع جلسات المستخدمين المباشرة، لأنّه المكوّن الأساسي الذي يعالج عمليات البث الثنائية الاتجاه للصوت والفيديو والنص.

حلقة النموذج والرمز والنموذج

لفهم طريقة عمل النظام في الوقت الفعلي، لنستعرض دورة حياة جلسة مهمة واحدة. تمثّل هذه الحلقة التبادل المستمر لعناصر LlmRequest وLlmResponse.

1. الرابط المرئي: يمكنك بدء الاتصال ومشاركة كاميرا الويب أو الشاشة. تبدأ إطارات JPEG عالية الدقة بالتدفّق في اتجاه المصدر عبر realtimeInput (باستخدام LiveRequestQueue).
2. التفعيل باستخدام كلمة التنبيه: يرسل النظام إشارة أولية "مرحبًا". وفقًا لتعليماته، يبدأ "وكيل الإرسال" على الفور في تشغيل monitor_for_hazard أداة البث. يبدأ هذا الإجراء تكرارًا حلقيًا في الخلفية يراقب كل إطار وارد بدون إصدار صوت.
3. أمر الطيار: تتحدث في جهاز الاتصال: "ابدأ التجميع".
4. بث الصوت: يتم تسجيل صوتك بجودة 16 كيلو هرتز وإرساله أثناء البث مع إطارات الفيديو.
5. التفويض (التطبيق إلى التطبيق): يستمع Dispatch إلى هدفك. يدرك الوكيل أنّه يفتقر إلى المخططات، لذا يستدعي وكيل المهندس المعماري باستخدام بروتوكول AgentTool (الوكيل كأداة).
6. استرجاع الحقائق: يرسل Architect طلب بحث إلى قاعدة بيانات Redis ويعرض قائمة الأجزاء إلى Dispatch. يظلّ الإرسال هو "المتحكّم في الجلسة"، حيث يتلقّى البيانات بدون أن يطلب منك اتّخاذ أي إجراء.
7. الإرسال المعلوماتي إلى الجهاز: يرسل مركز القيادة رسالة modelTurn (إلى الجهاز) تحتوي على نص وصوت أصلي: "تم تأكيد وصول المهندس المعماري. المجموعة الفرعية المطلوبة هي: Warp Core وFlux Pipe وIon Thruster".
8. الأزمة: فجأة، يصبح جزء من طاولة العمل غير ثابت ويبدأ بالتوهّج باللون الأبيض.
9. الرصد الذاتي: تلتقط حلقة monitor_for_hazard الخلفية (Sentinel) إطار JPEG المحدّد الذي يحتوي على التوهّج. تعالج هذه الميزة اللقطة من خلال استدعاء Gemini وتحديد المخاطر.
10. الأمان في اتجاه تدفق البيانات: تعرض أداة البث yields نتيجة. بما أنّ هذا العنصر هو وكيل Bidi-Streaming، يمكن لـ Dispatch مقاطعة حالته الحالية لإرسال تحذير أمان مهم في اتجاه واحد على الفور: "تم رصد خطر! نعمل الآن على إبطال مفعول "بلّورة البيانات". انقُل الملف إلى سلة المهملات الحمراء".

ضبط إعدادات وقت التشغيل للوكيل

تتيح RunConfig في "حزمة تطوير التطبيقات" إمكانية ضبط سلوك أحد البرامج الوسيطة بشكلٍ مفصّل، بما في ذلك طريقة تعامله مع بيانات البث والتفاعل مع الوسائط المختلفة.

يتم ضبط 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 لتفعيل جلسة الوكيل الثنائية الاتجاه.

ابحث عن العنصر النائب #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)، يتم تجميع البيانات الواردة من الوكيل في نوع معيّن من الأحداث التي يتم استخلاصها من البِنى الأساسية لحزمة تطوير البرامج (SDK) الخاصة بالذكاء الاصطناعي التوليدي. إنّ عنصر Event الذي تتلقّاه في حلقة async for event in runner.run_live(...) هو عنصر واحد يحتوي على عدّة حقول اختيارية، كلّ منها مخصّص لنوع مختلف من المعلومات:

طريقة تنظيم المحتوى:

• عندما يتحدث الوكيل (من خلال .server_content): الحقل ليس نصًا عاديًا. تحتوي على قائمة Parts. كل Part هو حاوية لنوع واحد من البيانات، إما سلسلة نصية (مثل "The part is stable.") أو مجموعة كبيرة من بيانات الصوت الخام (الصوت).
• عندما يتصرّف الوكيل (من خلال .tool_call): يحتوي الحقل على قائمة بعناصر FunctionCall. كل FunctionCall هو عنصر بسيط ومنظَّم يحدّد اسم الأداة ووسائط الإدخال بتنسيق واضح يمكن أن يقرأه رمز الخلفية وينفّذه بسهولة.

👀 إذا أردت الاطّلاع على 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 لإعادة توجيه بيانات الحدث الكاملة. تضمن هذه المنطقية تسجيل كل "فكرة" لدى الذكاء الاصطناعي في وحدة التحكم التشخيصية للسفينة وإرسالها كعنصر JSON واحد إلى واجهة المستخدم الأمامية.

ابحث عن العنصر النائب #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)

تنفيذ المهمة

بعد ربط المخزن الخلفي وضبط كلا الوكيلَين، تصبح جميع الأنظمة جاهزة للاستخدام. ستؤدي الخطوات التالية إلى تشغيل التطبيق الكامل، ما يتيح لك التفاعل مع نظام الوكيلَين الذي أنشأته للتو.

الهدف: تجميع محرك الدفع الفضائي المحدّد عشوائيًا والذي يظهر على منضدة العمل البروتوكول: يجب اتّباع الإرشادات الصوتية التي يقدّمها "وكيل الإرسال"، خاصةً تحذيرات المخاطر بشأن مكوّنات معيّنة.

تفعيل "الخبير" (The Architect)

👉💻 في نافذة الوحدة الطرفية الأولى، شغِّل وكيل Architect. ستتصل خدمة الخلفية هذه بمخزن 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

(اترك هذه الوحدة الطرفية قيد التشغيل. أصبح الآن "وكيل قاعدة البيانات" النشط لديك.)

تشغيل Cockpit (The Dispatcher)

👉💻 في نافذة طرفية جديدة (النافذة الطرفية B)، سننشئ واجهة المستخدم الأمامية ونبدأ تشغيل وكيل Dispatch الرئيسي الذي يعرض واجهة المستخدم ويتعامل مع جميع عمليات التواصل المباشر.

# 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. 👉 حلقة التجميع:
   • الطلب: لبدء تجميع محرك الأقراص، قُل: "بدء التجميع".
   • ردّ المهندس المعماري: سيقدّم لك الوكيل الأجزاء الصحيحة لتجميع المحرك.
   • التحقّق من الخطورة: عندما يبدو أنّ جزءًا ما خطير على منضدة العمل:
     • ستتعرّف أداة monitor_for_hazard الخاصة بموظف خدمة التوصيل على الحزمة بصريًا.
     • سيؤدي ذلك إلى ظهور "تنبيه بشأن خطر مرئي". (ستستغرق هذه العملية حوالي 30 ثانية)
     • سيتحقّق من الحاوية التي يجب استخدامها لإزالة الخطر.
   • الإجراء: سيصدر موظف مركز العمليات أمرًا مباشرًا: "تم تأكيد وجود خطر. ضَع XXX في الحاوية الحمراء على الفور". يجب اتّباع هذه التعليمات للمتابعة.

تمت المهمة. لقد أنشأت بنجاح نظامًا تفاعليًا يستند إلى عدّة وكلاء. الناجون بأمان، وقد تجاوز الصاروخ الغلاف الجوي، وتستمر رحلتك في "العودة إلى الوطن".

👉💻 اضغط على Ctrl+c في كلتا نافذتَي Terminal للخروج.

6. النشر في الإصدار العلني (اختياري)

لقد اختبرت الوكيل بنجاح على جهازك. الآن، يجب تحميل النواة العصبية للمهندس إلى الحواسيب المركزية للسفينة (Cloud Run). سيسمح ذلك بتشغيلها كخدمة دائمة ومستقلة يمكن لوكيل Dispatch الاستعلام عنها من أي مكان.

توفير 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: نفِّذ هذا الأمر وانسخ host عنوان IP. هذا هو العنوان الخاص لمثيل 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

نشر تطبيق Agent

تجميع صورة الوكيل وإنشاؤها

👉💻 انتقِل إلى دليل الخلفية وأنشئ ملف 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 Hub على الحاسوب المركزي للإنتاج

بعد أن أصبح Architect Agent جاهزًا للعمل على السحابة الإلكترونية، علينا الآن نشر Dispatch Hub. سيعمل هذا الوكيل كواجهة المستخدم الأساسية، وسيتعامل مع بث الصوت والفيديو المباشرين وسينقل طلبات البحث في قاعدة البيانات إلى نقطة النهاية الآمنة في Architect.

👉💻 شغِّل الأمر التالي في وحدة Cloud Shell الطرفية. سيؤدي ذلك إلى إنشاء ملف Dockerfile كامل ومتعدّد المستويات في دليل الخلفية.

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

تجميع وإنشاء صورة الوكيل/الواجهة الأمامية

👉💻 انتقِل إلى دليل الخلفية الذي يحتوي على رمز وكيل الإرسال (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. افتح Cockpit: ألصِق عنوان URL في متصفّح الويب.
3. بدء المحادثة: عند تحميل الواجهة، تأكَّد من السماح لها بالوصول إلى شاشتك وميكروفونك.
4. طلب البيانات: عند تعيين محرك أقراص، اطلب بدء التجميع. على سبيل المثال: "ابدأ التجميع"

أنت الآن تتفاعل مع نظام متعدّد الوكلاء تم نشره بالكامل ويعمل بالكامل على Google Cloud.

يقوم نظام الوكلاء المتعدّدين بتثبيت حلقة الاحتواء النهائية في مكانها، ويتحوّل الإشعاع غير المنتظم إلى صوت طنين ثابت.

محرك الدفع الفائق: تم تثبيته. مركبة الإنقاذ: تم تشغيل المحركات".

على شاشتك، يندفع طبق الفضائيين إلى الأعلى، ويهرب بالكاد من سطح كوكب أوزيماندياس المتداعي بينما ينهار الغلاف الجوي. يستقر في مدار آمن بجانب سفينتك، وتمتلئ قنوات الاتصال بأصوات الناجين، الذين يبدون متأثرين لكنهم على قيد الحياة. بعد إكمال عملية الاسترداد وإتاحة طريق العودة إلى المنزل، يتم قطع الرابط عن بُعد.

بفضل مساعدتك، تم إنقاذ الناجين.

إذا شاركت في المستوى 0، لا تنسَ التحقّق من مستوى تقدّمك في مهمة "العودة إلى الوطن".