Usa un agente del ADK

Antes de comenzar

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

• Crea un agente del Kit de desarrollo de agentes (ADK): para crear agent como una 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).

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 AdkApp que corresponde a un ID de recurso específico, haz lo siguiente:

SDK de Agent Platform

Ejecuta el siguiente código:

import agentplatform

client = agentplatform.Client(  # For service interactions via client.runtimes
    project="PROJECT_ID",
    location="LOCATION",
)

adk_app = client.runtimes.get(name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

print(adk_app)

donde

• PROJECT_ID es el ID del proyecto de Google Cloud en el que creas y implementas 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 usas el SDK de Agent Platform, el objeto adk_app corresponde a una clase AgentRuntime 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 que se devuelva la lista de operaciones que admite adk_app. Consulta Operaciones compatibles para obtener más detalles.
• adk_app.api_client que permite interacciones de servicio síncronas
• adk_app.async_api_client que permite interacciones asíncronas entre servicios

En el resto de esta sección, se supone que tienes una instancia de AgentRuntime, 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 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 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 desde la respuesta a la solicitud 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 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, usa el método AdkApp.async_create_session:

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 método AdkApp.async_list_sessions:

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 devuelve 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, usa el método AdkApp.async_get_session:

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.

Cómo borrar una sesión

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

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

Nota: La duración máxima de una solicitud de streamQuery es de 10 minutos. Para los flujos de trabajo que requieren tiempos de ejecución más largos, usa trabajos de consulta de larga duración en su lugar.

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

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 automáticamente una sesión nueva y se usa para entregar esa consulta. Si el ID de sesión proporcionado no existe, se devuelve una respuesta vacía. Te recomendamos que omitas este paso si solo quieres verificar si se puede establecer comunicación con 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

En el caso de las consultas que pueden tardar mucho tiempo 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 de reasoningEngine que se hayan creado el 22 de abril de 2026 o después. No puedes actualizar los recursos de reasoningEngine creados con anterioridad para ejecutar trabajos de larga duración.

Implementa un agente para consultas asíncronas

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

Si usas un extremo de API personalizado compilando 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 en el 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 agentplatform

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

response = client.runtimes.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 en el bucket de almacenamiento en el que se encuentran los archivos de entrada.

El 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.runtimes.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 devuelve del trabajo de consulta de larga duración.

SDK de Agent Platform

response = client.runtimes.cancel_query_job(
    name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    config={
        "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"
}'

La cancelación es asíncrona. La solicitud de cancelación se devuelve en cuanto se acepta, pero el trabajo puede seguir informando un estado de RUNNING desde check_query_job hasta que finalice el trabajo que ya está en curso, como una llamada de herramienta de bloqueo.

Una vez que se cancela el trabajo, su operación se completa con un código de error de 1 (CANCELLED) y el mensaje Cancelled by user.. Ten en cuenta que check_query_job informa cada operación que se completó con un error como un estado de FAILED, por lo que un trabajo cancelado se informa como FAILED en lugar de un estado cancelado distinto. Inspecciona el código de error para distinguir una cancelación de una falla genuina.

Administra los recuerdos

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, generar y recuperar recuerdos del agente a través de la implementación predeterminada del servicio de memoria del ADK.

Agrega Session a Memory

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

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

donde SESSION_DICT es la forma de diccionario de un objeto Session del ADK.

Cómo buscar recuerdos

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

SDK de Agent Platform

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

donde

• USER_ID es el permiso para 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 los agentes implementados.
• Obtén asistencia.