Usar um agente do Agent Development Kit

Antes de começar

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

• Desenvolver um agente do Kit de Desenvolvimento de Agente: para desenvolver agent como uma instância de AdkApp.
• Autenticação de usuário para se autenticar como um usuário e consultar o agente.
• Importe e inicialize o SDK para inicializar o cliente e receber uma instância implantada (se necessário).

Receber uma instância de um agente

Para consultar um AdkApp, primeiro crie uma instância ou acesse uma instância atual.

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

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

adk_app = client.agent_engines.get(name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

print(adk_app)

em que

• PROJECT_ID é o ID do projeto do Google Cloud em que você desenvolve e implanta agentes.
• LOCATION é uma das regiões com suporte, e
• RESOURCE_ID é o ID do agente implantado como um recurso reasoningEngine.

Biblioteca de solicitações do Python

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()}",
    },
)

API 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

Ao usar o SDK da Vertex AI para Python, o objeto adk_app corresponde a uma classe AgentEngine que contém o seguinte:

• um adk_app.api_resource com informações sobre o agente implantado. Também é possível chamar adk_app.operation_schemas() para retornar a lista de operações compatíveis com adk_app. Consulte Operações compatíveis para mais detalhes.
• um adk_app.api_client que permite interações de serviço síncronas
• um adk_app.async_api_client que permite interações assíncronas de serviço

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

Operações suportadas

As seguintes operações são compatíveis com AdkApp:

• async_stream_query: para transmitir uma resposta a uma consulta.

• async_create_session: para criar uma nova sessão.

• async_list_sessions: para listar as sessões disponíveis.

• async_get_session: para recuperar uma sessão específica.

• async_delete_session: para excluir uma sessão específica.

• async_add_session_to_memory: para gerar memórias de uma sessão.

• async_search_memory: para recuperar memórias.

Para listar todas as operações compatíveis:

SDK da Vertex AI para Python

Execute o seguinte código:

adk_app.operation_schemas()

Biblioteca de solicitações do Python

Execute o seguinte código:

import json

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

API REST

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

Gerenciar sessões

O AdkApp usa sessões gerenciadas baseadas na nuvem depois que você implanta o agente no mecanismo de agente da Vertex AI. Esta seção descreve como usar sessões gerenciadas.

Observação :estas instruções pressupõem que você não personalizou seu banco de dados usando session_service_builder ao desenvolver o agente.

Criar uma sessão

Para criar uma sessão para um usuário:

SDK da Vertex AI para Python

session = await adk_app.async_create_session(user_id="USER_ID")

print(session)

Biblioteca de solicitações do Python

Execute o seguinte código:

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_create_session",
    "input": {"user_id": "USER_ID"},
  }),
)
print(response.content)

API 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": "async_create_session", "input": {"user_id": "USER_ID"},}'

• USER_ID: escolha seu próprio ID de usuário com um limite de 128 caracteres. Por exemplo, user-123.

A sessão é criada como a representação de dicionário de um objeto de sessão do ADK.

Listar sessões

Para listar as sessões de um usuário:

SDK da Vertex AI para Python

response = await adk_app.async_list_sessions(user_id="USER_ID"):
for session in response.sessions:
    print(session)

Biblioteca de solicitações do Python

Execute o seguinte código:

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_list_sessions",
    "input": {"user_id": "USER_ID"},
  }),
)
print(response.content)

API 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": "async_list_sessions", "input": {"user_id": "USER_ID"},}'

em que USER_ID é o ID do usuário que você definiu. Por exemplo, user-123.

Se alguma sessão for retornada, ela usará a forma de dicionário de um objeto de sessão do ADK.

Acessar uma sessão

Para acessar uma sessão específica, você precisa do ID do usuário e do ID da sessão:

SDK da Vertex AI para Python

session = await adk_app.async_get_session(user_id="USER_ID", session_id="SESSION_ID")

print(session)

Biblioteca de solicitações do Python

