智能体开发套件快速入门

本教程演示了如何将记忆库与 ADK 搭配使用来管理长期记忆。将智能体开发套件 (ADK) 智能体配置为使用记忆库后，该智能体会为您编排对记忆库的调用，以管理长期记忆。

将记忆库与 ADK 搭配使用涉及以下步骤：

创建 ADK 智能体和运行程序。ADK 运行程序将您的代理连接到提供会话和记忆管理的服务。
与智能体互动，动态生成可在不同会话中访问的长期记忆。
清理.

如需在不使用 ADK 编排的情况下直接调用记忆库，请参阅 Agent Engine SDK 快速入门。使用 Agent Engine SDK 有助于了解记忆库如何生成记忆，或检查记忆库的内容。

使用 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_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 记忆库内存服务和运行时

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

创建 Agent Engine 实例

您首先需要创建一个 Agent Engine 实例，以便用于记忆库。如果您使用 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 传递给运行时或部署脚本，以便您的代理使用记忆库作为 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="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。例如，projects/my-project/locations/us-central1/reasoningEngines/456 中的 456。

Agent Engine

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

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

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

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 存储桶。

在本地运行时，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。例如，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

GKE

如需将代理部署到 Google Kubernetes Engine (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://AGENT_ENGINE_ID

替换以下内容：

PROJECT_ID：您的项目 ID。
LOCATION：您的区域。请参阅记忆库的支持的区域。
AGENT_ENGINE_ID：要用于记忆库的 Agent Engine 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 Engine

使用 Agent Engine ADK 模板时，您可以调用 Agent Engine 运行时与记忆库和会话进行互动。

# 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 的下一个会话。如果用户提供的信息与现有记忆相似或相矛盾，系统会将新信息与现有记忆整合。

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

清理

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

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

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

后续步骤

Agent Engine SDK 快速入门。

智能体开发套件快速入门 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

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

准备工作

设置环境变量

创建 ADK 智能体

定义内存生成回调

定义记忆检索工具

定义 ADK 记忆库内存服务和运行时

创建 Agent Engine 实例

创建 ADK 运行时

本地跑步者

Agent Engine

Cloud Run

GKE

ADK Web

与您的代理互动

本地跑步者

Agent Engine

Cloud Run

GKE

ADK Web

互动示例

清理

后续步骤

智能体开发套件快速入门