Way Back Home - Live Bidirectional Multi-Agent system

1. मिशन

कहानी

तुम अंतरिक्ष के शांत और अनजाने फैलाव में भटक रहे हो. एक बड़े सोलर पल्स ने आपके जहाज़ को डाइमेंशनल रिफ़्ट से गुज़ार दिया है. इससे आप ब्रह्मांड के एक ऐसे हिस्से में फंस गए हैं जहां कोई स्टार चार्ट मौजूद नहीं है.

कई दिनों तक मरम्मत करने के बाद, इंजन की जानी-पहचानी आवाज़ आखिरकार वापस आ जाती है. आपका रॉकेटशिप चालू है. आपने मदरशिप से लंबी दूरी का अपलिंक भी बना लिया है. प्रस्थान का समय नज़दीक है. अब आप घर जाने के लिए तैयार हैं.

लेकिन जंप ड्राइव को चालू करने से पहले, एक डिस्ट्रेस सिग्नल सुनाई देता है. आपके सेंसर को "ओज़िमैंडियास" ग्रह से मदद के लिए एक सिग्नल मिला है. इस ग्रह पर बचे हुए लोग फंस गए हैं. उनका जहाज़ भी यहीं है. आपका मिशन बहुत ज़रूरी है: ग्रह का वायुमंडल खत्म होने से पहले उन्हें बचाएं.

उनके पास बचने का सिर्फ़ एक ही तरीका है. उन्हें एलियन टेक्नोलॉजी से बने एक पुराने और बेकार रॉकेट का इस्तेमाल करना होगा. यह रॉकेट काम तो करता है, लेकिन इसका वॉर ड्राइव टूट गया है. बचे हुए लोगों को बचाने के लिए, आपको उनके Volatile Workbench से रिमोटली कनेक्ट करना होगा. इसके बाद, आपको मैन्युअल तरीके से ड्राइव को असेंबल करना होगा.

चैलेंज

आपके पास इस एलियन टेक्नोलॉजी का कोई अनुभव नहीं है. यह टेक्नोलॉजी बहुत नाज़ुक होती है. अस्थिर कॉम्पोनेंट, कुछ ही सेकंड में रेडियोऐक्टिव खतरा बन सकता है. आपके पास Volatile Workbench को इस्तेमाल करने के लिए एक मौका है. आपका मौजूदा एआई असिस्टेंट, विज़ुअल डेटा और तकनीकी मैन्युअल को एक साथ प्रोसेस नहीं कर पा रहा है. इस वजह से, वह गलत निर्देश दे रहा है और ख़तरों के बारे में चेतावनियां नहीं दे रहा है.

सफलता पाने के लिए, आपको अपने एआई को एक मोनोलिथिक इकाई से अपग्रेड करके, सहयोग करने वाले मल्टी-एजेंट सिस्टम में बदलना होगा.

आपके मिशन के मकसद:

अपने नए मल्टी-एजेंट सिस्टम से, रीयल-टाइम में मिलने वाले खास निर्देशों का पालन करके, Warp Drive को असेंबल करें.

मिशन ऐल्फ़ा

आपको क्या बनाना है

खास जानकारी

यह एक ऐसा एआई सिस्टम है जो रीयल-टाइम में दोनों तरफ़ से काम करता है. इसमें कई एजेंट होते हैं. इसमें एक सेंट्रल डिस्पैच एजेंट होता है, जो उपयोगकर्ता के इंटरैक्शन को मैनेज करता है और खास एजेंट के साथ समन्वय करता है.
एक आर्किटेक्ट एजेंट, जो स्कीमेटिक डेटा को वापस पाने और उसे उपलब्ध कराने के लिए Redis डेटाबेस से कनेक्ट होता है.
सुरक्षा मॉनिटर की सुविधा, स्ट्रीमिंग टूल का इस्तेमाल करके लाइव वीडियो फ़ीड का विश्लेषण करती है. इससे विज़ुअल खतरों का पता चलता है और रीयल-टाइम में चेतावनियां मिलती हैं.
React पर आधारित फ़्रंटएंड, जो सिस्टम के साथ इंटरैक्ट करने के लिए यूज़र इंटरफ़ेस उपलब्ध कराता है. साथ ही, बैकएंड एजेंट को वीडियो और ऑडियो स्ट्रीम करता है.

आपको क्या सीखने को मिलेगा

टेक्नोलॉजी / कॉन्सेप्ट	ब्यौरा
Google Agent Development Kit (ADK)	एजेंट बनाने, उनकी जांच करने, और उन्हें मैनेज करने के लिए, ADK का इस्तेमाल किया जाएगा. साथ ही, रीयल टाइम में कम्यूनिकेशन, टूल इंटिग्रेशन, और एजेंट के लाइफ़साइकल को मैनेज करने के लिए, 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 खाते का इस्तेमाल करें: आपको निजी खाते का इस्तेमाल करना होगा. जैसे, name@gmail.com). कंपनी या स्कूल के ज़रिए मैनेज किए जाने वाले खातों का इस्तेमाल नहीं किया जा सकेगा.

👉 तरीका:

क्रेडिट का दावा करने वाली साइट पर जाएं: फ़िलहाल उपलब्ध नहीं है
साइन इन करें: लिंक को पता बार में चिपकाएं और अपने निजी Gmail खाते से साइन इन करें.
शर्तें स्वीकार करें: Google Cloud Platform की सेवा की शर्तें स्वीकार करें.
क्रेडिट की पुष्टि करें: ऐसा मैसेज देखें जिसमें इस बात की पुष्टि की गई हो कि क्रेडिट लागू कर दिया गया है.
- *ध्यान दें: अगर आपसे क्रेडिट कार्ड की जानकारी डालने के लिए कहा जाता है, तो इसे अनदेखा करें और विंडो बंद करें.

अब आप इस विंडो को बंद कर सकते हैं

Cloud Shell ऐक्सेस करना

👉Google Cloud Console में सबसे ऊपर मौजूद, Cloud Shell चालू करें पर क्लिक करें. यह Cloud Shell पैनल में सबसे ऊपर मौजूद टर्मिनल के आकार का आइकॉन है,

👉 "एडिटर खोलें" बटन पर क्लिक करें. यह बटन, पेंसिल वाले खुले फ़ोल्डर की तरह दिखता है. इससे विंडो में Cloud Shell Code Editor खुल जाएगा. आपको बाईं ओर फ़ाइल एक्सप्लोरर दिखेगा.

👉क्लाउड आईडीई में टर्मिनल खोलें,

👉💻 टर्मिनल में, पुष्टि करें कि आपने पहले ही पुष्टि कर ली है और प्रोजेक्ट को अपने प्रोजेक्ट आईडी पर सेट किया गया है. इसके लिए, यह कमांड इस्तेमाल करें:

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 Console अब भी किसी पुराने या डिफ़ॉल्ट प्रोजेक्ट को देख रहा हो. आपको इसे मैन्युअल तरीके से स्विच करना होगा:

