Cómo usar un agente personalizado

Antes de comenzar

En este instructivo, se supone que leíste y seguiste las instrucciones que se indican a continuación:

• Crea un agente personalizado: Para crear un agent personalizado.
• Autenticación de usuarios: Para autenticarte como usuario para consultar el 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, incluso, obtener una instancia existente de un agente.

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

SDK de Agent Platform

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 creas y, luego, implementas agentes.
• LOCATION es una de las regiones admitidas.
• RESOURCE_ID es el ID del agente implementado como un reasoningEngine recurso.

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 Agent Platform para Python, el agent objeto corresponde a una AgentEngine clase que contiene lo siguiente:

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

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

Enumera las 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 Agent Platform

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

• Kit de desarrollo de agentes
• Agent2Agent
• LangChain
• LangGraph
• AG2
• LlamaIndex
• Personalizado

Como 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 denominada query).
• api_mode es el modo de API de la operación ("" para síncrono, "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 el agente con operaciones admitidas

Para 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 ciertos frameworks solo admiten operaciones de consulta o transmisión específicas:

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 el agente

Consulta el agente con la operación query:

SDK de Agent Platform

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

Transmite respuestas del agente

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

SDK de 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)

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

Agent Platform transmite 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 el agente de forma asíncrona

Si definiste una operación async_query cuando creaste el agente, el SDK de Agent Platform para Python admite la consulta asíncrona del agente del lado del cliente:

SDK de Agent Platform 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 el mismo 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."}

Transmite respuestas del agente de forma asíncrona

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

SDK de Agent Platform 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 streamQuery extremo 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 generadas durante las pruebas locales.

Trabajos de consulta de larga duración

Para las consultas que pueden tardar mucho en completarse (hasta siete días), puedes ejecutarlas como trabajos de larga duración. Para obtener más información, consulta Usa un agente de ADK.

Inicia un trabajo de consulta de larga duración

Para iniciar un trabajo de consulta de larga duración, haz lo siguiente:

SDK de Agent Platform para 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"
}'

Verifica el estado de un trabajo de consulta de larga duración

Para verificar el estado y recuperar los resultados de un trabajo de consulta de larga duración, haz lo siguiente:

SDK de Agent Platform

response = client.agent_engines.check_query_job(
    name="JOB_NAME",
    config={
        "retrieve_result": True,
    },
)
print(response)

¿Qué sigue?

• Usa un agente.
• Evalúa un agente.
• Administra agentes implementados.
• Obtén asistencia.