Cómo usar un agente personalizado

Nota: Esta página es para agentes personalizados implementados desde una plantilla de clase de Python (serializada). Si implementas contenedores personalizados o usas un Dockerfile, consulta el contrato del entorno de ejecución.

Antes de comenzar

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

• Crea un agente personalizado: para crear un agent personalizado.
• Autenticación de usuarios 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).

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

client = agentplatform.Client(  # For service interactions via client.runtimes
    project="PROJECT_ID",
    location="LOCATION",
)

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

print(agent)

donde

• PROJECT_ID es el ID del proyecto de Google Cloud en el que creas y implementas 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 usas el SDK de Agent Platform para Python, el objeto agent corresponde a una clase AgentRuntime que contiene lo siguiente:

• 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 agent. Consulta Operaciones compatibles para obtener más detalles.
• agent.api_client que permite interacciones de servicio síncronas
• agent.async_api_client que permite interacciones asíncronas entre servicios

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

Por ejemplo, el siguiente es el esquema para 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, "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 el formato de esquema de OpenAPI.

Consulta al agente con las operaciones admitidas

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

Consulta al 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 = client.runtimes.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 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 asíncrona al agente

Si definiste una operación async_query cuando creaste el agente, hay compatibilidad con las consultas asíncronas del agente del lado del cliente en el SDK de Agent Platform para Python:

SDK de Agent Platform para Python

agent = client.runtimes.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 creaste el agente, puedes transmitir de forma asíncrona una respuesta del agente con una de sus operaciones (p.ej., async_stream_query):

SDK de Agent Platform para Python

agent = client.runtimes.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 generadas durante las pruebas locales.

Trabajos de consulta de larga duración

En el caso de las consultas que pueden tardar mucho tiempo 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 agentplatform

client = agentplatform.Client(
    project="PROJECT_ID",
    location="LOCATION",
)

response = client.runtimes.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.runtimes.check_query_job(
    name="JOB_NAME",
    config={
        "retrieve_result": True,
    },
)
print(response)

¿Qué sigue?

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