使用 ADK 快速入門 Memory Bank

Agent Platform Memory Bank 可讓代理管理跨工作階段的長期記憶。搭配使用 Agent Development Kit (ADK) 時,代理程式可以自動協調對記憶庫的呼叫,根據使用者互動儲存及擷取記憶。

本文說明如何建立 ADK 代理程式、設定代理程式以使用 Memory Bank,以及與代理程式互動來產生及存取記憶。

如要瞭解如何在不使用 ADK 的情況下直接呼叫 API,請參閱 Memory Bank API 快速入門

使用 ADK 記憶體服務和 Memory Bank 管理記憶體

VertexAiMemoryBankService 是 ADK 針對 Memory Bank 的包裝函式,由 ADK 的 BaseMemoryService定義。 您可以定義與記憶體服務互動的回呼和工具,以便讀取及寫入記憶體。

VertexAiMemoryBankService 介面包含:

  • memory_service.add_session_to_memory 會使用提供的 adk.Session 中的所有事件做為來源內容,觸發對 Memory Bank 的 GenerateMemories 要求。您可以在回呼中使用 callback_context.add_session_to_memory,協調對這個方法的呼叫。

    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,這會使用部分事件觸發對 Memory Bank 的 GenerateMemories 要求。您可以在回呼中使用 callback_context.add_events_to_memory,協調對這個方法的呼叫。

    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 會觸發 RetrieveMemories 要求,向 Memory Bank 擷取目前 user_idapp_name 的相關記憶。您可以使用內建記憶體工具 (LoadMemoryToolPreloadMemoryTool) 或叫用 tool_context.search_memory 的自訂工具,協調對這個方法的呼叫。

事前準備

如要完成本教學課程中示範的步驟,請先按照「設定 Memory Bank」頁面的「開始使用」一節中的步驟操作。

設定環境變數

如要使用 ADK,請設定環境變數:

import os

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

更改下列內容:

  • PROJECT_ID:專案 ID。
  • LOCATION:您的區域。如需支援 Memory Bank 的區域,請參閱這篇文章

建立 ADK 代理

如要建立啟用記憶體功能的代理程式,請設定工具和回呼,協調對記憶體服務的呼叫。

定義記憶生成回呼

如要協調記憶體生成呼叫,請建立可觸發記憶體生成的回呼函式。您可以傳送部分事件 (使用 callback_context.add_events_to_memory) 或工作階段中的所有事件 (使用 callback_context.add_session_to_memory),在背景處理:

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

定義記憶體擷取工具

開發 ADK 代理程式時,請加入記憶體工具,控管代理程式擷取記憶體的時間,以及記憶體納入提示的方式。

如果您使用 PreloadMemoryTool,代理程式會在每個回合開始時擷取記憶,並將擷取的記憶納入系統指令,有助於建立使用者的基準背景資訊。如果您使用 LoadMemoryTool,模型會在判斷需要記憶內容才能回答使用者查詢時,呼叫這項工具。

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
)

或者,您也可以建立自己的自訂工具來擷取記憶內容,這有助於在您想向代理程式提供擷取記憶內容的時機相關指令時:

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
)

定義 ADK Memory Bank 記憶體服務和 Memory Bank 執行個體

建立啟用記憶體功能的代理程式後,您必須將其連結至記憶體服務。設定 ADK 記憶體服務的程序取決於 ADK 代理程式的執行位置。執行階段會自動調度管理代理、工具和回呼的執行作業。

建立 Memory Bank 執行個體

您必須先建立 Memory Bank 執行個體。如果您使用 Agent Runtime 部署代理,則可略過這個步驟。如要進一步瞭解如何自訂 Memory Bank 行為,請參閱「設定 Memory Bank」頁面的「設定 Memory Bank 執行個體」一節。

import vertexai

client = vertexai.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.agent_engines.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.api_resource.name)

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

更改下列內容:

  • PROJECT_ID:專案 ID。
  • LOCATION:您的區域。如要瞭解 Memory Bank 支援的區域,請參閱這篇文章

建立 ADK 執行階段

將 Memory Bank 執行個體 ID 傳遞至執行階段或部署指令碼,讓代理程式將 Memory Bank 做為 ADK 記憶體服務。

本機執行器

adk.Runner 通常用於本機環境,例如 Colab。在這種情況下,您需要直接建立記憶體服務和執行器。

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)

