Usar um agente

O código para consultar um agente é o mesmo, seja ele executado localmente ou implantado remotamente. Portanto, nesta página, o termo agent se refere a local_agent ou remote_agent de forma intercambiável. Como o conjunto de operações compatíveis varia entre os frameworks, fornecemos instruções de uso para modelos específicos de cada framework:

Framework Descrição
Kit de desenvolvimento de agente (prévia) Projetado com base nas práticas recomendadas internas do Google para desenvolvedores que criam aplicativos de IA ou equipes que precisam criar protótipos e implantar rapidamente soluções robustas baseadas em agentes.
LangChain Mais fácil de usar para casos de uso básicos devido às configurações e abstrações predefinidas.
LangGraph Abordagem baseada em gráficos para definir fluxos de trabalho, com recursos avançados de human-in-the-loop e de voltar/reproduzir.
AG2 (antigo AutoGen) O AG2 oferece um framework de conversa multiagente como uma abstração de alto nível para criar fluxos de trabalho de LLM.
LlamaIndex (pré-lançamento) O pipeline de consultas do LlamaIndex oferece uma interface de alto nível para criar fluxos de trabalho de geração aumentada de recuperação (RAG).

Para agentes personalizados que não são baseados em um dos modelos específicos da estrutura, siga estas etapas:

  1. Autenticação de usuário.
  2. Receba uma instância de agente.
  3. Pesquise as operações compatíveis.
  4. Consultar o agente.
  5. (Se aplicável) Transmita respostas do agente.

Etapa 1: autenticação do usuário

Siga as mesmas instruções de configuração do ambiente.

Etapa 2: acessar uma instância de um agente

Para consultar um agente, primeiro você precisa de uma instância dele. É possível criar uma instância ou acessar uma instância de um agente.

Para receber o agente correspondente a um ID de recurso específico:

SDK da Vertex AI para Python

Execute o seguinte código:

from vertexai import agent_engines

agent = agent_engines.get("RESOURCE_ID")

Como alternativa, forneça o nome completo do recurso do agente:

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

solicitações

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

O restante desta seção pressupõe que você tenha uma instância chamada agent.

Etapa 3: operações compatíveis

Ao desenvolver o agente localmente, você tem acesso e conhecimento das operações que ele oferece suporte. Para usar um agente implantado, enumere as operações compatíveis:

SDK da Vertex AI para Python

Execute o seguinte código:

agent.operation_schemas()

solicitações

Execute o seguinte código:

import json

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

REST

Representado em spec.class_methods da resposta à solicitação curl.

O esquema de cada operação é um dicionário que documenta as informações de um método para o agente que você pode chamar. Confira a seguir um exemplo do esquema de operação para uma operação síncrona:

O comando a seguir fornece uma lista de esquemas no formato JSON que correspondem às operações do objeto remote_app:

agent.operation_schemas()

Por exemplo, este é 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'}}

em que

  • name é o nome da operação (por exemplo, agent.query para uma operação chamada 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.

Etapa 4: consultar o agente

Para consultar o agente usando uma das operações compatíveis (por exemplo, query):

SDK da Vertex AI para Python 

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

solicitações 

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 à saída de um teste de aplicativo 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."}

Etapa 5: transmitir respostas do agente

Se aplicável, transmita uma resposta do agente usando uma das operações dele (por exemplo, stream_query):

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

solicitações 

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 de forma iterativa. Por exemplo, um conjunto de três respostas pode ser assim:

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

Etapa 6: consultar o agente de forma assíncrona

Se você definiu uma operação async_query ao desenvolver o agente, há suporte para consultas assíncronas do lado do cliente no SDK da Vertex AI para Python.

SDK da 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 à saída 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."}

Etapa 7: transmitir respostas do agente de forma assíncrona

Se você definiu uma operação async_stream_query ao desenvolver o agente, é possível transmitir uma resposta de forma assíncrona usando uma das operações dele (por exemplo, async_stream_query):

SDK da 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 endpoint streamQuery por baixo dos panos e transmite respostas de forma assíncrona como uma sequência de objetos gerados iterativamente. Por exemplo, um conjunto de três respostas pode ser assim:

{'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 precisam ser as mesmas geradas durante o teste local.

A seguir