Utiliser un agent personnalisé

Avant de commencer

Pour suivre ce tutoriel, vous devez avoir lu et suivi les instructions de :

Obtenir une instance d'un agent

Pour interroger un agent, vous avez d'abord besoin d'une instance d'agent. Vous pouvez soit créer une nouvelle instance soit obtenir une instance existante d'un agent.

Pour obtenir l'agent 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",
)

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

print(agent)

Où :

requêtes

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()}",
    },
)

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

Dans le reste de cette section, nous partons du principe que vous disposez d'une instance nommée agent.

Répertorier les opérations compatibles

Lorsque vous développez l'agent localement, vous avez accès aux opérations qu'il prend en charge et vous les connaissez. Pour utiliser un agent déployé, vous pouvez énumérer les opérations qu'il prend en charge :

SDK Agent Platform

Exécutez le code suivant :

print(agent.operation_schemas())

requêtes

Exécutez le code suivant :

import json

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

REST

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

Le schéma de chaque opération est un dictionnaire qui documente les informations d'une méthode pour l'agent que vous pouvez appeler. L'ensemble des opérations compatibles dépend du framework que vous avez utilisé pour développer votre agent :

Par exemple, voici le schéma de l'opération query d'un LangchainAgent :

{'api_mode': '',
 'name': 'query',
 'description': """Queries the Agent with the given input and config.
    Args:
        input (Union[str, Mapping[str, Any]]):
            Required. The input to be passed to the Agent.
        config (langchain_core.runnables.RunnableConfig):
            Optional. The config (if any) to be used for invoking the Agent.
    Returns:
        The output of querying the Agent with the given input and config.
""",            '        ',
 'parameters': {'$defs': {'RunnableConfig': {'description': 'Configuration for a Runnable.',
                                             'properties': {'configurable': {...},
                                                            'run_id': {...},
                                                            'run_name': {...},
                                                            ...},
                                             'type': 'object'}},
                'properties': {'config': {'nullable': True},
                               'input': {'anyOf': [{'type': 'string'}, {'type': 'object'}]}},
                'required': ['input'],
                'type': 'object'}}

Où :

  • name correspond au nom de l'opération (par exemple, agent.query pour une opération nommée query).
  • api_mode correspond au mode API de l'opération ("" pour synchrone, "stream" pour streaming).
  • description correspond à une description de l'opération basée sur la docstring de la méthode.
  • parameters correspond au schéma des arguments d'entrée au format de schéma OpenAPI.

Interroger l'agent à l'aide d'opérations compatibles

Pour les agents personnalisés, vous pouvez utiliser l'une des opérations de requête ou de streaming suivantes que vous avez définies lors du développement de votre agent :

Notez que certains frameworks ne sont compatibles qu'avec des opérations de requête ou de streaming spécifiques :

Framework Opérations de requête compatibles
Agent Development Kit async_stream_query
LangChain query, stream_query
LangGraph query, stream_query
AG2 query
LlamaIndex query

Interroger l'agent

Interrogez l'agent à l'aide de l'opération query :

SDK Agent Platform 

agent.query(input="What is the exchange rate from US dollars to Swedish Krona today?")

requêtes 

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:query",
    headers={
        "Content-Type": "application/json; charset=utf-8",
        "Authorization": f"Bearer {get_identity_token()}",
    },
    data=json.dumps({
        "class_method": "query",
        "input": {
            "input": "What is the exchange rate from US dollars to Swedish Krona today?"
        }
    })
)

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": "query",
  "input": {
    "input": "What is the exchange rate from US dollars to Swedish Krona today?"
  }
}'

La réponse à la requête est une chaîne semblable au résultat d'un test d'application locale :

{"input": "What is the exchange rate from US dollars to Swedish Krona today?",
 # ...
 "output": "For 1 US dollar you will get 10.7345 Swedish Krona."}

Diffuser en streaming les réponses de l'agent

Diffusez en streaming une réponse de l'agent à l'aide de l'opération stream_query :

SDK Agent Platform 

agent = agent_engines.get("projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

for response in agent.stream_query(
    input="What is the exchange rate from US dollars to Swedish Krona today?"
):
    print(response)

requêtes 

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": "stream_query",
        "input": {
            "input": "What is the exchange rate from US dollars to Swedish Krona today?"
        },
    }),
    stream=True,
)

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": "stream_query",
  "input": {
    "input": "What is the exchange rate from US dollars to Swedish Krona today?"
  }
}'

Agent Platform diffuse les réponses en streaming sous forme de séquence d'objets générés de manière itérative. Par exemple, un ensemble de trois réponses peut se présenter comme suit :

{'actions': [{'tool': 'get_exchange_rate', ...}]}  # first response
{'steps': [{'action': {'tool': 'get_exchange_rate', ...}}]}  # second response
{'output': 'The exchange rate is 11.0117 SEK per USD as of 2024-12-03.'}  # final response

Interroger l'agent de manière asynchrone

Si vous avez défini une opération async_query lors de la création de l'agent, le SDK Agent Platform pour Python est compatible avec l'interrogation asynchrone de l'agent côté client :

SDK Agent Platform pour Python 

agent = agent_engines.get("projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

response = await agent.async_query(
    input="What is the exchange rate from US dollars to Swedish Krona today?"
)
print(response)

La réponse à la requête est un dictionnaire identique à la sortie d'un test local :

{"input": "What is the exchange rate from US dollars to Swedish Krona today?",
 # ...
 "output": "For 1 US dollar you will get 10.7345 Swedish Krona."}

Diffuser en streaming de manière asynchrone les réponses de l'agent

Si vous avez défini une opération async_stream_query lors de la création de l'agent, vous pouvez diffuser en streaming de manière asynchrone une réponse de l'agent à l'aide de l'une de ses opérations (par exemple, async_stream_query) :

SDK Agent Platform pour Python 

agent = agent_engines.get("projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

async for response in agent.async_stream_query(
    input="What is the exchange rate from US dollars to Swedish Krona today?"
):
    print(response)

L'opération async_stream_query appelle le même streamQuery point de terminaison en arrière-plan et diffuse en streaming de manière asynchrone les réponses sous forme de séquence d'objets générés de manière itérative. Par exemple, un ensemble de trois réponses peut se présenter comme suit :

{'actions': [{'tool': 'get_exchange_rate', ...}]}  # first response
{'steps': [{'action': {'tool': 'get_exchange_rate', ...}}]}  # second response
{'output': 'The exchange rate is 11.0117 SEK per USD as of 2024-12-03.'}  # final response

Les réponses doivent être identiques à celles générées lors des tests locaux.

Tâches 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 tâches de longue durée. Pour en savoir plus, consultez Utiliser un agent ADK.

Démarrer une tâche de requête de longue durée

Pour démarrer une tâche de requête de longue durée :

SDK Agent Platform pour Python 

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": {"product": "Google"} }',
        "output_gcs_uri": "gs://GCS_BUCKET_NAME/OUTPUT_FILE",
    },
)
print(response)

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

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 tâche 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)

Étape suivante