Guía de inicio rápido de Memory Bank con el ADK

Agent Platform Memory Bank permite que tus agentes administren memorias a largo plazo en varias sesiones. Cuando se usa con el Kit de desarrollo de agentes (ADK), tu agente puede organizar automáticamente las llamadas a Memory Bank para almacenar y recuperar recuerdos según las interacciones del usuario.

En este documento, se explica cómo crear un agente del ADK, configurarlo para usar Memory Bank y, luego, interactuar con él para generar recuerdos y acceder a ellos.

Para obtener información sobre cómo realizar llamadas directas a la API sin el ADK, consulta la guía de inicio rápido de la API de Memory Bank.

Administra memorias con el servicio de memoria de ADK y Memory Bank

VertexAiMemoryBankService es un wrapper del ADK alrededor de Memory Bank que se define con BaseMemoryService del ADK. Puedes definir devoluciones de llamada y herramientas que interactúen con el servicio de memoria para leer y escribir recuerdos.

La interfaz de VertexAiMemoryBankService incluye lo siguiente:

• memory_service.add_session_to_memory activa una solicitud de GenerateMemories a Memory Bank con todos los eventos del adk.Session proporcionado como contenido fuente. Puedes organizar las llamadas a este método con callback_context.add_session_to_memory en tus devoluciones de llamada.

  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, que activa una solicitud de GenerateMemories a Memory Bank con un subconjunto de eventos. Puedes coordinar llamadas a este método con callback_context.add_events_to_memory en tus devoluciones de llamada.

  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 activa una solicitud de RetrieveMemories a Memory Bank para recuperar las memorias pertinentes para el user_id y el app_name actuales. Puedes coordinar llamadas a este método con herramientas de memoria integradas (LoadMemoryTool o PreloadMemoryTool) o una herramienta personalizada que invoque tool_context.search_memory.

Antes de comenzar

Para completar los pasos que se muestran en este instructivo, primero debes seguir los pasos de la sección de introducción de la página Configura Memory Bank.

Configura las variables de entorno

Para usar el ADK, configura tus variables de entorno:

import os

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

Reemplaza lo siguiente:

• PROJECT_ID: ID del proyecto
• LOCATION: Tu región. Consulta las regiones admitidas para Memory Bank.

Crea tu agente del ADK

Para crear un agente con memoria, configura herramientas y devoluciones de llamada que coordinen las llamadas a tu servicio de memoria.

Cómo definir una devolución de llamada de generación de memoria

Para coordinar las llamadas para la generación de memoria, crea una función de devolución de llamada que active la generación de memoria. Puedes enviar un subconjunto de eventos (con callback_context.add_events_to_memory) o todos los eventos de una sesión (con callback_context.add_session_to_memory) para que se procesen en segundo plano:

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

Define una herramienta de recuperación de memoria

Cuando desarrolles tu agente del ADK, incluye una herramienta de memoria que controle cuándo el agente recupera recuerdos y cómo se incluyen en la instrucción.

Si usas PreloadMemoryTool, tu agente recuperará recuerdos al comienzo de cada turno y los incluirá en la instrucción del sistema, lo que es útil para establecer un contexto de referencia sobre el usuario. Si usas LoadMemoryTool, el modelo llamará a esta herramienta cuando decida que los recuerdos son necesarios para responder la consulta del usuario.

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
)

Como alternativa, puedes crear tu propia herramienta personalizada para recuperar recuerdos, lo que es útil cuando quieres proporcionar instrucciones a tu agente sobre cuándo recuperar recuerdos:

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
)

Define un servicio de memoria de Memory Bank del ADK y una instancia de Memory Bank

Después de crear tu agente con memoria, debes vincularlo a un servicio de memoria. El proceso de configuración del servicio de memoria de ADK depende de dónde se ejecute el agente de ADK. El tiempo de ejecución coordina la ejecución de tus agentes, herramientas y devoluciones de llamada.

Crea una instancia de Memory Bank

Primero, debes crear una instancia de Memory Bank. Este paso es opcional si usas Agent Runtime para implementar tu agente. Para obtener más información sobre cómo personalizar el comportamiento de tu banco de memoria, consulta la sección Configura tu instancia de banco de memoria en la página Configurar banco de memoria.

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]

Reemplaza lo siguiente:

• PROJECT_ID: ID del proyecto
• LOCATION: Tu región. Consulta las regiones admitidas para Memory Bank.

Crea un entorno de ejecución del ADK

