1. The Mission
Sessiz ve keşfedilmemiş uzay boşluğunda sürükleniyorsunuz. Büyük bir Güneş Darbesi, geminizi boyutlar arası bir yarıktan geçirerek sizi evrenin yıldız haritasında yer almayan bir köşesinde mahsur bıraktı.
Günlerce süren zorlu onarımların ardından motorların tanıdık uğultusu nihayet geri döner. Roketiniz çalışır durumda. Hatta Ana Gemide uzun menzilli bir uplink bağlantısı kurmayı başardınız. Kalkış yakındır. Eve gitmeye hazırsınız.
Ancak flash sürücüyü kullanmaya hazırlanırken cızırtı sesinin arasından bir tehlike sinyali duyulur. Sensörleriniz, "Ozymandias" olarak adlandırılan bir gezegenden gelen yardım çağrısını tespit ediyor. Hayatta kalanlar, gemileri karaya oturmuş bir şekilde bu ölmekte olan dünyada mahsur kalır. Göreviniz çok önemli: Gezegenin atmosferi çökmek üzereyken onları kurtarmanız gerekiyor.
Tek kaçış yolları, uzaylı teknolojisiyle inşa edilmiş eski ve terk edilmiş bir roket. Roket çalışır durumda olsa da warp sürücüsü hasar görmüş. Hayatta kalanları kurtarmak için Volatile Workbench'lerine uzaktan bağlanıp yedek sürücüyü manuel olarak monte etmeniz gerekir.
Zorluk
Kırılganlığıyla bilinen bu uzaylı teknolojisiyle ilgili hiçbir deneyiminiz yok. Dengesiz bir bileşen saniyeler içinde radyoaktif tehlikeye dönüşebilir. Volatile Workbench'i çalıştırmak için bir deneme hakkınız var. Mevcut yapay zeka asistanınız, görsel verileri ve teknik kılavuzları aynı anda işlemekte zorlanıyor. Bu durum, halüsinasyon içeren talimatlara ve kaçırılan tehlike uyarılarına yol açıyor.
Başarılı olmak için yapay zekanızı tek bir varlık olmaktan çıkarıp işbirlikçi bir çoklu aracı sistemine yükseltmeniz gerekir.
Görev Hedefleriniz:
Yeni çoklu aracı sisteminizden gelen özel ve anlık talimatları uygulayarak Warp Drive'ı monte edin.
Ne oluşturacaksınız?
- Kullanıcı etkileşimini yöneten ve uzmanlaşmış temsilcilerle koordinasyon sağlayan merkezi bir Gönderim Temsilcisi'nin yer aldığı, gerçek zamanlı ve çift yönlü bir çoklu temsilci yapay zeka sistemi.
- Şematik verileri almak ve sunmak için bir Redis veritabanına bağlanan Mimar Aracısı.
- Canlı video akışını görsel tehlikeler açısından analiz etmek ve anlık uyarılar tetiklemek için akış araçlarını kullanan proaktif bir Güvenlik Monitörü.
- Sistemle etkileşim kurmak, video ve ses içeriğini arka uç aracılarında yayınlamak için kullanıcı arayüzü sağlayan React tabanlı bir ön uç.
Öğrenecekleriniz
Teknoloji / Kavram | Açıklama |
Google Agent Development Kit (ADK) | ADK'yı kullanarak temsilcileri oluşturur, test eder ve yönetirsiniz. Bu süreçte, gerçek zamanlı iletişimi, araç entegrasyonunu ve temsilci yaşam döngüsünü yönetmek için ADK'nın çerçevesinden yararlanırsınız. |
Çift Yönlü (Bidi) Akış | Doğal, düşük gecikmeli ve iki yönlü iletişime olanak tanıyan bir çift yönlü akış aracısı uygulayacaksınız. Bu sayede hem insanlar hem de yapay zeka, anlık olarak kesintiye uğratabilir ve yanıt verebilir. |
Çoklu aracı sistemleri | Birincil aracının görevleri uzmanlaşmış aracılara devrettiği, endişelerin ayrılmasına ve daha ölçeklenebilir bir mimariye olanak tanıyan dağıtılmış bir yapay zeka sistemi tasarlamayı öğreneceksiniz. |
Temsilciden Temsilciye (A2A) Protokolü | A2A protokolünü kullanarak Dispatch Agent ile Architect Agent arasında iletişimi etkinleştirirsiniz. Böylece, bu iki ajan birbirinin özelliklerini keşfedebilir ve veri alışverişinde bulunabilir. |
Yayın Araçları | Arka plan işlemi olarak çalışan, durum değişikliklerini (tehlikeler) izlemek için video feed'ini sürekli olarak analiz eden ve proaktif olarak sonuç veren bir akış aracı uygulayacaksınız. |
Google Cloud Run ve Memorystore | Aracı hizmetlerini barındırmak için Cloud Run'ı ve kalıcı veritabanı olarak Memorystore'u (Redis) kullanarak çok aracılı uygulamanın tamamını bir üretim ortamına dağıtacaksınız. |
FastAPI ve WebSocket'ler | Arka uç, ses, video ve aracı yanıtlarının aktarılması için gereken yüksek performanslı ve gerçek zamanlı iletişimi yönetmek üzere FastAPI ve WebSocket'ler kullanılarak oluşturulmuştur. |
React ön ucu | Kullanıcı medyalarını (ses/video) yakalayıp yayınlayan ve yapay zeka aracılarından gelen anlık yanıtları gösteren React tabanlı bir ön uçla çalışacaksınız. |
2. Ortamınızı ayarlama
Cloud Shell'e erişme
👉Google Cloud Console'un üst kısmında Cloud Shell'i etkinleştir'i tıklayın (Cloud Shell bölmesinin üst kısmındaki terminal şeklindeki simge). 
👉 "Open Editor" (Düzenleyiciyi aç) düğmesini tıklayın (kalemli açık bir klasöre benzer). Bu işlem, pencerede Cloud Shell kod düzenleyiciyi açar. Sol tarafta bir dosya gezgini görürsünüz. 
👉Bulut IDE'de terminali açın.
👉💻 Terminalde, aşağıdaki komutu kullanarak kimliğinizin doğrulandığını ve projenin proje kimliğinize ayarlandığını doğrulayın:
gcloud auth list
Hesabınız (ACTIVE) olarak listelenir.
Ön koşullar
ℹ️ 0. Düzey İsteğe Bağlıdır (Ancak Önerilir)
Bu görevi 0. seviyede tamamlayabilirsiniz ancak önce bu görevi tamamlamak daha sürükleyici bir deneyim sunar. Böylece ilerledikçe işaretinizin küresel haritada yandığını görebilirsiniz.
Proje ortamını ayarlama
Terminalinize geri dönerek etkin projeyi ayarlayıp gerekli Google Cloud hizmetlerini (Cloud Run, Vertex AI vb.) etkinleştirerek yapılandırmayı tamamlayın.
👉💻 Terminalinizde proje kimliğini ayarlayın:
gcloud config set project $(cat ~/project_id.txt) --quiet
👉💻 Zorunlu Hizmetleri Etkinleştirme:
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
Bağımlılıkları yükleme
👉💻 4. seviyeye gidin ve gerekli Python paketlerini yükleyin:
cd $HOME/way-back-home/level_4
uv sync
Temel bağımlılıklar şunlardır:
Paket | Amaç |
| Uydu İstasyonu ve SSE akışı için yüksek performanslı web çerçevesi |
| FastAPI uygulamasını çalıştırmak için ASGI sunucusu gerekir. |
| Formation Agent'ı oluşturmak için kullanılan Agent Development Kit |
| Standartlaştırılmış iletişim için aracıdan aracıya protokol kitaplığı |
| Gemini modellerine erişmek için yerel istemci |
| Şematik kasaya (Memorystore) bağlanmak için Python istemcisi |
| Gerçek zamanlı çift yönlü iletişim desteği |
| Ortam değişkenlerini ve yapılandırma sırlarını yönetir. |
| Veri doğrulama ve ayar yönetimi |
Kurulumu Doğrulama
Koda geçmeden önce tüm sistemlerin sorunsuz çalıştığından emin olalım. Google Cloud projenizi, API'lerinizi ve Python bağımlılıklarınızı denetlemek için doğrulama komut dosyasını çalıştırın.
👉💻 Doğrulama komut dosyasını çalıştırın:
cd $HOME/way-back-home/level_4/scripts
chmod +x verify_setup.sh
. verify_setup.sh
👀 Bir dizi yeşil onay işareti (✅) görmelisiniz.
- Kırmızı çarpı işaretleri (❌) görürseniz çıktıda önerilen düzeltme komutlarını uygulayın (ör.
gcloud services enable ...veyapip install ...). - Not:
.enviçin sarı uyarı şu an kabul edilebilir. Bu dosyayı sonraki adımda oluşturacağız.
🚀 Verifying Mission Bravo (Level 4) Infrastructure... ✅ Google Cloud Project: xxxxxxx ✅ Cloud APIs: Active ✅ Python Environment: Ready 🎉 SYSTEMS ONLINE. READY FOR MISSION.
3. Redis'te Şematik Kasa ve ADK ile Çift Yönlü Aracı Oluşturma
Terke edilmiş roketin planlarını içeren gezegen şeması deposunu buldunuz. Bu verileri doğru şekilde almak için, depodaki özel yönetim arayüzü olan Architect aracısıyla arayüz oluşturmanız gerekir.
Şematik Apps Kasası'nı (Redis) sağlama
Mimarın bize yardımcı olabilmesi için verilerin güvenli ve yüksek kullanılabilirlik sağlayan bir ortamda barındırıldığından emin olmamız gerekir. Uzaylı şemalarımız için hızlı bir veri deposu olarak Redis'i kullanacağız. Geliştirme kolaylığı için yerel bir Redis örneği oluşturacağız ancak Google Cloud Memorystore ile üretim ortamına nasıl dağıtım yapılacağıyla ilgili talimatlar daha sonra verilecektir.
👉💻 Redis örneğini sağlama için terminalinizde aşağıdaki komutları çalıştırın (Bu işlem 2-3 dakika sürebilir):
docker run -d --name ozymandias-vault -p 6379:6379 redis:8.6-rc1-alpine
👉💻 Ön verileri yüklemek için aşağıdaki komutu çalıştırarak Redis Shell'e girin:
docker exec -it ozymandias-vault redis-cli
(İsteminiz 127.0.0.1:6379 olarak değişecek.)
👉💻 Aşağıdaki komutları yapıştırın:
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"
👉💻 Normal kabuğunuza dönmek için exit yazın.
👉💻 Verilerin mevcut olup olmadığını kontrol etmek için doğrudan terminalinizden belirli bir gemiyi sorgulayarak şu komutu çalıştırın:
# Check 'TITAN-PRIME'
docker exec ozymandias-vault redis-cli LRANGE "TITAN-PRIME" 0 -1
👀 Beklenen çıktı:
1) "Ion Thruster" 2) "Quantum Cell" 3) "Warp Core"
Mimar Temsilciyi uygulama
Mimar Temsilci, Redis kasamızdan şematik planları almaktan sorumlu özel bir temsilcidir. Özel bir veri arayüzü olarak işlev görür. Temel veritabanı mantığını bilmeye gerek kalmadan ana Gönderme Aracısı'nın doğru ve yapılandırılmış bilgiler almasını sağlar.
Google Agent Development Kit (ADK), bu çoklu aracı kurulumunu mümkün kılan modüler çerçevedir. İki kritik katmanı işler:
- Bağlantı ve Oturum Yaşam Döngüsü: Gerçek zamanlı API'lerle etkileşim kurmak için karmaşık protokol yönetimi (el sıkışma, kimlik doğrulama ve canlı tutma sinyallerini işleme) gerekir.
- İşlev Çağırma: Bu, "Model-Kod-Model Gidiş Dönüşü"dür. LLM, veriye ihtiyaç duyduğuna karar verdiğinde yapılandırılmış bir işlev çağrısı oluşturur. ADK bunu yakalar, Python kodunuzu (
lookup_schematic_tool) yürütür ve sonucu milisaniyeler içinde modelin bağlamına geri besler.
Şimdi Mimar'ı oluşturacağız. Bu temsilcinin kamera erişimi yok. Yalnızca "Sürücü Adı" almak ve veritabanından "Parça Listesi" döndürmek için kullanılır.
👉💻 adk create komutunu kullanacağız. Bu, yeni bir aracı için ortak metin kodunu ve dosya yapısını otomatik olarak oluşturan, böylece kurulum süresini kısaltan bir Agent Development Kit (ADK) aracıdır.
cd $HOME/way-back-home/level_4/backend/
uv run adk create architect_agent
Aracıyı yapılandırma
CLI, etkileşimli bir kurulum sihirbazı başlatır. Aracınızı yapılandırmak için aşağıdaki yanıtları kullanın:
- Model seçin: 1. Seçenek'i (Gemini Flash) belirleyin.
- Not: Belirli sürüm (ör. 2,5, 3,0) kullanılabilirliğe bağlı olarak değişebilir. Hız için her zaman "Flash" varyantını seçin.
- Arka uç seçin: Seçenek 2'yi (Vertex AI) belirleyin.
- Google Cloud proje kimliğini girin: Varsayılanı (ortamınızdan algılanan) kabul etmek için Enter tuşuna basın.
- Google Cloud Bölgesi'ni girin: Varsayılanı (
us-central1) kabul etmek için Enter tuşuna basın.
👀 Terminal etkileşiminiz aşağıdaki gibi görünmelidir:
(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
Bu aşamada bir Agent created başarı mesajı gösterilir. Bu işlem, bir sonraki adımda değiştireceğimiz iskelet kodu oluşturur.
👉✏️ Düzenleyicinizde yeni oluşturulan $HOME/way-back-home/level_4/backend/architect_agent/agent.py dosyasına gidin ve dosyayı açın. Aracı ilk içe aktarma satırından sonra dosyaya ekleyin:
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 tanımındaki instruction satırının tamamını aşağıdakiyle değiştirin ve daha önce tanımladığımız aracı da ekleyin:
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'nın avantajı
Architect online olduğunda artık bir doğruluk kaynağımız var. Bunu birincil aracıya bağlamadan önce, Aracı Geliştirme Kiti (ADK), yapay zeka aracı oluşturma ve test etme işlemlerinin karmaşıklığını basitleştirerek önemli bir avantaj sağlar. Yerleşik adk web geliştirici konsolu sayesinde, Architect Agent'mizin işlevselliğini (özellikle araç çağırma özelliklerini) daha büyük çoklu aracı sistemine entegre etmeden önce izole edip doğrulayabiliriz. Geliştirme ve test sürecinde bu modüler yaklaşım, sağlam ve güvenilir yapay zeka uygulamaları oluşturmak için çok önemlidir.
👉💻 Terminalinizde şu komutu çalıştırın:
cd $HOME/way-back-home/level_4/
. scripts/check_redis.sh
cd $HOME/way-back-home/level_4/backend/
uv run adk web
👀 Şunları görene kadar bekleyin:
+-----------------------------------------------------------------------------+ | 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 araç çubuğunda Web önizlemesi simgesini tıklayın. Bağlantı noktasını değiştir'i seçin, 8000 olarak ayarlayın ve Değiştir ve Önizle'yi tıklayın.
- architect_agent'ı seçin.
- Aracı tetikleme: Sohbet arayüzünde
CHRONOS-ALPHA(veya şematik veritabanındaki herhangi bir Drive kimliği) yazın. - Davranışı Gözlemleyin:
- Mimar,
lookup_schematic_tool'yı hemen tetiklemelidir. - Sıkı sistem talimatlarımız nedeniyle, yalnızca parça listesini (ör.
['Shield Emitter', 'Data Crystal', 'Quantum Cell']) Sohbet dolgusu olmadan.
- Mimar,
- Günlükleri doğrulayın: Terminal pencerenize bakın. Başarılı yürütme günlüğünü görmeniz gerekir:
[ARCHITECT] Returning schematic for CHRONOS-ALPHA: ['Shield Emitter', 'Data Crystal', 'Quantum Cell']!(architect_agent adk)[img/03-02-adkweb.png]
Araç yürütme günlüğünü ve temiz veri yanıtını görüyorsanız uzman temsilciniz beklendiği gibi çalışıyor demektir. İstekleri işleyebilir, kasayı sorgulayabilir ve yapılandırılmış veriler döndürebilir.
👉💻 Çıkmak için Ctrl+C tuşuna basın.
A2A sunucusunu başlatma
Sevk Temsilcisi'ni Mimar'a bağlamak için Temsilciden Temsilciye (A2A) Protokolü'nü kullanırız.
MCP (Model Context Protocol) gibi protokoller, temsilcileri araçlara bağlamaya odaklanırken A2A, temsilcileri diğer temsilcilere bağlamaya odaklanır. Bu standart, Dispatcher'ımızın Architect'i "keşfetmesine" ve şemaları arama yeteneğini anlamasına olanak tanır.
A2A Akışı: Bu görevde istemci-sunucu modeli kullanıyoruz:
- Sunucu (Mimar): Veritabanı araçlarını barındırır ve bir Aracı Kartı aracılığıyla becerilerini "tanıtır".
- İstemci (Gönderme): Mimarın kartını okur, API'sini anlar ve şematik bir istek gönderir.
Temsilci kartı nedir?
Agent Card'ı dijital bir kartvizit veya yapay zeka için "sürücü belgesi" olarak düşünebilirsiniz. Bir A2A sunucusu başladığında aşağıdakileri içeren bu JSON nesnesini yayınlar:
- Kimlik: Temsilcinin adı (
architect_agent) ve kimliği. - Açıklama: Ne yaptığıyla ilgili, hem insanlar hem de makineler tarafından okunabilir bir özet ("Sistem Rolü: Veritabanı API'si...").
- Arayüz: Beklediği belirli giriş anahtarları (
drive_name) ve çıkış biçimleri.
Bu kart olmadan, Gönderme aracısı körü körüne çalışır ve Mimar ile nasıl iletişim kuracağını tahmin etmeye çalışır.
Sunucu kodunu oluşturma
👉✏️ Düzenleyicinizde, $HOME/way-back-home/level_4/backend/architect_agent dizini altında server.py adlı bir dosya oluşturun ve aşağıdaki kodu yapıştırın:
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)
👉💻 Terminalinize geri dönün, klasöre gidin ve sunucuyu başlatın:
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 sunucusunun başlatılıp başlatılmadığını onaylayın:
INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:8081 (Press CTRL+C to quit)
Temsilci kartını doğrulama
Yeni bir terminal sekmesi açın (+ simgesini tıklayın). Mimarın, Temsilci Kartını manuel olarak getirerek kimliğini doğru şekilde yayınladığını doğrularız.
👉💻 Aşağıdaki komutu çalıştırın:
curl -s http://localhost:8081/.well-known/agent.json | jq .
👀 Bir JSON yanıtı görmelisiniz. Çıkışta description alanını bulun. Bu yanıt, daha önce müşteri temsilcisine verdiğiniz talimatla ("SYSTEM ROLE: Database API...") eşleşmelidir.
{
"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"
}
Bu JSON'u görüyorsanız Architect canlıdır, A2A protokolü aktiftir ve Agent Card, Dispatcher tarafından keşfedilmeye hazırdır.
Architect artık uzak kaynak olarak hizmet vermeye hazır olduğuna göre, Dispatch Agent'a bağlamaya devam edebiliriz.
👉💻 A2A sunucusundan çıkmak için Ctrl+C tuşuna basın.
4. BIDI-Streams Agent'ı Remote Agent'a ve Streaming Tools'a Bağlama
Artık canlı veriler ile uzaktaki Architect arasındaki boşluğu kapatmak için birincil iletişim merkezini yapılandıracaksınız. Bu bağlantı, montaj tezgahının çalışma sırasında sabit kalmasını sağlamak için yüksek bant genişliğine ve düşük gecikme süresine sahip bir kanal gerektirir.
Çift yönlü yayın (canlı) aracıları hakkında
ADK'daki çift yönlü (Bidi) akış, Gemini Live API'nin düşük gecikmeli, iki yönlü ses ve video etkileşimi özelliğini yapay zeka temsilcilerine ekler. Bu, geleneksel yapay zeka etkileşimlerinden temel bir değişimi temsil ediyor. Katı "sor ve bekle" kalıbı yerine, hem insanın hem de yapay zekanın aynı anda konuşabileceği, dinleyebileceği ve yanıt verebileceği gerçek zamanlı, iki yönlü iletişime olanak tanır.
E-posta göndermek ile telefonda konuşmak arasındaki farkı düşünün. Geleneksel temsilci etkileşimleri e-postalara benzer: Tam bir mesaj gönderir, tam bir yanıt bekler ve ardından başka bir mesaj gönderirsiniz. Çift yönlü yayın, telefon görüşmesine benzer: akıcı, doğal, kesme, netleştirme ve anında yanıt verme olanağı sunar.
Başlıca Özellikler:
- İki yönlü iletişim: Tam yanıtları beklemeden sürekli veri alışverişi. Yapay zeka, kullanıcının konuşmayı bitirdiğini algıladığı anda yanıt verir.
- Duyarlı Kesme: Kullanıcılar, tıpkı gerçek bir kişiyle sohbet ederken olduğu gibi, yanıt verirken yeni bir girişle temsilcinin sözünü kesebilir. Bir yapay zeka karmaşık bir adımı açıklarken "Bir dakika, tekrar et" derseniz yapay zeka hemen durur ve isteğinizi yerine getirir.
- Çok formatlılık için optimize edildi: Çift yönlü akış, farklı giriş türlerini aynı anda işlemede mükemmeldir. Uzaylı parçalarını videoyla gösterirken temsilciyle konuşabilirsiniz. Temsilci, her iki akışı tek bir bağlantıda işler.
👀 İstemci mantığını uygulamadan önce Dispatch Agent için önceden oluşturulmuş iskeleti inceleyelim. Bu ajan, kullanıcıyla sesli ve görüntülü olarak iletişim kurar ve sorguları Mimar Ajanı'na yönlendirir.
__init__.py agent.py hazard_db.py
agent.py: Bu, "Beyin"dir. Şu anda temel bir Bidi-streaming kurulumu içeriyor. Bu dosyayı, mimarla iletişim kurabilmesi için A2A Client mantığını ekleyecek şekilde değiştireceğiz.hazard_db.py: Bu, güvenlik protokollerini içeren ve Dispatch Agent'a özel yerel bir araçtır. Mimarın şematik veritabanından ayrıdır.
A2A istemcisini uygulama
Gönderme aracısının uzak mimarımızla iletişim kurmasına izin vermek için Uzak A2A Aracısı tanımlamamız gerekir. Bu, Dispatch temsilcisine mimarın nerede bulunacağını ve "Temsilci Kartı"nın nasıl göründüğünü bildirir.
👉✏️ $HOME/way-back-home/level_4/backend/dispatch_agent/agent.py içindeki #REPLACE-REMOTEA2AAGENT ifadesini aşağıdakilerle değiştirin:
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,
)
Yayın araçlarının işleyiş şekli
Önceki temsilcide araçlar standart bir "İstek-Yanıt" modelini takip ediyordu. Temsilci soru soruyor, araç yanıt veriyor ve etkileşim sona eriyordu. Ancak Ozymandias'ta tehlikeler, var olup olmadıklarını sormanızı beklemez. Bunun için Yayın Aracı gerekir.
Akış araçları, işlevlerin ara sonuçları gerçek zamanlı olarak temsilciye geri aktarmasına olanak tanır. Böylece temsilci, değişikliklere anında tepki verebilir. Yaygın kullanım alanları arasında dalgalanan hisse senedi fiyatlarını izleme veya bizim durumumuzda canlı video akışını durum değişiklikleri açısından izleme yer alır.
Standart araçlardan farklı olarak, akış aracı AsyncGenerator olarak işlev gören bir Asynchronous Function'dır. Bu, tek bir değeri return yerine zaman içinde birden fazla güncellemeyi yield anlamına gelir.
ADK'da bir yayın aracı tanımlamak için aşağıdaki teknik şartlara uymanız gerekir:
- Asenkron İşlev: Araç,
async defile tanımlanmalıdır. - AsyncGenerator Dönüş Türü: İşlev,
AsyncGeneratordöndürecek şekilde yazılmalıdır. İlk parametre, elde edilen veri türüdür (ör.str) ve ikincisi genellikleNoneolur. - Giriş Akışları: Video Akışı Araçları'nı kullanırız. Bu modda, gerçek video/ses akışı (
LiveRequestQueue) doğrudan işleve aktarılır. Böylece araç, aracının gördüğü kareleri "görebilir".
Bir yayın aracını Sentinel olarak düşünebilirsiniz. Siz ve Gönderim temsilcisi planları tartışırken Sentinel, güvenliğinizi sağlamak için arka planda çalışır ve her video karesini sessizce işler.
Arka Plan İzleme Aracı'nı Uygulama
Şimdi monitor_for_hazard aracını uygulayacağız. Bu araç, input_stream (video kareleri) alır, bunları ayrı ve basit bir görsel işleme çağrısı kullanarak analiz eder ve yalnızca tehlike tespit edildiğinde yield uyarısı verir.
👉✏️ $HOME/way-back-home/level_4/backend/dispatch_agent/agent.py içinde #REPLACE_MONITOR_HAZARD yerine aşağıdaki mantığı kullanın:
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)
Dispatch aracısını uygulama
Dispatch Agent, birincil arayüzünüz ve düzenleyicinizdir. Çift yönlü yayın bağlantısını (canlı sesiniz ve videonuz) yönettiği için her zaman konuşmanın kontrolünü elinde tutması gerekir. Bunu yapmak için belirli bir ADK özelliğini kullanacağız: Agent-as-a-Tool.
Konsept: Araç Olarak Ajan ve Alt Ajanlar
Çok agent'lı sistemler oluştururken sorumluluğun nasıl paylaşılacağına karar vermeniz gerekir. Kurtarma görevimizde bu ayrım çok önemlidir:
- Agent-as-a-Tool: Bu, çift yönlü yayın merkezimiz için önerilen yaklaşımdır. Gönderme temsilcisi (A Temsilcisi) bir araç olarak Mimar temsilcisini (B Temsilcisi) aradığında Mimar'ın verileri Gönderme'ye geri aktarılır. Dispatch daha sonra bu verileri yorumlar ve sizin için bir yanıt oluşturur. Dispatch kontrolü elinde tutar ve sonraki tüm kullanıcı girişlerini işlemeye devam eder.
- Alt Aracı: Alt aracı ilişkisinde sorumluluk tamamen aktarılır. Gönderim, sizi alt aracı olarak Mimar'a yönlendirdiyse "vizyonu" ve sohbet becerileri olmayan bir veritabanı API'siyle doğrudan konuşuyor olursunuz. Birincil aracı (Gönderim) etkili bir şekilde bilgilendirilmemiş olur.
Agent-as-a-Tool'u kullanarak, Architect'in uzmanlık bilgisinden yararlanırken bidi-streaming aracısının akıcı ve insana benzer etkileşimini koruruz.
Yönlendirme Mantığını Kodlama
Artık architect_agent öğemizi AgentTool içine yerleştirecek ve Dispatch temsilcisine "Mantık Haritası" sağlayacağız. Bu harita, aracının kasadan ne zaman veri getireceğini ve arka plan sentry'sinden elde edilen bulguları ne zaman bildireceğini tam olarak belirtir.
Dispatch'e hiç göz kırpmayan "gözler" vermek için önceki adımda oluşturduğumuz Akış Aracı'na erişim izni vermeliyiz.
ADK'da tools listesine bir AsyncGenerator işlevi (ör. monitor_for_hazard) eklediğinizde aracı, bunu kalıcı bir arka plan işlemi olarak değerlendirir. Aracı tek seferlik yürütmek yerine, aracının çıkışına "abone olunur". Bu sayede, Sentinel arka planda sessizce tehlike uyarıları verirken Dispatch birincil görüşmesine devam edebilir.
👉✏️ $HOME/way-back-home/level_4/backend/dispatch_agent/agent.py içindeki #REPLACE_AGENT_TOOLS ifadesini aşağıdakilerle değiştirin:
tools=[AgentTool(agent=architect_agent), monitor_for_hazard],
Doğrulama
👉💻 Her iki aracı da yapılandırdıktan sonra canlı çoklu aracı etkileşimini test edebiliriz.
- A terminalinde Architect Agent'ı başlatın:
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
- Yeni bir terminalde (B terminali) Dispatch Agent'ı çalıştırın:
cd $HOME/way-back-home/level_4/backend/
cp architect_agent/.env .env
uv run adk web
gemini-live simülatöründe adk web gibi gerçek zamanlı, çok formatlı bir model kullanan çok agentlı bir sistemi test etmek için belirli bir iş akışı gerekir. Simülasyon aracı, araç çağrılarını incelemek için mükemmeldir ancak bu tür bir modelle ilk kez görüntü işlenirken bilinen bir uyumsuzluk söz konusudur.
- Cloud Shell araç çubuğunda Web önizlemesi simgesini tıklayın. Bağlantı noktasını değiştir'i seçin, 8000 olarak ayarlayın ve Değiştir ve Önizle'yi tıklayın.
👉dispatch_agent'ı seçin, Planı yükleyin ve Beklenen Hatayı İşleyin
Bu en önemli adımdır. Temsilciye resim bağlamını sağlamamız gerekir.
- Arayüz yüklendiğinde istendiğinde mikrofonunuza erişmesine izin verin.
- Bu plan görüntüsünü bilgisayarınıza indirin:
adk webarayüzünde ataş simgesini tıklayın ve az önce indirdiğiniz plan görüntüsünü yükleyin.
⚠️⚠️400 INVALID_ARGUMENT hatası görürsünüz. Bu beklenen bir durumdur.⚠️⚠️
Bu hata, adk web resim işleyicinin tek seferlik yükleme için gemini-live modelinin API'siyle tam olarak uyumlu olmamasından kaynaklanır. Ancak resim oturum bağlamına başarıyla eklenmiştir.
- 👉 Hatayı temizlemek için tarayıcı sayfasını yeniden yüklemeniz yeterlidir.
Assembly İşlemini Tetikleme
👉 Yeniden yükledikten sonra hata ortadan kalkar ve sohbet geçmişinde plan görüntüsünü görürsünüz. Temsilci artık ihtiyaç duyduğu görsel bağlama sahip.
- Mikrofonu açmak için mikrofon simgesini tıklayın. Arayüzde "Dinleniyor..." mesajı gösterilir.
- Sesli komutu söyleyin: "start to assemble" (monte etmeye başla).
- Temsilci, isteğinizi işler ve kullanıcı arayüzü "Konuşuluyor..." olarak değişir. Gerekli parçaların listelendiği yalnızca sesli bir yanıt duyarsınız.
4. Temsilciden temsilciye araç çağrılarını doğrulama
👉 İlk sesli yanıt, sistemin çalıştığını onaylar ancak asıl sihir, çoklu aracı iletişim izindedir.
- Mikrofonu kapatın.
- Sayfayı bir kez daha yenileyin.
Sol taraftaki "İzleme" paneli artık doldurulur. Başarıyla tamamlanan yürütme akışının tamamını görebilirsiniz:
dispatch_agentİlk aramalarmonitor_for_hazard- Ardından, şematik verileri almak için
architect_agent'ye birden fazlaexecute_architectçağrısı yapar.
Bu sıra, çoklu aracı iş akışının tamamının doğru şekilde çalıştığını onaylar: dispatch_agent, isteği aldı, veri alma görevini bir araç çağrısı aracılığıyla architect_agent'ye devretti ve kullanıcının komutunu yerine getirmek için verileri geri aldı.
Çift yönlü akış bağlantınız artık arka planda izleme ve çoklu temsilci işbirliğini destekliyor. Ardından, bu karmaşık yanıtları ön uçta nasıl ayrıştıracağımızı öğreneceğiz.
👉💻 Çıkmak için her iki terminalde de Ctrl+c tuşuna basın.
5. Çok formatlı canlı etkinlik yayınlarının ayrıntılı incelemesi
Önceki adımda, yerleşik geliştirme sunucusu adk web'yı kullanarak çok agentlı sistemimizi başarıyla doğruladık. Bu yardımcı program, oturumu, akışları ve aracı yaşam döngüsünü otomatik olarak yönetmek için varsayılan bir ADK çalıştırıcısı kullanır. Ancak FastAPI hizmetimiz (main.py) gibi bağımsız ve üretime hazır bir uygulama oluşturmak için açık kontrole ihtiyacımız var. Ses, video ve metin için çift yönlü akışları işleyen temel bileşen olduğundan, canlı kullanıcı oturumlarını yönetmek için ADK Runner'ı manuel olarak oluşturup yönetmemiz gerekir.
Model-Kod-Model Döngüsü
Sistemin gerçek zamanlı olarak nasıl çalıştığını anlamak için tek bir görev oturumunun yaşam döngüsünü inceleyelim. Bu döngü, LlmRequest ve LlmResponse nesnelerinin sürekli olarak değiş tokuş edilmesini temsil eder.
- Görsel Bağlantı: Bağlantıyı siz başlatır ve web kameranızı/ekranınızı paylaşırsınız. Yüksek kaliteli JPEG kareler,
realtimeInputüzerinden yukarı akış olarak gönderilmeye başlar (LiveRequestQueuekullanılarak). - Sentinel Etkinleştirme: Sistem, ilk "Merhaba" uyarısını gönderir. Talimatlarına göre, Dispatch Agent,
monitor_for_hazardStreaming Tool'u hemen tetikler. Bu işlem, gelen her kareyi sessizce izleyen bir arka plan döngüsü başlatır. - Pilot Komutu: İletişim cihazına "Montaja başla." diyorsunuz.
- Vocal Upstream: Sesiniz 16 kHz ses olarak kaydedilir ve video kareleriyle birlikte Upstream'e gönderilir.
- Yetki verme (A2A): Dispatch, amacınızı "duyar". Şemaların eksik olduğunu fark eden Mimar Aracısı,
AgentTool(Araç Olarak Aracı) protokolünü kullanarak çağrıda bulunur. - Bilgi Alma: Mimar, Redis veritabanını sorgular ve parça listesini Dispatch'e döndürür. Gönderme,"Oturumun Yöneticisi" olarak kalır ve verileri size aktarmadan alır.
- Bilgi Amaçlı Yayın: Dispatch, hem metin hem de doğal ses içeren bir
modelTurn(Yayın) gönderir: "Mimar Onayladı. Gerekli alt küme: Warp Çekirdeği, Akı Borusu, İyon İtici." - Kriz: Çalışma tezgahındaki bir parça aniden dengesini kaybeder ve beyaz renkte parlamaya başlar.
- Bağımsız Algılama: Arka plandaki
monitor_for_hazarddöngüsü (Sentinel), parıltıyı içeren belirli JPEG çerçevesini alır. Gemini'ı çağırarak kareyi işler ve tehlikeyi tanımlar. - Aşağı Akış Güvenliği: Yayın aracı
yieldssonuç. Bu bir Bidi-Streaming aracısı olduğundan Dispatch, mevcut durumu kesintiye uğratarak kritik bir güvenlik uyarısını hemen Downstream olarak gönderebilir: "Tehlike tespit edildi! Veri Kristali'ni etkisiz hâle getiriyoruz. KIRMIZI kutuya taşıyın."
Aracının çalışma zamanı yapılandırmasını ayarlama
ADK'daki RunConfig, bir aracının davranışının ayrıntılı şekilde yapılandırılmasına olanak tanır. Bu yapılandırmaya, aracının akış verilerini nasıl işlediği ve çeşitli yöntemlerle nasıl etkileşimde bulunduğu da dahildir.
streaming_mode, gerçek zamanlı ve çift yönlü iletişim için BIDI olarak ayarlanır. Böylece hem kullanıcı hem de temsilci aynı anda konuşup dinleyebilir. response_modalities parametresi, aracının üretebileceği çıkış türlerini (ör. ses ve metin) tanımlar. input_audio_transcription, aracının kullanıcının gelen konuşmasını nasıl işleyip yazıya dökeceğini yapılandırır. Daha esnek bir deneyim oluşturmak için session_resumption, bağlantı kesilirse temsilcinin sohbet bağlamını hatırlamasını ve devam etmesini sağlar. Son olarak, proactivity, aracının doğrudan kullanıcı komutu olmadan işlem veya konuşma başlatmasına (ör. spontane tehlike uyarısı verme) olanak tanırken enable_affective_dialog, aracının daha doğal ve empatik yanıtlar oluşturmasına olanak tanır. ADK'nın RunConfig'i hakkında daha fazla bilgiyi burada bulabilirsiniz.
👉✏️ $HOME/way-back-home/level_4/backend/main.py dosyanızda #REPLACE_RUN_CONFIG yer tutucusunu bulun ve aşağıdaki ayrıştırma mantığıyla değiştirin:
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,
)
İsteği Ajan'a iletme
Ardından, kullanıcının Volatile Workbench'inden Dispatch Agent'a WebSocket üzerinden gerçek zamanlı ve çok formatlı veriler aktaran temel iletişim bağlantısını uygulayacağız. Temsilci sürekli olarak "görür" (video kareleri) ve "duyar" (sesli komutlar). Mantık, veri akışını sürekli olarak alır, gelen ikili ses parçaları ile JSON ile sarmalanmış metin/resim paketleri arasında ayrım yapar ve bunları büyük ikili nesne (BLOB) (multimedya için) veya Content (metin için) nesneleri içine kapsülleyerek iki yönlü aracı oturumunu desteklemek üzere LiveRequestQueue'ya gönderir.
$HOME/way-back-home/level_4/backend/main.py dosyanızda #PROCESS_AGENT_REQUEST yer tutucusunu bulun ve aşağıdaki ayrıştırma mantığıyla değiştirin:
# 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
Çok formatlı veriler artık aracıya gönderiliyor.
Yanıtı uygulama: Aşağı akış etkinlik verileri yapısı
ADK ile çift yönlü (canlı) bir aracı çalıştırdığınızda, aracıdan gelen veriler temel GenAI SDK yapılarından devralınan belirli bir Etkinlik türünde paketlenir. Event döngünüzde aldığınız async for event in runner.run_live(...) nesnesi, her biri farklı bir bilgi türü için olan birkaç isteğe bağlı alan içeren tek bir nesnedir:
İçeriğin yapısı:
- Temsilci Konuştuğunda (
.server_content): Alan yalnızca düz metin değildir.Partslistesini içerir. HerPart, bir tür veri için kapsayıcıdır. Bu veri, metin dizesi (ör."The part is stable.") veya ham ses blob'u (ses) olabilir. - Aracı İşlem Yaptığında (
.tool_call): Alan,FunctionCallnesnelerinin listesini içerir. HerFunctionCall, aracın adını ve giriş bağımsız değişkenlerini, arka uç kodunuzun kolayca okuyup yürütebileceği temiz bir biçimde belirten basit ve yapılandırılmış bir nesnedir.
👀 run_live döngüsünün sağladığı tek bir Event öğesine bakarsanız event.model_dump(by_alias=True) tarafından üretilen JSON, GenAI SDK şekillerine sıkı sıkıya bağlı kalarak şu şekilde görünür:
{
"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" }
}
]
}
}
👉✏️ Artık tam etkinlik verilerini iletmek için main.py içindeki downstream_task değerini güncelleyeceğiz. Bu mantık, yapay zekanın her "düşüncesinin" geminin teşhis terminaline kaydedilmesini ve ön uç kullanıcı arayüzüne tek bir JSON nesnesi olarak gönderilmesini sağlar.
$HOME/way-back-home/level_4/backend/main.py dosyanızda #PROCESS_AGENT_RESPONSE yer tutucusunu bulun ve aşağıdaki ayrıştırma mantığıyla değiştirin:
# 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)
Görev Yürütme
Arka uç kasası bağlandıktan ve her iki aracı da yapılandırdıktan sonra tüm sistemler kullanıma hazır hâle gelir. Aşağıdaki adımlar, tam uygulamayı başlatarak yeni oluşturduğunuz iki aracılı sistemle etkileşim kurmanıza olanak tanır.
Amaç: Tezgahınızda görünen rastgele atanmış warp sürücüsünü monte edin. Protokol: Özellikle belirli bileşenlerle ilgili tehlike uyarıları olmak üzere, Gönderme Temsilcisi'nin sözlü talimatlarına uymanız gerekir.
Uzmanı (Mimar) etkinleştirme
👉💻 İlk terminal pencerenizde Architect aracını başlatın. Bu arka uç hizmeti, Redis kasasına bağlanır ve Dispatcher'dan gelen şematik istekleri bekler.
# 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
(Bu terminali çalışır durumda bırakın. Artık etkin "veritabanı aracınız" olur.)
Kokpiti (Gönderici) başlatma
👉💻 Yeni bir terminal penceresinde (Terminal B), ön uç kullanıcı arayüzünü oluşturup kullanıcı arayüzüne hizmet veren ve tüm canlı iletişimi yöneten ana Dispatch aracısını başlatacağız.
# 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
(Bu işlem, birincil sunucuyu 8080 bağlantı noktasında başlatır.)
Test Senaryosunu Çalıştırma
Sistem artık kullanıma sunuldu. Amacınız, montajı tamamlamak için temsilcinin talimatlarını uygulamak.
- 👉 Çalışma Tezgahı'na erişme:
- Cloud Shell araç çubuğunda Web önizlemesi simgesini tıklayın.
- Bağlantı noktasını değiştir'i seçin, 8080 olarak ayarlayın ve Değiştir ve Önizle'yi tıklayın.
- 👉 Görevi başlatın:
- Arayüz yüklendiğinde ekranınıza ve mikrofonunuza erişmesine izin verdiğinizden emin olun.
- Paylaşılacak bir sekme veya pencere seçmeniz istenir. Pencereyi paylaşıyorsanız sorun yaşamamak için pencerede TEK bir sekme olduğundan emin olun.
- Rastgele bir ada sahip sürücü (ör. "NOVA-V", "OMEGA-9") atanır.
- Arayüz yüklendiğinde ekranınıza ve mikrofonunuza erişmesine izin verdiğinizden emin olun.
- 👉 Assembly Döngüsü:
- İstek: Sürücüyü monte etmeye başlamak için "Monte etmeye başla" deyin.
- Mimar Yanıtı: Temsilci, sürücüyü monte etmek için doğru parçaları sağlar.
- Tehlike Kontrolü: Çalışma tezgahında tehlikeli gibi görünen bir parça olduğunda:
- Gönderim aracısının
monitor_for_hazardaracı, bunu görsel olarak tanımlar. - "GÖRSEL TEHLİKE UYARISI" sonucu elde edilir. (Bu işlem yaklaşık 30 saniye sürer.)
- Tehlikeyi devre dışı bırakmak için hangi çöp kutusunun kullanılacağını kontrol eder.
- Gönderim aracısının
- İşlem: Görevlendirme Temsilcisi size doğrudan bir komut verecek: "Tehlike doğrulandı. XXX'i hemen kırmızı kutuya koyun." Devam etmek için bu talimatı uygulamanız gerekir.
- İstek: Sürücüyü monte etmeye başlamak için "Monte etmeye başla" deyin.
Görev Tamamlandı. Başarıyla etkileşimli bir çoklu aracı sistemi oluşturdunuz. Hayatta kalanlar güvende, roket atmosferi aştı ve "Way Back Home" yolculuğunuz devam ediyor.
👉💻 Çıkmak için her iki terminalde de Ctrl+c tuşuna basın.
6. Üretim sürümüne dağıtma (isteğe bağlı)
Temsilciyi yerel olarak başarıyla test ettiniz. Şimdi, Mimar'ın sinirsel çekirdeğini geminin ana bilgisayarlarına (Cloud Run) yüklememiz gerekiyor. Bu sayede, Dispatch temsilcisinin her yerden sorgulayabileceği kalıcı ve bağımsız bir hizmet olarak çalışabilir.
Güvenli Depo'yu (altyapı) sağlama
Aracı dağıtmadan önce kalıcı belleğini (Memorystore) ve bu belleğe erişmek için güvenli kanalı (VPC bağlayıcısı) oluşturmamız gerekir.
👉💻 Memorystore örneğini (Redis Vault) oluşturun:
export REGION="us-central1"
gcloud redis instances create ozymandias-vault-prod --size=1 --tier=basic --region=${REGION}
👉💻 Vault'un Ağ Adresini Alma: Bu komutu çalıştırın ve host IP adresini kopyalayın. Bu, yeni Redis örneğinizin özel adresidir.
gcloud redis instances describe ozymandias-vault-prod --region=us-central1
👉💻 VPC erişim bağlayıcısını oluşturun (güvenli köprü): Bu bağlayıcı, özel bir köprü görevi görerek Cloud Run'ın VPC'nizdeki Redis örneğine erişmesine olanak tanır.
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
👉💻 Verileri yükleyin:
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
Aracı Uygulamasını Dağıtma
Aracı görüntüsünü derleme ve oluşturma
👉💻 Arka uç dizinine gidin ve Dockerfile'ı oluşturun.
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
👉💻 Uygulamayı bir container görüntüsüne paketleyin.
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'a dağıt
👉💻 Aracıyı Cloud Run'a dağıtın. Redis IP'sini ekleyip VPC bağlayıcısını doğrudan başlatma komutuna bağlayacağız. Bu, temsilcinin veritabanına güvenli ve gizli bir bağlantıyla başlamasını sağlar.
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 sunucusunun çalışıp çalışmadığını doğrulayın.
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
Komut tamamlandığında Hizmet URL'si gösterilir. Architect Agent artık bulutta kullanıma sunuldu. Kasasına kalıcı olarak bağlı ve diğer aracılara şematik veriler sunmaya hazır.
Sevk Merkezini Üretim Ana Bilgisayarına Dağıtma
Bulutta çalışan Architect Agent'ı dağıttıktan sonra Dispatch Hub'ı dağıtmamız gerekir. Bu aracı, canlı ses/video akışlarını işleyip veritabanı sorgularını Architect'in güvenli uç noktasına yönlendirerek birincil kullanıcı arayüzü olarak görev yapar.
👉💻 Cloud Shell terminalinizde aşağıdaki komutu çalıştırın. Arka uç dizininizde eksiksiz ve çok kademeli bir Dockerfile oluşturulur.
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
Aracı/ön uç görüntüsünü derleme ve oluşturma
👉💻 Dispatch aracısının kodunu (main.py) içeren arka uç dizinine gidin ve kodu bir container görüntüsüne paketleyin.
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'a dağıt
👉💻 Dispatch Hub'ı Cloud Run'a dağıtın. Mimarın URL'sini bir ortam değişkeni olarak ekleyerek iki bulutta yerel aracımız arasında kritik bağlantıyı oluşturacağız.
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}"
Komut tamamlandığında bir hizmet URL'si (ör. https://mission-bravo-...run.app). Uygulama artık bulutta yayında.
👉 Google Cloud Run sayfasına gidin ve listeden biometric-scout hizmetini seçin. 
👉 Hizmet ayrıntıları sayfasının üst kısmında gösterilen herkese açık URL'yi bulun. 
Son Sistem Kontrolü (Uçtan Uca Test)
👉 Artık canlı sistemle etkileşim kuracaksınız.
- URL'yi alma: Son dağıtım komutunun çıktısından Hizmet URL'sini kopyalayın (
run.appile bitmelidir). - Kokpiti açın: URL'yi web tarayıcınıza yapıştırın.
- İletişimi Başlatma: Arayüz yüklendiğinde ekranınıza ve mikrofonunuza erişmesine izin verdiğinizden emin olun.
- Veri İste: Bir sürücü atandığında montajın başlatılmasını isteyin. Örneğin: "Birleştirmeye başlayın"
Artık tamamen Google Cloud'da çalışan, tam olarak dağıtılmış bir çoklu aracı sistemiyle etkileşim kuruyorsunuz.
Çoklu ajan sistemi, son kapsama halkasını yerine kilitler ve düzensiz radyasyon, sabit bir uğultuya dönüşür.
"Warp Drive: STABILIZED. Rescue Craft: ENGINES IGNITED."
Monitörünüzde, atmosfer çökerken uzaylı gemisi Ozymandias'ın ufalanan yüzeyinden kıl payı kurtularak yukarı doğru hızla ilerliyor. Geminizin yanında güvenli bir yörüngeye yerleşiyor ve iletişim cihazı, hayatta kalanların sesleriyle doluyor. Kurtarma işlemi tamamlandığında ve eve dönüş yolunuz açıldığında uzaktan bağlantı kesilir.
Sayenizde hayatta kalanlar kurtarıldı.
0. seviyeye katıldıysanız Eve Dönüş görevindeki ilerleme durumunuzu kontrol etmeyi unutmayın.
