Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

ADK 记忆库快速入门

借助 Agent Platform 记忆库，智能体可以跨会话管理长期记忆。与智能体开发套件 (ADK) 搭配使用时，您的智能体可以自动编排对记忆库的调用，以根据用户互动存储和检索记忆。

本文档介绍了如何创建 ADK 代理、将其配置为使用记忆库，以及如何与该代理互动以生成和访问记忆。

如需了解如何在不使用 ADK 的情况下直接调用 API，请参阅记忆库 API 快速入门。

使用 ADK 记忆服务和记忆库管理记忆

VertexAiMemoryBankService 是 ADK 围绕记忆库的封装容器，由 ADK 的 BaseMemoryService 定义。您可以定义与记忆服务互动的回调和工具，以读取和写入记忆。

VertexAiMemoryBankService 接口包括：

memory_service.add_session_to_memory 会使用所提供的 adk.Session 中的所有事件作为源内容，触发对记忆库的 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，这会使用部分事件触发对记忆库的 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 请求，以提取当前 user_id 和 app_name 的相关记忆。您可以使用内置的记忆工具（LoadMemoryTool 或 PreloadMemoryTool）或调用 tool_context.search_memory 的自定义工具来编排对该方法的调用。

准备工作

如需完成本教程中演示的步骤，您必须先按照设置记忆库页面上的“准备工作”部分中的步骤操作。

设置环境变量

如需使用 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：您的区域。请参阅记忆库的支持的区域。

创建 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 记忆库记忆服务和记忆库实例

创建启用记忆功能的智能体后，您需要将其与记忆服务相关联。配置 ADK 内存服务的流程取决于 ADK 代理的运行位置。运行时负责对智能体、工具及回调的执行进行编排。

创建记忆库实例

您首先需要创建记忆库实例。如果您使用 Agent Runtime 部署代理，则此步骤为可选步骤。如需详细了解如何自定义记忆库行为，请参阅“设置记忆库”页面上的配置记忆库实例部分。

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]

替换以下内容：

PROJECT_ID：您的项目 ID。
LOCATION：您的区域。请参阅记忆库的支持的区域。

创建 ADK 运行时

将记忆库实例 ID 传递给运行时或部署脚本，以便您的代理使用记忆库作为 ADK 记忆库服务。

本地跑步者

adk.Runner 通常在本地环境（例如 Colab）中使用。在这种情况下，您需要直接创建内存服务和 runner。

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：您的区域。请参阅记忆库的支持的区域。
APP_NAME：ADK 应用名称。应用名称将包含在生成的记忆的 scope 字典中，以便在用户和应用之间隔离记忆。
MEMORY_BANK_ID：记忆库实例 ID。例如，projects/my-project/locations/us-central1/reasoningEngines/456 中的 456。
SESSIONS_ID：Agent Platform Sessions 实例 ID。例如，projects/my-project/locations/us-central1/reasoningEngines/789 中的 789。

Gemini Enterprise Agent Platform 上的 Agent Runtime

Agent Runtime ADK 模板 (AdkApp) 既可在本地使用，也可用于将 ADK 智能体部署到 Agent Runtime。部署在 Agent Platform 上时，记忆库 ADK 模板使用 VertexAiMemoryBankService 作为默认记忆库服务。因此，您只需一步即可创建记忆库实例并将其部署到运行时。

如需详细了解如何设置记忆库实例（包括如何自定义记忆库的行为），请参阅配置记忆库。

使用以下代码将启用记忆功能的 ADK 智能体部署到 Agent Runtime：

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)

替换以下内容：

PROJECT_ID：您的项目 ID。
LOCATION：您的区域。请参阅记忆库的支持的区域。
STAGING_BUCKET：用于暂存 Agent Runtime 的 Cloud Storage 存储桶。

在本地运行时，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.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)

替换以下内容：

PROJECT_ID：您的项目 ID。
LOCATION：您的区域。请参阅记忆库的支持的区域。
MEMORY_BANK_ID：要用于记忆库的记忆库实例 ID。例如，projects/my-project/locations/us-central1/reasoningEngines/456 中的 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_ID：记忆库实例 ID。例如，projects/my-project/locations/us-central1/reasoningEngines/456 中的 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：您的用户的标识符。由此会话生成的记忆会以此不透明标识符为键。生成的记忆的范围存储为 {"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：您的用户的标识符。从此会话生成的记忆会以此不透明标识符为键。生成的记忆的范围存储为 {"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 触发记忆生成。系统不会提取任何记忆。
第二轮对话：
- 用户：我感觉 22 度比较舒适。
- （工具调用）：ADK 尝试提取记忆，但没有可用的记忆。
- 模型：好的，我已经将温度更新为 71 度。
- （回调）：ADK 触发记忆生成。系统会创建“我喜欢 71 度的温度”记忆。

第二次会话

提取的记忆将可用于具有相同应用名称和用户 ID 的下一个会话。如果用户提供的信息与现有记忆相似或相矛盾，系统会将新信息与现有记忆整合。

第一轮交互
- 用户：Fix the temperature. 太不舒服了！
- （工具调用）：ADK 尝试提取记忆。系统会检索“我喜欢 71 度的温度”这条记忆。
- 模型：好的，我已经将温度更新为 71 度。
- （回调）：ADK 触发记忆生成。系统不会提取任何回忆，因为用户未分享任何有意义的持久内容。
第二轮
- 用户：实际上，我希望早上更暖和一些。
- （工具调用）：ADK 尝试提取记忆。系统会检索“我喜欢 71 度的温度”这条记忆。
- 模型：好的，我已调高温度。
- （回调）：ADK 触发记忆生成。系统会将现有记忆“我喜欢 71 度的温度”更新为“我一般喜欢 71 度的温度，但早上喜欢更暖和的温度”。

将多区域记忆库与区域运行时搭配使用

使用内置记忆库的运行时时，您的代理和记忆库默认部署在同一区域中。不过，您可以将它们解耦，以将多区域记忆库（例如 us）与区域运行时（例如 us-central1）搭配使用。此配置可让您在不同的区域级部署中维护一个中央记忆库。

如需使用多区域记忆库，您必须替换默认的 ADK 记忆服务构建器，以指向多区域位置和相应的记忆库 ID。

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

替换以下内容：

PROJECT_ID：您的项目 ID。
STAGING_BUCKET：用于暂存 Agent Runtime 的 Cloud Storage 存储桶。

清理

如需清理此项目中使用的所有资源，您可以删除用于本快速入门的 Google Cloud项目。

否则，您可以按如下所示逐个删除在本教程中创建的资源：

使用以下代码示例删除 Agent Runtime 实例，这也会删除属于该运行时的所有会话或记忆。

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

删除所有本地创建的文件。

后续步骤

快速入门

Memory Bank API 快速入门

开始使用记忆库 API 来管理长期记忆。

ADK 记忆库快速入门 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

使用 ADK 记忆服务和记忆库管理记忆

准备工作

设置环境变量

创建 ADK 智能体

定义记忆生成回调

定义记忆检索工具

定义 ADK 记忆库记忆服务和记忆库实例

创建记忆库实例

创建 ADK 运行时

本地跑步者

Gemini Enterprise Agent Platform 上的 Agent Runtime

Cloud Run

Google Kubernetes Engine (GKE)

ADK Web

与您的代理互动

本地跑步者

Agent Runtime

Cloud Run

GKE

ADK Web

互动示例

将多区域记忆库与区域运行时搭配使用

清理

后续步骤

Memory Bank API 快速入门

ADK 记忆库快速入门