Usar um agente do Kit de Desenvolvimento de Agente

Antes de começar

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

• Criar um agente do Kit de Desenvolvimento de Agente: para criar 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 que corresponde a um ID de recurso específico:

SDK do Agent Platform

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 Google Cloud ID do projeto em que você cria 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 Agent Platform, o objeto adk_app corresponde a uma classe AgentEngine que contém o seguinte:

• 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 o adk_app. Consulte Operações compatíveis para mais detalhes.
• adk_app.api_client que permite interações de serviço síncronas
• 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 recordações.

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

SDK do Agent Platform

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 na Agent Platform. 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, use o método AdkApp.async_create_session:

SDK do Agent Platform

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://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": "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, use o método AdkApp.async_list_sessions:

SDK do Agent Platform

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://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": "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á o formato de dicionário de um objeto de sessão do ADK.

Acessar uma sessão

Para acessar uma sessão específica, use o método AdkApp.async_get_session:

SDK do Agent Platform

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://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": "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, use o método AdkApp.async_delete_session:

SDK do Agent Platform

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://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": "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, use o método AdkApp.async_stream_query:

SDK do Agent Platform

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://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": "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 o ID da sessão fornecido não existir, uma resposta vazia será retornada. Recomendamos que você deixe isso de fora se estiver apenas verificando se um agente está disponível.

Se você estiver usando o SDK da Agent Platform, vai receber uma continuação da conversa, como esta 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',
 # ...
}

Jobs de consulta de longa duração

Para consultas que podem levar muito tempo para serem concluídas (até sete dias), execute-as como jobs de longa duração. Esses jobs são executados de forma assíncrona. Você pode verificar o status do job e recuperar os resultados mais tarde.

Observação: só é possível executar jobs de longa duração em recursos reasoningEngine criados a partir de 22 de abril de 2026. Não é possível atualizar recursos reasoningEngine criados anteriormente para executar jobs de longa duração.

Implantar um agente para consulta assíncrona

Para implantar um agente, siga as instruções gerais em Implantar um agente. Para implantação baseada em origem, defina o campo deploymentSpec.agentFramework como google-adk.

Se você usar um endpoint de API personalizado criando sua própria imagem de contêiner, adicione as seguintes variáveis de ambiente ao criar o agente usando o SDK:

"env_vars" = {
    "API_ENDPOINT_PREFIX": "/api/myendpoint"
}

Iniciar um job de consulta de longa duração

Como pré-requisito, conceda ao agente de serviço [e-mail] o papel roles/storage.objectCreator no bucket de armazenamento para arquivos de saída.

Para iniciar um job de consulta de longa duração:

SDK do Agent Platform

import vertexai

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

response = client.agent_engines.run_query_job(
    name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    config={
        "query": '{"input":{"user_id":"USER_ID", "message":"What is the exchange rate from US dollars to SEK today?"}}',
        "output_gcs_uri": "gs://GCS_BUCKET_NAME/OUTPUT_FILE",
    },
)
print(response)

Com o SDK, output_gcs_uri pode ser um diretório ou um nome de arquivo. Se for um nome de arquivo, o sistema vai usar esse arquivo para armazenar a resposta. Se for um diretório, o sistema vai gerar automaticamente um arquivo para a resposta. Nos dois casos, a consulta de entrada é armazenada no mesmo diretório com o mesmo prefixo de nome de arquivo do arquivo de saída.

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

Para a chamada de API REST, o campo input_gcs_uri precisa apontar para um arquivo que contenha a consulta. O conteúdo do arquivo precisa ser um objeto JSON com um campo input que corresponda ao campo input de QueryReasoningEngineRequest (como { "input": { "user_id": "hello", "message":"$QUERY"} }). Se esse arquivo de entrada estiver em um bucket diferente do local de saída, também será necessário conceder ao agente de serviço [e-mail] a função roles/storage.objectReader no bucket de armazenamento em que os arquivos de entrada estão localizados.

O output_gcs_uri precisa ser um nome de arquivo.

Verificar o status de um job de consulta de longa duração

Para verificar o status e recuperar os resultados de um job de consulta de longa duração:

SDK do Agent Platform

response = client.agent_engines.check_query_job(
    name="JOB_NAME",
    config={
        "retrieve_result": True,
    },
)
print(response)

Cancelar um job de consulta de longa duração

Para cancelar um job de consulta de longa duração, você precisa ter o nome do recurso LRO retornado pelo job.

SDK do Agent Platform

response = client.agent_engines.cancel_query_job(
    name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    operation_name="projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID",
)

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:cancelAsyncQuery -d \
'{
  "name": "projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
  "operation_name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID"
}'

Gerenciar recordações

O AdkApp usa o Memory Bank se você incluir um PreloadMemoryTool na definição do agente e implantar o agente na Agent Platform. Nesta seção, descrevemos como usar o recurso de gerar e recuperar memórias do agente usando a implementação padrão do serviço de memória do ADK.

Adicionar sessão à memória

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 do Agent Platform

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 do Agent Platform

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.