console.cloud.google.com पर जाएं.
पेज में सबसे ऊपर मौजूद, प्रोजेक्ट चुनने वाला टूल ड्रॉपडाउन पर क्लिक करें.
"सभी" टैब पर क्लिक करें. ऐसा हो सकता है कि आपका नया प्रोजेक्ट, "हाल ही के" टैब में अभी न दिखे.
स्क्रिप्ट से अभी-अभी बनाए गए प्रोजेक्ट आईडी को चुनें.

प्रोजेक्ट एनवायरमेंट सेट अप करना

अपने टर्मिनल पर वापस जाएं. इसके बाद, चालू प्रोजेक्ट सेट करके और ज़रूरी 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

डिपेंडेंसी इंस्टॉल करना

👉💻 चौथे लेवल पर जाएं और ज़रूरी Python पैकेज इंस्टॉल करें:

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

मुख्य डिपेंडेंसी ये हैं:

पैकेज	मकसद
`fastapi`	सैटलाइट स्टेशन और एसएसई स्ट्रीमिंग के लिए, ज़्यादा परफ़ॉर्मेंस वाला वेब फ़्रेमवर्क
`uvicorn`	FastAPI ऐप्लिकेशन को चलाने के लिए, ASGI सर्वर की ज़रूरत होती है
`google-adk`	फ़ॉर्मेशन एजेंट को बनाने के लिए इस्तेमाल की गई एजेंट डेवलपमेंट किट
`a2a-sdk`	स्टैंडर्ड कम्यूनिकेशन के लिए, एजेंट-टू-एजेंट प्रोटोकॉल लाइब्रेरी
`google-genai`	Gemini के मॉडल को ऐक्सेस करने के लिए नेटिव क्लाइंट
`redis`	Schematic Vault (Memorystore) से कनेक्ट करने के लिए Python क्लाइंट
`websockets`	रीयल-टाइम में दोनों तरफ़ से बातचीत करने की सुविधा
`python-dotenv`	यह एनवायरमेंट वैरिएबल और कॉन्फ़िगरेशन सीक्रेट मैनेज करता है
`pydantic`	डेटा की पुष्टि करना और सेटिंग मैनेज करना

सेटअप की पुष्टि करना

कोड को लॉन्च करने से पहले, आइए पक्का करें कि सभी सिस्टम ठीक से काम कर रहे हों. अपने Google Cloud प्रोजेक्ट, एपीआई, और Python डिपेंडेंसी का ऑडिट करने के लिए, पुष्टि करने वाली स्क्रिप्ट चलाएं.

👉💻 पुष्टि करने वाली स्क्रिप्ट चलाएं:

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

👀 आपको ग्रीन चेक (✅) की एक सीरीज़ दिखेगी.

अगर आपको लाल रंग के क्रॉस (❌) दिखते हैं, तो आउटपुट में दिए गए, समस्या ठीक करने के सुझावों का पालन करें. उदाहरण के लिए, gcloud services enable ... या pip install ...).
ध्यान दें: फ़िलहाल, .env के लिए पीले रंग की चेतावनी स्वीकार की जा सकती है. हम अगले चरण में यह फ़ाइल बनाएंगे.

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

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

🎉 SYSTEMS ONLINE. READY FOR MISSION.

3. Redis में स्कीमेटिक वॉल्ट और ADK के साथ BiDirectional एजेंट बनाना

आपको ग्रहों के बारे में जानकारी देने वाली ऐसी रिपॉज़िटरी मिली है जिसमें बेकार हो चुके रॉकेट के ब्लूप्रिंट मौजूद हैं. इस डेटा को सटीक तरीके से पाने के लिए, आपको रिपॉज़िटरी के मैनेजमेंट इंटरफ़ेस के साथ इंटरैक्ट करना होगा. इसे आर्किटेक्ट एजेंट कहा जाता है.

खास जानकारी

