Next ‘26 Developer Keynote: Enhancing Agents with Memory

1. Einführung

In diesem Codelab verbessern Sie Ihre ADK-Agenten, indem Sie ihnen dauerhaftes und spezialisiertes Wissen hinzufügen. Sie erfahren, wie Sie den Unterhaltungsstatus mit Agent Platform-Sitzungen verwalten, langfristiges Lernen mit der Memory Bank ermöglichen und komplexe Daten zu Stadtregeln mithilfe von Spark und AlloyDB für RAG (Retrieval-Augmented Generation) integrieren.

Aufgaben

• Agent Platform-Sitzungen für die Unterhaltungspersistenz konfigurieren.
• Eine Memory Bank implementieren, damit Agenten aus früheren Interaktionen lernen können.
• Spark Lightning Engine verwenden, um die Dokumentation zu Stadtregeln aufzunehmen und zu verarbeiten.
• Ein RAG-System mit AlloyDB und der Vektorsuche erstellen.
• Den verbesserten Agenten auf der Agent Platform bereitstellen.

Voraussetzungen

• Ein Webbrowser wie Chrome
• Ein Google Cloud-Projekt mit aktivierter Abrechnung
• Grundkenntnisse in Python und SQL

Geschätzte Dauer: 60 Minuten

Die in diesem Codelab erstellten Ressourcen sollten weniger als 5 $kosten.

2. Hinweis

Google Cloud-Projekt erstellen

1. Wählen Sie in der Google Cloud Console auf der Seite der Projektauswahl ein Google Cloud-Projekt aus oder erstellen Sie eines.
2. Die Abrechnung für das Cloud-Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist.

Cloud Shell starten

Die Cloud Shell ist eine Befehlszeilenumgebung, die in Google Cloud ausgeführt wird und in der die erforderlichen Tools vorinstalliert sind.

1. Klicken Sie oben in der Google Cloud Console auf Cloud Shell aktivieren.
2. Sobald die Verbindung mit der Cloud Shell hergestellt ist, prüfen Sie die Authentifizierung:

   gcloud auth list
   

3. Prüfen Sie, ob Ihr Projekt konfiguriert ist:

   gcloud config get project
   

4. Wenn Ihr Projekt nicht wie erwartet festgelegt ist, legen Sie es fest:

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

Authentifizierung prüfen:

gcloud auth list

Projekt bestätigen:

gcloud config get project

Bei Bedarf festlegen:

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

APIs aktivieren

Führen Sie diesen Befehl aus, um alle erforderlichen APIs für die Sitzungsverwaltung, die Spark-Verarbeitung und AlloyDB zu aktivieren:

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

Wenn Sie diese APIs aktivieren, haben Sie Zugriff auf die Agent Platform, verwaltete AlloyDB-Cluster und Dataproc Serverless für Ihre Spark-Pipelines.

3. Umgebung einrichten

In diesem Codelab verwenden Sie die vorkonfigurierte Umgebung im Keynote-Repository.

1. Klonen Sie das Repository und rufen Sie den Projektordner auf:

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

2. Richten Sie eine virtuelle Python-Umgebung ein und installieren Sie die erforderlichen ADK-Pakete:

uv venv
source .venv/bin/activate
uv sync

Umgebungsvariablen konfigurieren

Für den Agenten ist eine bestimmte Konfiguration erforderlich, um eine Verbindung zur Agent Platform und zu AlloyDB herzustellen.

1. Kopieren Sie die Beispieldatei für die Umgebung:

cp .env.example .env

2. Öffnen Sie .env und aktualisieren Sie die folgenden Felder:
   • GOOGLE_CLOUD_PROJECT: Ihre Projekt-ID
   • 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. Führen Sie das folgende Hilfsskript aus, um eine Agent Engine-Instanz zu erstellen, die für Unterhaltungssitzungen und langfristiges Lernen verwendet werden soll. Dadurch wird AGENT_ENGINE_ID automatisch in Ihrer Datei .env eingefügt:

uv run utils/setup_agent_engine.py

Nach erfolgreicher Ausführung sollte Folgendes angezeigt werden:

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

4. Agent mit Sitzungsverwaltung erstellen

In diesem Schritt initialisieren Sie einen Marathon Planner-Agenten, der den Unterhaltungsverlauf über mehrere Runden hinweg beibehalten kann. Dies wird mit der ADK-Klasse App und den Agent Platform-Sitzungen erreicht.

Agent- und Sitzungsdienst initialisieren

Öffnen Sie planner_agent/agent.py. Sie sehen, wie wir eine ADK-Klasse hinzufügen, um Agent Platform-Sitzungen zu integrieren. So können wir unsere Agenten im Laufe der Zeit zustandsorientiert machen und den Kontext nach Bedarf ändern.

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. Langfristiges Lernen mit der Memory Bank aktivieren

Während die Sitzungsverwaltung einzelne Unterhaltungen verfolgt, können Sie dasselbe für das langfristige Lernen tun. In diesem Schritt verknüpfen Sie den Agenten mit der Memory Bank der Agent Platform, einem unternehmenstauglichen und vollständig verwalteten Speicherdienst.

