Administra sesiones con el kit de desarrollo de agentes

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

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 crear una instancia de Vertex AI Agent Engine. No es necesario que implementes ningún código para comenzar a usar Sessions. Sin la implementación de código, la creación de una instancia de Vertex AI Agent Engine solo lleva unos segundos.

import vertexai
from vertexai import agent_engines

# Create an agent engine instance
agent_engine = agent_engines.create()

Desarrolla tu agente del ADK

Para crear tu agente del ADK, sigue las instrucciones que se indican 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 coordina la ejecución de tus agentes, herramientas y devoluciones de llamada, y coordina las llamadas para leer y escribir sesiones. Inicializa el ejecutor con VertexAiSessionService, que se conecta con las sesiones de Vertex AI Agent Engine.

from google.adk.sessions import VertexAiSessionService

app_name="AGENT_ENGINE_ID"
user_id="USER_ID"

# Create the ADK runner with VertexAiSessionService
session_service = VertexAiSessionService(
       "PROJECT_ID", "LOCATION")
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:

• PROJECT_ID: ID del proyecto

• LOCATION: Tu región.

• 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.name.split("/")[-1].

• USER_ID: Es un identificador único no vacío para el usuario, con una longitud máxima de 128 caracteres.

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.

agent_engine_id="AGENT_ENGINE_ID"

adk web --session_db_url=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)

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

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

# ListSessionsResponse(session_ids=['1122334455', '9988776655'])

# 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

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

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:

agent_engines.update(resource_name=agent_engine.name, agent_engine=AGENT, requirements=REQUIREMENTS)

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.

Limpia

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.