Utiliser un agent Agent Development Kit

Avant de commencer

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

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 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ù :

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 :

Le reste de cette section suppose 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 :

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é 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 Agent Platform. Cette section explique comment utiliser les sessions gérées.

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 du 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 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, 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 du 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 en continu 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?",
  }
}'

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',
 # ...
}

Tâches 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 pourrez vérifier l'état du job et récupérer les résultats plus tard.

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 de l'API personnalisé en créant votre propre image de conteneur, vous devez ajouter les variables d'environnement suivantes lorsque vous créez 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 ce faire, vous devez accorder à l'agent de service service-PROJECT_NUMBER@gcp-sa-aiplatform-re.s3ns-system.iam.gserviceaccount.com le rôle roles/storage.objectCreator pour le bucket de stockage des fichiers de sortie.

Pour démarrer une tâche 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 (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 au compte de service service-PROJECT_NUMBER@gcp-sa-aiplatform-re.s3ns-system.iam.gserviceaccount.com le rôle roles/storage.objectReader pour le bucket de stockage dans lequel se trouvent les fichiers d'entrée.

output_gcs_uri doit être un nom de fichier.

Vérifier l'état d'une tâche de requête de longue durée

Pour vérifier l'état et récupérer les résultats d'une 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 une tâche de requête de longue durée, vous devez disposer du nom de ressource LRO renvoyé par la tâche 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 demande d'annulation est renvoyée dès qu'elle est acceptée, mais le job peut continuer à signaler un état RUNNING 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 qui s'est terminée avec une erreur avec l'état FAILED. Par conséquent, une tâche annulée est signalée 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 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 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 correspond au champ d'application des infos mémorisées concernées.
  • QUERY est la requête pour laquelle effectuer une recherche de similarité.

Étapes suivantes