Memory Bank-Dienst initialisieren

Mit der Memory Bank kann sich der Agent den Kontext über verschiedene Sitzungen hinweg merken. Aktualisieren Sie planner_agent/agent.py, um den Speicherdienst einzufügen:

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
    )

Automatische Aufnahme von Informationen in die Memory Bank implementieren

Damit der Agent aus jeder Runde lernt, fügen wir einen after_agent_callback hinzu. Diese Funktion wird ausgelöst, nachdem der Agent eine Antwort erstellt hat. So kann er die Sitzung „verarbeiten“ und relevante Informationen in der Memory Bank speichern.

1. Definieren Sie die Callback-Funktion:

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. Verknüpfen Sie den Callback mit dem LlmAgent:

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

Durch die Integration von Agent Platform-Sitzungen und der Memory Bank ist Ihr Agent nicht mehr nur ein zustandsloses Modell, sondern ein zustandsorientierter Assistent, der den Kontext beibehält und im Laufe der Zeit von seinen Nutzern lernt.

6. AlloyDB für RAG einrichten

Bevor wir Daten zu Stadtregeln aufnehmen können, benötigen wir eine leistungsstarke Datenbank, in der sie gespeichert werden. In diesem Schritt erstellen Sie einen AlloyDB-Cluster und initialisieren das Datenbankschema für die Vektorsuche.

1. AlloyDB-Cluster und primäre Instanz erstellen

Führen Sie diese Befehle in der Cloud Shell aus, um Ihren Cluster und seine primäre Instanz zu erstellen:

# 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. Erforderliche IAM-Rollen zuweisen

Damit Sie den verwalteten AlloyDB MCP-Server verwenden können, benötigt Ihre Identität bestimmte Berechtigungen. Führen Sie diese Befehle aus, um die erforderlichen Rollen zuzuweisen:

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. Datenbank und Tabellen über AlloyDB Studio erstellen

Da AlloyDB-Datenbanken und ‑Tabellen über SQL verwaltet werden, verwenden wir AlloyDB Studio in der Google Cloud Console, um das Schema fertigzustellen.

1. Rufen Sie AlloyDB > Cluster auf und klicken Sie auf rules-db.
2. Klicken Sie im linken Navigationsmenü auf AlloyDB Studio.
3. Melden Sie sich mit dem Nutzer postgres und dem von Ihnen festgelegten Passwort (postgres) an.
4. Führen Sie die folgende SQL-Anweisung aus, um die Datenbank zu erstellen:

   CREATE DATABASE city_rules;
   

5. Wechseln Sie in AlloyDB Studio zu Ihrer Datenbankverbindung city_rules und führen Sie die folgende SQL-Anweisung aus, um Erweiterungen zu installieren und die Tabelle rules zu erstellen:

   -- 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. Daten zu Stadtregeln mit der Spark Lightning Engine aufnehmen

Für eine wirklich genaue Planung benötigt ein Agent mehr als nur einen gut formulierten Prompt. Er muss auf Daten und den organisatorischen Kontext abgestimmt sein. In diesem Schritt verwenden Sie die Spark Lightning Engine in Dataproc Serverless, um große PDFs mit Stadtregeln zu verarbeiten und in AlloyDB aufzunehmen.

Warum die Spark Lightning Engine?

Die Fundierung von Agenten im großen Maßstab erfordert die Verarbeitung riesiger Mengen unstrukturierter Daten. Die Spark Lightning Engine ist eine leistungsstarke Ausführungs-Engine für Spark, die diese Arbeitslasten erheblich beschleunigt. Wir verwenden sie hier, um mit Document AI von Google semantische Chunking auf Dokumente anzuwenden.

Spark-Pipeline untersuchen

Die Aufnahmelogik ist in spark-setup/spark_alloydb_processor.py definiert. Die Pipeline umfasst folgende Schritte:

1. PDFs auflisten: Ruft Dokument-URIs aus einem Google Cloud Storage-Bucket ab.
2. Semantische Extraktion: Ruft mit einer UDF (User Defined Function) die Document AI API auf.
3. In AlloyDB schreiben: Speichert die extrahierten Text-Chunks in der AlloyDB-Tabelle 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()

Aufnahmejob ausführen

Lösen Sie den Aufnahmeprozess mit dem bereitgestellten Skript aus:

./spark-setup/run_dataproc.sh

Wenn Sie dem Agenten mit der Spark Lightning Engine zusätzliche Daten und Kontext zur Verfügung stellen, sind seine Pläne nicht nur kreativ, sondern entsprechen auch vollständig den lokalen Regeln.

8. RAG mit AlloyDB

Nachdem sich die Daten zu den Stadtregeln in AlloyDB befinden, kann der Agent sie für Retrieval-Augmented Generation (RAG) verwenden. So wird sichergestellt, dass der Marathonplan bestimmten Stadtregeln entspricht.

Vorteile von AlloyDB für RAG

