Cómo usar un agente del Kit de desarrollo de agentes

Antes de comenzar

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

• Desarrolla un agente del Kit de desarrollo de agentes: Para desarrollar agent como una instancia de AdkApp
• Autenticación del usuario para autenticarse como usuario y consultar al agente
• Importa e inicializa el SDK para inicializar el cliente y obtener una instancia implementada (si es necesario).

Obtén una instancia de un agente

Para consultar un AdkApp, primero debes crear una instancia nueva o obtener una instancia existente.

Para obtener el objeto AdkApp correspondiente a un ID de recurso específico, haz lo siguiente:

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 del proyecto en el que desarrollas y despliegas agentes.
• LOCATION es una de las regiones admitidas.
• RESOURCE_ID es el ID del agente implementado como un 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 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

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 detalles.
• 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 supone 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 búsqueda.

• async_create_session: Para crear una sesión nueva

• async_list_sessions: Para enumerar las sesiones disponibles

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

• async_delete_session: Para borrar 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 de REST

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

Administra sesiones

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

Nota: En estas instrucciones, se supone que no personalizaste tu base de datos con session_service_builder cuando desarrollaste tu agente.

Crea una sesión

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

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

• USER_ID: Elige tu propio ID de usuario con un límite de 128 caracteres. Por ejemplo, user-123

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

Enumera sesiones

Para enumerar las sesiones de un usuario, haz lo siguiente:

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

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

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

Obtén 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 = 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 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": "async_get_session", "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},}'

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

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

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

Cómo transmitir una respuesta a una búsqueda

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

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 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: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 sesión nueva y se usa para entregar esa búsqueda.

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

Administra los recuerdos

AdkApp usa el banco de memoria de Vertex AI Agent Engine si incluyes un PreloadMemoryTool en la definición del agente y lo implementas en Vertex AI Agent Engine. En esta sección, se describe cómo usar la implementación predeterminada del servicio de memoria del ADK para generar y recuperar recuerdos del agente.

Agregar sesión a la memoria

Para conservar la memoria de la información significativa en 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")

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

Cómo buscar recuerdos

Para buscar en las memorias 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 alcance de los recuerdos pertinentes.
• QUERY es la búsqueda para la que se realizará la búsqueda de similitud.

¿Qué sigue?

• Usa un agente.
• Evalúa un agente.
• Administra los agentes implementados.
• Obtén asistencia.