Cette page explique comment connecter un agent Agent Development Kit (ADK) à des sessions Vertex AI Agent Engine et comment utiliser des sessions gérées dans l'environnement local et de production.
Avant de commencer
Assurez-vous que votre environnement est configuré en suivant les étapes Obtenir les rôles requis et Authentification de la section Configurer votre environnement.
Créer une instance Vertex AI Agent Engine
Pour accéder aux sessions Vertex AI Agent Engine, vous devez d'abord utiliser une instance Vertex AI Agent Engine. Vous n'avez pas besoin de déployer de code pour commencer à utiliser Sessions. Si vous avez déjà utilisé Agent Engine, la création d'une instance Vertex AI Agent Engine ne prend que quelques secondes, sans déploiement de code. Cela peut prendre plus de temps si vous utilisez Agent Engine pour la première fois.
Projet Google Cloud
import vertexai
client = vertexai.Client(
project="PROJECT_ID",
location="LOCATION"
)
# If you don't have an Agent Engine instance already, create an instance.
agent_engine = client.agent_engines.create()
# Print the agent engine ID, you will need it in the later steps to initialize
# the ADK `VertexAiSessionService`.
print(agent_engine.api_resource.name.split("/")[-1])
Remplacez les éléments suivants :
PROJECT_ID : ID de votre projet
LOCATION : votre région. Consultez les régions disponibles pour les sessions.
Mode Express Vertex AI
Vous pouvez utiliser le SDK Vertex AI pour Python avec votre clé API en mode Express :
import vertexai
client = vertexai.Client( # For service interactions via client.agent_engines
api_key="API_KEY",
)
# If you don't have an Agent Engine instance already, create an instance.
agent_engine = client.agent_engines.create()
# Print the agent engine ID, you will need it in the later steps to initialize
# the ADK `VertexAiSessionService`.
print(agent_engine.api_resource.name.split("/")[-1])
Développer votre agent ADK
Pour créer votre agent ADK, suivez les instructions de la section Agent Development Kit ou utilisez le code suivant pour créer un agent qui salue un utilisateur avec des salutations fixes :
from google import adk
def greetings(query: str):
"""Tool to greet user."""
if 'hello' in query.lower():
return {"greeting": "Hello, world"}
else:
return {"greeting": "Goodbye, world"}
# Define an ADK agent
root_agent = adk.Agent(
model="gemini-2.0-flash",
name='my_agent',
instruction="You are an Agent that greet users, always use greetings tool to respond.",
tools=[greetings]
)
Configurer l'exécuteur ADK
L'environnement d'exécution ADK orchestre l'exécution de vos agents, outils et rappels, ainsi que les appels pour lire et écrire des sessions. Initialisez le Runner avec VertexAiSessionService, qui se connecte aux sessions Vertex AI Agent Engine.
Projet Google Cloud
from google.adk.sessions import VertexAiSessionService
from google.genai import types
app_name="APP_NAME"
user_id="USER_ID"
# Create the ADK runner with VertexAiSessionService
session_service = VertexAiSessionService(
"PROJECT_ID",
"LOCATION",
"AGENT_ENGINE_ID"
)
runner = adk.Runner(
agent=root_agent,
app_name=app_name,
session_service=session_service)
# Helper method to send query to the runner
def call_agent(query, session_id, user_id):
content = types.Content(role='user', parts=[types.Part(text=query)])
events = runner.run(
user_id=user_id, session_id=session_id, new_message=content)
for event in events:
if event.is_final_response():
final_response = event.content.parts[0].text
print("Agent Response: ", final_response)
Remplacez les éléments suivants :
APP_NAME : nom de votre application d'agent.
USER_ID : choisissez votre propre ID utilisateur (128 caractères maximum). Exemple :
user-123AGENT_ENGINE_ID : ID de ressource d'une instance Vertex AI Agent Engine.
Pour les agents déployés, l'ID de ressource est indiqué en tant que variable d'environnement
GOOGLE_CLOUD_AGENT_ENGINE_ID.Pour les agents locaux, vous pouvez récupérer l'ID de ressource à l'aide de
agent_engine.api_resource.name.split("/")[-1].
Mode Express Vertex AI
import os
from google.adk.sessions import VertexAiSessionService
from google.genai import types
# Set environment variables
os.environ["GOOGLE_API_KEY"] = "API_KEY"
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"
# Create the ADK runner with VertexAiSessionService
session_service = VertexAiSessionService(
agent_engine_id="AGENT_ENGINE_ID"
)
runner = adk.Runner(
agent=root_agent,
app_name=app_name,
session_service=session_service)
# Helper method to send query to the runner
def call_agent(query, session_id, user_id):
content = types.Content(role='user', parts=[types.Part(text=query)])
events = runner.run(
user_id=user_id, session_id=session_id, new_message=content)
for event in events:
if event.is_final_response():
final_response = event.content.parts[0].text
print("Agent Response: ", final_response)
Remplacez les éléments suivants :
API_KEY : votre clé API Vertex AI Express Mode.
AGENT_ENGINE_ID : ID de ressource d'une instance Vertex AI Agent Engine.
Pour les agents déployés, l'ID de ressource est indiqué en tant que variable d'environnement
GOOGLE_CLOUD_AGENT_ENGINE_ID.Pour les agents locaux, vous pouvez récupérer l'ID de ressource à l'aide de
agent_engine.api_resource.name.split("/")[-1].
Interagir avec votre agent
Après avoir défini votre agent et configuré les sessions Vertex AI Agent Engine, vous pouvez interagir avec votre agent pour vérifier que l'historique et les états des sessions sont conservés.
Interface utilisateur ADK
Testez votre agent avec l'interface utilisateur de l'ADK et connectez-vous à la session Vertex AI Agent Engine à l'aide de l'option de ligne de commande session_service_uri :
agent_engine_id="AGENT_ENGINE_ID"
adk web --session_service_uri=agentengine://${agent_engine_id}
# Sample output
+-----------------------------------------------------------------------------+
| ADK Web Server started |
| |
| For local testing, access at http://localhost:8000. |
+-----------------------------------------------------------------------------+
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
Python
Utilisez le code Python de l'ADK pour gérer les sessions et les états.
Créer une session et interroger l'agent
Utilisez le code suivant pour créer une session et envoyer une requête à votre agent :
# Create a session
session = await session_service.create_session(
app_name=app_name,
user_id=user_id)
call_agent("Hello!", session.id, user_id)
# Agent response: "Hello, world"
call_agent("Thanks!", session.id, user_id)
# Agent response: "Goodbye, world"
Une fois la session créée et transmise à l'exécuteur, ADK l'utilise pour stocker les événements de l'interaction en cours. Vous pouvez également reprendre une session précédente en fournissant son ID.
Lister les sessions existantes
Répertorie toutes les sessions existantes associées à un ID utilisateur donné.
# List sessions
await session_service.list_sessions(app_name=app_name,user_id=user_id)
# ListSessionsResponse(session_ids=['1122334455', '9988776655'])
Gérer les états de session
Les états contiennent les informations dont l'agent a besoin pour une conversation. Vous pouvez fournir un état initial sous forme de dictionnaire lorsque vous créez une session :
# Create a session with state
session = await session_service.create_session(
app_name=app_name,
user_id=user_id,
state={'key': 'value'})
print(session.state['key'])
# value
Pour mettre à jour l'état de la session en dehors du lanceur, ajoutez un nouvel événement à la session à l'aide de state_delta :
from google.adk.events import Event, EventActions
import time
# Define state changes
state_changes = {'key': 'new_value'}
# Create event with actions
actions_with_update = EventActions(state_delta=state_changes)
system_event = Event(
invocation_id="invocation_id",
author="system", # Or 'agent', 'tool' etc.
actions=actions_with_update,
timestamp=time.time()
)
# Append the event
await session_service.append_event(session, system_event)
# Check updated state
updated_session = await session_service.get_session(
app_name=app_name,
user_id=user_id,
session_id=session.id)
# State is updated to new value
print(updated_session.state['key'])
# new_value
Supprimer une session
Supprimer une session spécifique associée à un ID utilisateur :
await session_service.delete_session(app_name=app_name, user_id=user_id, session_id=session.id)
Déployer votre agent dans Vertex AI Agent Engine
Après avoir testé votre agent en local, vous pouvez le déployer en production en mettant à jour l'instance Vertex AI Agent Engine avec les paramètres suivants :
Projet Google Cloud
client.agent_engines.update(
resource_name=agent_engine.api_resource.name,
agent=AGENT,
config={
"display_name": DISPLAY_NAME, # Optional.
"requirements": REQUIREMENTS, # Optional.
"staging_bucket": STAGING_BUCKET, # Required.
},
)
Remplacez les éléments suivants :
AGENT : application qui implémente la méthode
query / stream_query(par exemple,AdkApppour un agent ADK). Pour en savoir plus, consultez Éléments à prendre en compte pour le déploiement.DISPLAY_NAME : nom convivial pour votre agent.
REQUIREMENTS : liste des packages pip requis par votre agent. Exemple :
["google-cloud-storage", "google-cloud-aiplatform[agent_engines,adk]"]STAGING_BUCKET : bucket Cloud Storage précédé de
gs://.
Mode Express Vertex AI
Déployez votre agent à l'aide des fichiers sources en suivant ces étapes de déploiement.
Effectuer un nettoyage
Pour nettoyer toutes les ressources utilisées dans ce projet, vous pouvez supprimer l'instance Vertex AI Agent Engine ainsi que ses ressources enfants :
agent_engine.delete(force=True)