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 Agent Platform Sessions, habilitar el aprendizaje a largo plazo con Memory Bank y, luego, 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 interacciones anteriores.
• Usa Spark Lightning Engine para transferir y procesar la documentación de las reglas de la ciudad.
• Crea 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

Crea 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 se esperaba, 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úrala si es necesario:

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

Habilitar APIs

Ejecuta este comando para habilitar todas las APIs requeridas 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 asegurarás 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

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

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, luego, instala los paquetes del ADK necesarios:

uv venv
source .venv/bin/activate
uv sync

Configura variables de entorno

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

1. Copia el archivo de entorno de muestra:

cp .env.example .env

2. Abre .env y actualiza los siguientes campos:
   • GOOGLE_CLOUD_PROJECT: Es el ID de tu 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 completará automáticamente el AGENT_ENGINE_ID en tu archivo .env:

uv run utils/setup_agent_engine.py

Cuando se complete la operación, 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 Session Management

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

Inicializa el servicio de agente y sesión

Abre planner_agent/agent.py. Verás cómo agregamos una clase del ADK para integrar 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 con 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 empresas.

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 recuerdos relevantes 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. Conecta la devolución de llamada a LlmAgent:

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

Al integrar Agent Platform Sessions 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 transferir los datos de las 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, luego, inicializarás el esquema de la base de datos para la búsqueda de vectores.

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

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 los 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 a través de AlloyDB Studio

Dado que 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 > Clusters 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 la siguiente instrucción de 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 código 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. Cómo transferir 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 fundamentarse en los datos y el contexto de la organización. 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, es necesario 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. Aquí lo usamos para realizar la división semántica en fragmentos de documentos con Document AI de Google.

Explora la canalización de Spark

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

1. List PDFs: Recupera los URIs de documentos de un bucket de Google Cloud Storage.
2. Extracción semántica: Usa una UDF (función definida por el usuario) para llamar a la API de Document AI.
3. Write to 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

Al proporcionar 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 reglamentaciones 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 los 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 integrada embedding en AlloyDB para encontrar la información de reglas más pertinente.

Recuperación híbrida con Vector Search

Para darle acceso al agente 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 búsqueda 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 de 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 remoto administrado de AlloyDB de Google Cloud para conectarnos a nuestra base de datos.

1. Define la herramienta en planner_agent/tools.py con el patrón "Recuperación híbrida". 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 Agent Skills 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 sobrecargar tu instrucción del sistema con instrucciones complejas para cada herramienta, puedes encapsular esa experiencia en una habilidad que se cargue solo cuando sea necesario.

Google proporciona habilidades prediseñadas para los 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 Google Skills Depot. Encontrarás las habilidades básicas de AlloyDB aquí.

1. Explora el archivo de habilidad

Abre el archivo de habilidad preconfigurado en planner_agent/skills/get-local-and-traffic-rules/SKILL.md. A continuación, te mostramos cómo se ve:

---
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 skill

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

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 los agentes te permiten desacoplar la implementación de la herramienta de sus instrucciones. Esto mantiene limpio el mensaje 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 devolverá 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 que creaste 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 de Agent Runtime

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 Agent Runtime.
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: Integración de sesiones de Agent Platform para mantener el contexto de la conversación
• Aprendizaje a largo plazo: Se adjunta un Agent Platform Memory Bank para permitir que el agente aprenda de las interacciones del usuario.
• Transferencia de datos: Uso de Spark Lightning Engine y Document AI para procesar documentos no estructurados
• RAG: Se crea un sistema de búsqueda de vectores en AlloyDB para fundamentar el agente en 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.
• Análisis detallado de la búsqueda de vectores de AlloyDB para patrones avanzados de RAG
• Escala tus canalizaciones de transferencia con Dataproc Serverless.