स्कीमैटिक वॉल्ट (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"

आर्किटेक्ट एजेंट को लागू करना

आर्किटेक्ट एजेंट एक खास एजेंट होता है. यह हमारे Redis वॉल्ट से स्कीमेटिक ब्लूप्रिंट को वापस पाने के लिए ज़िम्मेदार होता है. यह एक खास डेटा इंटरफ़ेस के तौर पर काम करता है. इससे यह पक्का किया जाता है कि मुख्य डिसपैच एजेंट को सटीक और स्ट्रक्चर्ड जानकारी मिले. इसके लिए, उसे डेटाबेस के लॉजिक के बारे में जानने की ज़रूरत नहीं होती.

खास जानकारी

Google एजेंट डेवलपमेंट किट (एडीके) एक मॉड्यूलर फ़्रेमवर्क है. इसकी मदद से, एक से ज़्यादा एजेंट सेट अप किए जा सकते हैं. यह दो अहम लेयर को मैनेज करता है:

कनेक्शन और सेशन का लाइफ़साइकल: रीयल-टाइम एपीआई के साथ इंटरैक्ट करने के लिए, जटिल प्रोटोकॉल मैनेजमेंट की ज़रूरत होती है. जैसे, हैंडशेक, पुष्टि, और कीप-अलाइव सिग्नल को मैनेज करना.
फ़ंक्शन कॉलिंग: यह "मॉडल-कोड-मॉडल राउंड ट्रिप" है. जब एलएलएम को लगता है कि उसे डेटा की ज़रूरत है, तो वह स्ट्रक्चर्ड फ़ंक्शन कॉल आउटपुट करता है. ADK इस इंटरसेप्ट को रोकता है और आपके Python कोड (lookup_schematic_tool) को एक्ज़ीक्यूट करता है. इसके बाद, नतीजे को मॉडल के कॉन्टेक्स्ट में मिलीसेकंड में वापस भेजता है.

अब हम आर्किटेक्ट बनाएंगे. इस एजेंट के पास कैमरे का ऐक्सेस नहीं है. यह सिर्फ़ "ड्राइव का नाम" पाने और डेटाबेस से "पार्ट की सूची" वापस पाने के लिए मौजूद है.

👉💻 हम adk create कमांड का इस्तेमाल करेंगे. यह एजेंट डेवलपमेंट किट (एडीके) का एक टूल है. यह नए एजेंट के लिए, बॉयलरप्लेट कोड और फ़ाइल स्ट्रक्चर अपने-आप जनरेट करता है. इससे सेटअप करने में लगने वाला समय बच जाता है.

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

एजेंट को कॉन्फ़िगर करना

सीएलआई, इंटरैक्टिव सेटअप विज़र्ड लॉन्च करेगा. अपने एजेंट को कॉन्फ़िगर करने के लिए, इन जवाबों का इस्तेमाल करें:

कोई मॉडल चुनें: पहला विकल्प (Gemini Flash) चुनें.
- ध्यान दें: खास वर्शन (जैसे, 2.5, 3.0) की उपलब्धता के आधार पर अलग-अलग हो सकती है. तेज़ी से काम करने के लिए, हमेशा "फ़्लैश" वैरिएंट चुनें.
कोई बैकएंड चुनें: दूसरा विकल्प (Vertex AI) चुनें.
Google Cloud प्रोजेक्ट आईडी डालें: डिफ़ॉल्ट आईडी (आपके एनवायरमेंट से पता लगाया गया) स्वीकार करने के लिए, Enter दबाएं.
Google Cloud Region डालें: डिफ़ॉल्ट (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 'हो गया' मैसेज दिखेगा. इससे एक स्केलेटन कोड जनरेट होता है, जिसमें हम अगले चरण में बदलाव करेंगे.

👉✏️ अपने एडिटर में, नई बनाई गई $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 लाइन को इससे बदलें. साथ ही, वह टूल भी जोड़ें जिसे हमने पहले तय किया था:

    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 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 आईडी टाइप किया जा सकता है.
व्यवहार को मॉनिटर करना:
- आर्किटेक्ट को तुरंत 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) प्रोटोकॉल का इस्तेमाल करते हैं.

एमसीपी (मॉडल कॉन्टेक्स्ट प्रोटोकॉल) जैसे प्रोटोकॉल, एजेंटों को टूल से कनेक्ट करने पर फ़ोकस करते हैं. वहीं, A2A, एजेंटों को अन्य एजेंटों से कनेक्ट करने पर फ़ोकस करता है. यह एक ऐसा स्टैंडर्ड है जिसकी मदद से हमारा डिसपैचर, आर्किटेक्ट को "खोज" सकता है. साथ ही, यह समझ सकता है कि वह स्कीमेटिक्स को देखने में कितना सक्षम है.

A2A

A2A फ़्लो: इस मिशन में, हम क्लाइंट-सर्वर मॉडल का इस्तेमाल करते हैं:

सर्वर (आर्किटेक्ट): यह डेटाबेस टूल को होस्ट करता है और एजेंट कार्ड के ज़रिए अपनी क्षमताओं का "विज्ञापन" करता है.
क्लाइंट (डिस्पैच): यह आर्किटेक्ट के कार्ड को पढ़ता है, इसके एपीआई को समझता है, और स्कीमेटिक अनुरोध भेजता है.

एजेंट कार्ड क्या होता है?

एजेंट कार्ड को डिजिटल बिज़नेस कार्ड या एआई के लिए "ड्राइविंग लाइसेंस" के तौर पर देखा जा सकता है. 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 दिखता है, तो इसका मतलब है कि Architect लाइव है, A2A प्रोटोकॉल चालू है, और एजेंट कार्ड को डिसपैचर ढूंढ सकता है.

अब Architect, रिमोट रिसोर्स के तौर पर काम करने के लिए तैयार है. इसलिए, हम इसे Dispatch Agent में शामिल कर सकते हैं.

👉💻 A2A सर्वर से बाहर निकलने के लिए, Ctrl+C दबाएं.

4. BIDI-Streams Agent को Remote Agent और स्ट्रीमिंग टूल से कनेक्ट करना

अब आपको प्राइमरी कम्यूनिकेशन हब कॉन्फ़िगर करना होगा, ताकि लाइव डेटा और रिमोट आर्किटेक्ट के बीच के अंतर को कम किया जा सके. इस कनेक्शन के लिए, ज़्यादा बैंडविड्थ और कम समय में डेटा ट्रांसफ़र करने वाली पाइपलाइन की ज़रूरत होती है, ताकि ऑपरेशन के दौरान असेंबली बेंच स्थिर रहे.

दोनों तरफ़ से स्ट्रीम करने वाले (लाइव) एजेंट के बारे में जानकारी

ADK में दोनों तरफ़ से स्ट्रीम होने वाली सुविधा (बिडाइरेक्शनल स्ट्रीमिंग) जोड़ी गई है. इससे एआई एजेंट, Gemini Live API की मदद से कम समय में दोनों तरफ़ से आवाज़ और वीडियो के ज़रिए इंटरैक्ट कर सकते हैं. यह एआई के साथ इंटरैक्ट करने के पारंपरिक तरीकों से अलग है. "सवाल पूछो और जवाब का इंतज़ार करो" के पैटर्न के बजाय, यह रीयल-टाइम में दोनों तरफ़ से बातचीत करने की सुविधा देता है. इसमें इंसान और एआई, दोनों एक साथ बोल सकते हैं, सुन सकते हैं, और जवाब दे सकते हैं.

ईमेल भेजने और फ़ोन पर बातचीत करने के बीच के अंतर के बारे में सोचें. पारंपरिक एजेंट इंटरैक्शन, ईमेल की तरह होते हैं: इसमें पूरा मैसेज भेजा जाता है, पूरे जवाब का इंतज़ार किया जाता है, और फिर दूसरा मैसेज भेजा जाता है. द्विदिश स्ट्रीमिंग, फ़ोन पर होने वाली बातचीत की तरह होती है. इसमें बातचीत स्वाभाविक तरीके से होती है और बीच में टोकने, किसी बात को साफ़ तौर पर बताने, और रीयल-टाइम में जवाब देने की सुविधा होती है.

मुख्य बातें:

दोनों तरफ़ से बातचीत: जवाब पूरा होने का इंतज़ार किए बिना, लगातार डेटा का आदान-प्रदान होता है. उपयोगकर्ता के बोलना बंद करते ही, एआई जवाब देता है.
जवाब के बीच में टोकने की सुविधा: उपयोगकर्ता, एजेंट के जवाब के बीच में नया इनपुट देकर उसे टोक सकते हैं. यह सुविधा, किसी इंसान के साथ बातचीत करने के दौरान भी मिलती है. अगर कोई एआई किसी मुश्किल चरण के बारे में बता रहा है और आपने कहा, "रुको, इसे दोहराओ," तो एआई तुरंत रुक जाता है और आपकी बात सुनता है.
कई अलग-अलग काम करने के लिए ऑप्टिमाइज़ किया गया है: बिडी-स्ट्रीमिंग, अलग-अलग तरह के इनपुट को एक साथ प्रोसेस करने में बेहतर है. वीडियो के ज़रिए एजेंट को एलियन के हिस्सों को दिखाते हुए, उससे बात की जा सकती है. साथ ही, एजेंट एक ही कनेक्शन में दोनों स्ट्रीम को प्रोसेस करता है.

जीवनचक्र

👀 क्लाइंट लॉजिक लागू करने से पहले, आइए Dispatch Agent के लिए पहले से जनरेट किए गए स्केलेटन की जांच करें. यह एजेंट, उपयोगकर्ता से वॉइस और वीडियो के ज़रिए बातचीत करेगा. साथ ही, क्वेरी को आर्किटेक्ट एजेंट को सौंप देगा.

__init__.py
agent.py
hazard_db.py

agent.py: यह "ब्रेन" है. फ़िलहाल, इसमें बिडी-स्ट्रीमिंग का बुनियादी सेटअप शामिल है. हम इस फ़ाइल में बदलाव करके, A2A क्लाइंट लॉजिक जोड़ेंगे, ताकि यह आर्किटेक्ट से कम्यूनिकेट कर सके.
hazard_db.py: यह एक लोकल टूल है, जो खास तौर पर डिस्पैच एजेंट के लिए है. इसमें सुरक्षा से जुड़े प्रोटोकॉल शामिल होते हैं. यह आर्किटेक्ट के स्कीमेटिक डेटाबेस से अलग होता है.

A2A क्लाइंट लागू करना

डिसपैच एजेंट को हमारे रिमोट आर्किटेक्ट से कम्यूनिकेट करने की अनुमति देने के लिए, हमें रिमोट A2A एजेंट को तय करना होगा. इससे डिसपैच एजेंट को पता चलता है कि आर्किटेक्ट कहां मिलेगा और उसका "एजेंट कार्ड" कैसा दिखता है.

A2A क्लाइंट

👉✏️ $HOME/way-back-home/level_4/backend/dispatch_agent/agent.py में मौजूद #REPLACE-REMOTEA2AAGENT को इससे बदलें:

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 में स्ट्रीमिंग टूल तय करने के लिए, आपको इन तकनीकी शर्तों का पालन करना होगा:

एसिंक्रोनस फ़ंक्शन: टूल को async def के साथ तय किया जाना चाहिए.
AsyncGenerator का रिटर्न टाइप: फ़ंक्शन को AsyncGenerator को वापस लाने के लिए टाइप किया जाना चाहिए. पहला पैरामीटर, जनरेट किए जा रहे डेटा का टाइप होता है. उदाहरण के लिए, str) और दूसरा None होता है.
इनपुट स्ट्रीम: हम वीडियो स्ट्रीमिंग टूल का इस्तेमाल करते हैं. इस मोड में, वीडियो/ऑडियो स्ट्रीम (LiveRequestQueue) को सीधे तौर पर फ़ंक्शन में पास किया जाता है. इससे टूल को वही फ़्रेम "दिखते" हैं जो एजेंट को दिखते हैं.

