Administra sesiones con el Kit de desarrollo de agentes

En esta página, se describe cómo puedes conectar un agente del Kit de desarrollo de agentes (ADK) con las sesiones de la Plataforma de agentes y usar sesiones administradas en el entorno local y de producción.

Antes de comenzar

En estas instrucciones, se usa la siguiente estructura básica de archivos del proyecto para definir un agente del ADK y su código de implementación y de ejecución de asistencia:

my_agent/
    agent.py      # main agent code
    runner.py     # code for interacting with the agent
    deploy.py     # code for deploying the agent to Google Cloud

Asegúrate de que tu entorno esté configurado siguiendo los pasos de Obtén los roles necesarios y Autenticación en Configura tu entorno.

Configura las variables de entorno

Para usar el ADK, configura tus variables de entorno:

import os

os.environ["GOOGLE_GENAI_USE_ENTERPRISE"] = "TRUE"
os.environ["GOOGLE_CLOUD_PROJECT"] = "PROJECT_ID"
os.environ["GOOGLE_CLOUD_LOCATION"] = "LOCATION"

Reemplaza lo siguiente:

  • PROJECT_ID: ID del proyecto
  • LOCATION: Tu región. Consulta las regiones compatibles con Memory Bank.

Crea una instancia de Vertex AI Agent Engine

Para acceder a las sesiones de Agent Platform, primero debes usar una instancia de Vertex AI Agent Engine. No es necesario que implementes ningún código para comenzar a usar Sessions. Si ya usaste Agent Engine, crear una instancia de Vertex AI Agent Engine solo lleva unos segundos sin necesidad de implementar código. El proceso puede tardar más si es la primera vez que usas Agent Engine.

Proyecto de Google Cloud 

import vertexai

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

# If you don't have an Agent Engine instance already, create an instance.
agent_engine = client.agent_engines.create()

# Print the agent engine ID, you will need it in the later steps to initialize
# the ADK `VertexAiSessionService`.
print(agent_engine.api_resource.name.split("/")[-1])

Reemplaza lo siguiente:

  • PROJECT_ID: ID del proyecto

  • LOCATION: Tu región. Consulta las regiones admitidas para las Sesiones.

Desarrolla tu agente del ADK

Para crear tu agente de ADK, sigue las instrucciones en Kit de desarrollo de agentes o usa el siguiente código para crear un agente que salude a un usuario con saludos fijos. Guarda este código en un archivo llamado agent.py.

# file: my_agent/agent.py
from google import adk

def greetings(query: str):
  """Tool to greet user."""
  if 'hello' in query.lower():
    return {"greeting": "Hello, world"}
  else:
    return {"greeting": "Goodbye, world"}

# Define an ADK agent
root_agent = adk.Agent(
    model="gemini-2.0-flash",
    name='my_agent',
    instruction="You are an Agent that greet users, always use greetings tool to respond.",
    tools=[greetings]
)

Configura el ejecutor del ADK

El tiempo de ejecución del ADK organiza la ejecución de tus agentes, herramientas y devoluciones de llamada, y organiza las llamadas para leer y escribir sesiones. Inicializa el objeto Runner con VertexAiSessionService, que se conecta con las sesiones de Agent Platform. Guarda este código en un archivo llamado runner.py.

Proyecto de Google Cloud 

# file: my_agent/runner.py
import agent # Import from your agent.py
from google.adk import Runner
from google.adk.sessions import VertexAiSessionService
from google.genai import types

app_name="APP_NAME"
user_id="USER_ID"

# Create the ADK runner with VertexAiSessionService
session_service = VertexAiSessionService(
      project="PROJECT_ID",
      location="LOCATION",
      agent_engine_id="AGENT_ENGINE_ID"
)
runner = Runner(
    agent=agent.root_agent,
    app_name=app_name,
    session_service=session_service)

# Helper method to send query to the runner
async def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  async for event in runner.run_async(
      user_id=user_id, session_id=session_id, new_message=content):
      if event.is_final_response():
          final_response = event.content.parts[0].text
          print("Agent Response: ", final_response)

