快速入门：使用记忆库为 ADK 智能体添加长期记忆

Agent Platform 记忆库可为使用智能体开发套件 (ADK) 构建的智能体提供跨会话的长期记忆。在本快速入门中，您将运行一个智能体，该智能体会在一个会话中了解用户的首选温度，并在下一个会话中回忆该偏好设置。

本快速入门将引导您完成以下步骤：

• 设置环境。
• 创建记忆库实例。
• 在本地运行启用了记忆功能的 ADK 智能体。
• 验证代理是否会在新会话中记住该偏好设置。

如需在不使用 ADK 的情况下直接调用记忆库，请参阅记忆库 API 快速入门。如需选择其他记忆工具、回调或运行时，请参阅将记忆库与 ADK 搭配使用。

如需查看将记忆库与 ADK 搭配使用的示例，请在以下任一环境中运行“在 ADK 上开始使用记忆库”笔记本：

在 Colab 中打开 | 在 Colab Enterprise 中打开 | 在 Agent Platform Workbench 中打开 | 在 GitHub 上查看

准备工作

1. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Roles required to select or create a project

   • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
   • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.
   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

2. If you're using an existing project for this guide, verify that you have the permissions required to complete this guide. If you created a new project, then you already have the required permissions.

3. Verify that billing is enabled for your Google Cloud project.

4. Enable the Agent Platform API, if it is not already enabled.

   Roles required to enable APIs

   To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

   Enable the API

5. 设置 Python 开发环境。

6. 为您的用户账号创建本地身份验证凭证：

   gcloud auth application-default login

   注意：如果 gcloud CLI 输出警告，提示您的账号没有 serviceusage.services.use 权限，则某些 gcloud CLI 命令和客户端库可能无法正常运行。请让管理员授予您 Service Usage Consumer IAM 角色 (roles/serviceusage.serviceUsageConsumer)，然后运行以下命令：

   gcloud auth application-default set-quota-project PROJECT_ID

   如果返回身份验证错误，请确认您已使用联合身份登录 gcloud CLI。

注意：如需启用 API，您需要拥有 serviceusage.services.enable 权限。 如果您没有此权限，请让您的管理员为您授予 Service Usage Admin (roles/serviceusage.serviceUsageAdmin) 角色。

所需的角色

如需获得创建 Memory Bank 实例以及生成和检索记忆所需的权限，请让管理员向您授予项目的以下 IAM 角色：

• 创建或更新记忆库实例： Agent Platform User (`roles/aiplatform.user`)
• 读取、写入或生成记忆： Memory Bank User (`roles/aiplatform.memoryUser`)

如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

您也可以通过自定义角色或其他预定义角色来获取所需的权限。

设置环境

1. 在终端中，安装 ADK 和 Agent Platform SDK：

   pip install --upgrade "google-adk[gcp]" "google-cloud-agentplatform>=2.0.1"

2. 设置本快速入门中的脚本读取的环境变量：

   export GOOGLE_GENAI_USE_ENTERPRISE=TRUE
   export GOOGLE_CLOUD_PROJECT=PROJECT_ID
   export GOOGLE_CLOUD_LOCATION=global

   请将 PROJECT_ID 替换为您的 Google Cloud 项目 ID。

   本快速入门使用全球端点。如需改用区域或多区域位置，请将 global 替换为某个支持的位置。如果您需要客户管理的加密密钥 (CMEK)，请使用区域级或多区域级位置，因为 CMEK 不适用于全球端点。

   在同一终端会话和目录中运行本快速入门的其余部分，以便脚本可以读取这些变量。

创建记忆库实例

1. 创建一个名为 create_memory_bank.py 的文件，其中包含以下代码：

   import os
   
   import agentplatform
   
   client = agentplatform.Client(
       project=os.environ["GOOGLE_CLOUD_PROJECT"],
       location=os.environ["GOOGLE_CLOUD_LOCATION"],
   )
   
   memory_bank = client.memory_banks.create(
       config={"display_name": "Memory Bank quickstart"}
   )
   
   # Print only the instance ID, for example 1234567890.
   print(memory_bank.name.split("/")[-1])
   

