Guide de démarrage rapide de Memory Bank avec ADK

Agent Platform Memory Bank permet à vos agents de gérer les souvenirs à long terme d'une session à l'autre. Lorsqu'il est utilisé avec l'Agent Development Kit (ADK), votre agent peut orchestrer automatiquement les appels à Memory Bank pour stocker et récupérer des souvenirs en fonction des interactions des utilisateurs.

Ce document explique comment créer un agent ADK, le configurer pour qu'il utilise Memory Bank et interagir avec lui pour générer et accéder à des souvenirs.

Pour savoir comment effectuer des appels directs à l'API sans ADK, consultez le guide de démarrage rapide de l'API Memory Bank.

Gérer les informations mémorisées avec le service de mémoire ADK et Memory Bank

VertexAiMemoryBankService est un wrapper ADK autour de Memory Bank, défini par BaseMemoryService d'ADK. Vous pouvez définir des rappels et des outils qui interagissent avec le service de mémoire pour lire et écrire des souvenirs.

L'interface VertexAiMemoryBankService comprend :

• memory_service.add_session_to_memory déclenche une requête GenerateMemories vers Memory Bank en utilisant tous les événements du adk.Session fourni comme contenu source. Vous pouvez orchestrer les appels à cette méthode à l'aide de callback_context.add_session_to_memory dans vos rappels.

  from google.adk.agents.callback_context import CallbackContext
  
  async def add_session_to_memory_callback(callback_context: CallbackContext):
      await callback_context.add_session_to_memory()
      return None
  

• memory_service.add_events_to_memory, qui déclenche une requête GenerateMemories vers Memory Bank à l'aide d'un sous-ensemble d'événements. Vous pouvez orchestrer les appels à cette méthode à l'aide de callback_context.add_events_to_memory dans vos rappels.

  from google.adk.agents.callback_context import CallbackContext
  
  async def add_events_to_memory_callback(callback_context: CallbackContext):
      await callback_context.add_events_to_memory(events=callback_context.session.events[-5:-1])
      return None
  

• memory_service.search_memory déclenche une requête RetrieveMemories vers la banque de mémoire pour récupérer les souvenirs pertinents pour les user_id et app_name actuels. Vous pouvez orchestrer les appels à cette méthode à l'aide d'outils de mémoire intégrés (LoadMemoryTool ou PreloadMemoryTool) ou d'un outil personnalisé qui appelle tool_context.search_memory.

Avant de commencer

Pour suivre les étapes décrites dans ce tutoriel, vous devez d'abord suivre celles de la section "Premiers pas" de la page "Configurer Memory Bank".

Définir des variables d'environnement

Pour utiliser ADK, définissez vos variables d'environnement :

import os

os.environ["GOOGLE_GENAI_USE_ENTERPRISE"] = "TRUE"
os.environ["GOOGLE_CLOUD_PROJECT"] = "PROJECT_ID"
os.environ["GOOGLE_CLOUD_LOCATION"] = "LOCATION"

Remplacez les éléments suivants :

• PROJECT_ID : ID du projet.
• LOCATION : votre région. Consultez les régions disponibles pour la Banque de souvenirs.

Créer votre agent ADK

Pour créer un agent avec mémoire, configurez des outils et des rappels qui orchestrent les appels à votre service de mémoire.

Définir un rappel de génération de mémoire

Pour orchestrer les appels de génération de mémoire, créez une fonction de rappel qui déclenche la génération de mémoire. Vous pouvez envoyer un sous-ensemble d'événements (avec callback_context.add_events_to_memory) ou tous les événements d'une session (avec callback_context.add_session_to_memory) pour qu'ils soient traités en arrière-plan :

from google.adk.agents.callback_context import CallbackContext

async def generate_memories_callback(callback_context: CallbackContext):
    # Option 1 (Recommended): Send events to Memory Bank for memory generation,
    # which is ideal for incremental processing of events.
    await callback_context.add_events_to_memory(
      events=callback_context.session.events[-5:-1])

    # Option 2: Send the full session to Memory Bank for memory generation.
    # It's recommended to only call this at the end of a session to minimize
    # how many times a single event is re-processed.
    await callback_context.add_session_to_memory()

    return None

Définir un outil de récupération de mémoire

Lorsque vous développez votre agent ADK, incluez un outil de mémoire qui contrôle quand l'agent récupère les souvenirs et comment ils sont inclus dans la requête.