Pasa el ID de la instancia de Memory Bank al entorno de ejecución o a las secuencias de comandos de implementación para que tu agente use Memory Bank como el servicio de memoria del 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

Interactúa con el agente

Después de definir tu agente y configurar el banco de memoria, puedes interactuar con él. Si proporcionaste una devolución de llamada para activar la generación de memoria cuando inicializaste tu agente, la generación de memoria se activará cada vez que se invoque el agente.

Las memorias se almacenarán con el permiso {"user_id": USER_ID, "app_name": APP_NAME} correspondiente al ID de usuario y al nombre de la app que se usaron para ejecutar el agente.

El método para interactuar con tu agente depende de su entorno de ejecución:

# 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"
)

Interacción de ejemplo

Primera sesión

Si usaste PreloadMemoryTool, el agente intentará recuperar las memorias al comienzo de cada turno para acceder a las preferencias que el usuario comunicó anteriormente al agente. Durante la primera interacción del agente con el usuario, no hay recuerdos disponibles para recuperar. Por lo tanto, el agente no conoce las preferencias del usuario, como su temperatura preferida, como se muestra en el siguiente ejemplo:

1. Primer turno:

   • Usuario: "¿Puedes arreglar la temperatura?"

   • (Llamada a la herramienta): El ADK intenta recuperar recuerdos, pero no hay ninguno disponible.

   • Modelo: "¿Qué temperatura prefieres?"

   • (Devolución de llamada): El ADK activa la generación de memoria. No se extraen memorias.

2. Segundo turno:

   • Usuario: Me siento cómodo a 21 grados.

   • (Llamada a la herramienta): El ADK intenta recuperar recuerdos, pero no hay ninguno disponible.

   • Modelo: De acuerdo. Actualicé la temperatura a 22 °C.

   • (Devolución de llamada): El ADK activa la generación de memoria. Se crea el recuerdo "Me gusta la temperatura de 21 grados".

Segunda sesión

La memoria extraída estará disponible para la próxima sesión con el mismo nombre de la app y el mismo ID de usuario. Si el usuario proporciona información similar o contradictoria a los recuerdos existentes, la información nueva se consolidará con los recuerdos existentes.

1. Primer turno

   • Usuario: Fija la temperatura. ¡Es tan incómodo!

   • (Llamada a la herramienta): El ADK intenta recuperar recuerdos. Se recupera el recuerdo "Me gusta la temperatura de 21 grados".

   • Modelo: De acuerdo. Actualicé la temperatura a 22 °C.

   • (Devolución de llamada): El ADK activa la generación de memoria. No se extraen recuerdos porque el usuario no compartió nada significativo para conservar.

2. Segundo turno

   • Usuario: En realidad, prefiero que haga más calor por las mañanas.

   • (Llamada a la herramienta): El ADK intenta recuperar recuerdos. Se recupera el recuerdo "Me gusta la temperatura de 21 grados".

   • Modelo: De acuerdo, subí la temperatura.

   • (Devolución de llamada): El ADK activa la generación de memoria. La memoria existente "Me gusta la temperatura a 22 grados" se actualiza a "Por lo general, me gusta la temperatura a 22 grados, pero me gusta que esté más cálida por las mañanas".

Usa un Memory Bank multirregional con un entorno de ejecución regional

Cuando usas Runtime con Memory Bank integrado, tu agente y Memory Bank se implementan en la misma región de forma predeterminada. Sin embargo, puedes desvincularlos para usar un Memory Bank multirregional (por ejemplo, us) con un entorno de ejecución regional (por ejemplo, us-central1). Esta configuración te permite mantener un Memory Bank central en diferentes implementaciones regionales.

Para usar un Memory Bank multirregional, debes anular el compilador de servicios de memoria predeterminado del ADK para que apunte a la ubicación multirregional y al ID de Memory Bank correspondiente.

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]"]
    }
)

Reemplaza lo siguiente:

• PROJECT_ID: ID del proyecto
• STAGING_BUCKET: Es el bucket de Cloud Storage que usarás para preparar tu Agent Runtime.

Realiza una limpieza

Para limpiar todos los recursos que se usaron en este proyecto, puedes borrar el proyecto de Cloud de Confiance que usaste para la guía de inicio rápido.

De lo contrario, puedes borrar los recursos individuales que creaste en este instructivo de la siguiente manera:

1. Usa la siguiente muestra de código para borrar la instancia de Agent Runtime, lo que también borra cualquier sesión o memoria que pertenezca a ese entorno de ejecución.

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

2. Borra los archivos creados de forma local.

¿Qué sigue?