2. 运行脚本并将实例 ID 保存在 MEMORY_BANK_ID 环境变量中：

   export MEMORY_BANK_ID=$(python create_memory_bank.py)

   如果这是您的第一个记忆库实例，创建过程可能需要 1-2 分钟。

3. 确认该变量包含实例 ID：

   echo $MEMORY_BANK_ID

   输出是一个 ID，例如 1234567890。如果输出为空或返回 401 或 403 错误，请检查您的所需角色并运行 gcloud auth application-default login。如果新的 global 实例在首次使用时返回暂时性 403 错误，请等待 30 秒，然后再次运行脚本。

运行启用了记忆功能的代理

本部分中的脚本会为同一用户运行两个会话。在第一个会话中，用户说明了温度偏好。在第二个会话中，智能体从记忆库中调取了该偏好设置。

请完成以下步骤：

1. 创建一个名为 memory_agent 的目录：

   mkdir memory_agent

2. 创建一个名为 memory_agent/agent.py 的文件，其中包含以下代码。该代码会从您在设置环境和创建 Memory Bank 实例中设置的变量中读取项目、位置和实例 ID。您无需修改代码。

   import asyncio
   import os
   
   import agentplatform
   from google import adk
   from google.adk.agents.callback_context import CallbackContext
   from google.adk.memory import VertexAiMemoryBankService
   from google.adk.sessions import InMemorySessionService
   from google.adk.tools.preload_memory_tool import PreloadMemoryTool
   from google.genai import types
   
   PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
   LOCATION = os.environ["GOOGLE_CLOUD_LOCATION"]
   MEMORY_BANK_ID = os.environ["MEMORY_BANK_ID"]
   # Memory Bank scopes memories to this app name and user ID.
   APP_NAME = "memory_agent"
   USER_ID = "quickstart_user"
   
   async def generate_memories_callback(
       callback_context: CallbackContext,
   ) -> None:
       # Send the latest turn to Memory Bank and wait until memory generation
       # finishes, so that the next session can retrieve the new memory.
       await callback_context.add_events_to_memory(
           events=callback_context.session.events[-5:],
           custom_metadata={"wait_for_completion": True},
       )
       return None
   
   root_agent = adk.Agent(
       model="gemini-3.5-flash",
       name="memory_agent",
       instruction=(
           "You are a vehicle voice agent that controls in-vehicle settings. "
           "Use what you know about the user's preferences. "
           "If you don't know the user's preferred temperature, ask for it. "
           "Limit responses to under 30 words."
       ),
       tools=[PreloadMemoryTool()],
       after_agent_callback=generate_memories_callback,
   )
   
   memory_service = VertexAiMemoryBankService(
       project=PROJECT_ID,
       location=LOCATION,
       agent_engine_id=MEMORY_BANK_ID,
   )
   session_service = InMemorySessionService()
   runner = adk.Runner(
       agent=root_agent,
       app_name=APP_NAME,
       session_service=session_service,
       memory_service=memory_service,
   )
   
   async def send_message(session_id: str, text: str) -> None:
       print(f"User: {text}")
       message = types.Content(role="user", parts=[types.Part(text=text)])
       async for event in runner.run_async(
           user_id=USER_ID, session_id=session_id, new_message=message
       ):
           if event.is_final_response() and event.content:
               print(f"Agent: {event.content.parts[0].text}")
   
   async def main() -> None:
       print("--- Session 1: share a preference ---")
       session = await session_service.create_session(
           app_name=APP_NAME, user_id=USER_ID
       )
       await send_message(session.id, "Can you fix the temperature?")
       await send_message(session.id, "I'm comfortable at 71 degrees.")
   
       print("--- Session 2: recall the preference ---")
       session = await session_service.create_session(
           app_name=APP_NAME, user_id=USER_ID
       )
       await send_message(session.id, "Fix the temperature.")
   
       print("--- Memories stored in Memory Bank ---")
       client = agentplatform.Client(project=PROJECT_ID, location=LOCATION)
       memory_bank_name = (
           f"projects/{PROJECT_ID}/locations/{LOCATION}"
           f"/reasoningEngines/{MEMORY_BANK_ID}"
       )
       results = client.memory_banks.memories.retrieve(
           name=memory_bank_name,
           scope={"app_name": APP_NAME, "user_id": USER_ID},
       )
       for retrieved_memory in results.page:
           print(f"* {retrieved_memory.memory.fact}")
   
   if __name__ == "__main__":
       asyncio.run(main())
   

   注意：ADK 主要使用异步函数。 该脚本将其异步调用封装在 asyncio.run() 中，当您运行标准 Python 脚本时，该脚本会创建并管理事件循环。在 Colaboratory 等笔记本中，事件循环已在运行，因此请将 asyncio.run(main()) 替换为 await main()。

