Use um agente do Agent Development Kit

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

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 agent como uma instância de 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).

Para consultar uma aplicação ADK, primeiro tem de criar uma nova instância da aplicação ADK ou obter uma instância existente.

Para obter a aplicação ADK correspondente a um ID do recurso específico:

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

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.

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

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

onde USER_ID é um ID definido pelo utilizador com um limite de 128 carateres.

Apresentar sessões

Para listar as sessões de um utilizador:

SDK Vertex AI para Python

await adk_app.async_list_sessions(user_id="USER_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_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"},}'

onde USER_ID é um ID definido pelo utilizador com um limite de 128 carateres.

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

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

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.

O que se segue?

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