Se usó la API de Cloud Translation para traducir esta página.

Cómo usar un agente personalizado

Antes de comenzar

En este instructivo, se supone que leíste y seguiste las instrucciones que se indican en los siguientes documentos:

Desarrolla un agente personalizado: Para desarrollar un agent personalizado.
Autenticación del usuario para autenticarse como usuario y consultar al agente
Importa e inicializa el SDK para inicializar el cliente y obtener una instancia implementada (si es necesario).

Obtén una instancia de un agente

Para consultar un agente, primero necesitas una instancia de un agente. Puedes crear una instancia nueva o obtener una existente de un agente.

Para obtener el agente correspondiente a un ID de recurso específico, haz lo siguiente:

SDK de Vertex AI para Python

Ejecuta el siguiente código:

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)

donde

PROJECT_ID es el Google Cloud ID del proyecto en el que desarrollas y despliegas agentes.
LOCATION es una de las regiones admitidas.
RESOURCE_ID es el ID del agente implementado como un recurso reasoningEngine.

solicitudes

Ejecuta el siguiente código:

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

Cuando se usa el SDK de Vertex AI para Python, el objeto agent corresponde a una clase AgentEngine que contiene lo siguiente:

Un agent.api_resource con información sobre el agente implementado. También puedes llamar a agent.operation_schemas() para devolver la lista de operaciones que admite el agente. Consulta Operaciones admitidas para obtener más detalles.
un agent.api_client que permite interacciones de servicio síncronas
un agent.async_api_client que permite interacciones de servicio asíncronas

En el resto de esta sección, se supone que tienes una instancia llamada agent.

Enumera las operaciones admitidas

Cuando desarrollas el agente de forma local, tienes acceso a las operaciones que admite y sabes cuáles son. Para usar un agente implementado, puedes enumerar las operaciones que admite:

SDK de Vertex AI para Python

Ejecuta el siguiente código:

print(agent.operation_schemas())

solicitudes

Ejecuta el siguiente código:

import json

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

REST

Se representa en spec.class_methods a partir de la respuesta a la solicitud de curl.

El esquema de cada operación es un diccionario que documenta la información de un método para el agente al que puedes llamar. El conjunto de operaciones admitidas depende del framework que usaste para desarrollar tu agente:

Por ejemplo, el siguiente es el esquema de la operación query de 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'}}

donde

name es el nombre de la operación (es decir, agent.query para una operación llamada query).
api_mode es el modo de API de la operación ("" para síncrono y "stream" para transmisión).
description es una descripción de la operación basada en la cadena de documentación del método.
parameters es el esquema de los argumentos de entrada en formato de esquema de OpenAPI.

Consulta al agente con operaciones compatibles

En el caso de los agentes personalizados, puedes usar cualquiera de las siguientes operaciones de consulta o transmisión que definiste cuando desarrollaste tu agente:

query
stream_query
async_query
async_stream_query

Ten en cuenta que algunos frameworks solo admiten operaciones específicas de transmisión o consultas:

Framework	Operaciones de consulta admitidas
Kit de desarrollo de agentes	`async_stream_query`
LangChain	`query`, `stream_query`
LangGraph	`query`, `stream_query`
AG2	`query`
LlamaIndex	`query`

Consulta al agente

Consulta al agente con la operación query:

SDK de Vertex AI para Python

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

solicitudes

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 respuesta de la consulta es una cadena similar al resultado de una prueba de aplicación 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."}

Respuestas de transmisión del agente

Transmite una respuesta del agente con la operación stream_query:

SDK de Vertex AI para 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)

solicitudes

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 transmite las respuestas como una secuencia de objetos generados de forma iterativa. Por ejemplo, un conjunto de tres respuestas podría verse de la siguiente manera:

{'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

Consulta al agente de forma asíncrona

Si definiste una operación async_query cuando desarrollaste el agente, el SDK de Vertex AI para Python admite consultas asíncronas del agente del cliente:

SDK de Vertex AI para 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 respuesta de la consulta es un diccionario que es igual al resultado de una prueba 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."}

Transmite respuestas del agente de forma asíncrona

Si definiste una operación async_stream_query cuando desarrollaste el agente, puedes transmitir de forma asíncrona una respuesta del agente con una de sus operaciones (p.ej., async_stream_query):

SDK de Vertex AI para 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)

La operación async_stream_query llama al mismo extremo streamQuery de forma interna y transmite respuestas de forma asíncrona como una secuencia de objetos generados de forma iterativa. Por ejemplo, un conjunto de tres respuestas podría verse de la siguiente manera:

{'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

Las respuestas deben ser las mismas que las que se generan durante las pruebas locales.

Cómo usar un agente personalizado Organiza tus páginas con colecciones Guarda y categoriza el contenido según tus preferencias.

Antes de comenzar

Obtén una instancia de un agente

SDK de Vertex AI para Python

solicitudes

REST

Enumera las operaciones admitidas

SDK de Vertex AI para Python

solicitudes

REST

Consulta al agente con operaciones compatibles

Consulta al agente

SDK de Vertex AI para Python

solicitudes

REST

Respuestas de transmisión del agente

SDK de Vertex AI para Python

solicitudes

REST

Consulta al agente de forma asíncrona

SDK de Vertex AI para Python

Transmite respuestas del agente de forma asíncrona

SDK de Vertex AI para Python

¿Qué sigue?

Cómo usar un agente personalizado