स्ट्रीमिंग टूल को सेंटिनल की तरह समझें. जब आप और डिसपैच एजेंट, ब्लूप्रिंट पर चर्चा कर रहे होते हैं, तब सेंटिनल बैकग्राउंड में काम कर रहा होता है. यह आपकी सुरक्षा के लिए, हर वीडियो फ़्रेम को चुपचाप प्रोसेस करता है.

स्ट्रीमिंग टूल

बैकग्राउंड मॉनिटरिंग टूल लागू करना

अब हम 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)

डिस्पैच एजेंट को लागू करना

डिस्पैच एजेंट, आपका प्राइमरी इंटरफ़ेस और ऑर्केस्ट्रेटर होता है. यह बीडीआई-स्ट्रीमिंग लिंक (आपकी लाइव आवाज़ और वीडियो) को मैनेज करता है. इसलिए, इसे बातचीत का कंट्रोल हमेशा अपने पास रखना चाहिए. इसके लिए, हम एडीके की एक खास सुविधा का इस्तेमाल करेंगे: Agent-as-a-Tool.

कॉन्सेप्ट: एजेंट को टूल के तौर पर इस्तेमाल करना बनाम सब-एजेंट

मल्टी-एजेंट सिस्टम बनाते समय, आपको यह तय करना होगा कि ज़िम्मेदारी कैसे शेयर की जाती है. हमारे बचाव मिशन में, इन दोनों के बीच अंतर करना ज़रूरी है:

Agent-as-a-Tool: यह हमारे बिडी-स्ट्रीमिंग हब के लिए सुझाया गया तरीका है. जब डिस्पैच एजेंट (एजेंट A), आर्किटेक्ट एजेंट (एजेंट B) को टूल के तौर पर कॉल करता है, तो आर्किटेक्ट का डेटा वापस डिस्पैच को भेज दिया जाता है. इसके बाद, Dispatch उस डेटा को समझता है और आपके लिए जवाब जनरेट करता है. Dispatch के पास कंट्रोल रहता है और वह उपयोगकर्ता के सभी इनपुट को हैंडल करता रहता है.
सब-एजेंट: सब-एजेंट के तौर पर काम करने पर, ज़िम्मेदारी पूरी तरह से ट्रांसफ़र हो जाती है. अगर Dispatch ने आपको सब-एजेंट के तौर पर Architect को सौंप दिया है, तो आपको सीधे तौर पर एक डेटाबेस एपीआई से बात करनी होगी. इसमें न तो "विजन" है और न ही बातचीत करने की क्षमता है. इससे प्राइमरी एजेंट (डिस्पैच) को जानकारी नहीं मिल पाएगी.

Controle

Agent-as-a-Tool का इस्तेमाल करके, हम आर्किटेक्ट के खास ज्ञान का फ़ायदा उठाते हैं. साथ ही, हम बिडी-स्ट्रीमिंग एजेंट के साथ, इंसानों की तरह बातचीत करने की सुविधा को बनाए रखते हैं.

राउटिंग लॉजिक को कोड करना

अब हम अपने architect_agent को AgentTool में रैप करेंगे और डिसपैच एजेंट को "लॉजिक मैप" उपलब्ध कराएंगे. इस मैप से एजेंट को यह पता चलता है कि वॉल्ट से डेटा कब फ़ेच करना है और बैकग्राउंड सेंटिनल से नतीजे कब रिपोर्ट करने हैं.

Dispatch को "ऐसी नज़रें" देने के लिए जो कभी न झपकें, हमें उसे पिछले चरण में बनाए गए स्ट्रीमिंग टूल का ऐक्सेस देना होगा.

ADK में, tools सूची में AsyncGenerator फ़ंक्शन (जैसे कि monitor_for_hazard) जोड़ने पर, एजेंट इसे बैकग्राउंड में लगातार चलने वाली प्रोसेस के तौर पर मानता है. एक बार लागू करने के बजाय, एजेंट टूल के आउटपुट के लिए "सदस्यता लेता है". इससे Dispatch को अपनी मुख्य बातचीत जारी रखने में मदद मिलती है. वहीं, Sentinel बैकग्राउंड में बिना किसी सूचना के ख़तरे की चेतावनियां देता रहता है.

👉✏️ $HOME/way-back-home/level_4/backend/dispatch_agent/agent.py में मौजूद #REPLACE_AGENT_TOOLS को इससे बदलें:

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) में, डिसपैच एजेंट चलाएं:

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

adk web सिम्युलेटर में, एक से ज़्यादा एजेंट वाले ऐसे सिस्टम की टेस्टिंग के लिए एक खास वर्कफ़्लो की ज़रूरत होती है जो रीयल-टाइम में काम करने वाले मल्टीमॉडल मॉडल, जैसे कि gemini-live का इस्तेमाल करता है. टूल कॉल की जांच करने के लिए सिम्युलेटर बहुत अच्छा है. हालांकि, इस तरह के मॉडल के साथ पहली बार इमेज प्रोसेस करते समय, यह काम नहीं करता.

