Discurso de apertura para desarrolladores de Next ‘26: Mejora de los agentes con memoria

1. Introducción

En este codelab, llevarás tus agentes del ADK al siguiente nivel agregando conocimiento persistente y especializado. Aprenderás a administrar el estado de la conversación con las sesiones de Agent Platform, habilitar el aprendizaje a largo plazo con Memory Bank y, además, integrar datos complejos de reglas de la ciudad con Spark y AlloyDB para RAG (generación mejorada por recuperación).

Actividades

• Configura sesiones de Agent Platform para la persistencia de la conversación.
• Implementa un Memory Bank para permitir que los agentes aprendan de las interacciones anteriores.
• Usa Spark Lightning Engine para transferir y procesar la documentación de las reglas de la ciudad.
• Compila un sistema RAG con AlloyDB y la búsqueda de vectores.
• Implementa el agente mejorado en Agent Platform.

Requisitos

• Un navegador web, como Chrome
• Un proyecto de Google Cloud con la facturación habilitada.
• Conocimientos básicos de Python y SQL

Duración estimada: 60 minutos

Los recursos creados en este codelab deberían costar menos de USD 5.

2. Antes de comenzar

Cómo crear un proyecto de Google Cloud

1. En la consola de Google Cloud, en la página del selector de proyectos, selecciona o crea un proyecto de Google Cloud.
2. Asegúrate de que la facturación esté habilitada para tu proyecto de Cloud. Obtén información para verificar si la facturación está habilitada en un proyecto.

Inicie Cloud Shell

Cloud Shell es un entorno de línea de comandos que se ejecuta en Google Cloud y que viene precargado con las herramientas necesarias.

1. Haz clic en Activar Cloud Shell en la parte superior de la consola de Google Cloud.
2. Una vez que te conectes a Cloud Shell, verifica tu autenticación:

   gcloud auth list
   

3. Confirma que tu proyecto esté configurado:

   gcloud config get project
   

4. Si tu proyecto no está configurado como esperabas, configúralo:

   export PROJECT_ID=<YOUR_PROJECT_ID>
   gcloud config set project $PROJECT_ID
   

Verifica la autenticación:

gcloud auth list

Confirma tu proyecto:

gcloud config get project

Configúralo si es necesario:

export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID

Habilita las APIs

Ejecuta este comando para habilitar todas las APIs necesarias para la administración de sesiones, el procesamiento de Spark y AlloyDB:

gcloud services enable \
  aiplatform.googleapis.com \
  run.googleapis.com \
  alloydb.googleapis.com \
  dataproc.googleapis.com \
  documentai.googleapis.com \
  storage.googleapis.com \
  secretmanager.googleapis.com

Si habilitas estas APIs, te aseguras de tener acceso a Agent Platform, a los clústeres administrados de AlloyDB y a Dataproc Serverless para tus canalizaciones de Spark.

3. Configura tu entorno

Para este codelab, usarás el entorno preconfigurado en el repositorio de la presentación principal.

1. Clona el repositorio y navega a la carpeta del proyecto:

git clone https://github.com/GoogleCloudPlatform/next-26-keynotes
cd next-26-keynotes/devkey/enhancing-agents-with-memory

2. Configura un entorno virtual de Python y, además, instala los paquetes del ADK necesarios:

uv venv
source .venv/bin/activate
uv sync

Configura las variables de entorno

El agente requiere una configuración específica para conectarse a Agent Platform y AlloyDB.

1. Copia el archivo de entorno de muestra:

cp .env.example .env

2. Abre .env y actualiza los siguientes campos:
   • GOOGLE_CLOUD_PROJECT: ID del proyecto
   • GOOGLE_CLOUD_LOCATION: us-central1
   • ALLOYDB_CLUSTER_ID: rules-db

GOOGLE_CLOUD_PROJECT=<YOUR_PROJECT_ID>
GOOGLE_CLOUD_LOCATION=global
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_REGION=us-central1
ALLOYDB_CLUSTER_ID=rules-db

3. Ejecuta la siguiente secuencia de comandos de ayuda para crear una instancia de Agent Engine que se usará para las sesiones de conversación y la memoria a largo plazo. Esto propagará automáticamente el AGENT_ENGINE_ID en tu archivo .env:

uv run utils/setup_agent_engine.py

Una vez que se complete correctamente, deberías ver lo siguiente:

Creating Agent Engine instance...
Successfully created Agent Engine. ID: 1234567890
Updated .env with AGENT_ENGINE_ID=1234567890

4. Crea un agente con la administración de sesiones

En este paso, inicializarás un agente de planificación de maratón que puede mantener el historial de conversaciones en varios turnos. Esto se logra con la clase App del ADK y las sesiones de Agent Platform.

Inicializa el servicio de agente y sesión

Abre planner_agent/agent.py. Verás cómo agregamos una clase del ADK para integrar las sesiones de Agent Platform. Esto nos permite hacer que nuestros agentes tengan estado a lo largo del tiempo y modificar el contexto según sea necesario.

from google.adk.agents import LlmAgent
from google.adk.sessions import VertexAiSessionService
from vertexai.agent_engines import AdkApp

PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT")
REGION = os.environ.get("GOOGLE_CLOUD_REGION", "us-central1")

# Initialize Vertex AI for regional services
if PROJECT_ID:
    vertexai.init(project=PROJECT_ID, location=REGION)

# Define the agent logic
root_agent = LlmAgent(
    name="planner_agent",
    model="gemini-3-flash-preview",
    instruction="You are a helpful marathon planning assistant...",
    tools=[] # We will add tools in the next steps
)

def session_service_builder():
    """Builder for Agent Platform Sessions."""
    return VertexAiSessionService(project=PROJECT_ID, location=REGION)

# Wrap the agent in an AdkApp to manage stateful context
app = AdkApp(
    agent=root_agent,
    session_service_builder=session_service_builder
)

5. Habilita el aprendizaje a largo plazo con Memory Bank

Si bien la administración de sesiones hace un seguimiento de las conversaciones individuales, puedes hacer lo mismo para la memoria a largo plazo. En este paso, adjuntarás el agente a Memory Bank de Agent Platform, un servicio de memoria completamente administrado y listo para la empresa.

Inicializa el servicio de Memory Bank

Memory Bank permite que el agente recuerde el contexto en diferentes sesiones. Actualiza planner_agent/agent.py para incluir el servicio de memoria:

from google.adk.memory import VertexAiMemoryBankService

def memory_service_builder():
    """Builder for Agent Platform Memory Bank."""
    return VertexAiMemoryBankService(
        project=PROJECT_ID,
        location=REGION,
        agent_engine_id=AGENT_ENGINE_ID
    )

Implementa la transferencia automática de memoria

Para asegurarnos de que el agente aprenda de cada turno, agregamos un after_agent_callback. Esta función se activa después de que el agente completa una respuesta, lo que le permite "digerir" la sesión y guardar los recuerdos pertinentes en el banco.

1. Define la función de devolución de llamada:

async def auto_save_memories(callback_context):
    """Callback to ingest the session into the memory bank after the turn."""
    # In AdkApp, the memory service is available via the invocation context
    if hasattr(callback_context._invocation_context, 'memory_service') and callback_context._invocation_context.memory_service:
        await callback_context._invocation_context.memory_service.add_session_to_memory(
            callback_context._invocation_context.session
        )

2. Adjunta la devolución de llamada al LlmAgent:

root_agent = LlmAgent(
    # ... other params
    after_agent_callback=[auto_save_memories],
)

Si integras las sesiones de Agent Platform y Memory Bank, tu agente ya no es solo un modelo sin estado, sino que se convierte en un asistente con estado que mantiene el contexto y aprende de sus usuarios con el tiempo.

6. Configura AlloyDB para RAG

Antes de que podamos transferir datos de reglas de la ciudad, necesitamos una base de datos de alto rendimiento para almacenarlos. En este paso, crearás un clúster de AlloyDB y, además, inicializarás el esquema de la base de datos para la búsqueda de vectores.

1. Crea el clúster de AlloyDB y la instancia principal

Ejecuta estos comandos en Cloud Shell para crear tu clúster y su instancia principal:

# Create the cluster
gcloud alloydb clusters create rules-db \
  --password=postgres \
  --region=us-central1

# Create the primary instance with IAM authentication enabled
gcloud alloydb instances create rules-db-primary \
  --instance-type=PRIMARY \
  --cpu-count=2 \
  --region=us-central1 \
  --cluster=rules-db \
  --database-flags=alloydb.iam_authentication=on

2. Otorga roles de IAM obligatorios

