Usar un agente de Agent Development Kit

Antes de empezar

En este tutorial se da por hecho que has leído y seguido las instrucciones de los siguientes artículos:

• Desarrollar un agente de Agent Development Kit: para desarrollar agent como instancia de AdkApp.
• Autenticación de usuario: para autenticarte como usuario y consultar al agente.
• Importa e inicializa el SDK para inicializar el cliente y obtener una instancia implementada (si es necesario).

Obtener una instancia de un agente

Para consultar un AdkApp, primero debes crear una instancia o obtener una instancia que ya tengas.

Para obtener el AdkApp correspondiente a un ID de recurso específico, sigue estos pasos:

SDK de Vertex AI para Python

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

donde

• PROJECT_ID es el Google Cloud ID de proyecto en el que desarrollas y despliegas agentes.
• LOCATION es una de las regiones admitidas.
• RESOURCE_ID es el ID del agente implementado como recurso reasoningEngine.

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

Cuando se usa el SDK de Vertex AI para Python, el objeto adk_app corresponde a una clase AgentEngine que contiene lo siguiente:

• Un adk_app.api_resource con información sobre el agente implementado. También puedes llamar a adk_app.operation_schemas() para devolver la lista de operaciones que admite adk_app. Consulta Operaciones admitidas para obtener más información.
• un adk_app.api_client que permite interacciones de servicio síncronas
• un adk_app.async_api_client que permite interacciones de servicio asíncronas

En el resto de esta sección se presupone que tienes una instancia de AgentEngine llamada adk_app.

Operaciones admitidas

Se admiten las siguientes operaciones para AdkApp:

• async_stream_query: para transmitir una respuesta a una consulta.

• async_create_session: para crear una sesión.

• async_list_sessions: para mostrar las sesiones disponibles.

• async_get_session: para recuperar una sesión específica.

• async_delete_session: para eliminar una sesión específica.

• async_add_session_to_memory: para generar recuerdos de una sesión.

• async_search_memory: para recuperar recuerdos.

Para enumerar todas las operaciones admitidas, haz lo siguiente:

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 REST

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

Gestionar sesiones

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

Nota: En estas instrucciones se da por hecho que no has personalizado tu base de datos con session_service_builder al desarrollar tu agente.

Crear una sesión

Para crear una sesión de un usuario, sigue estos pasos:

SDK de Vertex AI para Python

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

print(session)

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": "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: elige tu propio ID de usuario (128 caracteres como máximo). Por ejemplo, user-123.

La sesión se crea como la representación de diccionario de un objeto de sesión del ADK.

Mostrar sesiones

Para enumerar las sesiones de un usuario, sigue estos pasos:

SDK de Vertex AI para Python

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

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

donde USER_ID es el ID de usuario que has definido. Por ejemplo, user-123.

Si se devuelve alguna sesión, se usa la forma de diccionario de un objeto de sesión de ADK.

Obtener una sesión

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

SDK de Vertex AI para Python

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

print(session)

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

session es la representación de diccionario de un objeto de sesión de ADK.

Eliminar una sesión

Para eliminar una sesión, necesita el ID de usuario y el ID de sesión:

SDK de Vertex AI para Python

await adk_app.async_delete_session(user_id="USER_ID", session_id="SESSION_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": "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 una respuesta a una consulta

Para transmitir respuestas de un agente en una sesión:

SDK de 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 solicitudes de 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: El argumento session_id= es opcional. Si no se especifica, se crea automáticamente una nueva sesión y se usa para servir esa consulta.

Si usas el SDK de Vertex AI para Python, deberías recibir una continuación de la conversación como la siguiente secuencia de diccionarios:

{'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',
 # ...
}

Gestionar recuerdos

AdkApp usa Vertex AI Agent Engine Memory Bank si incluyes un PreloadMemoryTool en la definición del agente y despliegas el agente en Vertex AI Agent Engine. En esta sección se describe cómo generar y recuperar recuerdos del agente mediante la implementación predeterminada del servicio de memoria del ADK.

Añadir sesión a la memoria

Para conservar la memoria de la información significativa de una sesión (que se puede usar en sesiones futuras), usa el método async_add_session_to_memory:

SDK de Vertex AI para Python

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

donde SESSION_DICT es la forma de diccionario de un objeto de sesión del ADK.

Buscar recuerdos

Para buscar en los recuerdos del agente, puedes usar el método async_search_memory:

SDK de Vertex AI para Python

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

donde

• USER_ID es el ámbito de los recuerdos relevantes.
• QUERY es la consulta para la que se va a realizar la búsqueda de similitud.

Siguientes pasos

• Usar un agente.
• Evalúa un agente.
• Gestionar agentes implementados.
• Obtener asistencia