Use um agente personalizado

Antes de começar

Este tutorial pressupõe que leu e seguiu as instruções em:

Obtenha uma instância de um agente

Para consultar um agente, primeiro precisa de uma instância de um agente. Pode criar uma nova instância ou obter uma instância existente de um agente.

Para obter o agente correspondente a um ID do recurso específico:

SDK Vertex AI para Python

Execute o seguinte 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)

onde

pedidos

Execute o seguinte 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

Quando usa o SDK Vertex AI para Python, o objeto agent corresponde a uma classe AgentEngine que contém o seguinte:

  • Um agent.api_resource com informações sobre o agente implementado. Também pode chamar agent.operation_schemas() para devolver a lista de operações que o agente suporta. Consulte o artigo Operações suportadas para ver detalhes.
  • Um agent.api_client que permite interações de serviço síncronas
  • Um agent.async_api_client que permite interações de serviço assíncronas

O resto desta secção pressupõe que tem uma instância com o nome agent.

Apresente as operações suportadas

Quando desenvolve o agente localmente, tem acesso e conhecimento das operações que suporta. Para usar um agente implementado, pode enumerar as operações que suporta:

SDK Vertex AI para Python

Execute o seguinte código:

print(agent.operation_schemas())

pedidos

Execute o seguinte código:

import json

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

REST

Representado em spec.class_methods a partir da resposta ao pedido curl.

O esquema de cada operação é um dicionário que documenta as informações de um método para o agente que pode chamar. O conjunto de operações suportadas depende da framework que usou para desenvolver o seu agente:

Por exemplo, segue-se o esquema da operação query de um 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'}}

onde

  • name é o nome da operação (ou seja, agent.query para uma operação denominada query).
  • api_mode é o modo da API da operação ("" para síncrono, "stream" para streaming).
  • description é uma descrição da operação com base na docstring do método.
  • parameters é o esquema dos argumentos de entrada no formato de esquema OpenAPI.

Consultar o agente através de operações suportadas

Para agentes personalizados, pode usar qualquer uma das seguintes operações de consulta ou streaming que definiu quando desenvolveu o seu agente:

Tenha em atenção que determinadas estruturas apenas suportam operações de consulta ou streaming específicas:

Framework Operações de consulta suportadas
Agent Development Kit async_stream_query
LangChain query, stream_query
LangGraph query, stream_query
AG2 query
LlamaIndex query

Consultar o agente

Consulte o agente através da operação query:

SDK Vertex AI para Python 

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

pedidos 

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

A resposta da consulta é uma string semelhante ao resultado de um teste de aplicação 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."}

Respostas graduais do agente

Transmita uma resposta do agente através da operação stream_query:

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

pedidos 

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

O Vertex AI Agent Engine transmite respostas como uma sequência de objetos gerados iterativamente. Por exemplo, um conjunto de três respostas pode ter o seguinte aspeto:

{'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 o agente de forma assíncrona

Se definiu uma operação async_query quando desenvolveu o agente, existe suporte para consultas assíncronas do lado do cliente do agente no SDK Vertex AI para Python:

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

A resposta da consulta é um dicionário igual ao resultado de um teste 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."}

Fazer stream de respostas do agente de forma assíncrona

Se definiu uma operação async_stream_query quando desenvolveu o agente, pode fazer streaming assíncrono de uma resposta do agente através de uma das respetivas operações (por exemplo, async_stream_query):

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

A operação async_stream_query chama o mesmo ponto final streamQuery nos bastidores e transmite respostas de forma assíncrona como uma sequência de objetos gerados iterativamente. Por exemplo, um conjunto de três respostas pode ter o seguinte aspeto:

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

As respostas devem ser as mesmas que as geradas durante os testes locais.

O que se segue?