Para usar el servidor MCP de AlloyDB administrado, tu identidad necesita permisos específicos. Ejecuta estos comandos para otorgar los roles necesarios:

export USER_EMAIL=$(gcloud config get-value account)

# Role to use MCP tools
gcloud projects add-iam-policy-binding $PROJECT_ID \
  --member="user:$USER_EMAIL" \
  --role="roles/mcp.toolUser"

# Role to execute SQL in AlloyDB
gcloud projects add-iam-policy-binding $PROJECT_ID \
  --member="user:$USER_EMAIL" \
  --role="roles/alloydb.admin"

# Role for IAM database authentication
gcloud projects add-iam-policy-binding $PROJECT_ID \
  --member="user:$USER_EMAIL" \
  --role="roles/alloydb.databaseUser"

# Create the IAM-based database user
gcloud alloydb users create "$USER_EMAIL" \
  --cluster=rules-db \
  --region=us-central1 \
  --type=IAM_BASED

3. Crea bases de datos y tablas con AlloyDB Studio

Como las bases de datos y las tablas de AlloyDB se administran a través de SQL, usaremos AlloyDB Studio en la consola de Google Cloud para finalizar el esquema.

1. Navega a AlloyDB > Clústeres y haz clic en rules-db.
2. En el menú de navegación de la izquierda, haz clic en AlloyDB Studio.
3. Accede con el usuario postgres y la contraseña que configuraste (postgres).
4. Ejecuta el siguiente SQL para crear la base de datos:

   CREATE DATABASE city_rules;
   

5. Cambia la conexión de la base de datos a city_rules en AlloyDB Studio y ejecuta el siguiente SQL para instalar extensiones y crear la tabla rules:

   -- Install extensions for vector search and ML
   CREATE EXTENSION IF NOT EXISTS vector;
   CREATE EXTENSION IF NOT EXISTS google_ml_integration CASCADE;
   
   -- Create the rules table
   CREATE TABLE IF NOT EXISTS rules (
       id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
       text TEXT NOT NULL,
       city TEXT NOT NULL,
       embedding vector(3072) DEFAULT NULL
   );
   
   -- Grant your IAM user access to the table (replace with your email)
   GRANT ALL PRIVILEGES ON TABLE rules TO "YOUR_EMAIL_ADDRESS";
   

7. Transfiere datos de reglas de la ciudad con Spark Lightning Engine

Para proporcionar una planificación realmente precisa, un agente necesita más que una instrucción bien elaborada; necesita fundamentación en datos y contexto organizacional. En este paso, usarás Spark Lightning Engine en Dataproc Serverless para procesar archivos PDF grandes de reglas de la ciudad y transferirlos a AlloyDB.

¿Por qué usar Spark Lightning Engine?

Para fundamentar agentes a gran escala, se requiere procesar grandes cantidades de datos no estructurados. Spark Lightning Engine es un motor de ejecución de alto rendimiento para Spark que acelera significativamente estas cargas de trabajo. Lo usamos aquí para realizar fragmentación semántica en documentos con Document AI de Google.

Explora la canalización de Spark

La lógica de transferencia se define en spark-setup/spark_alloydb_processor.py. La canalización sigue estos pasos:

1. Lista de PDFs: Recupera los URIs de documentos de un bucket de Cloud Storage de Google.
2. Extracción semántica: Usa una UDF (función definida por el usuario) para llamar a la API de Document AI.
3. Escribe en AlloyDB: Guarda los fragmentos de texto extraídos en la tabla de AlloyDB llamada rules.

# Extract from spark_alloydb_processor.py
def process_document(gcs_uri: str):
    # ... calls Document AI to parse PDF ...
    return chunks

# Parallel processing with Spark Lightning Engine
process_udf = udf(process_document, chunk_schema)
chunked_df = uri_df.withColumn("chunks", process_udf(col("gcs_uri"))) \
                   .select(explode(col("chunks")).alias("chunk")) \
                   .select("chunk.*")

# Save to AlloyDB for Vector Search
chunked_df.write.format("jdbc") \
    .option("url", jdbc_url) \
    .option("dbtable", "rules") \
    .mode("append") \
    .save()

Ejecuta el trabajo de transferencia

Activa el proceso de transferencia con la secuencia de comandos proporcionada:

./spark-setup/run_dataproc.sh