Reemplaza lo siguiente:

  • APP_NAME: Es el nombre de tu aplicación de agente.

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

  • AGENT_ENGINE_ID: Es el ID de recurso de una instancia de Vertex AI Agent Engine.

  • En el caso de los agentes implementados, el ID del recurso se muestra como la variable de entorno GOOGLE_CLOUD_AGENT_ENGINE_ID.

  • En el caso de los agentes locales, puedes recuperar el ID del recurso con agent_engine.api_resource.name.split("/")[-1].

Interactúa con el agente

Después de definir tu agente y configurar las sesiones de Agent Platform, puedes interactuar con él para verificar que el historial y los estados de la sesión persistan.

ADK UI

Para probar tu agente con la interfaz de usuario del ADK y conectarte a las sesiones de Agent Platform, ejecuta el comando adk web en tu terminal.

Si bien adk web suele leer las variables de entorno de un archivo .env, ignora este archivo cuando usas la marca --session_service_uri. En cambio, debes establecer las variables de entorno GOOGLE_CLOUD_PROJECT y GOOGLE_CLOUD_LOCATION en tu sesión de terminal antes de ejecutar el comando.

export GOOGLE_CLOUD_PROJECT="PROJECT_ID"
export GOOGLE_CLOUD_LOCATION="LOCATION"
adk web --session_service_uri=agentengine://AGENT_ENGINE_ID
# Sample output
+-----------------------------------------------------------------------------+
| ADK Web Server started                                                      |
|                                                                             |
| For local testing, access at http://localhost:8000.                         |
+-----------------------------------------------------------------------------+

INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

ADK UI

Python

Usar código de Python del ADK para administrar sesiones y estados Agrega el siguiente código al final de tu archivo runner.py para interactuar con el agente.

Para mayor brevedad, los siguientes fragmentos contienen llamadas de await de nivel superior. Para ejecutar este código como una secuencia de comandos de Python, coloca los fragmentos dentro de una función async y usa asyncio.run() para ejecutarla, como se muestra en este ejemplo:

import asyncio

async def main():
  # Place one or more snippets here.
  # For example:
  session = await session_service.create_session(
         app_name=app_name,
         user_id=user_id)

  await call_agent("Hello!", session.id, user_id)

asyncio.run(main())

Crea una sesión y consulta al agente

Usa el siguiente código para crear una sesión y enviar una consulta a tu agente. Especifica un ID de sesión personalizado para organizar las sesiones con tu propio esquema de nombres o para realizar una transición rápida desde otros servicios. Si no especificas un ID de sesión, Agent Platform generará uno automáticamente.

# file: my_agent/runner.py
# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_id,
       session_id=session_id)

await call_agent("Hello!", session.id, user_id)
# Agent response: "Hello, world"

await call_agent("Thanks!", session.id, user_id)
# Agent response: "Goodbye, world"

Para los IDs de sesión personalizados, ten en cuenta las siguientes restricciones para evitar colisiones con los IDs generados por el sistema:

  • Si el primer carácter es una letra, el ID puede tener hasta 63 caracteres. Los caracteres válidos son letras minúsculas, números y guiones ([a-z0-9-]). El último carácter debe ser una letra o un número.
  • Si el primer carácter es un número, el ID puede tener hasta 9 caracteres. Los caracteres válidos son números ([0-9]) sin ceros iniciales.

Después de que se crea la sesión y se pasa al ejecutor, el ADK usa la sesión para almacenar eventos de la interacción actual. También puedes reanudar una sesión anterior si proporcionas el ID de esa sesión.

Configura el tiempo de actividad (TTL) de la sesión

Todas las sesiones deben tener una fecha y hora de vencimiento, que puedes definir cuando creas o actualizas una sesión. La sesión y sus eventos secundarios se borran automáticamente después de que transcurre la fecha y hora de vencimiento. Puedes establecer la fecha y hora de vencimiento (expire_time) directamente o establecer el tiempo de actividad (ttl) en segundos. Si no se especifica ninguno de los dos, el sistema aplica un TTL predeterminado de 365 días.

Tiempo de actividad