Execute o seguinte código:

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_get_session",
    "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},
  }),
)
print(response.content)

API 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": "async_get_session", "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},}'

O session é a representação de dicionário de um objeto de sessão do ADK.

Excluir uma sessão

Para excluir uma sessão, você precisa do ID do usuário e do ID da sessão:

SDK da Vertex AI para Python

await adk_app.async_delete_session(user_id="USER_ID", session_id="SESSION_ID")

Biblioteca de solicitações do Python

Execute o seguinte código:

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_delete_session",
    "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},
  }),
)
print(response.content)

API 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": "async_delete_session", "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},}'

Transmitir uma resposta a uma consulta

Para transmitir respostas de um agente em uma sessão:

SDK da Vertex AI para Python

async for event in adk_app.async_stream_query(
    user_id="USER_ID",
    session_id="SESSION_ID",  # Optional
    message="What is the exchange rate from US dollars to SEK today?",
):
  print(event)

Biblioteca de solicitações do Python

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://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:streamQuery",
    headers={
        "Content-Type": "application/json",
        "Authorization": f"Bearer {get_identity_token()}",
    },
    data=json.dumps({
        "class_method": "async_stream_query",
        "input": {
            "user_id": "USER_ID",
            "session_id": "SESSION_ID",
            "message": "What is the exchange rate from US dollars to SEK today?",
        },
    }),
    stream=True,
)

API 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": "async_stream_query",
  "input": {
    "user_id": "USER_ID",
    "session_id": "SESSION_ID",
    "message": "What is the exchange rate from US dollars to SEK today?",
  }
}'

Observação: o argumento session_id= é opcional. Se não for especificado, uma nova sessão será criada automaticamente e usada para veicular essa consulta.

Se você estiver usando o SDK da Vertex AI para Python, vai receber uma continuação da conversa, como a seguinte sequência de dicionários:

{'author': 'currency_exchange_agent',
 'content': {'parts': [{'function_call': {'args': {'currency_date': '2025-04-03',
                                                   'currency_from': 'USD',
                                                   'currency_to': 'SEK'},
                                          'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
                                          'name': 'get_exchange_rate'}}],
             'role': 'model'},
 'id': 'bOPHtzji',
 # ...
}
{'author': 'currency_exchange_agent',
 'content': {'parts': [{'function_response': {'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
                                              'name': 'get_exchange_rate',
                                              'response': {'amount': 1.0,
                                                           'base': 'USD',
                                                           'date': '2025-04-03',
                                                           'rates': {'SEK': 9.6607}}}}],
             'role': 'user'},
 'id': '9AoDFmiL',
 # ...
}
{'author': 'currency_exchange_agent',
 'content': {'parts': [{'text': 'The exchange rate from USD to SEK on '
                                '2025-04-03 is 1 USD to 9.6607 SEK.'}],
             'role': 'model'},
 'id': 'hmle7trT',
 # ...
}

Gerenciar recordações

AdkApp usa o Memory Bank do Vertex AI Agent Engine se você incluir um PreloadMemoryTool na definição do agente e implantar o agente no Vertex AI Agent Engine. Esta seção descreve como usar o serviço de memória do ADK para gerar e recuperar memórias do agente.

Adicionar sessão à recordação

Para reter a memória de informações significativas em uma sessão (que podem ser usadas em sessões futuras), use o método async_add_session_to_memory:

SDK da Vertex AI para Python

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

em que SESSION_DICT é a forma de dicionário de um objeto de sessão do ADK.

Pesquisar recordações

Para pesquisar nas memórias do agente, use o método async_search_memory:

SDK da Vertex AI para Python

response = await adk_app.async_search_memory(
    user_id="USER_ID",
    query="QUERY",
)
print(response)

em que

• USER_ID é o escopo das recordações relevantes.
• QUERY é a consulta para realizar a pesquisa de similaridade.

A seguir

• Use um agente.
• Avalie um agente.
• Gerenciar agentes implantados.
• Receba suporte.