Esta página se ha traducido con Cloud Translation API.

Gestionar sesiones con Agent Development Kit

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

Antes de empezar

Para configurar tu entorno, sigue los pasos de las secciones Obtener los roles necesarios y Autenticación del artículo Configurar el entorno.

Crear 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 implemente ningún código para empezar a usar Sesiones. Si ya has usado Agent Engine, crear una instancia de Vertex AI Agent Engine solo te llevará unos segundos y no tendrás que desplegar código. Puede que tarde 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])

Haz los cambios siguientes:

PROJECT_ID: tu ID de proyecto.
LOCATION: tu región. Consulta las regiones admitidas para Sesiones.

Modo Express de Vertex AI

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

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

Desarrollar tu agente de ADK

Para crear tu agente de ADK, sigue las instrucciones de 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]
)

Configurar el ejecutor de ADK

El tiempo de ejecución del ADK coordina la ejecución de tus agentes, herramientas y retrollamadas, así como las llamadas para leer y escribir sesiones. Inicializa 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)

Haz los cambios siguientes:

APP_NAME: el nombre de tu aplicación de agente.
USER_ID: elige tu propio ID de usuario (128 caracteres como máximo). Por ejemplo, user-123.
AGENT_ENGINE_ID: el ID de recurso de una instancia de Vertex AI Agent Engine.
En el caso de los agentes implementados, el ID de recurso se indica como la variable de entorno GOOGLE_CLOUD_AGENT_ENGINE_ID.
En el caso de los agentes locales, puede obtener el ID de recurso mediante 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)

Haz los cambios siguientes:

API_KEY: tu clave de API del modo Express de Vertex AI.
AGENT_ENGINE_ID: el ID de recurso de una instancia de Vertex AI Agent Engine.
En el caso de los agentes implementados, el ID de recurso se indica como la variable de entorno GOOGLE_CLOUD_AGENT_ENGINE_ID.
En el caso de los agentes locales, puede obtener el ID de recurso mediante agent_engine.api_resource.name.split("/")[-1].

Interactuar con tu agente

Después de definir tu agente y configurar las sesiones de Vertex AI Agent Engine, puedes interactuar con él para comprobar que el historial y los estados de la sesión se conservan.

Interfaz de ADK

Prueba tu agente con la interfaz de usuario del ADK y conéctate a la sesión de Vertex AI Agent Engine mediante 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)

Interfaz de usuario de ADK

Python

Usa el código Python del ADK para gestionar sesiones y estados.

Crear una sesión y consultar 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"

Una vez que se crea la sesión y se transfiere al runner, 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.

Listar las sesiones

Lista todas las sesiones 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'])

Gestionar estados de sesión

Los estados contienen la información que necesita el agente para mantener una conversación. Puedes proporcionar un estado inicial como un diccionario al crear 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, añade un nuevo evento a la sesión mediante 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

Eliminar una sesión

Eliminar 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)

Implementar un agente en Vertex AI Agent Engine

Una vez que hayas probado tu agente de forma local, puedes desplegarlo 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.
    },
)

Haz los cambios siguientes:

AGENT: 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 el despliegue.
DISPLAY_NAME: nombre descriptivo del agente.
REQUIREMENTS: lista de paquetes pip que requiere tu agente. Por ejemplo, ["google-cloud-storage", "google-cloud-aiplatform[agent_engines,adk]"].
STAGING_BUCKET: un segmento de Cloud Storage con el prefijo gs://.

Modo Express de Vertex AI

Despliega tu agente a través de archivos de origen siguiendo estos pasos.

Limpieza

Para eliminar todos los recursos usados en este proyecto, puedes eliminar la instancia de Vertex AI Agent Engine junto con sus recursos secundarios:

agent_engine.delete(force=True)

Siguientes pasos

Gestionar sesiones mediante llamadas a la API.

Gestionar sesiones con Agent Development Kit Organízate con las colecciones Guarda y clasifica el contenido según tus preferencias.

Antes de empezar

Crear una instancia de Vertex AI Agent Engine

Proyecto de Google Cloud

Modo Express de Vertex AI

Desarrollar tu agente de ADK

Configurar el ejecutor de ADK

Proyecto de Google Cloud

Modo Express de Vertex AI

Interactuar con tu agente

Interfaz de ADK

Python

Crear una sesión y consultar al agente

Listar las sesiones

Gestionar estados de sesión

Eliminar una sesión

Implementar un agente en Vertex AI Agent Engine

Proyecto de Google Cloud

Modo Express de Vertex AI

Limpieza

Siguientes pasos

Gestionar sesiones con Agent Development Kit