Si agregas datos y contexto adicionales al agente con Spark Lightning Engine, te aseguras de que sus planes no solo sean creativos, sino que también cumplan con las reglas locales.

8. RAG con AlloyDB

Ahora que los datos de las reglas de la ciudad están en AlloyDB, el agente puede usarlos para realizar la generación mejorada por recuperación (RAG). Esto garantiza que el plan de maratón cumpla con códigos de ciudad específicos.

El poder de AlloyDB para RAG

AlloyDB se destaca en la búsqueda de vectores, lo que nos permite almacenar datos estructurados y embeddings de vectores en el mismo lugar. El agente puede usar la función embedding integrada en AlloyDB para encontrar la información de reglas más pertinente.

Recuperación híbrida con la búsqueda de vectores

Para darle al agente acceso a estos datos, proporcionamos una herramienta que consulta AlloyDB con la similitud de vectores. Puedes ver esta lógica en hybrid_recall.sql, que muestra cómo calcular la distancia entre una consulta y nuestras reglas almacenadas:

SELECT
    text,
    (embedding <=> 
     embedding('gemini-embedding-001', 
               'Restrictions for running a race on the Las Vegas strip')::vector) 
    as distance
FROM
    rules
WHERE city = 'Las Vegas'
ORDER BY
    distance ASC
LIMIT 5;

Fundamenta el agente en las reglas locales con una herramienta RAG

Para que la herramienta esté disponible para el agente, debes definirla en planner_agent/tools.py y, luego, registrarla en planner_agent/agent.py. Usaremos el servidor MCP de AlloyDB remoto administrado de Google Cloud para conectarnos a nuestra base de datos.

1. Define la herramienta en planner_agent/tools.py con el patrón "Hybrid Recall". Usaremos el protocolo streamable_http para conectarnos al servidor MCP de AlloyDB administrado:

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def get_local_and_traffic_rules(query: str) -> str:
    """Uses vector search in AlloyDB via managed MCP server."""
    # Vector search query using built-in AlloyDB embedding functions
    sql = f"SELECT text FROM rules WHERE city = 'Las Vegas' ORDER BY embedding <=> google_ml.embedding('gemini-embedding-001', '{query}')::vector ASC LIMIT 5;"
    
    # Establish a streamable HTTP connection to the MCP server
    async with streamablehttp_client(url, headers=get_auth_headers()) as (read_stream, write_stream, _):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            result = await session.call_tool(
                "execute_sql",
                arguments={
                    "instance": full_instance_name,
                    "database": "city_rules",
                    "sqlStatement": sql
                }
            )
            return "\n".join([c.text for c in result.content if hasattr(c, 'text')])

2. Registra la herramienta y finaliza planner_agent/agent.py:

# ... imports ...

# Assemble the Agent
root_agent = LlmAgent(
    name="planner_agent",
    model="gemini-3-flash-preview",
    instruction="You are a helpful marathon planning assistant...",
    tools=[
        get_local_and_traffic_rules,
    ],
    after_agent_callback=[auto_save_memories],
)

# 2. Wrap the agent in an AdkApp to manage the stateful lifecycle
app = AdkApp(
    agent=root_agent,
    session_service_builder=session_service_builder,
    memory_service_builder=memory_service_builder
)

9. Orientación de expertos con habilidades de agentes

Las habilidades de agentes son módulos independientes que proporcionan instrucciones, orientación y recursos específicos para ayudar a los agentes a realizar tareas de manera más eficaz. En lugar de saturar la instrucción del sistema con instrucciones complejas para cada herramienta, puedes encapsular esa experiencia en una habilidad que se carga solo cuando es necesario.

Google proporciona habilidades precompiladas para productos de Google (como AlloyDB y BigQuery) para garantizar que tus agentes sigan las prácticas recomendadas de la industria para consultar datos y administrar recursos. Puedes explorar estos y otros patrones especializados en el Google Skills Depot. Aquí encontrarás las habilidades básicas de AlloyDB aquí.

1. Explora el archivo de habilidades

Abre el archivo de habilidades preconfigurado en planner_agent/skills/get-local-and-traffic-rules/SKILL.md. Se ve de la siguiente manera:

---
name: get-local-and-traffic-rules
description: Retrieve local rules and traffic information for a specific jurisdiction.
---
# get_local_and_traffic_rules Skill

