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

ADK 记忆库快速入门

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

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

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

的错误。

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

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

VertexAiMemoryBankService 接口包括：

memory_service.add_session_to_memory 触发对记忆库的 GenerateMemories 请求使用提供的 adk.Session 中的所有事件作为源内容。您可以使用回调中的 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 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：您的区域。请参阅记忆库的支持的区域。

创建 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 会话实例 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 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：您的区域。请参阅记忆库支持的区域。
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.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_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 Web 界面，您可以直接在浏览器中测试智能体。

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，智能体会在每个回合开始时尝试检索记忆，以访问用户之前向智能体表达的偏好。在智能体与用户首次互动期间，没有可检索的记忆。因此，智能体不知道任何用户偏好，例如他们喜欢的温度，如以下示例所示：

第一轮交互：
- 用户: "Can you fix the temperature?"
- （工具调用）：ADK 尝试提取记忆；没有可用的记忆。
- 模型：“What temperature do you prefer?”
- （回调）：ADK 触发记忆生成。未提取任何记忆。
第二轮交互：
- 用户：“I'm comfortable at 71 degrees.”
- （工具调用）：ADK 尝试提取记忆；没有可用的记忆。
- 模型：“Ok, I've updated the temperature to 71 degrees.”
- （回调）：ADK 触发记忆生成。创建了记忆“I like the temperature 71 degrees”。

第二次会话

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

第一轮交互
- 用户：“Fix the temperature. It's so uncomfortable!” __
- （工具调用）：ADK 尝试提取记忆。检索到记忆“I like the temperature 71 degrees”。
- 模型：“Ok, I've updated the temperature to 71 degrees.”
- （回调）：ADK 触发记忆生成。未提取任何记忆，因为用户没有分享任何有意义的持久信息。
第二轮交互
- 用户：“Actually, I prefer it to be warmer in the mornings.”
- （工具调用）：ADK 尝试提取记忆。检索到记忆“I like the temperature 71 degrees”。
- 模型：“Ok, I've made the temperature warmer.”
- （回调）：ADK 触发记忆生成。现有记忆“I like the temperature 71 degrees”更新为“I generally like the temperature to be 71 degrees, but I like it to be warmer in the mornings”。

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

使用具有内置记忆库的运行时时，智能体和记忆库默认部署在同一区域中。但是，您可以将它们分离，以将多区域记忆库（例如 us）与区域运行时（例如 us-central1）搭配使用。借助此配置，您可以在不同的区域部署中维护一个中央记忆库。

如需使用多区域记忆库，您必须替换默认 ADK 记忆服务构建器，以指向多区域位置和相应的记忆库 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 存储桶。

清理

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

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

使用以下代码示例删除 Agent Runtime 实例，这也会删除属于该运行时的任何会话或记忆。
```
agent_engine.delete(force=True)
```
删除所有本地创建的文件。

后续步骤

快速入门

记忆库 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

互动示例

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

清理

后续步骤

记忆库 API 快速入门

ADK 记忆库快速入门