Gérer les sessions avec Agent Development Kit

Cette page explique comment connecter un agent Agent Development Kit (ADK) à des sessions Agent Platform et utiliser des sessions gérées dans l'environnement local et de production.

Avant de commencer

Ces instructions utilisent la structure de fichier de projet de base suivante pour définir un agent ADK, son exécuteur et son code de déploiement :

my_agent/
    agent.py      # main agent code
    runner.py     # code for interacting with the agent
    deploy.py     # code for deploying the agent to Google Cloud

Assurez-vous que votre environnement est configuré en suivant les étapes Obtenir les rôles requis et Authentification de la section Configurer votre environnement.

Définir des variables d'environnement

Pour utiliser ADK, définissez vos variables d'environnement :

import os

os.environ["GOOGLE_GENAI_USE_ENTERPRISE"] = "TRUE"
os.environ["GOOGLE_CLOUD_PROJECT"] = "PROJECT_ID"
os.environ["GOOGLE_CLOUD_LOCATION"] = "LOCATION"

Remplacez les éléments suivants :

  • PROJECT_ID : ID de votre projet
  • LOCATION : votre région Consultez les régions compatibles avec Memory Bank.

Créer une instance Vertex AI Agent Engine

Pour accéder aux sessions Agent Platform, vous devez d'abord utiliser une instance Vertex AI Agent Engine. Vous n'avez pas besoin de déployer de code pour commencer à utiliser les 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 compatibles avec les sessions.

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. Enregistrez ce code dans un fichier nommé agent.py.

# file: my_agent/agent.py
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

Le ADK Runtime orchestre l'exécution de vos agents, outils et rappels, ainsi que les appels pour lire et écrire des sessions. Initialisez l'exécuteur avec VertexAiSessionService, qui se connecte aux sessions Agent Platform. Enregistrez ce code dans un fichier nommé runner.py.

Projet Google Cloud 

# file: my_agent/runner.py
import agent # Import from your agent.py
from google.adk import Runner
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="PROJECT_ID",
      location="LOCATION",
      agent_engine_id="AGENT_ENGINE_ID"
)
runner = Runner(
    agent=agent.root_agent,
    app_name=app_name,
    session_service=session_service)

# Helper method to send query to the runner
async def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  async for event in runner.run_async(
      user_id=user_id, session_id=session_id, new_message=content):
      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 avec une limite de 128 caractères. Par exemple, user-123.

  • AGENT_ENGINE_ID : ID de ressource d'une instance Vertex AI Agent Engine.

  • Pour les agents déployés, l'ID de ressource est répertorié 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 Agent Platform, vous pouvez interagir avec votre agent pour vérifier que l'historique et les états de la session sont conservés.

UI ADK

Pour tester votre agent avec l'interface utilisateur ADK et vous connecter aux sessions Agent Platform, exécutez la commande adk web dans votre terminal.

Bien que adk web lise généralement les variables d'environnement à partir d'un fichier .env, il ignore ce fichier lorsque vous utilisez l'indicateur --session_service_uri. Au lieu de cela, vous devez définir les variables d'environnement GOOGLE_CLOUD_PROJECT et GOOGLE_CLOUD_LOCATION dans votre session de terminal avant d'exécuter la commande.

export GOOGLE_CLOUD_PROJECT="PROJECT_ID"
export GOOGLE_CLOUD_LOCATION="LOCATION"
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)

UI ADK

Python

Utilisez le code Python ADK pour gérer les sessions et les états. Ajoutez le code suivant à la fin de votre fichier runner.py pour interagir avec l'agent.

Les extraits suivants contiennent des appels await de premier niveau pour plus de concision. Pour exécuter ce code en tant que script Python, placez les extraits dans une fonction async et utilisez asyncio.run() pour l'exécuter, comme illustré dans cet exemple :

import asyncio

async def main():
  # Place one or more snippets here.
  # For example:
  session = await session_service.create_session(
         app_name=app_name,
         user_id=user_id)

  await call_agent("Hello!", session.id, user_id)

asyncio.run(main())

Créer une session et interroger l'agent