Cloud Shell टूलबार में, वेब प्रीव्यू आइकॉन पर क्लिक करें. पोर्ट बदलें को चुनें. इसके बाद, इसे 8000 पर सेट करें. इसके बाद, बदलें और झलक देखें पर क्लिक करें.

👉dispatch_agent को चुनें और ब्लूप्रिंट अपलोड करें. साथ ही, संभावित गड़बड़ी को ठीक करें

यह सबसे अहम चरण है. हमें एजेंट को इमेज के बारे में जानकारी देनी होगी.

इंटरफ़ेस लोड होने पर, मांगे जाने पर उसे अपने माइक्रोफ़ोन का ऐक्सेस अनुमति दें.
इस ब्लूप्रिंट इमेज को अपने कंप्यूटर पर डाउनलोड करें:
adk web इंटरफ़ेस में, पेपरक्लिप आइकॉन पर क्लिक करें और अभी डाउनलोड की गई ब्लूप्रिंट इमेज अपलोड करें.

⚠️⚠️आपको 400 INVALID_ARGUMENT गड़बड़ी दिखेगी. ऐसा होता है.⚠️⚠️

गड़बड़ी का अनुमानित मैसेज

यह गड़बड़ी इसलिए होती है, क्योंकि एक बार अपलोड करने के लिए adk web इमेज हैंडलर, gemini-live मॉडल के एपीआई के साथ पूरी तरह से काम नहीं करता. हालांकि, इमेज को सेशन के कॉन्टेक्स्ट में जोड़ दिया गया है.

👉 गड़बड़ी को ठीक करने के लिए, ब्राउज़र पेज को फिर से लोड करें.

असेंबली प्रोसेस ट्रिगर करना

👉 फिर से लोड करने के बाद, गड़बड़ी ठीक हो जाएगी. इसके बाद, आपको चैट के इतिहास में ब्लूप्रिंट की इमेज दिखेगी. अब एजेंट के पास ज़रूरी विज़ुअल कॉन्टेक्स्ट है.

माइक्रोफ़ोन चालू करने के लिए, माइक्रोफ़ोन आइकॉन पर क्लिक करें. इंटरफ़ेस पर "सुना जा रहा है..." दिखेगा.
बोलकर निर्देश दें: "असेंबल करना शुरू करो".
एजेंट आपके अनुरोध को प्रोसेस करेगा. इसके बाद, यूज़र इंटरफ़ेस (यूआई) "बोल रहा है..." में बदल जाएगा. आपको सिर्फ़ ऑडियो में जवाब मिलेगा. इसमें ज़रूरी हिस्सों की सूची दी गई होगी.

एजेंट के बोलने का जवाब

4. एजेंट-टू-एजेंट टूल कॉल की पुष्टि करना

👉 शुरुआती ऑडियो जवाब से पुष्टि होती है कि सिस्टम काम कर रहा है. हालांकि, असली कमाल मल्टी-एजेंट कम्यूनिकेशन ट्रेस में है.

माइक्रोफ़ोन बंद करें.
पेज को एक बार और रीफ़्रेश करें.

बाईं ओर मौजूद "ट्रेस करें" पैनल में अब जानकारी दिखने लगेगी. यहां बदलाव होने का पूरा फ़्लो देखा जा सकता है:

dispatch_agent पहली बार कॉल करने वाले लोग monitor_for_hazard.
इसके बाद, यह स्कीमेटिक डेटा को वापस पाने के लिए, architect_agent को कई execute_architect कॉल करता है.

टूल कॉल की पुष्टि करना

इस क्रम से पुष्टि होती है कि मल्टी-एजेंट वर्कफ़्लो सही तरीके से काम कर रहा है: dispatch_agent को अनुरोध मिला, उसने टूल कॉल के ज़रिए डेटा वापस पाने का टास्क architect_agent को सौंपा, और उपयोगकर्ता के निर्देश को पूरा करने के लिए डेटा वापस मिला.

अब आपका बिडirectional स्ट्रीमिंग लिंक, बैकग्राउंड में मॉनिटर करने और कई एजेंट के साथ मिलकर काम करने की सुविधा देता है. इसके बाद, हम फ़्रंटएंड पर इन मुश्किल रिस्पॉन्स को पार्स करने का तरीका जानेंगे.

👉💻 बाहर निकलने के लिए, दोनों टर्मिनल में Ctrl+c दबाएं.

5. लाइव मल्टीमॉडल इवेंट स्ट्रीम के बारे में ज़्यादा जानकारी

पिछले चरण में, हमने पहले से मौजूद डेवलपमेंट सर्वर adk web का इस्तेमाल करके, अपने मल्टी-एजेंट सिस्टम की पुष्टि कर ली है. यह यूटिलिटी, सेशन, स्ट्रीम, और एजेंट के लाइफ़साइकल को अपने-आप मैनेज करने के लिए, डिफ़ॉल्ट एडीके रनर का इस्तेमाल करती है. हालांकि, हमारी FastAPI सेवा (main.py) की तरह, प्रोडक्शन के लिए तैयार स्टैंडअलोन ऐप्लिकेशन बनाने के लिए, हमें साफ़ तौर पर कंट्रोल की ज़रूरत होती है. लाइव उपयोगकर्ता सेशन को मैनेज करने के लिए, हमें ADK Runner को मैन्युअल तरीके से बनाना और मैनेज करना होगा. ऐसा इसलिए, क्योंकि यह मुख्य कॉम्पोनेंट है. यह ऑडियो, वीडियो, और टेक्स्ट के लिए दोनों दिशाओं में स्ट्रीम को प्रोसेस करता है.

मॉडल-कोड-मॉडल लूप

सिस्टम के रीयल-टाइम में काम करने के तरीके को समझने के लिए, आइए एक मिशन सेशन के लाइफ़साइकल को फ़ॉलो करें. इस लूप में, LlmRequest और LlmResponse ऑब्जेक्ट के लगातार आदान-प्रदान को दिखाया गया है.