更改下列內容:

  • PROJECT_ID:專案 ID。
  • LOCATION:您的區域。如要瞭解支援 Memory Bank 的區域,請參閱這篇文章
  • APP_NAME:ADK 應用程式名稱。應用程式名稱會納入產生的記憶內容的 scope 字典,確保記憶內容不會與其他使用者和應用程式混淆。
  • MEMORY_BANK_ID:Memory Bank 執行個體 ID。例如,456 位於 projects/my-project/locations/us-central1/reasoningEngines/456 中。
  • SESSIONS_ID:Agent Platform Sessions 執行個體 ID。例如,789 位於 projects/my-project/locations/us-central1/reasoningEngines/789 中。

Gemini Enterprise Agent Platform 的 Agent Runtime

Agent Runtime ADK 範本 (AdkApp) 可用於本機,也可將 ADK 代理部署至 Agent Runtime。在 Agent Platform 上部署時,Memory Bank ADK 範本會使用 VertexAiMemoryBankService 做為預設記憶體服務。因此,您只需一個步驟,即可建立 Memory Bank 執行個體並部署至執行階段。

如要進一步瞭解如何設定 Memory Bank 執行個體,包括如何自訂 Memory Bank 的行為,請參閱「設定 Memory Bank」。

使用下列程式碼,將啟用記憶體的 ADK 代理部署至 Agent Runtime:

import asyncio

import vertexai
from vertexai.agent_engines import AdkApp

