Usa 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 a continuación:

• Crea un agente del Kit de desarrollo de agentes: Para crear agent como una instancia de AdkApp.
• Autenticación de usuarios para autenticarte como usuario para consultar el agente
• Importa e inicializa el SDK para inicializar el cliente para 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 existente.

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

SDK de Agent Platform

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 bajo el cual crea e implementa agentes.
• LOCATION es una de las regiones admitidas.
• RESOURCE_ID es el ID del agente implementado como un reasoningEngine recurso.

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 Agent Platform, el objeto adk_app corresponde a una clase AgentEngine que contiene lo siguiente:

• adk_app.api_resource con información sobre el agente implementado. También puedes llamar a adk_app.operation_schemas() para mostrar la lista de operaciones que admite adk_app. Consulta Operaciones admitidas para obtener más detalles.
• adk_app.api_client que permite interacciones de servicio síncronas
• 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, denominada 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 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 memorias de una sesión

• async_search_memory: Para recuperar memorias

Para enumerar todas las operaciones admitidas, haz lo siguiente:

SDK de Agent Platform

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 implementar el agente en Agent Platform. En esta sección, se describe cómo usar 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, usa el AdkApp.async_create_session método:

SDK de Agent Platform

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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, usa el AdkApp.async_list_sessions método:

SDK de Agent Platform

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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 el formulario de diccionario de un objeto de sesión del ADK.

Obtén una sesión

Para obtener una sesión específica, usa el AdkApp.async_get_session método:

SDK de Agent Platform

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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 session es la representación de diccionario de un objeto de sesión del ADK.

Borra una sesión

Para borrar una sesión, usa el AdkApp.async_delete_session método:

SDK de Agent Platform

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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"},}'

Transmite una respuesta a una consulta

Para transmitir respuestas de un agente en una sesión, usa el AdkApp.async_stream_query método:

SDK de Agent Platform

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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 y se usa automáticamente una sesión nueva para entregar esa consulta. Si no existe el ID de sesión proporcionado, se muestra una respuesta vacía. Te recomendamos que no lo incluyas si solo verificas si se puede acceder a un agente.

Si usas el SDK de Agent Platform, 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',
 # ...
}

Trabajos de consulta de larga duración

Para las consultas que pueden tardar mucho en completarse (hasta siete días), puedes ejecutarlas como trabajos de larga duración. Estos trabajos se ejecutan de forma asíncrona. Puedes verificar el estado del trabajo y recuperar los resultados más tarde.

Nota: Solo puedes ejecutar trabajos de larga duración en recursos reasoningEngine que se crearon el 22 de abril de 2026 o después. No puedes actualizar los recursos reasoningEngine creados anteriormente para ejecutar trabajos de larga duración.

Implementa un agente para la consulta asíncrona

Para implementar un agente, sigue las instrucciones generales que se indican en Implementa un agente. Para la implementación basada en la fuente, establece el campo deploymentSpec.agentFramework en google-adk.

Si usas un extremo de API personalizado mediante la compilación de tu propia imagen de contenedor, debes agregar las siguientes variables de entorno cuando crees el agente con el SDK:

"env_vars" = {
    "API_ENDPOINT_PREFIX": "/api/myendpoint"
}

Inicia un trabajo de consulta de larga duración

Como requisito previo, debes otorgar al agente de servicio [e-mail] el rol roles/storage.objectCreator al bucket de almacenamiento para los archivos de salida.

Para iniciar un trabajo de consulta de larga duración, haz lo siguiente:

SDK de Agent Platform

import vertexai

client = vertexai.Client(
    project="PROJECT_ID",
    location="LOCATION",
)

response = client.agent_engines.run_query_job(
    name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    config={
        "query": '{"input":{"user_id":"USER_ID", "message":"What is the exchange rate from US dollars to SEK today?"}}',
        "output_gcs_uri": "gs://GCS_BUCKET_NAME/OUTPUT_FILE",
    },
)
print(response)

Con el SDK, output_gcs_uri puede ser un directorio o un nombre de archivo. Si es un nombre de archivo, el sistema usa este archivo para almacenar la respuesta. Si es un directorio, el sistema genera automáticamente un archivo para la respuesta. En ambos casos, la consulta de entrada se almacena en el mismo directorio con el mismo prefijo de nombre de archivo que el archivo de salida.

REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1beta1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:asyncQuery -d \
'{
  "input_gcs_uri": "gs://GCS_BUCKET_NAME/INPUT_FILE",
  "output_gcs_uri": "gs://GCS_BUCKET_NAME/OUTPUT_FILE"
}'

Para la llamada a la API de REST, el campo input_gcs_uri debe apuntar a un archivo que contenga la consulta. El contenido del archivo debe ser un objeto JSON con un campo input que coincida con el campo input de QueryReasoningEngineRequest (como { "input": { "user_id": "hello", "message":"$QUERY"} }). Si este archivo de entrada se encuentra en un bucket diferente de la ubicación de salida, también debes otorgar al agente de servicio [e-mail] el rol roles/storage.objectReader al bucket de almacenamiento en el que se encuentran los archivos de entrada.

output_gcs_uri debe ser un nombre de archivo.

Verifica el estado de un trabajo de consulta de larga duración

Para verificar el estado y recuperar los resultados de un trabajo de consulta de larga duración, haz lo siguiente:

SDK de Agent Platform

response = client.agent_engines.check_query_job(
    name="JOB_NAME",
    config={
        "retrieve_result": True,
    },
)
print(response)

Cancela un trabajo de consulta de larga duración

Para cancelar un trabajo de consulta de larga duración, debes tener el nombre del recurso de LRO que se muestra en el trabajo de consulta de larga duración.

SDK de Agent Platform

response = client.agent_engines.cancel_query_job(
    name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    operation_name="projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID",
)

REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1beta1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:cancelAsyncQuery -d \
'{
  "name": "projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
  "operation_name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID"
}'

Administra memorias

AdkApp usa Memory Bank si incluyes un PreloadMemoryTool en la definición del agente y lo implementas en Agent Platform. En esta sección, se describe cómo usar la implementación predeterminada del servicio de memoria del ADK para generar y recuperar memorias del agente.

Agrega Session a Memory

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

SDK de Agent Platform

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.

Busca memorias

Para buscar en las memorias del agente, puedes usar el async_search_memory método:

SDK de Agent Platform

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

donde

• USER_ID es el alcance de las memorias relevantes.
• QUERY es la consulta para la que se realizará la búsqueda por similitud.

¿Qué sigue?

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