Administra sesiones con el kit de desarrollo de agentes

Versión preliminar

Esta función está sujeta a las “Condiciones de la Oferta de la Fase Previa a la DG” en la sección Condiciones del Servicio Generales de las Condiciones Específicas del Servicio. Las funciones de la fase previa a la DG están disponibles sin modificaciones y pueden tener asistencia limitada. Para obtener más información, consulta las descripciones de la etapa de lanzamiento.

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

Nota: Si ya seguiste las instrucciones en Cómo desarrollar un agente de Agent Development Kit, no es necesario que sigas esta guía, ya que la plantilla AdkApp ya está conectada a las sesiones de Vertex AI Agent Engine a través de session_service.

Antes de comenzar

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

Crea una instancia de Vertex AI Agent Engine

Para acceder a las sesiones de Vertex AI Agent Engine, 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.

Modo Express de Vertex AI

Puedes usar el SDK de Vertex AI para Python con tu clave de API del modo Express de la siguiente manera:

import vertexai

client = vertexai.Client(  # For service interactions via client.agent_engines
    api_key="API_KEY",
)

# 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])

Desarrolla tu agente del ADK

Nota: Asegúrate de haber instalado la versión 1.0.0 del ADK o una posterior. Esta versión se incluye en google-cloud-aiplatform[adk,agent_engine].

Para crear tu agente del ADK, sigue las instrucciones en Agent Development Kit o usa el siguiente código para crear un agente que salude a un usuario con saludos fijos:

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 entorno 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 Runner con VertexAiSessionService, que se conecta con las sesiones de Vertex AI Agent Engine.

Proyecto de Google Cloud

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_ID",
      "LOCATION",
      "AGENT_ENGINE_ID"
)
runner = adk.Runner(
    agent=root_agent,
    app_name=app_name,
    session_service=session_service)

# Helper method to send query to the runner
def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  events = runner.run(
      user_id=user_id, session_id=session_id, new_message=content)

  for event in events:
      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 indica 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].

Modo Express de Vertex AI

import os
from google.adk.sessions import VertexAiSessionService
from google.genai import types

# Set environment variables
os.environ["GOOGLE_API_KEY"] = "API_KEY"
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"

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

# Helper method to send query to the runner
def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  events = runner.run(
      user_id=user_id, session_id=session_id, new_message=content)

  for event in events:
      if event.is_final_response():
          final_response = event.content.parts[0].text
          print("Agent Response: ", final_response)

Reemplaza lo siguiente:

• API_KEY: Es tu clave de API del modo Express de Vertex AI.

• 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 indica 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 Vertex AI Agent Engine, puedes interactuar con tu agente para verificar que el historial y los estados de la sesión persistan.

IU del ADK

Prueba tu agente con la interfaz de usuario del ADK y conéctate a la sesión de Vertex AI Agent Engine con la opción de línea de comandos session_service_uri:

agent_engine_id="AGENT_ENGINE_ID"

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)

Python

Usa código de Python del ADK para administrar sesiones y estados.

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:

# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_id)

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

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

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

Enumera las sesiones existentes

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

# List sessions
await session_service.list_sessions(app_name=app_name,user_id=user_id)

# 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:

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

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 Vertex AI Agent Engine

Después de probar tu agente de forma local, puedes implementarlo en producción actualizando la instancia de Vertex AI Agent Engine con los siguientes 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). Para obtener más información, consulta Consideraciones sobre la implementación.

• 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://.

Modo Express de Vertex AI

Para implementar tu agente a través de archivos fuente, sigue estos pasos de implementación.

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?

• Administra sesiones con llamadas a la API.