विज़ुअल लिंक: इसमें कनेक्शन शुरू करने और वेबकैम/स्क्रीन शेयर करने का विकल्प होता है. हाई-फ़िडेलिटी वाले JPEG फ़्रेम, realtimeInput के ज़रिए अपस्ट्रीम होने लगते हैं. इसके लिए, LiveRequestQueue का इस्तेमाल किया जाता है.
सेंटिनल ऐक्टिवेशन: सिस्टम, शुरुआती "हैलो" स्टिमुलस भेजता है. Dispatch Agent, निर्देशों के मुताबिक तुरंत monitor_for_hazard Streaming Tool को ट्रिगर करता है. इससे बैकग्राउंड में एक लूप शुरू हो जाता है. यह लूप, हर आने वाले फ़्रेम को चुपचाप देखता है.
पायलट का निर्देश: पायलट कम्यूनिकेशन सिस्टम में बोलता है: "असेंबल करना शुरू करो."
Vocal Upstream: इसमें आपकी आवाज़ को 16 किलोहर्ट्ज़ के ऑडियो के तौर पर रिकॉर्ड किया जाता है. इसके बाद, इसे वीडियो फ़्रेम के साथ अपस्ट्रीम किया जाता है.
डेलिगेशन (A2A): Dispatch को आपके इरादे के बारे में "पता" चलता है. इसे पता चलता है कि इसके पास स्कीमेटिक्स नहीं है. इसलिए, यह AgentTool (एजेंट-एज़-ए-टूल) प्रोटोकॉल का इस्तेमाल करके, आर्किटेक्ट एजेंट को कॉल करता है.
तथ्य पाना: आर्किटेक्ट, Redis डेटाबेस से क्वेरी करता है और डिस्पैच को पार्ट की सूची दिखाता है. Dispatch,"मास्टर ऑफ़ द सेशन" बना रहता है. इससे आपको डेटा मिलता रहता है.
जानकारी देने वाला डाउनस्ट्रीम: डिसपैच, टेक्स्ट और नेटिव ऑडियो, दोनों के साथ modelTurn (डाउनस्ट्रीम) भेजता है: "आर्किटेक्ट ने पुष्टि कर दी है. ज़रूरी सबसेट यह है: Warp Core, Flux Pipe, Ion Thruster."
समस्या: अचानक, वर्कबेंच का एक हिस्सा अस्थिर हो जाता है और सफ़ेद रंग में चमकने लगता है.
अपने-आप पता लगाने की सुविधा: बैकग्राउंड monitor_for_hazard लूप (सेंटिनल), चमक वाला खास JPEG फ़्रेम चुनता है. यह Gemini को कॉल करके फ़्रेम को प्रोसेस करता है और खतरे की पहचान करता है.
सुरक्षा डाउनस्ट्रीम: स्ट्रीमिंग टूल yields एक नतीजा है. यह बिडाइरेक्शनल स्ट्रीमिंग एजेंट है. इसलिए, Dispatch इसकी मौजूदा स्थिति में रुकावट डाल सकता है, ताकि डाउनस्ट्रीम को सुरक्षा से जुड़ी गंभीर चेतावनी तुरंत भेजी जा सके: "खतरे का पता चला! डेटा क्रिस्टल को अब न्यूट्रलाइज़ किया जा रहा है. इसे लाल रंग के बिन में डालो."

Flow

एजेंट के रनटाइम कॉन्फ़िगरेशन को सेट करना

ADK में मौजूद RunConfig की मदद से, एजेंट के व्यवहार को पूरी तरह से कॉन्फ़िगर किया जा सकता है. इसमें यह भी शामिल है कि एजेंट, स्ट्रीमिंग डेटा को कैसे मैनेज करता है और अलग-अलग मोड के साथ कैसे इंटरैक्ट करता है.

रीयल-टाइम में दोनों तरफ़ से बातचीत करने के लिए, streaming_mode को BIDI पर सेट किया जाता है. इससे उपयोगकर्ता और एजेंट, दोनों एक साथ बोल और सुन सकते हैं. response_modalities पैरामीटर से यह तय होता है कि एजेंट किस तरह का आउटपुट दे सकता है. जैसे, आवाज़ और टेक्स्ट. input_audio_transcription कॉन्फ़िगर करता है कि एजेंट, उपयोगकर्ता के इनपुट किए गए भाषण को कैसे प्रोसेस और ट्रांसक्राइब करता है. session_resumption की मदद से, एजेंट को बातचीत का कॉन्टेक्स्ट याद रखने में मदद मिलती है. साथ ही, कनेक्शन कट जाने पर बातचीत को फिर से शुरू किया जा सकता है. इससे बेहतर अनुभव मिलता है. आखिर में, proactivity की मदद से एजेंट, उपयोगकर्ता के सीधे तौर पर निर्देश दिए बिना कार्रवाइयां शुरू कर सकता है या बोल सकता है. जैसे, अचानक किसी खतरे के बारे में चेतावनी देना. वहीं, enable_affective_dialog की मदद से एजेंट, ज़्यादा स्वाभाविक और सहानुभूति भरे जवाब जनरेट कर सकता है. ADK के RunConfig के बारे में ज़्यादा जानने के लिए यहां जाएं.

👉✏️ अपनी $HOME/way-back-home/level_4/backend/main.py फ़ाइल में #REPLACE_RUN_CONFIG प्लेसहोल्डर ढूंढें और उसे यहां दिए गए डिससेक्शन लॉजिक से बदलें:

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 के ज़रिए, उपयोगकर्ता के Volatile Workbench से Dispatch Agent को रीयल-टाइम में मल्टीमॉडल डेटा स्ट्रीम करता है. इससे एजेंट को लगातार "देखने" (वीडियो फ़्रेम) और "सुनने" (आवाज़ वाले निर्देश) की सुविधा मिलती है. यह लॉजिक, डेटा स्ट्रीम को लगातार रिसीव करता है. साथ ही, बाइनरी ऑडियो के आने वाले चंक और JSON-रैप किए गए टेक्स्ट/इमेज पैकेट के बीच अंतर करता है. इसके बाद, इसे Blob (मल्टीमीडिया के लिए) या Content (टेक्स्ट के लिए) ऑब्जेक्ट में शामिल करता है. साथ ही, इसे LiveRequestQueue में भेजता है, ताकि एजेंट के साथ दोनों तरफ़ से बातचीत की जा सके.

BIDI

अपनी $HOME/way-back-home/level_4/backend/main.py फ़ाइल में #PROCESS_AGENT_REQUEST प्लेसहोल्डर ढूंढें और उसे यहां दिए गए डिसैक्शन लॉजिक से बदलें:

# Start the loop
        try:
            while True:
                # Receive message from WebSocket (text or binary)
                message = await websocket.receive()

                # Handle binary frames (audio data)
                if "bytes" in message:
                    audio_data = message["bytes"]
                    audio_blob = types.Blob(
                        mime_type="audio/pcm;rate=16000", data=audio_data
                    )
                    live_request_queue.send_realtime(audio_blob)

                # Handle text frames (JSON messages)
                elif "text" in message:
                    text_data = message["text"]
                    json_message = json.loads(text_data)

                    # Extract text from JSON and send to LiveRequestQueue
                    if json_message.get("type") == "text":
                        logger.info(f"User says: {json_message['text']}")
                        content = types.Content(
                            parts=[types.Part(text=json_message["text"])]
                        )
                        live_request_queue.send_content(content)

                    # Handle audio data (microphone)
                    elif json_message.get("type") == "audio":
                        # logger.info("Received AUDIO packet") # Uncomment for verbose debugging
                        import base64
                        # Decode base64 audio data
                        audio_data = base64.b64decode(json_message.get("data", ""))
                        
                        # logger.info(f"Received Audio Chunk: {len(audio_data)} bytes")
                        
                        import math
                        import struct
                        # Calculate RMS to debug silence
                        count = len(audio_data) // 2
                        shorts = struct.unpack(f"<{count}h", audio_data)
                        sum_squares = sum(s*s for s in shorts)
                        rms = math.sqrt(sum_squares / count) if count > 0 else 0
                        
                        # logger.info(f"RMS: {rms:.2f} | Bytes: {len(audio_data)}")

                        # Send to Live API as PCM 16kHz
                        audio_blob = types.Blob(
                            mime_type="audio/pcm;rate=16000", 
                            data=audio_data
                        )
                        live_request_queue.send_realtime(audio_blob)

                    # Handle image data
                    elif json_message.get("type") == "image":
                        import base64
                        
                        # Decode base64 image data
                        image_data = base64.b64decode(json_message["data"])
                        # logger.info(f"Received Image Frame: {len(image_data)} bytes")
                        
                        mime_type = json_message.get("mimeType", "image/jpeg")

                        # Send image as blob
                        image_blob = types.Blob(mime_type=mime_type, data=image_data)
                        live_request_queue.send_realtime(image_blob)
                        
                        frame_count += 1
                        
        finally:
             pass

