Gérer les sessions avec Agent Development Kit

Projet Google Cloud

import agentplatform

client = agentplatform.Client(
  project="PROJECT_ID",
  location="LOCATION"
)

# If you don't have an Agent Runtime instance already, create an instance.
remote_agent = client.runtimes.create(
    config={
      "display_name": "My Session Store",
    }
)

# Print the runtime ID, you will need it in the later steps to initialize
# the ADK `VertexAiSessionService`.
print(remote_agent.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.

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 (128 caractères maximum). Exemple :user-123

• AGENT_ENGINE_ID : ID de ressource d'une instance Agent Runtime.

• 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 remote_agent.api_resource.name.split("/")[-1].

UI ADK

Pour tester votre agent avec l'interface utilisateur ADK et vous connecter à Agent Platform Sessions, 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. Vous devez plutôt 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)

Python

Utiliser 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 indiqué 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 système de dénomination ou pour passer rapidement à d'autres services. Si vous ne spécifiez pas d'ID de session, Agent Platform en génère un automatiquement.

# 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 des chiffres ([0-9]) sans zéros au début.

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.

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

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

  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"
  )
  ```

  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()
  )

Lister les sessions existantes

Lister 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 du runner, 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

L'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'agents ou pour stocker des données d'événements personnalisées.

Lorsque vous ajoutez un événement avec des champs personnalisés, l'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

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)

Projet Google Cloud

client.runtimes.update(
    resource_name=remote_agent.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 pour votre agent.

• REQUIREMENTS : liste des packages pip requis par votre agent. Exemple :["google-cloud-storage", "google-cloud-agentplatform[runtimes,adk]"]

• STAGING_BUCKET : bucket Cloud Storage précédé de gs://.