Memory：使用 `MemoryService` 的长期知识¶

我们已经看到，Session 用于跟踪单次、持续对话的历史（events）和临时数据（state）。但如果智能体需要回忆过去对话的信息，或访问外部知识库怎么办？这就是长期知识和 MemoryService 概念的用武之地。

可以这样理解：

Session / State： 就像你在一次特定聊天中的短期记忆。
长期知识（MemoryService）：就像智能体可以查询的可搜索档案或知识库，可能包含来自许多过去聊天或其他来源的信息。

`MemoryService` 角色¶

BaseMemoryService 定义了管理这种可搜索、长期知识存储的接口。其主要职责是：

摄入信息（add_session_to_memory）： 获取（通常是已完成的）Session 的内容，并将相关信息添加到长期知识存储中。
搜索信息（search_memory）： 允许智能体（通常通过 Tool）查询知识存储并基于搜索查询检索相关片段或上下文。

选择合适的记忆服务¶

ADK 提供两种不同的 MemoryService 实现，每种都针对不同的用例。使用下表来决定哪种最适合你的智能体。

功能	InMemoryMemoryService	[新功能！] VertexAiMemoryBankService
持久性	无（重启时数据丢失）	是（由 Vertex AI 管理）
主要用例	原型设计、本地开发和简单测试。	从用户对话中构建有意义、不断发展的记忆。
记忆提取	存储完整对话	从对话中提取有意义的信息并与现有记忆合并（由 LLM 支持）
搜索能力	基本关键词匹配。	高级语义搜索。
设置复杂度	无。这是默认选项。	低。需要在 Vertex AI 中创建智能体引擎。
依赖项	无。	Google Cloud 项目，Vertex AI API
何时使用	当你想要在多个会话的聊天历史中搜索以进行原型设计时。	当你希望智能体记住并从过去的交互中学习时。

内存中的记忆¶

InMemoryMemoryService 将会话信息存储在应用程序的内存中，并对搜索执行基本的关键词匹配。它不需要设置，最适合原型设计和不需要持久性的简单测试场景。

from google.adk.memory import InMemoryMemoryService
memory_service = InMemoryMemoryService()

示例：添加和搜索记忆

此示例演示了使用 InMemoryMemoryService 的基本流程以便简化理解。

完整代码

import asyncio
from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService, Session
from google.adk.memory import InMemoryMemoryService # 导入 MemoryService
from google.adk.runners import Runner
from google.adk.tools import load_memory # 查询内存的工具
from google.genai.types import Content, Part

# --- 常量 ---
APP_NAME = "memory_example_app"
USER_ID = "mem_user"
MODEL = "gemini-2.0-flash" # 使用有效模型

# --- 智能体定义 ---
# 智能体 1：简单的信息捕获智能体
info_capture_agent = LlmAgent(
    model=MODEL,
    name="InfoCaptureAgent",
    instruction="确认用户的陈述。",
    # output_key="captured_info" # 也可以选择保存到状态
)

# 智能体 2：可以使用内存的智能体
memory_recall_agent = LlmAgent(
    model=MODEL,
    name="MemoryRecallAgent",
    instruction="回答用户的问题。如果答案可能在过去的对话中，请使用 'load_memory' 工具",
    tools=[load_memory] # 给智能体提供工具
)

# --- 服务和 Runner ---
session_service = InMemorySessionService()
memory_service = InMemoryMemoryService() # 用于演示的内存服务

runner = Runner(
    # 从信息捕获智能体开始
    agent=info_capture_agent,
    app_name=APP_NAME,
    session_service=session_service,
    memory_service=memory_service # 向 Runner 提供内存服务
)

# --- 场景 ---

# 轮次 1：在会话中捕获一些信息
print("--- 轮次 1：捕获信息 ---")
session1_id = "session_info"
session1 = await runner.session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=session1_id)
user_input1 = Content(parts=[Part(text="My favorite project is Project Alpha.")], role="user")

# 运行代理
final_response_text = "(No final response)"
async for event in runner.run_async(user_id=USER_ID, session_id=session1_id, new_message=user_input1):
    if event.is_final_response() and event.content and event.content.parts:
        final_response_text = event.content.parts[0].text
print(f"智能体 1 响应: {final_response_text}")

# 获取已完成的 session
completed_session1 = await runner.session_service.get_session(app_name=APP_NAME, user_id=USER_ID, session_id=session1_id)

# 将此 session 的内容添加到 Memory Service
print("\n--- 添加 Session 1 到记忆中 ---")
memory_service = await memory_service.add_session_to_memory(completed_session1)
print("会话已添加到记忆中。")

Vertex AI 记忆库¶

VertexAiMemoryBankService 将你的智能体连接到 Vertex AI 记忆库，这是一个完全托管的 Google Cloud 服务，为对话智能体提供复杂的持久记忆功能。

工作原理¶

该服务自动处理两个关键操作：

生成记忆： 对话结束时，ADK 将会话事件发送到记忆库，记忆库智能地处理并将信息存储为"记忆"。
检索记忆： 你的智能体代码可以对记忆库发出搜索查询，从过去的对话中检索相关记忆。

先决条件¶

在使用此功能之前，你必须具备：

Google Cloud 项目： 启用了 Vertex AI API。
智能体引擎： 你需要在 Vertex AI 中创建智能体引擎。这将为你提供配置所需的智能体引擎 ID。
身份验证： 确保你的本地环境已通过身份验证以访问 Google Cloud 服务。最简单的方法是运行：
```
gcloud auth application-default login
```
环境变量： 该服务需要你的 Google Cloud 项目 ID 和位置。将它们设置为环境变量：
```
export GOOGLE_CLOUD_PROJECT="your-gcp-project-id"
export GOOGLE_CLOUD_LOCATION="your-gcp-location"
```