अब मल्टीमॉडल डेटा, एजेंट को भेजा जा रहा है.

जवाब लागू करना: डाउनस्ट्रीम इवेंट डेटा स्ट्रक्चर

ADK के साथ लाइव एजेंट का इस्तेमाल करने पर, एजेंट से मिलने वाले डेटा को एक खास तरह के इवेंट में पैकेज किया जाता है. यह इवेंट, GenAI SDK के मुख्य स्ट्रक्चर से इनहेरिट होता है. आपको async for event in runner.run_live(...) लूप में जो Event ऑब्जेक्ट मिलता है वह एक ही ऑब्जेक्ट होता है. इसमें कई वैकल्पिक फ़ील्ड होते हैं. हर फ़ील्ड में अलग-अलग तरह की जानकारी होती है:

इवेंट

कॉन्टेंट का स्ट्रक्चर कैसा है:

जब एजेंट बोलता है (.server_content के ज़रिए): यह फ़ील्ड सिर्फ़ सादा टेक्स्ट नहीं है. इसमें Parts की सूची शामिल होती है. हर Part एक कंटेनर होता है, जिसमें एक तरह का डेटा होता है. जैसे, टेक्स्ट स्ट्रिंग ("The part is stable." की तरह) या रॉ ऑडियो बीएलओबी (आवाज़).
जब एजेंट कार्रवाई करता है (.tool_call के ज़रिए): इस फ़ील्ड में FunctionCall ऑब्जेक्ट की सूची होती है. हर FunctionCall एक आसान, स्ट्रक्चर्ड ऑब्जेक्ट होता है. इसमें टूल का नाम और इनपुट आर्ग्युमेंट, साफ़ तौर पर दिए जाते हैं. इससे आपका बैकएंड कोड, इन्हें आसानी से पढ़ और लागू कर सकता है.

👀 अगर आपको run_live लूप से मिले किसी एक Event को देखना है, तो event.model_dump(by_alias=True) से जनरेट किया गया JSON इस तरह दिखेगा. इसमें 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 ऑब्जेक्ट के तौर पर भेजा गया है.

अपनी $HOME/way-back-home/level_4/backend/main.py फ़ाइल में #PROCESS_AGENT_RESPONSE प्लेसहोल्डर ढूंढें और उसे यहां दिए गए डिसैक्शन लॉजिक से बदलें:

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

मिशन को पूरा करना

बैकएंड वॉल्ट कनेक्ट होने और दोनों एजेंट कॉन्फ़िगर होने के बाद, सभी सिस्टम अब मिशन के लिए तैयार हैं. नीचे दिए गए चरणों को पूरा करने पर, पूरा ऐप्लिकेशन लॉन्च हो जाएगा. इससे, आपको अभी-अभी बनाए गए दो एजेंट वाले सिस्टम के साथ इंटरैक्ट करने की सुविधा मिलेगी.

मकसद: आपको अपनी वर्कबेंच पर दिखने वाली, रैंडम तरीके से असाइन की गई वॉर्प ड्राइव को असेंबल करना है. प्रोटोकॉल: आपको डिस्पैच एजेंट के निर्देशों का पालन करना होगा. खास तौर पर, कुछ कॉम्पोनेंट के लिए ख़तरे की चेतावनियों का पालन करना ज़रूरी है.

स्पेशलिस्ट (आर्किटेक्ट) को ऐक्टिवेट करना

👉💻 अपनी पहली टर्मिनल विंडो में, आर्किटेक्ट एजेंट लॉन्च करें. यह बैकएंड सेवा, Redis वॉल्ट से कनेक्ट होगी और डिसपैचर से स्कीमेटिक अनुरोधों का इंतज़ार करेगी.

# 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) में, हम फ़्रंटएंड यूज़र इंटरफ़ेस (यूआई) बनाएंगे. साथ ही, मुख्य डिसपैच एजेंट को शुरू करेंगे. यह एजेंट, यूज़र इंटरफ़ेस (यूआई) उपलब्ध कराता है और लाइव बातचीत को मैनेज करता है.

# 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 पर प्राइमरी सर्वर शुरू हो जाता है.)

टेस्ट का सिनेरियो चलाना

यह सिस्टम अब लाइव है. आपका लक्ष्य, एजेंट के निर्देशों का पालन करके असेंबली पूरी करना है.

👉 Workbench ऐक्सेस करना:
- Cloud Shell टूलबार में, वेब प्रीव्यू आइकॉन पर क्लिक करें.
- पोर्ट बदलें को चुनें. इसके बाद, इसे 8080 पर सेट करें. इसके बाद, बदलें और झलक देखें पर क्लिक करें.
👉 मिशन शुरू करें:
- इंटरफ़ेस लोड होने के बाद, पक्का करें कि आपने उसे अपनी स्क्रीन और माइक्रोफ़ोन को ऐक्सेस करने की अनुमति दी हो.
- आपको शेयर करने के लिए कोई टैब या विंडो चुनने के लिए कहा जाएगा. अगर आपको विंडो शेयर करनी है, तो समस्या से बचने के लिए पक्का करें कि विंडो में सिर्फ़ यही टैब हो.
- बिना सोचे-समझे दिए गए नाम वाली ड्राइव (उदाहरण के लिए, "NOVA-V", "OMEGA-9") असाइन किए जाएंगे.
👉 असेंबली लूप:
- अनुरोध: ड्राइव को असेंबल करना शुरू करने के लिए, "असेंबल करना शुरू करो" कहें
- आर्किटेक्ट का जवाब: एजेंट, ड्राइव को असेंबल करने के लिए सही पार्ट देगा.
- खतरनाक कॉन्टेंट की जांच: वर्कबेंच पर किसी हिस्से के खतरनाक होने पर:
  - डिस्पैच एजेंट का monitor_for_hazard टूल, इसे विज़ुअल तौर पर पहचान लेगा.
  - इससे "विज़ुअल हैज़र्ड अलर्ट" मिलेगा. (इसमें करीब 30 सेकंड लगेंगे)
  - यह देखेगा कि खतरे को कम करने के लिए, किस बिन का इस्तेमाल करना है.