Si estableces el tiempo de actividad (TTL), el servidor calcula la fecha de vencimiento como create_time + ttl para las sesiones recién creadas o update_time + ttl para las sesiones actualizadas.

  session = await session_service.create_session(
        app_name=app_name,
        user_id=user_id,
        # Session will be deleted 10 days after creation time.
        ttl=f"{24 * 60 * 60 * 10}s"
  )
  ```

Hora de vencimiento 

  import datetime

  expire_time = datetime.datetime.now(
        tz=datetime.timezone.utc) + datetime.timedelta(seconds=24 * 60 * 60 * 10)

  session = await session_service.create_session(
        app_name=app_name,
        user_id=user_id,
        # Session will be deleted at the provided time (10 days after current time).
        expire_time=expire_time.isoformat()
  )

Enumera las sesiones existentes

Enumera todas las sesiones existentes asociadas a un ID de usuario determinado.

# List sessions
sessions = await session_service.list_sessions(app_name=app_name,user_id=user_id)
print(sessions)
# ListSessionsResponse(session_ids=['1122334455', '9988776655'])

Administra los estados de sesión

Los estados contienen información que el agente necesita para una conversación. Puedes proporcionar un estado inicial como un diccionario cuando creas una sesión:

# Create a session with state
session = await session_service.create_session(
    app_name=app_name,
    user_id=user_id,
    state={'key': 'value'})

print(session.state['key'])
# value

Para actualizar el estado de la sesión fuera del ejecutor, agrega un evento nuevo a la sesión con state_delta:

# file: my_agent/runner.py
from google.adk.events import Event, EventActions
import time

# Define state changes
state_changes = {'key': 'new_value'}

# Create event with actions
actions_with_update = EventActions(state_delta=state_changes)
system_event = Event(
invocation_id="invocation_id",
author="system", # Or 'agent', 'tool' etc.
actions=actions_with_update,
timestamp=time.time()
)

# Append the event
await session_service.append_event(session, system_event)

# Check updated state
updated_session = await session_service.get_session(
    app_name=app_name,
    user_id=user_id,
    session_id=session.id)
# State is updated to new value
print(updated_session.state['key'])
# new_value

Campos personalizados para eventos

El ADK admite un esquema de eventos flexible para que puedas incluir datos arbitrarios en los eventos de sesión. Esto es útil para la interoperabilidad con otros frameworks de agentes o para almacenar datos de eventos personalizados.

Cuando agregas un evento con campos personalizados, el ADK serializa los datos del evento en el campo raw_event del evento de sesión almacenado.

# Create an event with custom fields
custom_event = Event(
  invocation_id="invocation_id",
  author="user",
  timestamp=time.time(),
  custom_field="custom_value",
  another_field={"nested_key": "nested_value"}
)

# Append the event
await session_service.append_event(session, custom_event)

Cuando recuperas la sesión, los eventos recuperados conservan estos campos personalizados.

Cómo borrar una sesión

Borra una sesión específica asociada a un ID de usuario:

await session_service.delete_session(app_name=app_name, user_id=user_id, session_id=session.id)

Implementa tu agente en Agent Platform

Después de probar tu agente de forma local, puedes implementarlo en producción actualizando la instancia de Agent Platform con parámetros:

Proyecto de Google Cloud

client.agent_engines.update(
    resource_name=agent_engine.api_resource.name,
    agent=AGENT,
    config={
      "display_name": DISPLAY_NAME,      # Optional.
      "requirements": REQUIREMENTS,      # Optional.
      "staging_bucket": STAGING_BUCKET,  # Required.
    },
)

Reemplaza lo siguiente:

  • AGENT: Es la aplicación que implementa el método query / stream_query (por ejemplo, AdkApp para un agente de ADK).

  • DISPLAY_NAME: Es un nombre fácil de usar para tu agente.

  • REQUIREMENTS: Es una lista de paquetes de pip que requiere tu agente. Por ejemplo, ["google-cloud-storage", "google-cloud-aiplatform[agent_engines,adk]"]

  • STAGING_BUCKET: Es un bucket de Cloud Storage con el prefijo gs://.

Realiza una limpieza

Para limpiar todos los recursos que se usan en este proyecto, puedes borrar la instancia de Vertex AI Agent Engine junto con sus recursos secundarios:

agent_engine.delete(force=True)

¿Qué sigue?