Utilisez le code suivant pour créer une session et envoyer une requête à votre agent. Spécifiez un ID de session personnalisé pour organiser les sessions à l'aide de votre propre schéma de nommage ou pour passer rapidement d'autres services. Si vous ne spécifiez pas d'ID de session, Agent Platform en génère automatiquement un.

# file: my_agent/runner.py
# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_id,
       session_id=session_id)

await call_agent("Hello!", session.id, user_id)
# Agent response: "Hello, world"

await call_agent("Thanks!", session.id, user_id)
# Agent response: "Goodbye, world"

Pour les ID de session personnalisés, tenez compte des restrictions suivantes afin d'éviter les conflits avec les ID générés par le système :

  • Si le premier caractère est une lettre, l'ID peut comporter jusqu'à 63 caractères. Les caractères valides sont les lettres minuscules, les chiffres et les traits d'union ([a-z0-9-]). Le dernier caractère doit être une lettre ou un chiffre.
  • Si le premier caractère est un chiffre, l'ID peut comporter jusqu'à neuf caractères. Les caractères valides sont les chiffres ([0-9]) sans zéro non significatif.

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 l'ID de cette session.

Configurer la valeur TTL (Time To Live) de la session

Toutes les sessions doivent avoir une date d'expiration. Vous pouvez définir cette date d'expiration lorsque vous créez ou mettez à jour une session. La session et ses événements enfants sont automatiquement supprimés une fois la date d'expiration dépassée. Vous pouvez définir directement la date d'expiration (expire_time) ou définir la valeur TTL (ttl) en secondes. Si aucune des deux n'est spécifiée, le système applique une valeur TTL par défaut de 365 jours.

Valeur TTL

Si vous définissez la valeur TTL, le serveur calcule la date d'expiration comme suit : create_time + ttl pour les sessions nouvellement créées ou update_time + ttl pour les sessions mises à jour.

  session = await session_service.create_session(
        app_name=app_name,
        user_id=user_id,
        # Session will be deleted 10 days after creation time.
        ttl=f"{24 * 60 * 60 * 10}s"
  )
  ```

Date/Heure d'expiration 

  import datetime

  expire_time = datetime.datetime.now(
        tz=datetime.timezone.utc) + datetime.timedelta(seconds=24 * 60 * 60 * 10)

  session = await session_service.create_session(
        app_name=app_name,
        user_id=user_id,
        # Session will be deleted at the provided time (10 days after current time).
        expire_time=expire_time.isoformat()
  )

Répertorier les sessions existantes

Répertoriez toutes les sessions existantes associées à un ID utilisateur donné.

# List sessions
sessions = await session_service.list_sessions(app_name=app_name,user_id=user_id)
print(sessions)
# 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 de l'exécuteur, ajoutez un nouvel événement à la session à l'aide de state_delta :

# file: my_agent/runner.py
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

Champs personnalisés pour les événements

ADK est compatible avec un schéma d'événement flexible qui vous permet d'inclure des données arbitraires dans les événements de session. Cela est utile pour l'interopérabilité avec d'autres frameworks d'agent ou pour stocker des données d'événement personnalisées.

Lorsque vous ajoutez un événement avec des champs personnalisés, ADK sérialise les données d'événement dans le champ raw_event de l'événement de session stocké.

# Create an event with custom fields
custom_event = Event(
  invocation_id="invocation_id",
  author="user",
  timestamp=time.time(),
  custom_field="custom_value",
  another_field={"nested_key": "nested_value"}
)

# Append the event
await session_service.append_event(session, custom_event)

Lorsque vous récupérez la session, les événements récupérés conservent ces champs personnalisés.

Supprimer une session

Supprimez 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 sur Agent Platform

Après avoir testé votre agent en local, vous pouvez le déployer en production en mettant à jour l'instance Agent Platform avec des paramètres :

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, AdkApp pour un agent ADK)

  • DISPLAY_NAME : nom convivial de votre agent

  • REQUIREMENTS : liste des packages pip requis par votre agent Par exemple, ["google-cloud-storage", "google-cloud-aiplatform[agent_engines,adk]"].

  • STAGING_BUCKET : bucket Cloud Storage préfixé par gs://

Libérer de l'espace

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)

Étape suivante