配置¶

要将你的智能体连接到记忆库，你需要在启动 ADK 服务器（adk web 或 adk api_server）时使用 --memory_service_uri 标志。URI 必须采用 agentengine://<agent_engine_id> 格式。

bash

adk web path/to/your/agents_dir --memory_service_uri="agentengine://1234567890"

或者，你可以通过手动实例化 VertexAiMemoryBankService 并将其传递给 Runner 来配置智能体使用记忆库。

from google.adk.memory import VertexAiMemoryBankService

agent_engine_id = agent_engine.api_resource.name.split("/")[-1]

memory_service = VertexAiMemoryBankService(
    project="PROJECT_ID",
    location="LOCATION",
    agent_engine_id=agent_engine_id
)

runner = adk.Runner(
    ...
    memory_service=memory_service
)

在智能体中使用记忆¶

配置服务后，ADK 会自动将会话数据保存到记忆库。要让你的智能体使用这些记忆，你需要从智能体代码中调用 search_memory 方法。

这通常在轮次开始时完成，以便在生成响应之前获取相关上下文。

示例：

from google.adk.agents import Agent
from google.genai import types

class MyAgent(Agent):
    async def run(self, request: types.Content, **kwargs) -> types.Content:
        # 获取用户的最新消息
        user_query = request.parts[0].text

        # 在记忆中搜索与用户查询相关的上下文
        search_result = await self.search_memory(query=user_query)

        # 创建包含检索记忆的提示
        prompt = f"根据我的记忆，以下是我对你查询的回忆：{search_result.memories}\n\n现在，请回应：{user_query}"

        # 使用增强提示调用 LLM
        return await self.llm.generate_content_async(prompt)

高级概念¶

记忆在实践中的工作原理¶

记忆工作流程在内部涉及以下步骤：

会话交互： 用户通过 SessionService 管理的 Session 与智能体交互。添加事件，状态可能会更新。
摄入到记忆中： 在某个时候（通常当会话被认为完成或产生了重要信息时），你的应用程序调用 memory_service.add_session_to_memory(session)。这从会话事件中提取相关信息并将其添加到长期知识存储（内存字典或 RAG 语料库）中。
后续查询： 在不同（或相同）会话中，用户可能会问需要过去上下文的问题（例如，"我们上周讨论的项目 X 是什么？"）。
智能体使用记忆工具： 配备记忆检索工具（如内置的 load_memory 工具）的智能体识别到需要过去的上下文。它调用工具，提供搜索查询（例如，"讨论项目 X 上周"）。
搜索执行： 工具内部调用 memory_service.search_memory(app_name, user_id, query)。
返回结果： MemoryService 搜索其存储（使用关键词匹配或语义搜索）并返回相关片段作为包含 MemoryResult 对象列表的 SearchMemoryResponse（每个对象可能包含相关过去会话的事件）。
智能体使用结果： 工具将这些结果返回给智能体，通常作为上下文或函数响应的一部分。智能体然后可以使用这些检索到的信息来制定对用户的最终答案。

智能体可以访问多个记忆服务吗？¶

通过标准配置：否。 框架（adk web、adk api_server）设计为一次只能通过 --memory_service_uri 标志配置一个记忆服务。然后将这个单一服务提供给智能体，并通过内置的 self.search_memory() 方法访问。从配置角度来看，你只能为该进程服务的所有智能体选择一个后端（InMemory、VertexAiMemoryBankService）。
在智能体代码中：是的，绝对可以。 没有什么可以阻止你直接在智能体代码内手动导入和实例化另一个记忆服务。这允许你在单个智能体轮次内访问多个记忆源。

例如，你的智能体可以使用框架配置的 VertexAiMemoryBankService 来回忆对话历史，同时也手动实例化一个 InMemoryMemoryService 来查找技术手册中的信息。

示例：使用两个记忆服务¶

以下是如何在智能体代码中实现这一点：

from google.adk.agents import Agent
from google.adk.memory import InMemoryMemoryService, VertexAiMemoryBankService
from google.genai import types

class MultiMemoryAgent(Agent):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

        self.memory_service = InMemoryMemoryService()
        # 手动实例化第二个记忆服务用于文档查找
        self.vertexai_memorybank_service = VertexAiMemoryBankService(
            project="PROJECT_ID",
            location="LOCATION",
            agent_engine_id="AGENT_ENGINE_ID"
        )

    async def run(self, request: types.Content, **kwargs) -> types.Content:
        user_query = request.parts[0].text

        # 1. 使用框架提供的记忆搜索对话历史
        #    （如果配置的话，这将是 InMemoryMemoryService）
        conversation_context = await self.search_memory(query=user_query)

        # 2. 使用手动创建的服务搜索文档知识库
        document_context = await self.vertexai_memorybank_service.search_memory(query=user_query)

        # 结合两个来源的上下文生成更好的响应
        prompt = "从我们过去的对话中，我记得：\n"
        prompt += f"{conversation_context.memories}\n\n"
        prompt += "从技术手册中，我找到了：\n"
        prompt += f"{document_context.memories}\n\n"
        prompt += f"基于这些信息，以下是我对'{user_query}'的回答："

        return await self.llm.generate_content_async(prompt)

Memory：使用 MemoryService 的长期知识¶

MemoryService 角色¶