Usa un agente

El código para consultar un agente es el mismo, independientemente de si se ejecuta localmente o se implementa de forma remota. Por lo tanto, en esta página, el término agent se refiere a local_agent o remote_agent de forma indistinta. Como el conjunto de operaciones admitidas varía según los frameworks, proporcionamos instrucciones de uso para las plantillas específicas de cada framework:

Framework Descripción
Kit de desarrollo de agentes (vista previa) Diseñado según las prácticas recomendadas internas de Google para los desarrolladores que crean aplicaciones basadas en IA o los equipos que necesitan crear prototipos e implementar rápidamente soluciones sólidas basadas en agentes
LangChain Es más fácil de usar para casos de uso básicos debido a sus configuraciones y abstracciones predefinidas.
LangGraph Enfoque basado en gráficos para definir flujos de trabajo, con capacidades avanzadas de interacción humana y de retroceso/reproducción.
AG2 (anteriormente AutoGen) AG2 proporciona un framework de conversación multiagente como una abstracción de alto nivel para crear flujos de trabajo de LLM.
LlamaIndex (versión preliminar) La canalización de consultas de LlamaIndex ofrece una interfaz de alto nivel para crear flujos de trabajo de generación mejorada por recuperación (RAG).

En el caso de los agentes personalizados que no se basan en una de las plantillas específicas del framework, puedes seguir estos pasos:

  1. Autenticación del usuario
  2. Obtén una instancia del agente.
  3. Busca operaciones admitidas.
  4. Realiza una consulta al agente.
  5. (Si corresponde) Transmite las respuestas del agente.

Paso 1: Autenticación del usuario

Sigue las mismas instrucciones que para configurar tu entorno.

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

from vertexai import agent_engines

agent = agent_engines.get("RESOURCE_ID")

Como alternativa, puedes proporcionar el nombre completo del recurso del agente:

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

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

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

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

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. A continuación, se muestra un ejemplo del esquema de operación para una operación síncrona:

Con el siguiente comando, se proporciona una lista de esquemas en formato JSON que corresponden a las operaciones del objeto remote_app:

agent.operation_schemas()

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.

Paso 4: Hazle preguntas al agente

Para consultar el agente con una de sus operaciones compatibles (p.ej., query), haz lo siguiente:

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

Paso 5: Transmite las respuestas del agente

Si corresponde, puedes transmitir una respuesta del agente con una de sus operaciones (p.ej., 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

Paso 6: Realiza consultas 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 lado 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."}

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

¿Qué sigue?