Utiliser un agent ADK

Avant de commencer

Ce tutoriel suppose que vous avez lu et suivi les instructions des documents suivants :

• Créer un agent Agent Development Kit (ADK) : pour créer agent en tant qu'instance de AdkApp.
• Authentification des utilisateurs 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 l'AdkApp correspondant à un ID de ressource spécifique :

SDK Agent Platform

Exécutez le code suivant :

import agentplatform

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

adk_app = client.runtimes.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 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 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 Agent Platform, l'objet adk_app correspond à une classe AgentRuntime 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 compatibles.
• adk_app.api_client qui permet des interactions de service synchrones
• adk_app.async_api_client qui permet des interactions de service asynchrones

Dans la suite de cette section, nous partons du principe que vous disposez d'une instance AgentRuntime, nommée adk_app.

Opérations prises en charge

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 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é par spec.class_methods dans 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 méthode AdkApp.async_create_session :

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

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

Répertorier les sessions

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

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. 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, utilisez la méthode AdkApp.async_get_session :

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"},}'

session est la représentation sous forme 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

Remarque : La durée maximale d'une requête streamQuery est de 10 minutes. Pour les workflows qui nécessitent des temps d'exécution plus longs, utilisez plutôt des jobs de requête de longue durée.

Pour diffuser les réponses d'un agent dans une session, utilisez la méthode AdkApp.async_stream_query :

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. Si elle n'est pas spécifiée, une nouvelle session est automatiquement créée et utilisée pour diffuser 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 l'inclure si vous souhaitez simplement vérifier si un agent est joignable.

Si vous utilisez le SDK Agent Platform, vous devriez recevoir une suite 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 qui peuvent 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 plus tard.

Remarque : Vous ne pouvez exécuter des jobs de longue durée que sur les 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 un 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

Pour cela, vous devez attribuer le rôle roles/storage.objectCreator à l'agent de service [e-mail] sur le bucket de stockage des fichiers de sortie.

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

SDK Agent Platform

import agentplatform

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

response = client.runtimes.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 correspondant au champ input de QueryReasoningEngineRequest (tel que { "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 attribuer à l'agent de service [e-mail] le rôle roles/storage.objectReader sur 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.runtimes.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.runtimes.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, comme un appel d'outil bloquant, se termine.

Une fois le job annulé, son opération se termine avec un code d'erreur 1 (CANCELLED) et le message Cancelled by user.. Notez que check_query_job signale chaque opération qui s'est terminée avec une erreur comme ayant l'état FAILED. Par conséquent, un job annulé est signalé comme FAILED plutôt que par un état d'annulation distinct. Examinez 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 infos mémorisées à partir de l'agent via l'implémentation par défaut du service de mémoire ADK.

Ajouter la session à la mémoire

Pour conserver en mémoire les informations importantes d'une session (qui pourront être utilisées lors de sessions ultérieures), 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 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 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 informations mémorisées pertinentes.
• QUERY est la requête pour laquelle effectuer une recherche de similarité.

Étapes suivantes

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