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

ADK 记忆库快速入门

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

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

如需了解如何在不使用 ADK 的情况下直接调用 API，请参阅 Memory Bank 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）中使用。在这种情况下，您需要直接创建记忆服务和运行器。

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 触发记忆生成。未提取任何记忆。
第二轮：
- 用户：我觉得 71 度很舒适。
- （工具调用）：ADK 尝试提取记忆，但没有可用的记忆。
- 模型：好的，我已将温度调到 71 度。
- （回调）：ADK 触发记忆生成。已创建记忆“我喜欢 71 度的温度”。

第二次会话

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

第一轮交互
- 用户：修正温度。太不舒服了！
- （工具调用）：ADK 尝试提取记忆。检索到记忆“我喜欢 71 度的温度”。
- 模型：好的，我已将温度调到 71 度。
- （回调）：ADK 触发记忆生成。未提取任何记忆，因为用户未分享任何有意义的内容。
第二轮
- 用户：实际上，我希望早上更暖和一些。
- （工具调用）：ADK 尝试提取记忆。检索到记忆“我喜欢 71 度的温度”。
- Model: Ok, I've made the temperature warmer.
- （回调）：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 快速入门

开始使用 Memory Bank 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 记忆库快速入门