Cette page a été traduite par l'API Cloud Translation.

Utiliser un agent personnalisé

Avant de commencer

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

Développer un agent personnalisé : pour développer un agent personnalisé.
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 agent, vous avez d'abord besoin d'une instance d'agent. Vous pouvez créer une instance ou obtenir une instance existante d'un agent.

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

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

print(agent)

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.

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

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

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

Lister 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 Vertex AI pour Python

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 est le 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 le mode synchrone et "stream" pour le mode streaming).
description est 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 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 que vous avez définies lors du développement de votre agent :

query
stream_query
async_query
async_stream_query

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 acceptées
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 Vertex AI pour Python

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 à la sortie 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 les réponses de l'agent

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

SDK Vertex AI pour Python

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

Vertex AI Agent Engine diffuse 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

Interroger l'agent de manière asynchrone

Si vous avez défini une opération async_query lors du développement de l'agent, la requête asynchrone côté client de l'agent est compatible avec le SDK Vertex AI pour Python :

SDK Vertex AI 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 de manière asynchrone les réponses de l'agent

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

SDK Vertex AI 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 point de terminaison streamQuery en coulisses et diffuse les réponses de manière asynchrone sous la forme d'une 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.

Utiliser un agent personnalisé Restez organisé à l'aide des collections Enregistrez et classez les contenus selon vos préférences.

Avant de commencer

Obtenir une instance d'un agent

SDK Vertex AI pour Python

requêtes

REST

Lister les opérations compatibles

SDK Vertex AI pour Python

requêtes

REST

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

Interroger l'agent

SDK Vertex AI pour Python

requêtes

REST

Diffuser les réponses de l'agent

SDK Vertex AI pour Python

requêtes

REST

Interroger l'agent de manière asynchrone

SDK Vertex AI pour Python

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

SDK Vertex AI pour Python

Étapes suivantes

Utiliser un agent personnalisé