This skill provides guidelines on how to effectively use the `get_local_and_traffic_rules` tool.

## Overview
The `get_local_and_traffic_rules` tool interfaces with an AlloyDB database to perform vector similarity searches on a corpus of rules and traffic information using a provided natural language query.

## Usage Guidelines
1. **Query Specificity**: When calling the tool, provide specific details in the `query` argument. For example, instead of querying "food rules", use "rules regarding food vendors during public events".
2. **Contextual Use**: Use the tool when planning events or activities that require adherence to local municipal or state rules (e.g., street closures, noise ordinances, environmental rules).
3. **Handling Results**: The tool returns a string containing the text of the top 5 most relevant rules. If no error occurs, parse the returned string to inform your planning tasks.
4. **Error Handling**: If an error string is returned (e.g., "Error querying rules: ..."), you must report this failure or attempt an alternative approach if applicable.

## Underlying Mechanism
- The tool uses `google_ml.embedding` to convert the query into a vector representation.
- It calculates distance (`<=>`) against the `embedding` column in the `rules` table on an AlloyDB instance.
- Results are fetched in descending order of similarity, limited to 5 results.

2. Cómo se registra la habilidad

En planner_agent/agent.py, la habilidad se carga desde el directorio y se agrega a las herramientas del agente. El código se ve de la siguiente manera:

import pathlib
from google.adk.skills import load_skill_from_dir
from google.adk.tools import skill_toolset

# Load the AlloyDB skill from its directory
alloydb_skill = load_skill_from_dir(pathlib.Path(__file__).parent / "skills" / "get-local-and-traffic-rules")

# Assemble the Agent with the Skill Toolset
root_agent = LlmAgent(
    name="planner_agent",
    model="gemini-3-flash-preview",
    instruction="You are a helpful marathon planning assistant...",
    tools=[
        get_local_and_traffic_rules,
        skill_toolset.SkillToolset(skills=[alloydb_skill])
    ],
    after_agent_callback=[auto_save_memories],
)

Las habilidades de agentes te permiten separar la implementación de la herramienta de la instrucción de la herramienta. Esto mantiene limpia la instrucción del sistema y garantiza que el agente tenga el contexto exacto que necesita solo cuando usa esa capacidad específica.

10. Prueba el agente

1. Inicia el agente de forma local:

uv run adk run planner_agent

2. Haz una pregunta sobre las reglas de la ciudad: [user]: What are the rules for running a race on the Las Vegas strip?

El agente llamará a la herramienta get_local_and_traffic_rules, realizará una búsqueda de vectores en AlloyDB y mostrará una respuesta basada en los fragmentos de reglas oficiales procesados por Spark.

11. Implementa el agente

Implementa en Agent Platform

uv run adk deploy agent_engine \
  --env_file .env \
  planner_agent

12. Limpia

Para evitar cargos continuos, borra los recursos creados durante este codelab.

Borra el clúster de AlloyDB

# Delete the AlloyDB Cluster
gcloud alloydb clusters delete rules-db --region=us-central1 --force

Borra la app del entorno de ejecución del agente

Puedes borrar la instancia de Reasoning Engine a través de la consola o con el comando gcloud (si tienes el nombre del recurso). Para simplificar, usa la consola:

1. Ve a la página Entorno de ejecución del agente.
2. Selecciona planner_agent -> haz clic en el botón de tres puntos en el lado derecho.
3. Haz clic en Borrar.

13. Felicitaciones

¡Felicitaciones! Mejoraste correctamente un agente del ADK con capacidades avanzadas de memoria y fundamentación de datos.

Qué aprendiste

• Agentes con estado: Integra sesiones de Agent Platform para mantener el contexto de la conversación.
• Aprendizaje a largo plazo: Conecta un Memory Bank de Agent Platform para permitir que el agente aprenda de las interacciones del usuario.
• Transferencia de datos: Usa Spark Lightning Engine y Document AI para procesar documentos no estructurados.
• RAG: Compila un sistema de búsqueda de vectores en AlloyDB para fundamentar el agente en las reglas del mundo real.

Próximos pasos

• Explora la documentación de Agent Platform para obtener más información sobre la implementación administrada.
• Analiza en detalle la búsqueda de vectores de AlloyDB para obtener patrones avanzados de RAG.
• Escala tus canalizaciones de transferencia con Dataproc Serverless.