Usar un agente personalizado

Antes de empezar

En este tutorial se da por hecho que has leído y seguido las instrucciones de los siguientes artículos:

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

Obtener una instancia de un agente

Para consultar un agente, primero necesitas una instancia de un agente. Puedes crear una instancia o obtener una instancia 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 de proyecto en el que desarrollas y despliegas agentes.
• LOCATION es una de las regiones admitidas.
• RESOURCE_ID es el ID del agente implementado como 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 que se devuelva la lista de operaciones que admite el agente. Consulta Operaciones admitidas para obtener más información.
• 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 presupone que tienes una instancia llamada agent.

Mostrar operaciones admitidas

Cuando desarrollas el agente de forma local, tienes acceso y conocimiento de las operaciones que admite. 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 de la respuesta a la solicitud curl.

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

• Agent Development Kit
• Agent2Agent
• LangChain
• LangGraph
• AG2
• LlamaIndex
• Personalizado

Por ejemplo, a continuación se muestra 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 (por ejemplo, agent.query para una operación llamada query).
• api_mode es el modo de la API de la operación ("" para síncrono y "stream" para streaming).
• description es una descripción de la operación basada en el docstring del método.
• parameters es el esquema de los argumentos de entrada en formato de esquema OpenAPI.

Consultar el agente mediante operaciones admitidas

En el caso de los agentes personalizados, puedes usar cualquiera de las operaciones de consulta o de streaming que hayas definido al desarrollar el agente:

• query

• stream_query

• async_query

• async_stream_query

Ten en cuenta que algunos frameworks solo admiten consultas u operaciones de streaming específicas:

Framework	Operaciones de consulta admitidas
Agent Development Kit	async_stream_query
LangChain	query, stream_query
LangGraph	query, stream_query
AG2	query
LlamaIndex	query

Consultar el agente

Consulta el agente mediante 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 a la salida 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 graduales del agente

Transmite una respuesta del agente mediante 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 tener el siguiente aspecto:

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

Consultar el agente de forma asíncrona

Si has definido una operación async_query al desarrollar el agente, el SDK de Vertex AI para Python admite consultas asíncronas del lado del cliente del agente:

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 igual que el 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."}

Transmitir respuestas del agente de forma asíncrona

Si has definido una operación async_stream_query al desarrollar el agente, puedes transmitir de forma asíncrona una respuesta del agente mediante una de sus operaciones (por ejemplo, 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 endpoint 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 tener el siguiente aspecto:

{'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 generadas durante las pruebas locales.

Siguientes pasos

• Usar un agente.
• Evalúa un agente.
• Gestionar agentes implementados.
• Obtener asistencia