Si vous utilisez PreloadMemoryTool, votre agent récupère les souvenirs au début de chaque tour et les inclut dans l'instruction système, ce qui est utile pour établir un contexte de base sur l'utilisateur. Si vous utilisez LoadMemoryTool, le modèle appellera cet outil lorsqu'il décidera que des souvenirs sont nécessaires pour répondre à la requête de l'utilisateur.

from google import adk
from google.adk.tools.load_memory_tool import LoadMemoryTool
from google.adk.tools.preload_memory_tool import PreloadMemoryTool

memory_retrieval_tools = [
  # Option 1: Retrieve memories at the start of every turn.
  PreloadMemoryTool(),
  # Option 2: Retrieve memories via tool calls. The model will only call this tool
  # when it decides that memories are necessary to respond to the user query.
  LoadMemoryTool()
]

agent = adk.Agent(
    model="gemini-3.5-flash",
    name='stateful_agent',
    instruction="""You are a Vehicle Voice Agent, designed to assist users with information and in-vehicle actions.

1.  **Direct Action:** If a user requests a specific vehicle function (e.g., "turn on the AC"), execute it immediately using the corresponding tool. You don't have the outcome of the actual tool execution, so provide a hypothetical tool execution outcome.
2.  **Information Retrieval:** Respond concisely to general information requests with your own knowledge (e.g., restaurant recommendation).
3.  **Clarity:** When necessary, try to seek clarification to better understand the user's needs and preference before taking an action.
4.  **Brevity:** Limit responses to under 30 words.
""",
    tools=memory_retrieval_tools,
    after_agent_callback=generate_memories_callback
)

Vous pouvez également créer votre propre outil personnalisé pour récupérer des souvenirs. Cela est utile lorsque vous souhaitez fournir des instructions à votre agent sur le moment où récupérer des souvenirs :

from google import adk
from google.adk.tools import ToolContext, FunctionTool

async def search_memories(query: str, tool_context: ToolContext):
  """Query this tool when you need to fetch information about user preferences."""
  return await tool_context.search_memory(query)

agent = adk.Agent(
    model="gemini-3.5-flash",
    name='stateful_agent',
    instruction="""...""",
    tools=[FunctionTool(func=search_memories)],
    after_agent_callback=generate_memories_callback
)

Définir un service de mémoire Memory Bank ADK et une instance Memory Bank

Une fois que vous avez créé votre agent avec mémoire, vous devez l'associer à un service de mémoire. La procédure de configuration de votre service de mémoire ADK dépend de l'emplacement d'exécution de votre agent ADK. Le runtime orchestre l'exécution de vos agents, outils et rappels.

Créer une instance Memory Bank

Vous devez d'abord créer une instance Memory Bank. Cette étape est facultative si vous utilisez Agent Runtime pour déployer votre agent. Pour en savoir plus sur la personnalisation du comportement de votre Memory Bank, consultez la section Configurer votre instance Memory Bank sur la page "Configurer Memory Bank".

import agentplatform

client = agentplatform.Client(
  project="PROJECT_ID",
  location="LOCATION"
)
# If you don't have a Memory Bank instance already, create a
# Memory Bank instance using the default configuration.
memory_bank = client.memory_banks.create()

# Optionally, print out the resource name. You will need the
# resource name if you want to interact with your Memory Bank instance later on.
print(memory_bank.name)

memory_bank_id = memory_bank.name.split("/")[-1]

Remplacez les éléments suivants :

• PROJECT_ID : ID du projet.
• LOCATION : votre région. Consultez les régions prises en charge pour Memory Bank.

Créer un environnement d'exécution ADK

Transmettez l'ID d'instance Memory Bank aux scripts d'exécution ou de déploiement afin que votre agent utilise Memory Bank comme service de mémoire ADK.

import asyncio

from google.adk.memory import VertexAiMemoryBankService
from google.adk.sessions import VertexAiSessionService
from google.genai import types

memory_service = VertexAiMemoryBankService(
    project="PROJECT_ID",
    location="LOCATION",
    agent_engine_id="MEMORY_BANK_ID",
)

# You can use any ADK session service. This example uses Sessions.
session_service = VertexAiSessionService(
    project="PROJECT_ID",
    location="LOCATION",
    agent_engine_id="SESSIONS_ID",
)

runner = adk.Runner(
    agent=agent,
    app_name="APP_NAME",
    session_service=session_service,
    memory_service=memory_service
)

