Next '26 Developer Keynote: Enhancing Agents with Memory

1. Introduzione

In questo codelab, porterai i tuoi agenti ADK a un livello superiore aggiungendo conoscenze persistenti e specializzate. Imparerai a gestire lo stato della conversazione con le sessioni di Agent Platform, ad abilitare l'apprendimento a lungo termine con Memory Bank e a integrare dati complessi sulle regole della città utilizzando Spark e AlloyDB per RAG (Retrieval-Augmented Generation).

Attività previste

• Configura le sessioni di Agent Platform per la persistenza delle conversazioni.
• Implementa un Memory Bank per consentire agli agenti di apprendere dalle interazioni precedenti.
• Utilizza Spark Lightning Engine per importare ed elaborare la documentazione sulle regole della città.
• Crea un sistema RAG utilizzando AlloyDB e la ricerca vettoriale.
• Esegui il deployment dell'agente migliorato su Agent Platform.

Che cosa ti serve

• Un browser web come Chrome
• Un progetto Google Cloud con la fatturazione abilitata
• Una conoscenza di base di Python e SQL

Durata stimata: 60 minuti

Le risorse create in questo codelab dovrebbero costare meno di 5 $.

2. Prima di iniziare

Crea un progetto Google Cloud

1. Nella console Google Cloud, nella pagina di selezione del progetto, seleziona o crea un progetto Google Cloud.
2. Verifica che la fatturazione sia attivata per il tuo progetto Cloud. Scopri come verificare se la fatturazione è abilitata per un progetto.

Avvia Cloud Shell

Cloud Shell è un ambiente a riga di comando in esecuzione in Google Cloud che viene fornito con gli strumenti necessari precaricati.

1. Fai clic su Attiva Cloud Shell nella parte superiore della console Google Cloud.
2. Una volta eseguita la connessione a Cloud Shell, verifica l'autenticazione:

   gcloud auth list
   

3. Verifica che il progetto sia configurato:

   gcloud config get project
   

4. Se il progetto non è impostato come previsto, impostalo:

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

Verifica l'autenticazione:

gcloud auth list

Conferma il progetto:

gcloud config get project

Impostalo se necessario:

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

Abilita API

Esegui questo comando per abilitare tutte le API richieste per la gestione delle sessioni, l'elaborazione di Spark e 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

L'abilitazione di queste API ti garantisce l'accesso ad Agent Platform, ai cluster AlloyDB gestiti e a Dataproc Serverless per le pipeline Spark.

3. Configura l'ambiente

Per questo codelab, utilizzerai l'ambiente preconfigurato nel repository keynote.

1. Clona il repository e vai alla cartella del progetto:

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

2. Configura un ambiente Python virtuale e installa i pacchetti ADK richiesti:

uv venv
source .venv/bin/activate
uv sync

Configura le variabili di ambiente

L'agente richiede una configurazione specifica per connettersi ad Agent Platform e AlloyDB.

1. Copia il file di ambiente di esempio:

cp .env.example .env

2. Apri .env e aggiorna i seguenti campi:
   • GOOGLE_CLOUD_PROJECT: il tuo ID progetto.
   • 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. Esegui il seguente script di assistenza per creare un'istanza di Agent Engine da utilizzare per le sessioni di conversazione e la memoria a lungo termine. Il campo AGENT_ENGINE_ID nel file .env verrà compilato automaticamente:

uv run utils/setup_agent_engine.py

Se l'operazione va a buon fine, dovresti visualizzare:

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

4. Crea un agente con la gestione delle sessioni

In questo passaggio, inizializzerai un agente di pianificazione della maratona in grado di mantenere la cronologia delle conversazioni in più turni. Questo risultato si ottiene utilizzando la classe App dell'ADK e le sessioni di Agent Platform.

Inizializza l'agente e il servizio di sessione

Apri planner_agent/agent.py. Vedrai come stiamo aggiungendo una classe ADK per integrare le sessioni di Agent Platform. In questo modo, possiamo rendere gli agenti con stato nel tempo e modificare il contesto in base alle esigenze.

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. Abilita l'apprendimento a lungo termine con Memory Bank

Mentre la gestione delle sessioni tiene traccia delle singole conversazioni, puoi fare la stessa cosa per la memoria a lungo termine. In questo passaggio, collegherai l'agente a Memory Bank di Agent Platform, un servizio di memoria completamente gestito e pronto per l'uso aziendale.

