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 en los siguientes documentos:

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

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 AgentRuntime que contiene lo siguiente:

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:

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 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 Agent Platform. En esta sección, se describe cómo usar las sesiones administradas.

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

Cómo transmitir una respuesta a una búsqueda

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

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

Implementa un agente para la búsqueda asíncrona

Para implementar un agente, sigue las instrucciones generales 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 creando 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 service-PROJECT_NUMBER@gcp-sa-aiplatform-re.s3ns-system.iam.gserviceaccount.com 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 búsqueda 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 búsqueda. 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 otorgarle al agente de servicio service-PROJECT_NUMBER@gcp-sa-aiplatform-re.s3ns-system.iam.gserviceaccount.com el rol de 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 muestra 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 a una 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 a través de un estado cancelado distinto. Inspecciona el código de error para distinguir una cancelación de una falla genuina.

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

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

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 los recuerdos 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 alcance de los recuerdos relevantes.
  • QUERY es la búsqueda para la que se realizará la búsqueda por similitud.

¿Qué sigue?