3. 运行脚本：

   python memory_agent/agent.py

   回调会在每轮对话后等待记忆生成完成，以便在第二个会话开始时可以使用该记忆。

验证代理是否记住了

脚本运行完毕后，会将两个对话和存储的记忆输出到终端。将该输出与以下示例进行比较，以确认智能体在新会话中记起了偏好设置。

在第一个会话中，智能体没有任何关于用户的记忆，因此它会询问温度：

--- Session 1: share a preference ---
User: Can you fix the temperature?
Agent: Sure. What temperature would you like?
User: I'm comfortable at 71 degrees.
Agent: Okay, I've set the temperature to 71 degrees.

在第二个会话中，智能体从记忆库中检索记忆，并在不询问的情况下将温度设置为 71 度。存储的记忆会显示在输出的末尾：

--- Session 2: recall the preference ---
User: Fix the temperature.
Agent: Okay, I've set the temperature to 71 degrees, as you prefer.
--- Memories stored in Memory Bank ---
* I like the temperature to be 71 degrees.

智能体回答和记忆的措辞因智能体和记忆库使用的模型而异。

可选：在 ADK Web 中与智能体 Chat

ADK Web 是一个基于浏览器的界面，用于在开发期间测试智能体。

重要提示：ADK Web 不适用于生产部署。仅将 ADK Web 用于开发和调试。

1. 在同一终端会话中，从包含 memory_agent 目录的目录中运行以下命令：

   adk web --memory_service_uri=agentengine://$MEMORY_BANK_ID memory_agent

2. 在浏览器中，打开 http://localhost:8000?userId=quickstart_user。

   userId 查询参数与脚本中的用户 ID 匹配，因此智能体可以检索脚本创建的记忆。

3. 发送消息 Fix the temperature.，并确认智能体在不询问的情况下将温度设置为 71 度。智能体使用脚本在运行启用了记忆功能的智能体中存储的偏好设置。

4. 如需停止 ADK Web，请在终端中按 Control+C。

清理

为避免因本页面中使用的资源导致您的 Google Cloud 账号产生费用，您可以删除 Google Cloud项目。否则，请逐个删除资源：

1. 创建一个名为 clean_up.py 的文件，其中包含以下代码：

   import os
   
   import agentplatform
   
   project = os.environ["GOOGLE_CLOUD_PROJECT"]
   location = os.environ["GOOGLE_CLOUD_LOCATION"]
   memory_bank_id = os.environ["MEMORY_BANK_ID"]
   
   client = agentplatform.Client(project=project, location=location)
   client.memory_banks.delete(
       name=f"projects/{project}/locations/{location}/reasoningEngines/{memory_bank_id}",
       force=True,
   )
   

2. 运行脚本以删除 Memory Bank 实例：

   python clean_up.py

3. 删除您创建的本地文件：

   rm -r memory_agent create_memory_bank.py clean_up.py

后续步骤

快速入门

Memory Bank API 快速入门

开始使用 Memory Bank API 管理长期记忆。

指南

在 ADK 中使用记忆库

为 ADK 选择回调、工具和运行时。

指南

生成回忆

了解如何在记忆库中生成记忆。

指南

设置记忆库

了解如何设置记忆库。