Inizializza il servizio Memory Bank

Memory Bank consente all'agente di richiamare il contesto in diverse sessioni. Aggiorna planner_agent/agent.py per includere il servizio di 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 l'importazione automatica della memoria

Per garantire che l'agente apprenda da ogni turno, aggiungiamo un after_agent_callback. Questa funzione viene attivata dopo che l'agente completa una risposta, consentendogli di "digerire" la sessione e salvare i ricordi pertinenti nella banca.

1. Definisci la funzione di callback:

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. Collega il callback a LlmAgent:

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

Integrando le sessioni di Agent Platform e Memory Bank, l'agente non è più solo un modello senza stato, ma diventa un assistente con stato che mantiene il contesto e apprende dagli utenti nel tempo.

6. Configura AlloyDB per RAG

Prima di poter importare i dati sulle regole della città, abbiamo bisogno di un database ad alte prestazioni per archiviarli. In questo passaggio, creerai un cluster AlloyDB e inizializzerai lo schema del database per la ricerca vettoriale.

1. Crea il cluster AlloyDB e l'istanza principale

Esegui questi comandi in Cloud Shell per creare il cluster e la relativa istanza principale:

# 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. Concedi i ruoli IAM richiesti

Per utilizzare il server MCP AlloyDB gestito, la tua identità deve disporre di autorizzazioni specifiche. Esegui questi comandi per concedere i ruoli richiesti:

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 database e tabelle tramite AlloyDB Studio

Poiché i database e le tabelle AlloyDB vengono gestiti tramite SQL, utilizzeremo AlloyDB Studio nella console Google Cloud per finalizzare lo schema.

1. Vai a AlloyDB > Cluster e fai clic su rules-db.
2. Nel menu di navigazione a sinistra, fai clic su AlloyDB Studio.
3. Accedi utilizzando l'utente postgres e la password che hai impostato (postgres).
4. Esegui il seguente SQL per creare il database:

   CREATE DATABASE city_rules;
   

5. Passa alla connessione al database city_rules in AlloyDB Studio ed esegui il seguente SQL per installare le estensioni e creare la tabella 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. Importa i dati sulle regole della città con Spark Lightning Engine

Per fornire una pianificazione davvero accurata, un agente ha bisogno di più di un prompt ben elaborato: ha bisogno di grounding nel contesto dei dati e dell'organizzazione. In questo passaggio, utilizzerai Spark Lightning Engine su Dataproc Serverless per elaborare PDF di grandi dimensioni sulle regole della città e importarli in AlloyDB.

Perché Spark Lightning Engine?

Il grounding degli agenti su larga scala richiede l'elaborazione di enormi quantità di dati non strutturati. Spark Lightning Engine è un motore di esecuzione ad alte prestazioni per Spark che accelera notevolmente questi workload. Lo utilizziamo qui per eseguire il chunking semantico sui documenti utilizzando Document AI di Google.

Esplora la pipeline Spark

La logica di importazione è definita in spark-setup/spark_alloydb_processor.py. La pipeline segue questi passaggi:

1. Elenca PDF: recupera gli URI dei documenti da un bucket Cloud Storage.
2. Estrazione semantica: utilizza una UDF (funzione definita dall'utente) per chiamare l'API Document AI.
3. Scrivi in AlloyDB: salva i blocchi di testo estratti nella tabella AlloyDB denominata 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()

Esegui il job di importazione

Attiva il processo di importazione utilizzando lo script fornito:

./spark-setup/run_dataproc.sh

Aggiungendo dati e contesto all'agente utilizzando Spark Lightning Engine, ti assicuri che i suoi piani non siano solo creativi, ma anche pienamente conformi alle regole locali.

8. RAG con AlloyDB

Ora che i dati sulle regole della città sono in AlloyDB, l'agente può utilizzarli per eseguire Retrieval-Augmented Generation (RAG). In questo modo, il piano della maratona rispetta i codici specifici della città.

La potenza di AlloyDB per RAG

AlloyDB eccelle nella ricerca vettoriale, consentendoci di archiviare sia i dati strutturati sia gli embedding vettoriali nello stesso posto. L'agente può utilizzare la funzione embedding integrata in AlloyDB per trovare le informazioni sulle regole più pertinenti.

Richiamo ibrido con la ricerca vettoriale

Per consentire all'agente di accedere a questi dati, forniamo uno strumento che esegue query su AlloyDB utilizzando la somiglianza vettoriale. Puoi vedere questa logica in hybrid_recall.sql, che mostra come calcolare la distanza tra una query e le regole archiviate:

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;

Fornisci grounding all'agente nelle regole locali con uno strumento RAG

Per rendere lo strumento disponibile per l'agente, devi definirlo in planner_agent/tools.py e poi registrarlo in planner_agent/agent.py. Utilizzeremo il server MCP AlloyDB remoto gestito di Google Cloud per connetterci al nostro database.

1. Definisci lo strumento in planner_agent/tools.py utilizzando il pattern "Richiamo ibrido". Utilizzeremo il protocollo streamable_http per connetterci al server MCP AlloyDB gestito:

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 lo strumento e finalizza 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. Assistenza da parte di esperti con le skill dell'agente

Le skill dell'agente sono moduli autonomi che forniscono istruzioni, indicazioni e risorse specifiche per aiutare gli agenti a svolgere le attività in modo più efficace. Anziché riempire il prompt di sistema con istruzioni complesse per ogni strumento, puoi incapsulare questa competenza in una skill che viene caricata solo quando necessario.

Google fornisce skill predefinite per i prodotti Google (come AlloyDB e BigQuery) per garantire che gli agenti seguano le best practice del settore per l'esecuzione di query sui dati e la gestione delle risorse. Puoi esplorare queste e altre pattern specializzate nel Google Skills Depot. Qui troverai le skill di base di AlloyDB qui.

1. Esplora il file di skill

Apri il file di skill preconfigurato all'indirizzo planner_agent/skills/get-local-and-traffic-rules/SKILL.md. Ecco come appare:

---
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. Come viene registrata la skill

In planner_agent/agent.py, la skill viene caricata dalla directory e aggiunta agli strumenti dell'agente. Ecco come appare il codice:

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

Le skill dell'agente ti consentono di separare l'implementazione dello strumento dalle istruzioni dello strumento. In questo modo, il prompt di sistema rimane pulito e l'agente dispone del contesto esatto di cui ha bisogno solo quando utilizza la funzionalità specifica.

10. Testa l'agente

1. Avvia l'agente localmente:

uv run adk run planner_agent

2. Fai una domanda sulle regole della città: [user]: What are the rules for running a race on the Las Vegas strip?

L'agente chiamerà lo strumento get_local_and_traffic_rules, eseguirà una ricerca vettoriale in AlloyDB e restituirà una risposta basata sui blocchi di regole ufficiali elaborati da Spark.

11. Esegui il deployment dell'agente

Esegui il deployment su Agent Platform

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

12. Libera spazio

Per evitare addebiti continui, elimina le risorse create durante questo codelab.

Elimina il cluster AlloyDB

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

Elimina l'app di runtime dell'agente

Puoi eliminare l'istanza di Reasoning Engine tramite la console o utilizzando il comando gcloud (se hai il nome della risorsa). Per semplicità, utilizza la console:

1. Vai alla pagina Runtime dell'agente.
2. Seleziona planner_agent -> fai clic sul pulsante con tre puntini sul lato destro.
3. Fai clic su Elimina.

13. Complimenti

Complimenti! Hai migliorato un agente ADK con funzionalità avanzate di memoria e grounding dei dati.

Che cosa hai imparato

• Agenti con stato: integrazione delle sessioni di Agent Platform per mantenere il contesto della conversazione.
• Apprendimento a lungo termine: collegamento di un Memory Bank di Agent Platform per consentire all'agente di apprendere dalle interazioni degli utenti.
• Importazione dei dati: utilizzo di Spark Lightning Engine e Document AI per elaborare documenti non strutturati.
• RAG: creazione di un sistema di ricerca vettoriale in AlloyDB per fornire grounding all'agente nelle regole del mondo reale.

Passaggi successivi

• Esplora la documentazione di Agent Platform per scoprire di più sul deployment gestito.
• Approfondimento sulla ricerca vettoriale di AlloyDB per pattern RAG avanzate.
• Scala le pipeline di importazione con Dataproc Serverless.