async def call_agent(query, session, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  events = runner.run_async(
    user_id=user_id, session_id=session, new_message=content)

  async for event in events:
      if event.is_final_response():
          final_response = event.content.parts[0].text
          print("Agent Response: ", final_response)

import asyncio

import agentplatform
from agentplatform.frameworks import AdkApp

client = agentplatform.Client(
  project="PROJECT_ID",
  location="LOCATION"
)

adk_app = AdkApp(agent=agent)

# Create a new resource with your agent deployed to Agent Runtime.
# The Agent Runtime instance will also include an empty Memory Bank instance.
runtime_instance = client.runtimes.create(
      agent=adk_app,
      config={
            "staging_bucket": "STAGING_BUCKET",
            "requirements": ["google-cloud-agentplatform[runtimes,adk]"]
      }
)

# Alternatively, update an existing resource to deploy your agent to Agent Platform.
# Your agent will have access to the Runtime instance's existing memories.
runtime_instance = client.runtimes.update(
      name=agent_engine.api_resource.name,
      agent=adk_app,
      config={
            "staging_bucket": "STAGING_BUCKET",
            "requirements": ["google-cloud-agentplatform[runtimes,adk]"]
      }
)

async def call_agent(query, session_id, user_id):
    async for event in agent_engine.async_stream_query(
        user_id=user_id,
        session_id=session_id,
        message=query,
    ):
        print(event)

def memory_bank_service_builder():
    return VertexAiMemoryBankService(
        project="PROJECT_ID",
        location="LOCATION",
        agent_engine_id="MEMORY_BANK_ID"
    )

adk_app = AdkApp(
      agent=adk_agent,
      # Override the default memory service.
      memory_service_builder=memory_bank_service_builder
)

async def call_agent(query, session_id, user_id):
  # adk_app is a local agent. If you want to deploy it to Agent Runtime,
  # use `client.runtimes.create(...)` or `client.runtimes.update(...)`
  # and call the returned Agent Runtime instance instead.
  async for event in adk_app.async_stream_query(
      user_id=user_id,
      session_id=session_id,
      message=query,
  ):
      print(event)

adk deploy cloud_run \
    ...
    --memory_service_uri=agentengine://AGENT_ENGINE_ID

adk deploy gke \
    ...
    --memory_service_uri=agentengine://AGENT_ENGINE_ID

export GOOGLE_CLOUD_PROJECT="PROJECT_ID"
export GOOGLE_CLOUD_LOCATION="LOCATION"

adk web --memory_service_uri=agentengine://MEMORY_BANK_ID

Interagir avec votre agent

Une fois votre agent défini et la Memory Bank configurée, vous pouvez interagir avec votre agent. Si vous avez fourni un rappel pour déclencher la génération de mémoire lors de l'initialisation de votre agent, la génération de mémoire sera déclenchée chaque fois que l'agent sera invoqué.

Les informations mémorisées seront stockées à l'aide du champ d'application {"user_id": USER_ID, "app_name": APP_NAME} correspondant à l'ID utilisateur et au nom de l'application utilisés pour exécuter votre agent.

La méthode d'interaction avec votre agent dépend de son environnement d'exécution :

# Use `asyncio.run(session_service.create(...))` if you're running this
# code as a standard Python script.
session = await session_service.create_session(
    app_name="APP_NAME",
    user_id="USER_ID"
)

# Use `asyncio.run(call_agent(...))` if you're running this code as a
# standard Python script.
await call_agent(
    "Can you fix the temperature?",
    session.id,
    "USER_ID"
)

# Use `asyncio.run(agent_engine.async_create_session(...))` if you're
# running this code as a standard Python script.
session = await agent_engine.async_create_session(user_id="USER_ID")

# Use `asyncio.run(call_agent(...))` if you're running this code as a
# standard Python script.
await call_agent(
    "Can you fix the temperature?",
    session.get("id"),
    "USER_ID"
)

Exemple d'interaction

Première session

Si vous avez utilisé PreloadMemoryTool, l'agent tentera de récupérer des informations mémorisées au début de chaque tour pour accéder aux préférences que l'utilisateur lui a précédemment communiquées. Lors de la première interaction de l'agent avec l'utilisateur, aucune mémoire n'est disponible pour être récupérée. L'agent ne connaît donc aucune préférence de l'utilisateur, comme sa température préférée, comme le montre l'exemple suivant :

1. Premier tour :

   • Utilisateur : "Peux-tu régler la température ?"

   • (Appel d'outil) : ADK tente de récupérer des souvenirs, mais aucun n'est disponible.

   • Modèle : "Quelle température préférez-vous ?"

   • (Rappel) : ADK déclenche la génération de mémoire. Aucune information n'est extraite.

2. Deuxième tour :

   • Utilisateur : Je me sens bien à 21 degrés.

   • (Appel d'outil) : ADK tente de récupérer des infos mémorisées, mais aucune n'est disponible.

   • Modèle : OK, j'ai réglé la température sur 21 degrés.

   • (Rappel) : ADK déclenche la génération de mémoire. L'information mémorisée "J'aime quand la température est de 21 degrés" est créée.

Deuxième séance

La mémoire extraite sera disponible pour la prochaine session avec le même nom d'application et le même ID utilisateur. Si l'utilisateur fournit des informations similaires ou contradictoires à celles des souvenirs existants, les nouvelles informations seront consolidées avec les souvenirs existants.

1. Premier tour

   • Utilisateur : Règle la température. C'est tellement inconfortable !

   • (Appel d'outil) : ADK tente de récupérer des informations mémorisées. La mémoire "J'aime la température de 22 degrés" est récupérée.

   • Modèle : D'accord, j'ai réglé la température sur 21 degrés.

   • (Rappel) : ADK déclenche la génération de mémoire. Aucun souvenir n'est extrait, car l'utilisateur n'a partagé aucun élément pertinent à conserver.

2. Deuxième tour

   • Utilisateur : En fait, je préférerais qu'il fasse plus chaud le matin.

   • (Appel d'outil) : ADK tente de récupérer des informations mémorisées. La mémoire "J'aime la température de 22 degrés" est récupérée.

   • Modèle : OK, j'ai augmenté la température.

   • (Rappel) : l'ADK déclenche la génération de mémoire. La mémoire existante "J'aime la température à 22 degrés" est mise à jour et devient "J'aime généralement la température à 22 degrés, mais je préfère qu'il fasse plus chaud le matin".

Utiliser une Memory Bank multirégionale avec un environnement d'exécution régional

Lorsque vous utilisez l'environnement d'exécution avec Memory Bank intégré, votre agent et Memory Bank sont déployés par défaut dans la même région. Toutefois, vous pouvez les dissocier pour utiliser un Memory Bank multirégional (par exemple, us) avec un environnement d'exécution régional (par exemple, us-central1). Cette configuration vous permet de gérer un Memory Bank centralisé sur différents déploiements régionaux.

Pour utiliser un Memory Bank multirégional, vous devez remplacer le générateur de service de mémoire ADK par défaut afin de pointer vers l'emplacement multirégional et l'ID Memory Bank correspondant.

import agentplatform
from google.adk.memory import VertexAiMemoryBankService
from agentplatform.frameworks import AdkApp

# Create the Memory Bank instance in a multi-region location (for example, 'us')
client_mb = agentplatform.Client(project="PROJECT_ID", location="us")
memory_bank = client_mb.memory_banks.create()
memory_bank_id = memory_bank.name.split("/")[-1]


# Point your memory service to the 'us' location, 'us' Memory Bank
def memory_bank_service_builder():
    return VertexAiMemoryBankService(
        project="PROJECT_ID",
        location="us",
        agent_engine_id=memory_bank_id
    )

# Create the AdkApp with the overridden builder
adk_app = AdkApp(
    agent=agent,
    memory_service_builder=memory_bank_service_builder
)

# Deploy the runtime to a specific region (for example, 'us-central1')
client_runtime = agentplatform.Client(project="PROJECT_ID", location="us-central1")
runtime_instance = client_runtime.runtimes.create(
    agent=adk_app,
    config={
        "staging_bucket": "STAGING_BUCKET",
        "requirements": ["google-cloud-agentplatform[runtimes,adk]"]
    }
)

Remplacez les éléments suivants :

• PROJECT_ID : ID du projet.
• STAGING_BUCKET : bucket Cloud Storage à utiliser pour l'environnement de préproduction de votre Agent Runtime.

Effectuer un nettoyage

Pour nettoyer toutes les ressources utilisées dans ce projet, vous pouvez supprimer le projet Cloud de Confiance que vous avez utilisé pour ce guide de démarrage rapide.

Vous pouvez également supprimer les ressources individuelles que vous avez créées, comme suit :

1. Utilisez l'exemple de code suivant pour supprimer l'instance Agent Runtime, ce qui supprime également toutes les sessions ou mémoires appartenant à ce runtime.

   client_runtime.runtimes.delete(name=runtime_instance.api_resource.name, force=True)
   client_mb.memory_banks.delete(name=memory_bank.name, force=True)
   

2. Supprimez tous les fichiers créés localement.

Étapes suivantes