Utiliser un agent Agent Development Kit

Avant de commencer

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

• Créer un agent Agent Development Kit : pour créer agent en tant qu'instance de AdkApp.
• Authentification de l'utilisateur pour vous authentifier en tant qu'utilisateur afin d'interroger l'agent.
• Importer et initialiser 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 une AdkApp, vous devez d'abord créer une instance ou en obtenir une existante.

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

SDK Agent Platform

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 est l' Google Cloud ID de projet sous lequel vous créez et déployez des agents.
• LOCATION désigne l'une des régions compatibles.
• RESOURCE_ID est l'ID de l'agent déployé en tant que reasoningEngine ressource.

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 Agent Platform, l'objet adk_app correspond à une AgentEngine classe qui contient les éléments suivants :

• adk_app.api_resource avec 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 prises en charge.
• adk_app.api_client qui permet les interactions de service synchrones
• adk_app.async_api_client qui permet les interactions de service asynchrones

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

Opérations prises en charge

Les opérations suivantes sont compatibles avec AdkApp :

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

• async_create_session: pour créer une session.

• async_list_sessions: pour répertorier 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 répertorier toutes les opérations prises en charge :

SDK Agent Platform

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 dans le cloud après le déploiement de l'agent sur Agent Platform. 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, utilisez la AdkApp.async_create_session méthode :

SDK Agent Platform

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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 avec une limite de 128 caractères. Par exemple, user-123.

La session est créée en tant que représentation de dictionnaire d'un objet de session ADK.

Répertorier les sessions

Pour répertorier les sessions d'un utilisateur, utilisez la AdkApp.async_list_sessions méthode :

SDK Agent Platform

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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 est l'ID utilisateur que vous avez défini. Par exemple, user-123.

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

Obtenir une session

Pour obtenir une session spécifique, utilisez la AdkApp.async_get_session méthode :

SDK Agent Platform

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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"},}'

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

Supprimer une session

Pour supprimer une session, utilisez la méthode AdkApp.async_delete_session :

SDK Agent Platform

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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 des réponses d'un agent dans une session, utilisez la AdkApp.async_stream_query méthode :

SDK Agent Platform

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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. S'il n'est pas spécifié, une session est automatiquement créée et utilisée pour traiter cette requête. Si l'ID de session fourni n'existe pas, une réponse vide est renvoyée. Nous vous recommandons de ne pas inclure cet argument si vous vérifiez uniquement si un agent est accessible.

Si vous utilisez le SDK Agent Platform, 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',
 # ...
}

Jobs de requête de longue durée

Pour les requêtes dont l'exécution peut prendre beaucoup de temps (jusqu'à sept jours), vous pouvez les exécuter en tant que jobs de longue durée. Ces jobs s'exécutent de manière asynchrone. Vous pouvez vérifier l'état du job et récupérer les résultats ultérieurement.

Remarque : Vous ne pouvez exécuter des jobs de longue durée que sur des ressources reasoningEngine créées à partir du 22 avril 2026. Vous ne pouvez pas mettre à jour les ressources reasoningEngine créées précédemment pour exécuter des jobs de longue durée.

Déployer un agent pour une requête asynchrone

Pour déployer un agent, suivez les instructions générales de la section Déployer un agent. Pour le déploiement basé sur la source, définissez le champ deploymentSpec.agentFramework sur google-adk.

Si vous utilisez un point de terminaison d'API personnalisé en créant votre propre image de conteneur, vous devez ajouter les variables d'environnement suivantes lors de la création de l'agent à l'aide du SDK :

"env_vars" = {
    "API_ENDPOINT_PREFIX": "/api/myendpoint"
}

Démarrer un job de requête de longue durée

Vous devez au préalable accorder le rôle roles/storage.objectCreator à l'agent de service [e-mail] pour le bucket de stockage des fichiers de sortie.

Pour démarrer un job de requête de longue durée :

SDK Agent Platform

import vertexai

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

response = client.agent_engines.run_query_job(
    name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    config={
        "query": '{"input":{"user_id":"USER_ID", "message":"What is the exchange rate from US dollars to SEK today?"}}',
        "output_gcs_uri": "gs://GCS_BUCKET_NAME/OUTPUT_FILE",
    },
)
print(response)

Avec le SDK, output_gcs_uri peut être un répertoire ou un nom de fichier. S'il s'agit d'un nom de fichier, le système utilise ce fichier pour stocker la réponse. S'il s'agit d'un répertoire, le système génère automatiquement un fichier pour la réponse. Dans les deux cas, la requête d'entrée est stockée dans le même répertoire avec le même préfixe de nom de fichier que le fichier de sortie.

REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1beta1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:asyncQuery -d \
'{
  "input_gcs_uri": "gs://GCS_BUCKET_NAME/INPUT_FILE",
  "output_gcs_uri": "gs://GCS_BUCKET_NAME/OUTPUT_FILE"
}'

Pour l'appel d'API REST, le champ input_gcs_uri doit pointer vers un fichier contenant la requête. Le contenu du fichier doit être un objet JSON avec un champ input qui correspond au champ input de QueryReasoningEngineRequest (par exemple, { "input": { "user_id": "hello", "message":"$QUERY"} }). Si ce fichier d'entrée se trouve dans un bucket différent de l'emplacement de sortie, vous devez également accorder le rôle roles/storage.objectReader à l'agent de service [e-mail] pour le bucket de stockage où se trouvent les fichiers d'entrée.

output_gcs_uri doit être un nom de fichier.

Vérifier l'état d'un job de requête de longue durée

Pour vérifier l'état et récupérer les résultats d'un job de requête de longue durée :

SDK Agent Platform

response = client.agent_engines.check_query_job(
    name="JOB_NAME",
    config={
        "retrieve_result": True,
    },
)
print(response)

Annuler un job de requête de longue durée

Pour annuler un job de requête de longue durée, vous devez disposer du nom de ressource LRO renvoyé par le job de requête de longue durée.

SDK Agent Platform

response = client.agent_engines.cancel_query_job(
    name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    config={
        "operation_name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID",
    },
)

REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1beta1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:cancelAsyncQuery -d \
'{
  "name": "projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
  "operation_name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID"
}'

L'annulation est asynchrone. La requête d'annulation est renvoyée dès qu'elle est acceptée, mais le job peut continuer à signaler l'état RUNNING à partir de check_query_job jusqu'à ce que le travail déjà en cours, tel qu'un appel d'outil bloquant, soit terminé.

Une fois le job annulé, son opération se termine avec le code d'erreur 1 (CANCELLED) et le message Cancelled by user.. Notez que check_query_job signale chaque opération terminée avec une erreur comme état FAILED. Par conséquent, un job annulé est signalé comme FAILED plutôt que par un état annulé distinct. Inspectez le code d'erreur pour distinguer une annulation d'un véritable échec.

Gérer les souvenirs

AdkApp utilise Memory Bank si vous incluez un PreloadMemoryTool dans la définition de l'agent et que vous déployez l'agent sur Agent Platform. Cette section explique comment générer et récupérer des souvenirs à partir de l'agent via 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 significatives dans une session (qui peuvent être utilisées dans des sessions futures ), utilisez la méthode async_add_session_to_memory :

SDK Agent Platform

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

où SESSION_DICT est la représentation 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 async_search_memory méthode :

SDK Agent Platform

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

Où :

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

Étape suivante

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