Usa un agente del kit de desarrollo de agentes

Además de las instrucciones generales para usar un agente, en esta página se describen las funciones específicas de AdkApp.

Antes de comenzar

En este instructivo, se supone que leíste y seguiste las instrucciones de los siguientes documentos:

Para consultar una aplicación del ADK, primero debes crear una nueva instancia de la aplicación del ADK o obtener una instancia existente.

Para obtener la aplicación del ADK correspondiente a un ID de recurso específico, sigue estos pasos:

SDK de Vertex AI para Python

Ejecuta el siguiente código:

from vertexai import agent_engines

adk_app = agent_engines.get(RESOURCE_ID)

Como alternativa, puedes proporcionar el nombre completo del recurso del agente:

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

Biblioteca de solicitudes de Python

Ejecuta el siguiente 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 de 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

Operaciones admitidas

Las siguientes operaciones son compatibles con AdkApp:

Para enumerar todas las operaciones admitidas, sigue estos pasos:

SDK de Vertex AI para Python

Ejecuta el siguiente código:

adk_app.operation_schemas()

Biblioteca de solicitudes de Python

Ejecuta el siguiente código:

import json

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

API de REST

Se representa en spec.class_methods de la respuesta a la solicitud de curl.

Administra sesiones

AdkApp usa sesiones administradas basadas en la nube después de que implementes el agente en Vertex AI Agent Engine. En esta sección, se describe cómo usar las sesiones administradas.

Crea una sesión

Para crear una sesión para un usuario, haz lo siguiente:

SDK de Vertex AI para Python 

session = adk_app.create_session(user_id="USER_ID")

Biblioteca de solicitudes de Python

Ejecuta el siguiente 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": "create_session",
    "input": {"user_id": "USER_ID"},
  }),
)
print(response.content)

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

en el que USER_ID es un ID definido por el usuario con un límite de 128 caracteres.

Enumera sesiones

Para mostrar la lista de sesiones de un usuario, haz lo siguiente:

SDK de Vertex AI para Python 

adk_app.list_sessions(user_id="USER_ID")

solicitudes

Ejecuta el siguiente 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": "list_sessions",
    "input": {"user_id": "USER_ID"},
  }),
)
print(response.content)

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

en el que USER_ID es un ID definido por el usuario con un límite de 128 caracteres.

Cómo obtener una sesión

Para obtener una sesión específica, necesitas el ID de usuario y el ID de sesión:

SDK de Vertex AI para Python 

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

solicitudes

Ejecuta el siguiente 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": "get_session",
    "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},
  }),
)
print(response.content)

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

Cómo borrar una sesión

Para borrar una sesión, necesitas el ID de usuario y el ID de sesión:

SDK de Vertex AI para Python 

adk_app.delete_session(user_id="USER_ID", session_id="SESSION_ID")

solicitudes

Ejecuta el siguiente 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": "delete_session",
    "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},
  }),
)
print(response.content)

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

Cómo transmitir una respuesta a una consulta

Para transmitir respuestas de un agente en una sesión, haz lo siguiente:

SDK de Vertex AI para Python 

for event in adk_app.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)

solicitudes 

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

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

¿Qué sigue?