client = vertexai.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.
agent_engine = client.agent_engines.create(
      agent_engine=adk_app,
      config={
            "staging_bucket": "STAGING_BUCKET",
            "requirements": ["google-cloud-aiplatform[agent_engines,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.
agent_engine = client.agent_engines.update(
      name=agent_engine.api_resource.name,
      agent_engine=adk_app,
      config={
            "staging_bucket": "STAGING_BUCKET",
            "requirements": ["google-cloud-aiplatform[agent_engines,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)

更改下列內容:

  • PROJECT_ID:專案 ID。
  • LOCATION:您的區域。如需支援 Memory Bank 的區域,請參閱這篇文章
  • STAGING_BUCKET:用於暫存 Agent Runtime 的 Cloud Storage bucket。

在本機執行時,ADK 範本會使用 InMemoryMemoryService 做為預設記憶體服務。不過,您可以覆寫預設記憶體服務,改用 VertexAiMemoryBankService

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.agent_engines.create(...)` or `client.agent_engines.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)

更改下列內容:

  • PROJECT_ID:專案 ID。
  • LOCATION:您的區域。如需支援 Memory Bank 的區域,請參閱這篇文章
  • MEMORY_BANK_ID:用於 Memory Bank 的 Memory Bank 執行個體 ID。例如,456 位於 projects/my-project/locations/us-central1/reasoningEngines/456 中。

Cloud Run

如要將代理部署至 Cloud Run,請參閱 ADK 說明文件中的操作說明,瞭解如何定義要部署至 Cloud Run 的代理。

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

Google Kubernetes Engine (GKE)

如要將代理程式部署至 GKE,請參閱 ADK 說明文件中的操作說明,瞭解如何定義要部署至 GKE 的代理程式。

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

ADK Web

ADK 網頁介面可讓您直接在瀏覽器中測試代理程式。

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

adk web --memory_service_uri=agentengine://MEMORY_BANK_ID

更改下列內容:

  • PROJECT_ID:專案 ID。
  • LOCATION:您的區域。如需支援 Memory Bank 的區域,請參閱這篇文章
  • MEMORY_BANK_ID:Memory Bank 執行個體 ID。例如,456 位於 projects/my-project/locations/us-central1/reasoningEngines/456 中。

與代理程式互動

定義代理程式並設定 Memory Bank 後,即可與代理程式互動。如果您在初始化代理程式時,提供回呼來觸發記憶體生成,每次叫用代理程式時,系統都會觸發記憶體生成。

系統會使用與執行代理程式時所用的使用者 ID 和應用程式名稱相應的範圍 {"user_id": USER_ID, "app_name": APP_NAME} 儲存記憶。

與代理互動的方法取決於執行環境:

本機執行器

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

更改下列內容:

  • APP_NAME:執行器的應用程式名稱。
  • USER_ID:使用者的 ID。系統會使用這個不透明的 ID,為這個工作階段產生的回憶內容建立索引。產生的記憶體範圍會儲存為 {"user_id": "USER_ID"}

Agent Runtime

使用 ADK 範本時,您可以呼叫 Agent Runtime,與記憶體和工作階段互動。

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

更改下列內容:

  • USER_ID:使用者的 ID。這個工作階段產生的記憶體會以這個不透明的 ID 做為鍵。系統會將生成回憶集錦的範圍儲存為 {"user_id": "USER_ID"}

Cloud Run

請參閱 ADK Cloud Run 部署說明文件的「測試代理」一節。

GKE

請參閱 ADK GKE 部署說明文件的「測試代理程式」一節。

ADK Web

如要使用 ADK Web,請前往 http://localhost:8000 的本機伺服器。

根據預設,ADK Web 會將使用者 ID 設為 user。如要覆寫預設使用者 ID,請在查詢參數中加入 userId,例如 http://localhost:8000?userId=YOUR_USER_ID

詳情請參閱 ADK 說明文件中的 ADK Web 頁面。

互動樣本

第一次工作階段

如果您使用 PreloadMemoryTool,代理程式會在每一回合開始時嘗試擷取記憶,存取使用者先前向代理程式傳達的偏好設定。服務專員與使用者初次互動時,系統不會擷取任何記憶內容。因此,代理程式不知道任何使用者偏好設定,例如偏好的溫度,如下列範例所示:

  1. 第一回合:

    • 使用者:「可以調整溫度嗎?」

    • (工具呼叫)ADK 嘗試擷取記憶內容,但沒有可用的記憶內容。

    • 模型:「你喜歡的溫度是幾度?」

    • (回呼)ADK 會觸發記憶體生成作業。不會擷取任何記憶。

  2. 第二回合:

    • 使用者:I'm comfortable at 71 degrees. (71 度很舒適。)

    • (工具呼叫)ADK 嘗試擷取記憶內容,但沒有可用的記憶內容。

    • 模型:好的,已將溫度參數調為 22 度。

    • (回呼)ADK 會觸發記憶體生成作業。系統會建立「我喜歡 71 度的溫度」記憶內容。

第二個工作階段

系統會將擷取的記憶體用於下一個相同應用程式名稱和使用者 ID 的工作階段。如果使用者提供與現有記憶內容相似或矛盾的資訊,系統會將新資訊與現有記憶內容合併

  1. 第一回合

    • 使用者:調整溫度。這太不舒服了!

    • (工具呼叫)ADK 嘗試擷取記憶內容。系統會擷取「我喜歡 22 度的溫度」記憶體。

    • 模型:好的,已將溫度參數調為 22 度。

    • (回呼)ADK 會觸發記憶體生成作業。使用者未分享任何值得保留的內容,因此系統不會擷取記憶內容。

  2. 第二輪

    • 使用者:其實我比較喜歡早上暖一點。

    • (工具呼叫)ADK 嘗試擷取記憶內容。系統會擷取「我喜歡 22 度的溫度」記憶體。

    • 模型:好的,我已調高溫度。

    • (回呼)ADK 會觸發記憶體生成作業。現有記憶內容「我喜歡 71 度的溫度」會更新為「我通常喜歡 71 度的溫度,但早上喜歡暖一點」。

搭配區域執行階段使用多區域 Memory Bank

使用內建 Memory Bank 的執行階段時,代理程式和 Memory Bank 預設會部署在相同區域。不過,您可以將兩者分離,搭配使用多區域 Memory Bank (例如 us) 和區域執行階段 (例如 us-central1)。這樣一來,您就能在不同區域部署作業中,維護中央 Memory Bank。

如要使用多區域 Memory Bank,您必須覆寫預設 ADK 記憶體服務建構工具,指向多區域位置和對應的 Memory Bank ID。

import vertexai
from google.adk.memory import VertexAiMemoryBankService
from vertexai.agent_engines import AdkApp

# Create the Memory Bank instance in a multi-region location (for example, 'us')
client_mb = vertexai.Client(project="PROJECT_ID", location="us")
memory_bank = client_mb.agent_engines.create()
memory_bank_id = memory_bank.api_resource.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 = vertexai.Client(project="PROJECT_ID", location="us-central1")
agent_engine = client_runtime.agent_engines.create(
    agent=adk_app,
    config={
        "staging_bucket": "STAGING_BUCKET",
        "requirements": ["google-cloud-aiplatform[agent_engines,adk]"]
    }
)

更改下列內容:

  • PROJECT_ID:專案 ID。
  • STAGING_BUCKET:用於暫存 Agent Runtime 的 Cloud Storage bucket。

清除所用資源

如要清理此專案中使用的所有資源,請刪除用於本快速入門導覽課程的專案。 Google Cloud

或者,您也可以按照下列步驟,刪除在本教學課程中建立的個別資源:

  1. 使用下列程式碼範例刪除 Agent Runtime 執行個體,這也會刪除屬於該執行階段的所有工作階段或記憶體。

    agent_engine.delete(force=True)
    
  2. 刪除所有在本機建立的檔案。

後續步驟

快速入門

開始使用 Memory Bank API 管理長期記憶。