AlloyDB zeichnet sich durch die Vektorsuche aus. So können wir sowohl strukturierte Daten als auch Vektoreinbettungen an derselben Stelle speichern. Der Agent kann die integrierte Funktion embedding in AlloyDB verwenden, um die relevantesten Informationen zu den Regeln zu finden.

Hybrider Abruf mit der Vektorsuche

Damit der Agent auf diese Daten zugreifen kann, stellen wir ein Tool bereit, das AlloyDB mithilfe der Vektorähnlichkeit abfragt. Diese Logik finden Sie in hybrid_recall.sql. Dort wird gezeigt, wie Sie den Abstand zwischen einer Abfrage und unseren gespeicherten Regeln berechnen:

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;

Agent mit einem RAG-Tool auf lokale Regeln abstimmen

Damit das Tool für den Agenten verfügbar ist, müssen Sie es in planner_agent/tools.py definieren und dann in planner_agent/agent.py registrieren. Wir verwenden den verwalteten AlloyDB MCP-Remoteserver von Google Cloud, um eine Verbindung zu unserer Datenbank herzustellen.

1. Definieren Sie das Tool in planner_agent/tools.py mit dem Muster „Hybrider Abruf“. Wir verwenden das Protokoll streamable_http, um eine Verbindung zum verwalteten AlloyDB MCP-Server herzustellen:

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. Registrieren Sie das Tool und schließen Sie planner_agent/agent.py ab:

# ... 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. Unterstützung durch Experten mit Agent Skills

Agent Skills sind in sich geschlossene Module, die spezifische Anweisungen, Anleitungen und Ressourcen enthalten, damit Agenten Aufgaben effektiver ausführen können. Anstatt Ihren Systemprompt mit komplexen Anweisungen für jedes Tool zu überladen, können Sie dieses Fachwissen in einem Skill zusammenfassen, der nur bei Bedarf geladen wird.

Google bietet vorgefertigte Skills für Google-Produkte (z. B. AlloyDB und BigQuery), damit Ihre Agenten Best Practices der Branche für das Abfragen von Daten und das Verwalten von Ressourcen einhalten. Diese und andere spezialisierte Muster finden Sie im Google Skills Depot. Hier finden Sie die grundlegenden AlloyDB-Skills hier.

1. Skill-Datei untersuchen

Öffnen Sie die vorkonfigurierte Skill-Datei unter planner_agent/skills/get-local-and-traffic-rules/SKILL.md. Das sieht so aus:

---
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. Skill registrieren

In planner_agent/agent.py wird der Skill aus dem Verzeichnis geladen und den Tools des Agenten hinzugefügt. So sieht der Code aus:

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

Mit Agent Skills können Sie die Tool-Implementierung von der Tool-Anweisung entkoppeln. So bleibt Ihr Systemprompt übersichtlich und der Agent hat nur dann den genauen Kontext, den er benötigt, wenn er diese spezielle Funktion verwendet.

10. Agent testen

1. Starten Sie den Agenten lokal:

uv run adk run planner_agent

2. Stellen Sie eine Frage zu den Stadtregeln: [user]: What are the rules for running a race on the Las Vegas strip?

Der Agent ruft das Tool get_local_and_traffic_rules auf, führt eine Vektorsuche in AlloyDB durch und gibt eine Antwort basierend auf den offiziellen Regel-Chunks zurück, die von Spark verarbeitet wurden.

11. Agent bereitstellen

Auf der Agent Platform bereitstellen

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

12. Bereinigen

Löschen Sie die in diesem Codelab erstellten Ressourcen, um laufende Kosten zu vermeiden.

AlloyDB-Cluster löschen

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

Agent Runtime-App löschen

Sie können die Reasoning Engine-Instanz über die Console oder mit dem Befehl gcloud löschen (wenn Sie den Ressourcennamen haben). Verwenden Sie zur Vereinfachung die Console:

1. Rufen Sie die Seite Agent Runtime auf.
2. Wählen Sie planner_agent aus und klicken Sie rechts auf das Dreipunkt-Menü.
3. Klicken Sie auf Löschen.

13. Glückwunsch

Glückwunsch! Sie haben einen ADK-Agenten erfolgreich mit erweiterten Funktionen für Gemerkte Informationen und Datenfundierung verbessert.

Lerninhalte

• Zustandsbehaftete Agenten: Integration von Agent Platform-Sitzungen, um den Unterhaltungskontext beizubehalten.
• Langfristiges Lernen: Verknüpfen einer Agent Platform Memory Bank , damit der Agent aus Nutzerinteraktionen lernen kann.
• Datenaufnahme: Verwenden der Spark Lightning Engine und von Document AI zum Verarbeiten unstrukturierter Dokumente.
• RAG: Erstellen eines Vektorsuchsystems in AlloyDB, um den Agenten auf reale Regeln abzustimmen.

Nächste Schritte

• Weitere Informationen zur verwalteten Bereitstellung finden Sie in der Dokumentation zur Agent Platform.
• Weitere Informationen zu erweiterten RAG-Mustern finden Sie unter AlloyDB Vector Search.
• Skalieren Sie Ihre Aufnahmepipelines mit Dataproc Serverless.