Usar um agente do Kit de desenvolvimento de agentes

Além das instruções gerais para usar um agente, esta página descreve recursos específicos do AdkApp.

Antes de começar

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

• Desenvolver um agente do Agent Development Kit: 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).

Para consultar um aplicativo do ADK, primeiro é necessário criar uma nova instância de aplicativo do ADK ou receber uma instância atual.

Para receber o aplicativo do ADK correspondente a um ID de recurso específico:

SDK da Vertex AI para Python

Execute o seguinte código:

adk_app = client.agent_engines.get(name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_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

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

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.

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 Vertex AI Agent Engine. 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")

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

em que USER_ID é um ID definido pelo usuário com um limite de 128 caracteres.

Listar sessões

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

SDK da Vertex AI para Python

await adk_app.async_list_sessions(user_id="USER_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_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 é um ID definido pelo usuário com um limite de 128 caracteres.

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

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

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.

A seguir

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