Utiliser un agent Agent Development Kit

Avant de commencer

Ce tutoriel suppose que vous avez lu et suivi les instructions de :

• Développer un agent Agent Development Kit : pour développer agent en tant qu'instance de AdkApp.
• Authentification de l'utilisateur pour s'authentifier en tant qu'utilisateur afin d'interroger l'agent.
• Importez et initialisez le SDK pour initialiser le client afin d'obtenir une instance déployée (si nécessaire).

Obtenir une instance d'un agent

Pour interroger un AdkApp, vous devez d'abord créer une instance ou obtenir une instance existante.

Pour obtenir le AdkApp correspondant à un ID de ressource spécifique :

SDK Vertex AI pour Python

Exécutez le code suivant :

import vertexai

client = vertexai.Client(  # For service interactions via client.agent_engines
    project="PROJECT_ID",
    location="LOCATION",
)

adk_app = client.agent_engines.get(name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

print(adk_app)

Où :

• PROJECT_ID correspond à l'ID de projet Google Cloud sous lequel vous développerez et déploierez des agents.
• LOCATION désigne l'une des régions compatibles.
• RESOURCE_ID est l'ID de l'agent déployé en tant que ressource reasoningEngine.

Bibliothèque de requêtes Python

Exécutez le code suivant :

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests

def get_identity_token():
    credentials, _ = google_auth.default()
    auth_request = google_requests.Request()
    credentials.refresh(auth_request)
    return credentials.token

response = requests.get(
f"https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    headers={
        "Content-Type": "application/json; charset=utf-8",
        "Authorization": f"Bearer {get_identity_token()}",
    },
)

API REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID

Lorsque vous utilisez le SDK Vertex AI pour Python, l'objet adk_app correspond à une classe AgentEngine qui contient les éléments suivants :

• un adk_app.api_resource contenant des informations sur l'agent déployé. Vous pouvez également appeler adk_app.operation_schemas() pour renvoyer la liste des opérations compatibles avec adk_app. Pour en savoir plus, consultez Opérations compatibles.
• un adk_app.api_client qui permet des interactions de service synchrones.
• un adk_app.async_api_client qui permet des interactions de service asynchrones.

Le reste de cette section suppose que vous disposez d'une instance AgentEngine nommée adk_app.

Opérations compatibles

Les opérations suivantes sont acceptées pour AdkApp :

• async_stream_query : pour diffuser une réponse à une requête.

• async_create_session : pour créer une session.

• async_list_sessions : pour lister les sessions disponibles.

• async_get_session : pour récupérer une session spécifique.

• async_delete_session : pour supprimer une session spécifique.

• async_add_session_to_memory : pour générer des souvenirs d'une session.

• async_search_memory : pour récupérer des souvenirs.

Pour lister toutes les opérations compatibles :

SDK Vertex AI pour Python

Exécutez le code suivant :

adk_app.operation_schemas()

Bibliothèque de requêtes Python

Exécutez le code suivant :

import json

json.loads(response.content).get("spec").get("classMethods")

API REST

Représenté dans spec.class_methods à partir de la réponse à la requête curl.

Gérer des sessions

AdkApp utilise des sessions gérées basées sur le cloud après le déploiement de l'agent sur Vertex AI Agent Engine. Cette section explique comment utiliser les sessions gérées.

Remarque : Ces instructions supposent que vous n'avez pas personnalisé votre base de données à l'aide de session_service_builder lors du développement de votre agent.

Créer une session

Pour créer une session pour un utilisateur :

SDK Vertex AI pour Python

session = await adk_app.async_create_session(user_id="USER_ID")

print(session)

Bibliothèque de requêtes Python

Exécutez le code suivant :

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_create_session",
    "input": {"user_id": "USER_ID"},
  }),
)
print(response.content)

API REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_create_session", "input": {"user_id": "USER_ID"},}'

• USER_ID : choisissez votre propre ID utilisateur (128 caractères maximum). Exemple :user-123

La session est créée sous la forme d'une représentation de dictionnaire d'un objet de session ADK.

Répertorier les sessions

Pour lister les sessions d'un utilisateur :

SDK Vertex AI pour Python

response = await adk_app.async_list_sessions(user_id="USER_ID"):
for session in response.sessions:
    print(session)

Bibliothèque de requêtes Python

Exécutez le code suivant :

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_list_sessions",
    "input": {"user_id": "USER_ID"},
  }),
)
print(response.content)

API REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_list_sessions", "input": {"user_id": "USER_ID"},}'

où USER_ID correspond à l'ID utilisateur que vous avez défini. Exemple :user-123

Si des sessions sont renvoyées, elles utilisent la forme de dictionnaire d'un objet de session ADK.

Obtenir une session

