使用 Agent Development Kit 的快速入門

本教學課程說明如何使用 ADK 的 Memory Bank 管理長期記憶。將 Agent Development Kit (ADK) 代理程式設定為使用記憶體庫後，代理程式就會協調對記憶體庫的呼叫，為您管理長期記憶。

使用 ADK 的記憶體庫時，請按照下列步驟操作：

建立 ADK 代理和執行器。ADK 執行器會將代理程式連結至提供工作階段和記憶體管理功能的服務。
與代理互動，動態生成可跨工作階段存取的長期記憶。
清除所用資源。

如要直接呼叫 Memory Bank，而不使用 ADK 編排，請參閱「Agent Engine SDK 快速入門」。使用 Agent Engine SDK 有助於瞭解記憶體庫如何生成記憶，或檢查記憶體庫的內容。

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

事前準備

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

設定環境變數

如要使用 ADK，請設定環境變數：

import os

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

更改下列內容：

PROJECT_ID：專案 ID。
LOCATION：您的區域。請參閱「支援記憶體空間的區域」。

建立 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-2.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-2.5-flash",
      name='stateful_agent',
      instruction="""...""",
      tools=[FunctionTool(func=search_memories)],
      after_agent_callback=generate_memories_callback
  )

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

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

建立 Agent Engine 執行個體

您必須先建立 Agent Engine 執行個體，才能使用 Memory Bank。如果您使用 Agent Engine 執行階段部署代理程式，則可略過這個步驟。如要進一步瞭解如何自訂記憶體庫行為，請參閱「設定記憶體庫」頁面的「設定 Agent Engine 執行個體以使用記憶體庫」一節。

import vertexai

client = vertexai.Client(
  project="PROJECT_ID",
  location="LOCATION"
)
# If you don't have an Agent Engine instance already, create an Agent Engine
# Memory Bank instance using the default configuration.
agent_engine = client.agent_engines.create()

# Optionally, print out the Agent Engine resource name. You will need the
# resource name if you want to interact with your Agent Engine instance later on.
print(agent_engine.api_resource.name)

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

更改下列內容：

PROJECT_ID：專案 ID。
LOCATION：您的區域。請參閱「支援記憶體空間的區域」。

建立 ADK 執行階段

將 Agent Engine 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="AGENT_ENGINE_ID",
)

# You can use any ADK session service. This example uses Agent Engine Sessions.
session_service = VertexAiSessionService(
    project="PROJECT_ID",
    location="LOCATION",
    agent_engine_id="AGENT_ENGINE_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：您的區域。請參閱「支援記憶體空間的區域」。
APP_NAME：ADK 應用程式名稱。系統會在產生的回憶集錦scope字典中加入應用程式名稱，確保回憶集錦不會與其他使用者和應用程式混淆。
AGENT_ENGINE_ID：用於記憶體庫和工作階段的 Agent Engine ID。例如，456 位於 projects/my-project/locations/us-central1/reasoningEngines/456 中。

Agent Engine

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

如要進一步瞭解如何設定 Agent Engine 執行階段，包括如何自訂記憶體庫的行為，請參閱「設定 Agent Engine」。

使用下列程式碼，將啟用記憶體的 ADK 代理程式部署至 Agent Engine 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 Agent Engine with your agent deployed to Agent Engine Runtime.
# The Agent Engine instance will also include an empty Memory Bank.
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 Agent Engine to deploy your agent to Agent Engine Runtime.
# Your agent will have access to the Agent Engine 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：您的區域。請參閱「支援記憶體空間的區域」。
STAGING_BUCKET：用於暫存 Agent Engine Runtime 的 Cloud Storage bucket。

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

def memory_bank_service_builder():
    return VertexAiMemoryBankService(
        project="PROJECT_ID",
        location="LOCATION",
        agent_engine_id="AGENT_ENGINE_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 Engine Runtime,
  # use `client.agent_engines.create(...)` or `client.agent_engines.update(...)`
  # and call the returned Agent Engine 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：您的區域。請參閱「支援記憶體空間的區域」。
AGENT_ENGINE_ID：用於記憶體庫的 Agent Engine 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

GKE

如要在 Google Kubernetes Engine (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://AGENT_ENGINE_ID

更改下列內容：

PROJECT_ID：專案 ID。
LOCATION：您的區域。請參閱「支援記憶體空間的區域」。
AGENT_ENGINE_ID：用於記憶體庫的 Agent Engine ID。例如，456 位於 projects/my-project/locations/us-central1/reasoningEngines/456 中。

與虛擬服務專員互動

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

記憶內容會使用與執行代理程式時所用使用者 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 Engine

使用 Agent Engine ADK 範本時，您可以呼叫 Agent Engine 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，專員會在每一回合開始時嘗試擷取記憶內容，存取使用者先前向專員傳達的偏好設定。服務專員與使用者初次互動時，沒有可供擷取的記憶內容。因此，如以下範例所示，智慧助理不會知道任何使用者偏好設定，例如偏好的溫度：

第一回合：
- 使用者：「可以調整溫度嗎？」
- (工具呼叫)：ADK 嘗試擷取回憶內容，但沒有可用的回憶內容。
- 模型：「你喜歡的溫度是幾度？」
- (回呼)：ADK 會觸發記憶生成作業。不會擷取任何記憶。
第二回合：
- 使用者：I'm comfortable at 71 degrees. (71 度很舒適。)
- (工具呼叫)：ADK 嘗試擷取回憶內容，但沒有可用的回憶內容。
- 模型：好的，已將溫度調為 22 度。
- (回呼)：ADK 會觸發記憶生成作業。系統會建立「我喜歡 71 度的溫度」記憶。

第二個工作階段

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

第一回合
- 使用者：調整溫度。這太不舒服了！
- (工具呼叫)：ADK 嘗試擷取記憶內容。系統會擷取「我喜歡 22 度的溫度」記憶。
- 模型：好的，已將溫度調為 22 度。
- (回呼)：ADK 會觸發記憶生成作業。使用者未分享任何有意義的內容，因此系統不會擷取回憶集。
第二輪
- 使用者：其實我比較喜歡早上暖一點。
- (工具呼叫)：ADK 嘗試擷取記憶內容。系統會擷取「我喜歡 22 度的溫度」記憶。
- 模型：好的，我已調高溫度。
- (回呼)：ADK 會觸發記憶生成作業。現有記憶內容「我喜歡 71 度的溫度」會更新為「我通常喜歡 71 度的溫度，但早上喜歡溫暖一點」。

清除所用資源

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

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

使用下列程式碼範例刪除 Vertex AI Agent Engine 執行個體，這也會刪除屬於該 Vertex AI Agent Engine 的任何工作階段或記憶體。
```
agent_engine.delete(force=True)
```
刪除所有在本機建立的檔案。

後續步驟

Agent Engine SDK 快速入門導覽課程。

使用 Agent Development Kit 的快速入門 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

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

事前準備

設定環境變數

建立 ADK 代理

定義記憶體生成回呼

定義記憶體擷取工具

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

建立 Agent Engine 執行個體

建立 ADK 執行階段

本機執行器

Agent Engine

Cloud Run

GKE

ADK Web

與虛擬服務專員互動

本機執行器

Agent Engine

Cloud Run

GKE

ADK Web

互動範例

清除所用資源

後續步驟

使用 Agent Development Kit 的快速入門