- कार्रवाई: डिस्पैच एजेंट आपको सीधे तौर पर यह निर्देश देगा: "खतरे की पुष्टि हो गई है. XXX को तुरंत लाल बिन में डालो." आगे बढ़ने के लिए, आपको इस निर्देश का पालन करना होगा.

मिशन पूरा हुआ. आपने इंटरैक्टिव, मल्टी-एजेंट सिस्टम बना लिया है. सभी लोग सुरक्षित हैं, रॉकेट वायुमंडल से बाहर निकल गया है, और "घर वापसी" जारी है.

👉💻 बाहर निकलने के लिए, दोनों टर्मिनल में Ctrl+c दबाएं.

6. प्रोडक्शन में डिप्लॉय करें (ज़रूरी नहीं)

आपने एजेंट की स्थानीय तौर पर जांच कर ली है. अब हमें आर्किटेक्ट के न्यूरल कोर को शिप के मेनफ़्रेम (Cloud Run) पर अपलोड करना होगा. इससे यह एक स्थायी और स्वतंत्र सेवा के तौर पर काम कर पाएगा. डिसपैच एजेंट, इस सेवा से कहीं से भी क्वेरी कर सकता है.

खास जानकारी

सुरक्षित वॉल्ट (इंफ़्रास्ट्रक्चर) की सुविधा चालू करना

एजेंट को डिप्लॉय करने से पहले, हमें उसकी परसिस्टेंट मेमोरी (Memorystore) और उसे ऐक्सेस करने के लिए सुरक्षित चैनल (VPC कनेक्टर) बनाना होगा.

👉💻 Memorystore इंस्टेंस (Redis Vault) बनाएं:

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

👉💻 Vault का नेटवर्क पता वापस पाएं: यह निर्देश चलाएं और host आईपी पता कॉपी करें. यह आपके नए Redis इंस्टेंस का निजी पता है.

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

👉💻 वीपीसी ऐक्सेस कनेक्टर (सुरक्षित ब्रिज) बनाएं: यह कनेक्टर, एक निजी ब्रिज के तौर पर काम करता है. इससे Cloud Run को आपके वीपीसी में मौजूद Redis इंस्टेंस को ऐक्सेस करने की अनुमति मिलती है.

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

एजेंट ऐप्लिकेशन को डिप्लॉय करना

एजेंट की इमेज को कंपाइल और बिल्ड करना

👉💻 बैकएंड डायरेक्ट्री पर जाएं और 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 पर डिप्लॉय करें. हम Redis आईपी को इंजेक्ट करेंगे और वीपीसी कनेक्टर को सीधे लॉन्च कमांड से लिंक करेंगे. इससे यह पक्का होता है कि एजेंट, अपने डेटाबेस से सुरक्षित और निजी कनेक्शन के साथ शुरू हो.

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

कमांड पूरी होने के बाद, आपको सेवा का यूआरएल दिखेगा. आर्किटेक्ट एजेंट अब क्लाउड में लाइव है. यह अपने वॉल्ट से हमेशा के लिए कनेक्ट हो गया है और अन्य एजेंट को स्कीमेटिक डेटा देने के लिए तैयार है.

Dispatch Hub को प्रोडक्शन मेनफ़्रेम पर डिप्लॉय करना

क्लाउड में Architect Agent के चालू होने के बाद, अब हमें Dispatch Hub को डिप्लॉय करना होगा. यह एजेंट, प्राइमरी यूज़र इंटरफ़ेस के तौर पर काम करेगा. यह लाइव वॉइस/वीडियो स्ट्रीम को मैनेज करेगा और डेटाबेस क्वेरी को आर्किटेक्ट के सुरक्षित एंडपॉइंट को सौंपेगा.

👉💻 अपने 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 पर डिप्लॉय करें. हम आर्किटेक्ट के यूआरएल को एनवायरमेंट वैरिएबल के तौर पर इंजेक्ट करेंगे. इससे हमारे दो क्लाउड-नेटिव एजेंट के बीच ज़रूरी लिंक बन जाएगा.

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

कमांड पूरी होने के बाद, आपको सेवा का यूआरएल दिखेगा. जैसे, https://mission-bravo-...run.app). ऐप्लिकेशन अब क्लाउड में लाइव है.

👉 Google Cloud Run पेज पर जाएं और सूची से, बायोमेट्रिक-स्काउट सेवा चुनें. CloudRun

👉 सेवा की जानकारी वाले पेज पर सबसे ऊपर मौजूद, सार्वजनिक यूआरएल ढूंढें. CloudRun

सिस्टम की फ़ाइनल जांच (शुरू से अंत तक का टेस्ट)

👉 अब आपको लाइव सिस्टम से इंटरैक्ट करना होगा.

यूआरएल पाएं: पिछले डिप्लॉयमेंट कमांड के आउटपुट से सेवा का यूआरएल कॉपी करें. यह run.app पर खत्म होना चाहिए.
Cockpit खोलें: यूआरएल को अपने वेब ब्राउज़र में चिपकाएं.
संपर्क शुरू करें: इंटरफ़ेस लोड होने पर, पक्का करें कि आपने इसे अपनी स्क्रीन और माइक्रोफ़ोन को ऐक्सेस करने की अनुमति दी हो.
डेटा का अनुरोध करें: जब किसी ड्राइवर को डिलीवरी असाइन की जाती है, तब उससे डेटा इकट्ठा करने के लिए कहें. उदाहरण के लिए: "असेंबल करना शुरू करें"

CloudRun

अब आपके पास पूरी तरह से डिप्लॉय किए गए, मल्टी-एजेंट सिस्टम के साथ इंटरैक्ट करने का विकल्प है. यह सिस्टम पूरी तरह से Google Cloud पर काम करता है.

मल्टी-एजेंट सिस्टम, फ़ाइनल कंटेनमेंट रिंग को लॉक कर देता है. साथ ही, अनियमित रेडिएशन एक स्थिर आवाज़ में बदल जाता है.

"वॉर ड्राइव: स्टेबल हो गई है. Rescue Craft: ENGINES IGNITED."

खत्म हो रहा है

मॉनिटर पर, एलियन का जहाज़ ऊपर की ओर उड़ता है. ओज़िमैंडियास की सतह के ढहने के साथ-साथ, वायुमंडल के खत्म होने से वह बाल-बाल बचता है. यह आपके जहाज़ के साथ सुरक्षित ऑर्बिट में सेटल हो जाता है. साथ ही, कम्यूनिकेशन सिस्टम में बचे हुए लोगों की आवाज़ें सुनाई देती हैं. वे डरे हुए हैं, लेकिन ज़िंदा हैं. समस्या ठीक हो जाने और घर का रास्ता साफ़ हो जाने पर, रिमोट लिंक बंद हो जाता है.

आपकी वजह से, सर्वाइवर को बचा लिया गया है.

अगर आपने लेवल 0 में हिस्सा लिया था, तो यह देखना न भूलें कि 'घर वापसी' मिशन में आपकी प्रोग्रेस कहां तक पहुंची है!

फ़ाइनल