Pour obtenir une session spécifique, vous avez besoin de l'ID utilisateur et de l'ID de session :

SDK Vertex AI pour Python

session = await adk_app.async_get_session(user_id="USER_ID", session_id="SESSION_ID")

print(session)

Bibliothèque de requêtes Python

Exécutez le code suivant :

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_get_session",
    "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},
  }),
)
print(response.content)

API REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_get_session", "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},}'

session est la représentation du dictionnaire d'un objet de session ADK.

Supprimer une session

Pour supprimer une session, vous avez besoin de l'ID utilisateur et de l'ID de session :

SDK Vertex AI pour Python

await adk_app.async_delete_session(user_id="USER_ID", session_id="SESSION_ID")

Bibliothèque de requêtes Python

Exécutez le code suivant :

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_delete_session",
    "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},
  }),
)
print(response.content)

API REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_delete_session", "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},}'

Diffuser une réponse à une requête

Pour diffuser en streaming les réponses d'un agent dans une session :

SDK Vertex AI pour Python

async for event in adk_app.async_stream_query(
    user_id="USER_ID",
    session_id="SESSION_ID",  # Optional
    message="What is the exchange rate from US dollars to SEK today?",
):
  print(event)

Bibliothèque de requêtes Python

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests

def get_identity_token():
    credentials, _ = google_auth.default()
    auth_request = google_requests.Request()
    credentials.refresh(auth_request)
    return credentials.token

requests.post(
    f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:streamQuery",
    headers={
        "Content-Type": "application/json",
        "Authorization": f"Bearer {get_identity_token()}",
    },
    data=json.dumps({
        "class_method": "async_stream_query",
        "input": {
            "user_id": "USER_ID",
            "session_id": "SESSION_ID",
            "message": "What is the exchange rate from US dollars to SEK today?",
        },
    }),
    stream=True,
)

API REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:streamQuery?alt=sse -d '{
  "class_method": "async_stream_query",
  "input": {
    "user_id": "USER_ID",
    "session_id": "SESSION_ID",
    "message": "What is the exchange rate from US dollars to SEK today?",
  }
}'

Remarque : L'argument session_id= est facultatif. Si elle n'est pas spécifiée, une nouvelle session est automatiquement créée et utilisée pour diffuser cette requête.

Si vous utilisez le SDK Vertex AI pour Python, vous devriez recevoir une continuation de la conversation, comme la séquence de dictionnaires suivante :

{'author': 'currency_exchange_agent',
 'content': {'parts': [{'function_call': {'args': {'currency_date': '2025-04-03',
                                                   'currency_from': 'USD',
                                                   'currency_to': 'SEK'},
                                          'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
                                          'name': 'get_exchange_rate'}}],
             'role': 'model'},
 'id': 'bOPHtzji',
 # ...
}
{'author': 'currency_exchange_agent',
 'content': {'parts': [{'function_response': {'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
                                              'name': 'get_exchange_rate',
                                              'response': {'amount': 1.0,
                                                           'base': 'USD',
                                                           'date': '2025-04-03',
                                                           'rates': {'SEK': 9.6607}}}}],
             'role': 'user'},
 'id': '9AoDFmiL',
 # ...
}
{'author': 'currency_exchange_agent',
 'content': {'parts': [{'text': 'The exchange rate from USD to SEK on '
                                '2025-04-03 is 1 USD to 9.6607 SEK.'}],
             'role': 'model'},
 'id': 'hmle7trT',
 # ...
}

Gérer les souvenirs

AdkApp utilise la banque de mémoire Vertex AI Agent Engine si vous incluez un PreloadMemoryTool dans la définition de l'agent et que vous déployez l'agent sur Vertex AI Agent Engine. Cette section explique comment générer et récupérer des souvenirs de l'agent à l'aide de l'implémentation par défaut du service de mémoire ADK.

Ajouter une session à la mémoire

Pour conserver en mémoire des informations importantes dans une session (qui pourront être utilisées dans de futures sessions), utilisez la méthode async_add_session_to_memory :

SDK Vertex AI pour Python

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

où SESSION_DICT est la forme de dictionnaire d'un objet de session ADK.

Rechercher des souvenirs

Pour effectuer une recherche dans les souvenirs de l'agent, vous pouvez utiliser la méthode async_search_memory :

SDK Vertex AI pour Python

response = await adk_app.async_search_memory(
    user_id="USER_ID",
    query="QUERY",
)
print(response)

Où :

• USER_ID correspond au champ d'application des souvenirs pertinents.
• QUERY est la requête pour laquelle effectuer une recherche de similarités.

Étapes suivantes

• Utiliser un agent
• Évaluez un agent.
• Gérer les agents déployés.
• Accédez à l'assistance.