Use um agente do Agent Development Kit

Antes de começar

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

• Desenvolver um agente do Agent Development Kit: para desenvolver o agent como uma instância do AdkApp.
• Autenticação do utilizador para autenticar como utilizador para consultar o agente.
• Importe e inicialize o SDK para inicializar o cliente para obter uma instância implementada (se necessário).

Obtenha uma instância de um agente

Para consultar um AdkApp, primeiro tem de criar uma nova instância ou obter uma instância existente.

Para obter o AdkApp 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",
)

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

print(adk_app)

onde

• PROJECT_ID é o Google Cloud ID do projeto no qual desenvolve e implementa agentes,
• LOCATION é uma das regiões suportadas e
• RESOURCE_ID é o ID do agente implementado como um recurso reasoningEngine.

Biblioteca de pedidos 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

Modo expresso do Vertex AI

Pode usar o SDK Vertex AI para Python com a chave da API do modo expresso:

import vertexai

client = vertexai.Client(  # For service interactions via client.agent_engines
    api_key="API_KEY",
)

adk_app = client.agent_engines.get(name="AGENT_ENGINE_NAME")

print(adk_app)

onde

• AGENT_ENGINE_NAME é o nome totalmente qualificado do recurso Agent Engine.

Nota: as seguintes operações suportadas são todas suportadas para utilizadores do modo expresso do Vertex AI através do SDK Vertex AI para Python

Quando usa o SDK 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 implementado. Também pode chamar adk_app.operation_schemas() para devolver a lista de operações que o adk_app suporta. Consulte o artigo Operações suportadas para ver 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 de serviço assíncronas

O resto desta secção pressupõe que tem uma instância do AgentEngine, denominada adk_app.

Operações compatíveis

As seguintes operações são suportadas para AdkApp:

• async_stream_query: para fazer streaming de uma resposta a uma consulta.

• async_create_session: para criar uma nova sessão.

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

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

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

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

• async_search_memory: para aceder a memórias.

Para apresentar uma lista de todas as operações suportadas:

SDK Vertex AI para Python

Execute o seguinte código:

adk_app.operation_schemas()

Biblioteca de pedidos Python

Execute o seguinte código:

import json

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

API REST

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

Faça a gestão de sessões

AdkApp usa sessões geridas baseadas na nuvem depois de implementar o agente no Vertex AI Agent Engine. Esta secção descreve como usar sessões geridas.

Nota: estas instruções partem do princípio de que não personalizou a sua base de dados com session_service_builder durante o desenvolvimento do seu agente.

Crie uma sessão

Para criar uma sessão para um utilizador:

SDK Vertex AI para Python

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

print(session)

Biblioteca de pedidos 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 o seu próprio ID do utilizador com um limite de 128 carateres. Por exemplo, user-123.

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

Apresentar sessões

Para listar as sessões de um utilizador:

SDK Vertex AI para Python

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

Biblioteca de pedidos 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 utilizador que definiu. Por exemplo, user-123.

Se forem devolvidas sessões, estas usam o formato de dicionário de um objeto de sessão ADK.

Receba uma sessão

Para obter uma sessão específica, precisa do ID do utilizador e do ID da sessão:

SDK Vertex AI para Python

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

print(session)

Biblioteca de pedidos 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 ADK.

Elimine uma sessão

Para eliminar uma sessão, precisa do ID do utilizador e do ID da sessão:

SDK Vertex AI para Python

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

Biblioteca de pedidos 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 numa sessão:

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

Nota: o argumento session_id= é opcional. Se não for especificado, é criada automaticamente uma nova sessão e usada para publicar essa consulta.

Se estiver a usar o SDK Vertex AI para Python, deve 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',
 # ...
}

Faça a gestão das memórias

AdkApp usa o Memory Bank do Vertex AI Agent Engine se incluir um PreloadMemoryTool na definição do agente e implementar o agente no Vertex AI Agent Engine. Esta secção descreve como usar a geração e a obtenção de memórias do agente através da implementação predefinida do serviço de memória do ADK.

Adicione a sessão à memória

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

SDK Vertex AI para Python

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

onde SESSION_DICT é a forma de dicionário de um objeto de sessão ADK.

Pesquise memórias

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

SDK Vertex AI para Python

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

onde

• USER_ID é o âmbito das memórias relevantes.
• QUERY é a consulta para a qual vai ser feita a pesquisa de semelhanças.

O que se segue?

• Use um agente.
• Avalie um agente.
• Faça a gestão dos agentes implementados.
• Receba apoio técnico.