1. Présentation
Dans cet atelier de programmation, vous allez créer un agent Supply Chain Orchestrator. Cette application permet aux utilisateurs d'analyser l'inventaire, de suivre la logistique et de gérer les risques liés à la chaîne d'approvisionnement en langage naturel.
Nous allons exploiter l'Agent Development Kit (ADK) de Google pour créer une architecture multi-agents qui maintient le contexte, mémorise les préférences des utilisateurs via Vertex AI Memory Bank et interagit avec un ensemble de données volumineux stocké dans AlloyDB via MCP Toolbox.
Ce que vous allez faire
Une application Python Flask comprenant :
Agent orchestrateur global : agent racine qui gère le flux de conversation et la délégation.
Agents spécialisés : "InventorySpecialist" et "LogisticsManager" pour les tâches spécifiques au domaine.
Intégration de la mémoire : mémoire de session à court terme et mémoire à long terme à l'aide de Vertex AI Memory Bank.
UI narrative : interface Web qui visualise le processus de raisonnement de l'agent (contexte de trace).
Points abordés
- Découvrez comment utiliser Google ADK pour créer des agents et des sous-agents spécialisés.
- Découvrez comment intégrer Vertex AI Memory Bank pour la mémoire à long terme des agents.
- Découvrez comment utiliser MCP Toolbox pour connecter des agents aux outils de données AlloyDB.
- Implémenter des rappels ADK pour suivre et visualiser le raisonnement de l'agent
- Découvrez comment déployer la solution à l'aide de Cloud Run ou l'exécuter en local.
Architecture
Pile technologique
- AlloyDB pour PostgreSQL : sert de base de données opérationnelle hautes performances contenant plus de 50 000 enregistrements de la chaîne d'approvisionnement. Il alimente la recherche et la récupération vectorielles.
- MCP Toolbox for Databases : agit comme un "maître d'orchestre" en exposant les données AlloyDB sous forme d'outils exécutables que les agents peuvent appeler.
- Agent Development Kit (ADK) : framework utilisé pour définir les agents, les instructions et les outils.
- Vertex AI Memory Bank : fournit une mémoire à long terme, permettant à l'agent de se souvenir des préférences de l'utilisateur et des interactions passées au fil des sessions.
- Service de session Vertex AI : gère le contexte de conversation à court terme.
Le Flow
- Requête de l'utilisateur : l'utilisateur pose une question (par exemple, "Vérifie le stock de glace Premium").
- Vérification de la mémoire : l'orchestrateur vérifie la mémoire pour trouver des informations pertinentes sur le passé (par exemple, "L'utilisateur est responsable régional pour la zone EMEA").
- Délégation : l'orchestrateur délègue la tâche à l'InventorySpecialist.
- Exécution de l'outil : le spécialiste utilise les outils fournis par MCP Toolbox pour interroger AlloyDB.
- Réponse : l'agent traite les données et renvoie un tableau au format Markdown.
- Stockage de la mémoire : les interactions importantes sont enregistrées dans la mémoire.
Conditions requises
2. Avant de commencer
Créer un projet
- Dans la console Google Cloud, sur la page du sélecteur de projet, sélectionnez ou créez un projet Google Cloud.
- Assurez-vous que la facturation est activée pour votre projet Cloud. Découvrez comment vérifier si la facturation est activée sur un projet.
- Vous allez utiliser Cloud Shell, un environnement de ligne de commande exécuté dans Google Cloud. Cliquez sur "Activer Cloud Shell" en haut de la console Google Cloud.
- Une fois connecté à Cloud Shell, vérifiez que vous êtes déjà authentifié et que le projet est défini sur votre ID de projet à l'aide de la commande suivante :
gcloud auth list
- Exécutez la commande suivante dans Cloud Shell pour vérifier que la commande gcloud connaît votre projet.
gcloud config list project
- Si votre projet n'est pas défini, utilisez la commande suivante pour le définir :
gcloud config set project <YOUR_PROJECT_ID>
- Activez les API requises : suivez ce lien et activez les API.
Vous pouvez également utiliser la commande gcloud. Consultez la documentation pour connaître les commandes gcloud ainsi que leur utilisation.
Problèmes et dépannage
Syndrome du projet fantôme | Vous avez exécuté |
Barricade de facturation | Vous avez activé le projet, mais vous avez oublié le compte de facturation. AlloyDB est un moteur hautes performances. Il ne démarrera pas si le "réservoir" (la facturation) est vide. |
Décalage de la propagation de l'API | Vous avez cliqué sur "Activer les API", mais la ligne de commande indique toujours |
Quags de quota | Si vous utilisez un tout nouveau compte d'essai, vous pouvez atteindre un quota régional pour les instances AlloyDB. Si |
Agent du service"Masqué" | Il arrive que le compte de service AlloyDB ne reçoive pas automatiquement le rôle |
3. Configuration de la base de données
AlloyDB pour PostgreSQL est au cœur de notre application. Nous avons exploité ses puissantes fonctionnalités vectorielles et son moteur columnar intégré pour générer des embeddings pour plus de 50 000 enregistrements SCM. Cela permet une analyse vectorielle en temps quasi réel, ce qui permet à nos agents d'identifier les anomalies d'inventaire ou les risques logistiques dans de grands ensembles de données en quelques millisecondes.
Dans cet atelier, nous allons utiliser AlloyDB comme base de données pour les données de test. Il utilise des clusters pour stocker toutes les ressources, telles que les bases de données et les journaux. Chaque cluster possède une instance principale qui fournit un point d'accès aux données. Les tables contiennent les données réelles.
Commençons par créer un cluster, une instance et une table AlloyDB dans lesquels l'ensemble de données de test sera chargé.
- Cliquez sur le bouton ou copiez le lien ci-dessous dans le navigateur dans lequel l'utilisateur de la console Google Cloud est connecté.
Vous pouvez également accéder au terminal Cloud Shell depuis le projet dans lequel vous avez utilisé le compte de facturation, cloner le dépôt GitHub et accéder au projet à l'aide des commandes ci-dessous :
git clone https://github.com/AbiramiSukumaran/easy-alloydb-setup
cd easy-alloydb-setup
- Une fois cette étape terminée, le dépôt sera cloné dans votre éditeur Cloud Shell local. Vous pourrez ensuite exécuter la commande ci-dessous à partir du dossier du projet (assurez-vous d'être dans le répertoire du projet) :
sh run.sh
- Utilisez maintenant l'UI (en cliquant sur le lien dans le terminal ou sur le lien "Prévisualiser sur le Web" dans le terminal).
- Saisissez les détails de l'ID de projet, du nom du cluster et du nom de l'instance pour commencer.
- Allez prendre un café pendant que les journaux défilent. Pour en savoir plus sur le fonctionnement en coulisses, cliquez ici.
Problèmes et dépannage
Le problème de la patience | Les clusters de bases de données sont une infrastructure lourde. Si vous actualisez la page ou mettez fin à la session Cloud Shell parce qu'elle semble bloquée, vous risquez de vous retrouver avec une instance "fantôme" partiellement provisionnée et impossible à supprimer sans intervention manuelle. |
Région non concordante | Si vous avez activé vos API dans |
Clusters de zombies | Si vous avez déjà utilisé le même nom pour un cluster et que vous ne l'avez pas supprimé, le script peut indiquer que le nom du cluster existe déjà. Les noms de clusters doivent être uniques dans un projet. |
Délai d'inactivité de Cloud Shell | Si votre pause-café dure 30 minutes, Cloud Shell peut se mettre en veille et déconnecter le processus |
4. Provisionnement de schémas
Une fois votre cluster et votre instance AlloyDB en cours d'exécution, accédez à l'éditeur SQL AlloyDB Studio pour activer les extensions d'IA et provisionner le schéma.
Vous devrez peut-être attendre que votre instance soit créée. Une fois le cluster créé, connectez-vous à AlloyDB à l'aide des identifiants que vous avez créés. Utilisez les données suivantes pour vous authentifier auprès de PostgreSQL :
- Nom d'utilisateur : "
postgres" - Base de données : "
postgres" - Mot de passe : "
alloydb" (ou celui que vous avez défini lors de la création)
Une fois l'authentification réussie dans AlloyDB Studio, les commandes SQL sont saisies dans l'éditeur. Vous pouvez ajouter plusieurs fenêtres de l'éditeur en cliquant sur le signe plus à droite de la dernière fenêtre.
Vous saisirez des commandes pour AlloyDB dans des fenêtres d'éditeur, en utilisant les options "Exécuter", "Mettre en forme" et "Effacer" selon les besoins.
Activer les extensions
Pour créer cette application, nous allons utiliser les extensions pgvector et google_ml_integration. L'extension pgvector vous permet de stocker et de rechercher des embeddings vectoriels. L'extension google_ml_integration fournit les fonctions que vous utilisez pour accéder aux points de terminaison de prédiction Vertex AI afin d'obtenir des prédictions en SQL. Activez ces extensions en exécutant les LDD suivants :
CREATE EXTENSION IF NOT EXISTS google_ml_integration CASCADE;
CREATE EXTENSION IF NOT EXISTS vector;
Créer une table
Vous pouvez créer une table à l'aide de l'instruction LDD ci-dessous dans AlloyDB Studio :
DROP TABLE IF EXISTS shipments;
DROP TABLE IF EXISTS products;
-- 1. Product Inventory Table
CREATE TABLE products (
id SERIAL PRIMARY KEY,
name VARCHAR(255) NOT NULL,
category VARCHAR(100),
stock_level INTEGER,
distribution_center VARCHAR(100),
region VARCHAR(50),
embedding vector(768),
last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 2. Logistics & Shipments
CREATE TABLE shipments (
shipment_id SERIAL PRIMARY KEY,
product_id INTEGER REFERENCES products(id),
status VARCHAR(50), -- 'In Transit', 'Delayed', 'Delivered', 'Pending'
estimated_arrival TIMESTAMP,
route_efficiency_score DECIMAL(3, 2)
);
La colonne embedding permettra de stocker les valeurs vectorielles de certains champs de texte.
Ingestion de données
Exécutez l'ensemble d'instructions SQL ci-dessous pour insérer 50 000 enregistrements dans la table "products" :
-- We use a CROSS JOIN pattern with realistic naming segments to create meaningful variety
DO $$
DECLARE
brand_names TEXT[] := ARRAY['Artisan', 'Nature', 'Elite', 'Pure', 'Global', 'Eco', 'Velocity', 'Heritage', 'Aura', 'Summit'];
product_types TEXT[] := ARRAY['Ice Cream', 'Body Wash', 'Laundry Detergent', 'Shampoo', 'Mayonnaise', 'Deodorant', 'Tea', 'Soup', 'Face Cream', 'Soap'];
variants TEXT[] := ARRAY['Classic', 'Gold', 'Premium', 'Eco-Friendly', 'Organic', 'Night-Repair', 'Extra-Fresh', 'Zero-Sugar', 'Sensitive', 'Maximum-Strength'];
regions TEXT[] := ARRAY['EMEA', 'APAC', 'LATAM', 'NAMER'];
dcs TEXT[] := ARRAY['London-Hub', 'Mumbai-Central', 'Sao-Paulo-Logistics', 'Singapore-Port', 'Rotterdam-Gate', 'New-York-DC'];
BEGIN
INSERT INTO products (name, category, stock_level, distribution_center, region)
SELECT
b || ' ' || v || ' ' || t as name,
CASE
WHEN t IN ('Ice Cream', 'Mayonnaise', 'Tea', 'Soup') THEN 'Food & Refreshment'
WHEN t IN ('Body Wash', 'Shampoo', 'Deodorant', 'Face Cream', 'Soap') THEN 'Personal Care'
ELSE 'Home Care'
END as category,
floor(random() * 20000 + 100)::int as stock_level,
dcs[floor(random() * 6 + 1)] as distribution_center,
regions[floor(random() * 4 + 1)] as region
FROM
unnest(brand_names) b,
unnest(variants) v,
unnest(product_types) t,
generate_series(1, 50); -- 10 * 10 * 10 * 50 = 50,000 records
END $$;
Insérons des enregistrements spécifiques à la démo pour garantir des réponses prévisibles aux questions de type "executive".
-- These ensure you have predictable answers for specific "Executive" questions
INSERT INTO products (name, category, stock_level, distribution_center, region) VALUES
('Magnum Ultra Gold Limited Edition', 'Food & Refreshment', 45, 'Rotterdam-Gate', 'EMEA'),
('Dove Pro-Health Deep Moisture', 'Personal Care', 12000, 'Mumbai-Central', 'APAC'),
('Hellmanns Real Organic Mayonnaise', 'Food & Refreshment', 8000, 'London-Hub', 'EMEA');
Insérer des données d'expédition
-- Shipments Generation (More shipments than products)
INSERT INTO shipments (product_id, status, estimated_arrival, route_efficiency_score)
SELECT
id,
CASE
WHEN random() > 0.8 THEN 'Delayed'
WHEN random() > 0.4 THEN 'In Transit'
ELSE 'Delivered'
END,
NOW() + (random() * 10 || ' days')::interval,
(random() * 0.5 + 0.5)::decimal(3,2)
FROM products
WHERE random() > 0.3; -- Create shipments for ~70% of products
-- Add duplicate shipments for some products to show complex logistics
INSERT INTO shipments (product_id, status, estimated_arrival, route_efficiency_score)
SELECT id, 'In Transit', NOW() + INTERVAL '12 days', 0.88
FROM products
LIMIT 5000;
Accorder l'autorisation
Exécutez l'instruction ci-dessous pour accorder l'exécution de la fonction "embedding" :
GRANT EXECUTE ON FUNCTION embedding TO postgres;
Attribuer le RÔLE Utilisateur Vertex AI au compte de service AlloyDB
Dans la console Google Cloud IAM, accordez au compte de service AlloyDB (qui ressemble à ceci : service-<<PROJECT_NUMBER>>@gcp-sa-alloydb.iam.gserviceaccount.com) l'accès au rôle "Utilisateur Vertex AI". PROJECT_NUMBER correspondra au numéro de votre projet.
Vous pouvez également exécuter la commande ci-dessous à partir du terminal Cloud Shell :
PROJECT_ID=$(gcloud config get-value project)
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:service-$(gcloud projects describe $PROJECT_ID --format="value(projectNumber)")@gcp-sa-alloydb.iam.gserviceaccount.com" \
--role="roles/aiplatform.user"
Générer des embeddings
Ensuite, générons des embeddings vectoriels pour des champs de texte spécifiques et pertinents :
WITH
rows_to_update AS (
SELECT
id
FROM
products
WHERE
embedding IS NULL
LIMIT
5000 )
UPDATE
products
SET
embedding = ai.embedding('text-embedding-005', name || ' ' || category || ' ' || distribution_center || ' ' || region)::vector
FROM
rows_to_update
WHERE
products.id = rows_to_update.id
AND embedding IS null;
Dans l'instruction ci-dessus, nous avons défini la limite sur 5 000. Veillez donc à l'exécuter plusieurs fois jusqu'à ce qu'il n'y ait plus de ligne dans le tableau avec l'intégration de colonne définie sur NULL.
Problèmes et dépannage
La boucle "Mot de passe oublié" | Si vous avez utilisé la configuration "En un clic" et que vous ne vous souvenez plus de votre mot de passe, accédez à la page "Informations de base sur l'instance" dans la console, puis cliquez sur "Modifier" pour réinitialiser le mot de passe |
Erreur "Extension introuvable" | Si |
Délai de propagation IAM | Vous avez exécuté la commande IAM |
Incompatibilité de la dimension du vecteur | La table |
Faute de frappe dans l'ID du projet | Dans l'appel |
5. Configuration des outils et de la boîte à outils
MCP Toolbox for Databases est un serveur MCP Open Source pour les bases de données. Il vous permet de développer des outils plus facilement, plus rapidement et de manière plus sécurisée en gérant les complexités telles que le regroupement de connexions, l'authentification et plus encore. La boîte à outils vous aide à créer des outils d'IA générative qui permettent à vos agents d'accéder aux données de votre base de données.
Nous utilisons la boîte à outils MCP (Model Context Protocol) pour les bases de données comme "chef d'orchestre". Il sert de middleware standardisé entre nos agents et AlloyDB. En définissant une configuration tools.yaml, la boîte à outils expose automatiquement des opérations de base de données complexes sous forme d'outils propres et exécutables tels que search_products_by_context ou check_inventory_levels. Cela élimine le besoin de regroupement de connexions manuel ou de code SQL récurrent dans la logique de l'agent.
Installer le serveur Toolbox
Dans le terminal Cloud Shell, créez un dossier pour enregistrer votre nouveau fichier YAML d'outils et le fichier binaire de la boîte à outils :
mkdir scm-agent-toolbox
cd scm-agent-toolbox
Dans ce nouveau dossier, exécutez l'ensemble de commandes suivant :
# see releases page for other versions
export VERSION=0.27.0
curl -L -o toolbox https://storage.googleapis.com/genai-toolbox/v$VERSION/linux/amd64/toolbox
chmod +x toolbox
Ensuite, créez le fichier tools.yaml dans ce nouveau dossier en accédant à l'éditeur Cloud Shell, puis copiez le contenu de ce fichier de dépôt dans le fichier tools.yaml.
sources:
supply_chain_db:
kind: "alloydb-postgres"
project: "YOUR_PROJECT_ID"
region: "us-central1"
cluster: "YOUR_CLUSTER"
instance: "YOUR_INSTANCE"
database: "postgres"
user: "postgres"
password: "YOUR_PASSWORD"
tools:
search_products_by_context:
kind: postgres-sql
source: supply_chain_db
description: Find products in the inventory using natural language search and vector embeddings.
parameters:
- name: search_text
type: string
description: Description of the product or category the user is looking for.
statement: |
SELECT name, category, stock_level, distribution_center, region
FROM products
ORDER BY embedding <=> ai.embedding('text-embedding-005', $1)::vector
LIMIT 5;
check_inventory_levels:
kind: postgres-sql
source: supply_chain_db
description: Get precise stock levels for a specific product name.
parameters:
- name: product_name
type: string
description: The exact or partial name of the product.
statement: |
SELECT name, stock_level, distribution_center, last_updated
FROM products
WHERE name ILIKE '%' || $1 || '%'
ORDER BY stock_level DESC;
track_shipment_status:
kind: postgres-sql
source: supply_chain_db
description: Retrieve real-time logistics and shipping status for a specific region or product.
parameters:
- name: region
type: string
description: The geographical region to filter shipments (e.g., EMEA, APAC).
statement: |
SELECT p.name, s.status, s.estimated_arrival, s.route_efficiency_score
FROM shipments s
JOIN products p ON s.product_id = p.id
WHERE p.region = $1
ORDER BY s.estimated_arrival ASC;
analyze_supply_chain_risk:
kind: postgres-sql
source: supply_chain_db
description: Rerank and filter shipments based on risk profiles and efficiency scores using Google ML reranker.
parameters:
- name: risk_context
type: string
description: The business context for risk analysis (e.g., 'heatwave impact' or 'port strike').
statement: |
WITH initial_ranking AS (
SELECT s.shipment_id, p.name, s.status, p.distribution_center,
ROW_NUMBER() OVER () AS ref_number
FROM shipments s
JOIN products p ON s.product_id = p.id
WHERE s.status != 'Delivered'
LIMIT 10
),
reranked_results AS (
SELECT index, score FROM
ai.rank(
model_id => 'semantic-ranker-default-003',
search_string => $1,
documents => (SELECT ARRAY_AGG(name || ' at ' || distribution_center ORDER BY ref_number) FROM initial_ranking)
)
)
SELECT i.name, i.status, i.distribution_center, r.score
FROM initial_ranking i, reranked_results r
WHERE i.ref_number = r.index
ORDER BY r.score DESC;
toolsets:
supply_chain_toolset:
- search_products_by_context
- check_inventory_levels
- track_shipment_status
- analyze_supply_chain_risk
Testez maintenant le fichier tools.yaml sur le serveur local :
./toolbox --tools-file "tools.yaml"
Vous pouvez également le tester dans l'UI.
./toolbox --ui
Parfait ! Une fois que vous êtes sûr que tout fonctionne, déployez-le dans Cloud Run comme suit.
Déploiement Cloud Run
- Définissez la variable d'environnement PROJECT_ID :
export PROJECT_ID="my-project-id"
- Initialisez la gcloud CLI :
gcloud init
gcloud config set project $PROJECT_ID
- Vous devez activer les API suivantes :
gcloud services enable run.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
iam.googleapis.com \
secretmanager.googleapis.com
- Créez un compte de service backend si vous n'en avez pas déjà un :
gcloud iam service-accounts create toolbox-identity
- Accordez des autorisations pour utiliser Secret Manager :
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member serviceAccount:toolbox-identity@$PROJECT_ID.iam.gserviceaccount.com \
--role roles/secretmanager.secretAccessor
- Accorder des autorisations supplémentaires au compte de service, spécifiques à notre source AlloyDB (roles/alloydb.client et roles/serviceusage.serviceUsageConsumer)
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member serviceAccount:toolbox-identity@$PROJECT_ID.iam.gserviceaccount.com \
--role roles/alloydb.client
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member serviceAccount:toolbox-identity@$PROJECT_ID.iam.gserviceaccount.com \
--role serviceusage.serviceUsageConsumer
- Importez tools.yaml en tant que secret :
gcloud secrets create tools-scm-agent --data-file=tools.yaml
- Si vous avez déjà un secret et que vous souhaitez mettre à jour sa version, exécutez la commande suivante :
gcloud secrets versions add tools-scm-agent --data-file=tools.yaml
- Définissez une variable d'environnement sur l'image de conteneur que vous souhaitez utiliser pour Cloud Run :
export IMAGE=us-central1-docker.pkg.dev/database-toolbox/toolbox/toolbox:latest
- Déployez la boîte à outils sur Cloud Run à l'aide de la commande suivante :
Si vous avez activé l'accès public dans votre instance AlloyDB (ce qui n'est pas recommandé), suivez la commande ci-dessous pour le déploiement sur Cloud Run :
gcloud run deploy toolbox-scm-agent \
--image $IMAGE \
--service-account toolbox-identity \
--region us-central1 \
--set-secrets "/app/tools.yaml=tools-scm-agent:latest" \
--args="--tools-file=/app/tools.yaml","--address=0.0.0.0","--port=8080" \
--allow-unauthenticated
Si vous utilisez un réseau VPC, exécutez la commande ci-dessous :
gcloud run deploy toolbox-scm-agent \
--image $IMAGE \
--service-account toolbox-identity \
--region us-central1 \
--set-secrets "/app/tools.yaml=tools-scm-agent:latest" \
--args="--tools-file=/app/tools.yaml","--address=0.0.0.0","--port=8080" \
# TODO(dev): update the following to match your VPC details
--network <<YOUR_NETWORK_NAME>> \
--subnet <<YOUR_SUBNET_NAME>> \
--allow-unauthenticated
6. Configuration de l'agent
À l'aide de l'Agent Development Kit (ADK), nous avons abandonné les requêtes monolithiques au profit d'une architecture multi-agents spécialisée :
- InventorySpecialist : axé sur les métriques concernant le stock de produits et l'entrepôt.
- LogisticsManager : expert en itinéraires de transport mondiaux et en analyse des risques.
- GlobalOrchestrator : le "cerveau" qui utilise le raisonnement pour déléguer des tâches et synthétiser les résultats.
Clonez ce dépôt dans votre projet et examinons-le.
Pour cloner ce dépôt, exécutez la commande suivante depuis le terminal Cloud Shell (dans le répertoire racine ou à l'emplacement où vous souhaitez créer ce projet) :
git clone https://github.com/AbiramiSukumaran/scm-memory-agent
- Le projet devrait être créé. Vous pouvez le vérifier dans l'éditeur Cloud Shell.
- Veillez à mettre à jour le fichier .env avec les valeurs de votre projet et de votre instance.
Tutoriel du code
Aperçu rapide de l'agent Orchestrator
Go to app.py and you should be able to see the following snippet:
orchestrator = adk.Agent(
name="GlobalOrchestrator",
model="gemini-2.5-flash",
description="Global Supply Chain Orchestrator root agent.",
instruction="""
You are the Global Supply Chain Brain. You are responsible for products, inventory and logistics.
You also have access to the memory tool, remember to include all the information that the tool can provide you with about the user before you respond.
1. Understand intent and delegate to specialists. As the Global Orchestrator, you have access to the full conversation history with the user.
When you transfer a query to a specialist agent, sub agent or tool, share the important facts and information from your memory to them so they can operate with the full context.
2. Ensure the final response is professional and uses Markdown tables for data.
3. If a specialist provides a long list, ensure only the top 10 items are shown initially.
4. Conclude with a brief, high-level executive summary of what the data implies.
""",
tools=[adk.tools.preload_memory_tool.PreloadMemoryTool()],
sub_agents=[inventory_agent, logistics_agent],
#after_agent_callback=auto_save_session_to_memory_callback,
)
Cet extrait correspond à la définition de la racine, qui est l'agent d'orchestration qui reçoit la conversation ou la demande de l'utilisateur et la transmet au sous-agent ou à l'utilisateur correspondant aux outils appropriés en fonction de la tâche.
- Examinons l'agent d'inventaire.
inventory_agent = adk.Agent(
name="InventorySpecialist",
model="gemini-2.5-flash",
description="Specialist in product stock and warehouse data.",
instruction="""
Analyze inventory levels.
1. Use 'search_products_by_context' or 'check_inventory_levels'.
2. ALWAYS format results as a clean Markdown table.
3. If there are many results, display only the TOP 10 most relevant ones.
4. At the end, state: 'There are additional records available. Would you like to see more?'
""",
tools=tools
)
Ce sous-agent est spécialisé dans les activités d'inventaire, comme la recherche contextuelle de produits et la vérification des niveaux d'inventaire.
- Ensuite, il y a le sous-agent logistique :
logistics_agent = adk.Agent(
name="LogisticsManager",
model="gemini-2.5-flash",
description="Expert in global shipping routes and logistics tracking.",
instruction="""
Check shipment statuses.
1. Use 'track_shipment_status' or 'analyze_supply_chain_risk'.
2. ALWAYS format results as a clean Markdown table.
3. Limit initial output to the top 10 shipments.
4. Ask if the user needs the full manifest if more results exist.
""",
tools=tools
)
Ce sous-agent est spécialisé dans les activités logistiques, comme le suivi des expéditions et l'analyse des risques dans la chaîne d'approvisionnement.
- Les trois agents dont nous avons parlé jusqu'à présent utilisent des outils. Ces outils sont référencés via le serveur Toolbox que nous avons déjà déployé dans la section précédente. Reportez-vous à l'extrait ci-dessous :
from toolbox_core import ToolboxSyncClient
TOOLBOX_SERVER = os.environ["TOOLBOX_SERVER"]
TOOLBOX_TOOLSET = os.environ["TOOLBOX_TOOLSET"]
# --- ADK TOOLBOX CONFIGURATION ---
toolbox = ToolboxSyncClient(TOOLBOX_SERVER)
tools = toolbox.load_toolset(TOOLBOX_TOOLSET)
Ce sous-agent est spécialisé dans les activités logistiques, comme le suivi des expéditions et l'analyse des risques dans la chaîne d'approvisionnement.
7. Agent Engine
Lors de la première exécution, créez le moteur de l'agent.
import vertexai
GOOGLE_CLOUD_PROJECT = os.environ["GOOGLE_CLOUD_PROJECT"]
GOOGLE_CLOUD_LOCATION = os.environ["GOOGLE_CLOUD_LOCATION"]
client = vertexai.Client(
project=GOOGLE_CLOUD_PROJECT,
location=GOOGLE_CLOUD_LOCATION
)
agent_engine = client.agent_engines.create()
- Pour la prochaine exécution, mettez à jour la configuration Agent Engine avec la banque de mémoire :
agent_engine = client.agent_engines.update(
name=APP_NAME,
config={
"context_spec": {
"memory_bank_config": {
"generation_config": {
"model": f"projects/{PROJECT_ID}/locations/{GOOGLE_CLOUD_LOCATION}/publishers/google/models/gemini-2.5-flash"
}
}
}
})
8. Contexte, exécution et mémoire
La gestion du contexte est divisée en deux couches distinctes pour que l'agent ressemble à un partenaire continu plutôt qu'à un robot sans état :
Mémoire à court terme (sessions) : gérée via VertexAiSessionService, elle suit l'historique des événements immédiats (messages utilisateur, réponses des outils) au cours d'une même interaction.
Mémoire à long terme (Memory Bank) : optimisée par Vertex AI Memory Bank via adk.memorybankservice. Cette couche extrait les informations "pertinentes" (comme la préférence d'un utilisateur pour certains transporteurs ou les retards récurrents dans l'entrepôt) et les conserve d'une session à l'autre.
Initialiser la session pour la mémoire de session dans le champ d'application de la conversation
Il s'agit de la partie de l'extrait qui crée la session pour l'application actuelle pour l'utilisateur actuel.
from google.adk.sessions import VertexAiSessionService
...
session_service = VertexAiSessionService(
project=PROJECT_ID,
location=GOOGLE_CLOUD_LOCATION,
)
...
# Initialize the session *outside* of the route handler to avoid repeated creation
session = None
session_lock = threading.Lock()
async def initialize_session():
global session
try:
session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID)
print(f"Session {session.id} created successfully.") # Add a log
except Exception as e:
print(f"Error creating session: {e}")
session = None # Ensure session is None in case of error
# Create the session on app startup
asyncio.run(initialize_session())
Initialiser Vertex AI Memory Bank pour la mémoire à long terme
Il s'agit de la partie de l'extrait qui instancie l'objet de service Vertex AI Memory Bank pour le moteur d'agent.
from google.adk.memory import InMemoryMemoryService
from google.adk.memory import VertexAiMemoryBankService
...
try:
memory_bank_service = adk.memory.VertexAiMemoryBankService(
agent_engine_id=AGENT_ENGINE_ID,
project=PROJECT_ID,
location=GOOGLE_CLOUD_LOCATION,
)
#in_memory_service = InMemoryMemoryService()
print("Memory Bank Service initialized successfully.")
except Exception as e:
print(f"Error initializing Memory Bank Service: {e}")
memory_bank_service = None
runner = adk.Runner(
agent=orchestrator,
app_name=APP_NAME,
session_service=session_service,
memory_service=memory_bank_service,
)
...
Qu'est-ce qui est configuré ?
Dans cette partie de l'extrait, nous configurons le service Vertex AI Memory Bank pour la mémoire à long terme. Il stocke de manière contextuelle la session pour l'application spécifique et l'utilisateur spécifique en tant que mémoire dans Vertex AI Memory Bank.
Qu'est-ce qui est exécuté dans le cadre de l'exécution de l'agent ?
async def run_and_collect():
final_text = ""
try:
async for event in runner.run_async(
new_message=content,
user_id=user_id,
session_id=session_id
):
if hasattr(event, 'author') and event.author:
if not any(log['agent'] == event.author for log in execution_logs):
execution_logs.append({
"agent": event.author,
"action": "Analyzing data requirements...",
"type": "orchestration_event"
})
if hasattr(event, 'text') and event.text:
final_text = event.text
elif hasattr(event, 'content') and hasattr(event.content, 'parts'):
for part in event.content.parts:
if hasattr(part, 'text') and part.text:
final_text = part.text
except Exception as e:
print(f"Error during runner.run_async: {e}")
raise # Re-raise the exception to signal failure
finally:
gc.collect()
return final_text
Il traite le contenu saisi par l'utilisateur dans l'objet new_message avec l'ID utilisateur et l'ID de session dans le champ d'application. L'agent prend ensuite le relais, et sa réponse est traitée et renvoyée.
Qu'est-ce qui est stocké dans la mémoire à long terme ?
Les détails de la session dans le champ d'application de l'application et de l'utilisateur sont extraits dans la variable de session.
Cette session est ensuite ajoutée en tant que mémoire pour l'utilisateur actuel pour l'application actuelle de l'objet Vertex AI Memory Bank à l'aide de la méthode "add_session_to_memory".
session = asyncio.run(session_service.get_session(app_name=APP_NAME, user_id=USER_ID, session_id=session.id))
if memory_bank_service and session: # Check memory service AND session
try:
#asyncio.run(in_memory_service.add_session_to_memory(session))
asyncio.run(memory_bank_service.add_session_to_memory(session))
'''
client.agent_engines.memories.generate(
scope={"app_name": APP_NAME, "user_id": USER_ID},
name=APP_NAME,
direct_contents_source={
"events": [
{"content": content}
]
},
config={"wait_for_completion": True},
)
'''
print("Successfully added session to memory.******")
print(session.id)
except Exception as e:
print(f"Error adding session to memory: {e}")
Récupération de la mémoire
Nous devons récupérer la mémoire à long terme stockée en utilisant le nom de l'application et le nom d'utilisateur comme portée (puisque c'est la portée pour laquelle nous avons stocké les mémoires) afin de pouvoir la transmettre dans le contexte à l'orchestrateur et aux autres agents, le cas échéant.
results = client.agent_engines.memories.retrieve(
name=APP_NAME,
scope={"app_name": APP_NAME, "user_id": USER_ID}
)
# RetrieveMemories returns a pager. You can use `list` to retrieve all pages' memories.
list(results)
print(list(results))
Comment la mémoire récupérée est-elle chargée dans le contexte ?
Nous utilisons l'attribut suivant dans la définition de l'agent Orchestrator, qui permet à l'agent racine de précharger le contexte à partir de la banque de mémoire. Cela s'ajoute aux outils auxquels nous accédons depuis le serveur de la boîte à outils pour les sous-agents.
tools=[adk.tools.preload_memory_tool.PreloadMemoryTool()],
Contexte de rappel
Dans une chaîne d'approvisionnement d'entreprise, vous ne pouvez pas avoir de "boîte noire". Nous utilisons CallbackContext de l'ADK pour créer un moteur narratif. En nous connectant à l'exécution de l'agent, nous capturons chaque processus de réflexion et chaque appel d'outil, et les diffusons dans une barre latérale de l'UI.
- Événement de trace : "GlobalOrchestrator is analyzing data requirements..." (GlobalOrchestrator analyse les exigences en matière de données…)
- Événement de trace : "Délégation à InventorySpecialist pour les niveaux de stock..."
- Événement Trace : "Récupération des tendances historiques de retard des fournisseurs à partir de la banque de données..."
Cet historique est très utile pour le débogage et permet aux opérateurs humains de faire confiance aux décisions autonomes de l'agent.
from google.adk.agents.callback_context import CallbackContext
...
# --- ADK CALLBACKS (Narrative Engine) ---
execution_logs = []
async def trace_callback(context: CallbackContext):
"""
Captures agent and tool invocation flow for the UI narrative.
"""
agent_name = context.agent.name
event = {
"agent": agent_name,
"action": "Processing request steps...",
"type": "orchestration_event"
}
execution_logs.append(event)
return None
...
Et voilà ! Nous avons cloné le projet et examiné en détail l'agent, la mémoire et le contexte.
Pour le tester, accédez au dossier du projet du dépôt cloné et exécutez les commandes suivantes :
>> pip install -r requirements.txt
>> python app.py
Votre agent devrait démarrer en local et vous devriez pouvoir le tester.
9. Déployons-la sur Cloud Run.
- Déployez-le sur Cloud Run en exécutant la commande suivante à partir du terminal Cloud Shell où le projet est cloné. Assurez-vous d'être dans le dossier racine du projet.
Exécutez la commande suivante dans votre terminal Cloud Shell :
gcloud run deploy supply-chain-agent --source . --platform managed --region us-central1 --allow-unauthenticated --set-env-vars GOOGLE_CLOUD_PROJECT=<<YOUR_PROJECT>>,GOOGLE_CLOUD_LOCATION=us-central1,GOOGLE_GENAI_USE_VERTEXAI=TRUE,REASONING_ENGINE_APP_NAME=<<YOUR_APP_ENGINE_URL>>,TOOLBOX_SERVER=<<YOUR_TOOLBOX_SERVER>>,TOOLBOX_TOOLSET=supply_chain_toolset,AGENT_ENGINE_ID=<<YOUR_AGENT_ENGINE_ID>>
Remplacez les valeurs des espaces réservés <<YOUR_PROJECT>>, <<YOUR_APP_ENGINE_URL>>, <<YOUR_TOOLBOX_SERVER>> et <<YOUR_AGENT_ENGINE_ID>>.
Une fois la commande terminée, une URL de service est générée. Copiez-le.
- Attribuez le rôle Client AlloyDB au compte de service Cloud Run.Cela permet à votre application sans serveur de se connecter de manière sécurisée à la base de données.
Exécutez la commande suivante dans votre terminal Cloud Shell :
# 1. Get your Project ID and Project Number
PROJECT_ID=$(gcloud config get-value project)
PROJECT_NUMBER=$(gcloud projects describe $PROJECT_ID --format="value(projectNumber)")
# 2. Grant the AlloyDB Client role
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$PROJECT_NUMBER-compute@developer.gserviceaccount.com" \
--role="roles/alloydb.client"
Utilisez maintenant l'URL du service (point de terminaison Cloud Run que vous avez copié précédemment) et testez l'application.
Remarque : Si vous rencontrez un problème de service et que la mémoire est indiquée comme cause, essayez d'augmenter la limite de mémoire allouée à 1 Gio pour le tester.
10. Effectuer un nettoyage
Une fois cet atelier terminé, n'oubliez pas de supprimer le cluster et l'instance AlloyDB.
Il devrait nettoyer le cluster ainsi que ses instances.
11. Félicitations
En combinant la vitesse d'AlloyDB, l'efficacité d'orchestration de MCP Toolbox et la "mémoire institutionnelle" de Vertex AI Memory Bank, nous avons créé un système de chaîne d'approvisionnement qui évolue. Il ne se contente pas de répondre à vos questions : il se souvient que votre entrepôt à Singapour est souvent confronté à des retards liés à la mousson et vous suggère de réacheminer vos colis de manière proactive, avant même que vous ne le demandiez.