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

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

可以这样想：

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

`MemoryService` 角色¶

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

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

`MemoryService` 实现¶

ADK 提供了不同的方式来实现这种长期知识存储：

InMemoryMemoryService {: #inmemorymemoryservice}
- 工作原理： 在应用程序内存中存储会话信息并为搜索执行基本关键词匹配。
- 持久性： 无。如果应用程序重启，所有存储的知识都会丢失。
- 要求： 无额外要求。
- 最适用于： 原型设计、简单测试、仅需要基本关键词回忆且不需要持久性的场景。
```
from google.adk.memory import InMemoryMemoryService
memory_service = InMemoryMemoryService()
```
VertexAiRagMemoryService {: #vertexairagmemoryservice}
- 工作原理： 利用 Google Cloud 的 Vertex AI RAG（检索增强生成）服务。它将会话数据摄入到指定的 RAG Corpus 中，并使用强大的语义搜索能力进行检索。
- 持久性： 是。知识在配置的 Vertex AI RAG Corpus 中持久存储。
- 要求： Google Cloud 项目、适当的权限、必要的 SDK（pip install google-adk[vertexai]）以及预先配置的 Vertex AI RAG Corpus 资源名称/ID。
- 最适用于： 需要可扩展、持久且语义相关的知识检索的生产应用程序，特别是在 Google Cloud 上部署时。
```
# 要求：pip install google-adk[vertexai]
# 以及 GCP 设置、RAG Corpus 和认证
from google.adk.memory import VertexAiRagMemoryService

# RAG Corpus 名称或 ID
RAG_CORPUS_RESOURCE_NAME = "projects/your-gcp-project-id/locations/us-central1/ragCorpora/your-corpus-id"
# 检索的可选配置
SIMILARITY_TOP_K = 5
VECTOR_DISTANCE_THRESHOLD = 0.7

memory_service = VertexAiRagMemoryService(
    rag_corpus=RAG_CORPUS_RESOURCE_NAME,
    similarity_top_k=SIMILARITY_TOP_K,
    vector_distance_threshold=VECTOR_DISTANCE_THRESHOLD
)
```

Memory 在实践中如何工作¶

典型的工作流程包括以下步骤：

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

示例：添加和搜索 Memory¶

此示例演示了使用 InMemory 服务的基本流程，以简单起见。

完整代码

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 = session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=session1_id)
user_input1 = Content(parts=[Part(text="我最喜欢的项目是Project Alpha。")], role="user")

# 运行代理
final_response_text = "(No final response)"
for event in runner.run(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}")

# Get the completed session
completed_session1 = session_service.get_session(app_name=APP_NAME, user_id=USER_ID, session_id=session1_id)

# 将此会话的内容添加到 Memory Service
print("\n--- 将会话 1 添加到内存 ---")
memory_service.add_session_to_memory(completed_session1)
print("会话已添加到内存。")

# 回合 2：在*新*（或相同）会话中，提出需要内存的问题
print("\n--- 回合 2：回忆信息 ---")
session2_id = "session_recall" # 可以是相同或不同的会话 ID
session2 = session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=session2_id)

# 将 runner 切换到回忆智能体
runner.agent = memory_recall_agent
user_input2 = Content(parts=[Part(text="我最喜欢的项目是什么？")], role="user")

# 运行回忆智能体
print("运行 MemoryRecallAgent...")
final_response_text_2 = "(无最终响应)"
for event in runner.run(user_id=USER_ID, session_id=session2_id, new_message=user_input2):
    print(f"  事件: {event.author} - 类型: {'Text' if event.content and event.content.parts and event.content.parts[0].text else ''}"
        f"{'FuncCall' if event.get_function_calls() else ''}"
        f"{'FuncResp' if event.get_function_responses() else ''}")
    if event.is_final_response() and event.content and event.content.parts:
        final_response_text_2 = event.content.parts[0].text
        print(f"智能体 2 最终响应: {final_response_text_2}")
        break # 在最终响应后停止

# 回合 2 的预期事件序列：
# 1. 用户发送 "我最喜欢的项目是什么？"
# 2. 智能体（LLM）决定使用 `load_memory` 工具，查询如 "favorite project"。
# 3. Runner 执行 `load_memory` 工具，它调用 `memory_service.search_memory`。
# 4. `InMemoryMemoryService` 找到来自 session1 的相关文本（"我最喜欢的项目是 Project Alpha。"）。
# 5. 工具在 FunctionResponse 事件中返回此文本。
# 6. 智能体（LLM）接收函数响应，处理检索到的文本。
# 7. 智能体生成最终答案（例如，"你最喜欢的项目是 Project Alpha。"）。

Memory：使用 MemoryService 的长期知识¶

MemoryService 角色¶

MemoryService 实现¶

Memory 在实践中如何工作¶

示例：添加和搜索 Memory¶

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

`MemoryService` 角色¶

`MemoryService` 实现¶