Memory: Long-Term Knowledge with MemoryService¶
记忆:使用 MemoryService 的长期知识¶
We've seen how Session tracks the history (events) and temporary data (state) for a single, ongoing conversation. But what if an agent needs to recall information from past conversations? This is where to concept of Long-Term Knowledge and MemoryService come into play.
我们已经了解了 Session 如何跟踪单个、正在进行的对话的历史记录(events)和临时数据(state)。但是,如果智能体需要从过去的对话中回忆信息怎么办?这就是长期知识和MemoryService的概念发挥作用的地方。
Think of it this way:
可以这样理解:
Session/State: Like your short-term memory during one specific chat.Session/State: 类似于您在特定聊天期间的短期记忆。- Long-Term Knowledge (
MemoryService): Like a searchable archive or knowledge library agent can consult, potentially containing information from many past chats or other sources. 长期知识(MemoryService): 类似于智能体可以查询的可搜索档案或知识库,可能包含来自许多过去聊天或其他来源的信息。
The MemoryService Role¶
MemoryService 的作用¶
The BaseMemoryService defines the interface for managing this searchable, long-term knowledge store. Its primary responsibilities are:
BaseMemoryService 定义了管理此可搜索长期知识存储的接口。其主要职责是:
- Ingesting Information (
add_session_to_memory): Taking the contents of a (usually completed)Sessionand adding relevant information to the long-term knowledge store. 摄取信息(add_session_to_memory): 获取(通常已完成的)Session的内容,并将相关信息添加到长期知识存储中。 - Searching Information (
search_memory): Allowing an agent (typically via aTool) to query to knowledge store and retrieve relevant snippets or context based on a search query. 搜索信息(search_memory): 允许智能体(通常通过Tool)查询知识存储,并根据搜索查询检索相关片段或上下文。
Choosing to Right Memory Service¶
选择正确的记忆服务¶
The ADK offers two distinct MemoryService implementations, each tailored to different use cases. Use to table below to decide which is to best fit for your agent.
ADK 提供两种不同的 MemoryService 实现,每种都针对不同的用例。使用下面的表格来决定哪种最适合您的智能体。
| Feature | InMemoryMemoryService | VertexAiMemoryBankService | |
|---|---|---|---|
| Persistence | None (data is lost on restart) | Yes (Managed by Vertex AI) | |
| 持久性 | 无(重启时数据丢失) | 是(由 Vertex AI 管理) | |
| Primary Use Case | Prototyping, local development, and simple testing. | Building meaningful, evolving memories from user conversations. | |
| 主要用例 | 原型设计、本地开发和简单测试。 | 从用户对话中构建有意义的、不断发展的记忆。 | |
| Memory Extraction | Stores full conversation | Extracts meaningful information from conversations and consolidates it with existing memories (powered by LLM) | |
| 记忆提取 | 存储完整对话 | 从对话中提取有意义的信息并将其与现有记忆整合(由 LLM 提供支持) | |
| Search Capability | Basic keyword matching. | Advanced semantic search. | |
| 搜索能力 | 基本关键字匹配。 | 高级语义搜索。 | |
| Setup Complexity | None. It's to default. | Low. Requires an Agent Engine instance in Vertex AI. | |
| 设置复杂性 | 无。它是默认设置。 | 低。需要在 Vertex AI 中创建一个Agent Engine实例。 | |
| Dependencies | None. | Google Cloud Project, Vertex AI API | |
| 依赖项 | 无。 | Google Cloud 项目、Vertex AI API | |
| When to use it | When you want to search across multiple sessions' chat histories for prototyping. | When you want your agent to remember and learn from past interactions. | |
| 何时使用 | 当您想在原型的多个会话聊天历史中搜索时。 | 当您希望您的智能体记住并从过去的交互中学习时。 |
In-Memory Memory¶
内存记忆¶
The InMemoryMemoryService stores session information in the application's memory and performs basic keyword matching for searches. It requires no setup and is best for prototyping and simple testing scenarios where persistence isn't required.
InMemoryMemoryService 将会话信息存储在应用程序的内存中,并执行基本的关键字匹配进行搜索。它不需要设置,最适合不需要持久性的原型设计和简单测试场景。
Example: Adding and Searching Memory 示例:添加和搜索记忆
This example demonstrates to basic flow using InMemoryMemoryService for simplicity.
此示例演示了使用 InMemoryMemoryService 的基本流程。
import asyncio
from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService, Session
from google.adk.memory import InMemoryMemoryService # Import MemoryService
# 导入 MemoryService
from google.adk.runners import Runner
from google.adk.tools import load_memory # Tool to query memory
# 用于查询记忆的工具
from google.genai.types import Content, Part
# --- Constants ---
# --- 常量 ---
APP_NAME = "memory_example_app"
USER_ID = "mem_user"
MODEL = "gemini-2.0-flash" # Use to valid model
# 使用有效的模型
# --- Agent Definitions ---
# --- 智能体定义 ---
# Agent 1: Simple agent to capture information
# 智能体 1:简单的信息捕获智能体
info_capture_agent = LlmAgent(
model=MODEL,
name="InfoCaptureAgent",
instruction="Acknowledge to user's statement.",
# 确认用户的陈述。
)
# Agent 2: Agent that can use memory
# 智能体 2:可以使用记忆的智能体
memory_recall_agent = LlmAgent(
model=MODEL,
name="MemoryRecallAgent",
instruction="Answer to user's question. Use to 'load_memory' tool "
"if answer might be in past conversations.",
# 回答用户的问题。如果答案可能在过去的对话中,请使用 'load_memory' 工具。
tools=[load_memory] # Give to agent to tool
# 给智能体工具
)
# --- Services ---
# --- 服务 ---
# Services must be shared across runners to share state and memory
# 服务必须在运行程序之间共享以共享状态和记忆
session_service = InMemorySessionService()
memory_service = InMemoryMemoryService() # Use in-memory for demo
# 演示使用内存
async def run_scenario():
# --- Scenario ---
# --- 场景 ---
# Turn 1: Capture some information in a session
# 第 1 轮:在会话中捕获一些信息
print("--- Turn 1: Capturing Information ---")
runner1 = Runner(
# Start with the info capture agent
# 从信息捕获智能体开始
agent=info_capture_agent,
app_name=APP_NAME,
session_service=session_service,
memory_service=memory_service # Provide to memory service to to Runner
# 向运行程序提供记忆服务
)
session1_id = "session_info"
await runner1.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")
# Run to agent
# 运行智能体
final_response_text = "(No final response)"
async for event in runner1.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"Agent 1 Response: {final_response_text}")
# Get to completed session
# 获取已完成的会话
completed_session1 = await runner1.session_service.get_session(app_name=APP_NAME, user_id=USER_ID, session_id=session1_id)
# Add this session's content to the Memory Service
# 将此会话的内容添加到记忆服务
print("\n--- Adding Session 1 to Memory ---")
await memory_service.add_session_to_memory(completed_session1)
print("Session added to memory.")
# Turn 2: Recall to information in a new session
# 第 2 轮:在新会话中回忆信息
print("\n--- Turn 2: Recalling Information ---")
runner2 = Runner(
# Use to second agent, which has to memory tool
# 使用第二个智能体,它具有记忆工具
agent=memory_recall_agent,
app_name=APP_NAME,
session_service=session_service, # Reuse to same service
# 重用相同的服务
memory_service=memory_service # Reuse to same service
# 重用相同的服务
)
session2_id = "session_recall"
await runner2.session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=session2_id)
user_input2 = Content(parts=[Part(text="What is my favorite project?")], role="user")
# Run to second agent
# 运行第二个智能体
final_response_text_2 = "(No final response)"
async for event in runner2.run_async(user_id=USER_ID, session_id=session2_id, new_message=user_input2):
if event.is_final_response() and event.content and event.content.parts:
final_response_text_2 = event.content.parts[0].text
print(f"Agent 2 Response: {final_response_text_2}")
# To run this example, you can use to following snippet:
# 要运行此示例,您可以使用以下代码片段:
# asyncio.run(run_scenario())
# await run_scenario()
import (
"context"
"fmt"
"log"
"strings"
"google.golang.org/adk/agent"
"google.golang.org/adk/agent/llmagent"
"google.golang.org/adk/memory"
"google.golang.org/adk/model/gemini"
"google.golang.org/adk/runner"
"google.golang.org/adk/session"
"google.golang.org/adk/tool"
"google.golang.org/adk/tool/functiontool"
"google.golang.org/genai"
)
const (
appName = "go_memory_example_app"
userID = "go_mem_user"
modelID = "gemini-2.5-pro"
)
// Args defines the input structure for the memory search tool.
type Args struct {
Query string `json:"query" jsonschema:"The query to search for in the memory."`
}
// Result defines the output structure for the memory search tool.
type Result struct {
Results []string `json:"results"`
}
// memorySearchToolFunc is the implementation of the memory search tool.
// This function demonstrates accessing memory via tool.Context.
func memorySearchToolFunc(tctx tool.Context, args Args) (Result, error) {
fmt.Printf("Tool: Searching memory for query: '%s'\n", args.Query)
// The SearchMemory function is available on the context.
searchResults, err := tctx.SearchMemory(context.Background(), args.Query)
if err != nil {
log.Printf("Error searching memory: %v", err)
return Result{}, fmt.Errorf("failed memory search")
}
var results []string
for _, res := range searchResults.Memories {
if res.Content != nil {
results = append(results, textParts(res.Content)...)
}
}
return Result{Results: results}, nil
}
// Define a tool that can search memory.
var memorySearchTool = must(functiontool.New(
functiontool.Config{
Name: "search_past_conversations",
Description: "Searches past conversations for relevant information.",
},
memorySearchToolFunc,
))
// This example demonstrates how to use the MemoryService in the Go ADK.
// It covers two main scenarios:
// 1. Adding a completed session to memory and recalling it in a new session.
// 2. Searching memory from within a custom tool using the tool.Context.
func main() {
ctx := context.Background()
// --- Services ---
// Services must be shared across runners to share state and memory.
sessionService := session.InMemoryService()
memoryService := memory.InMemoryService() // Use in-memory for this demo.
// --- Scenario 1: Capture information in one session ---
fmt.Println("--- Turn 1: Capturing Information ---")
infoCaptureAgent := must(llmagent.New(llmagent.Config{
Name: "InfoCaptureAgent",
Model: must(gemini.NewModel(ctx, modelID, nil)),
Instruction: "Acknowledge the user's statement.",
}))
runner1 := must(runner.New(runner.Config{
AppName: appName,
Agent: infoCaptureAgent,
SessionService: sessionService,
MemoryService: memoryService, // Provide the memory service to the Runner
}))
session1ID := "session_info"
must(sessionService.Create(ctx, &session.CreateRequest{AppName: appName, UserID: userID, SessionID: session1ID}))
userInput1 := genai.NewContentFromText("My favorite project is Project Alpha.", "user")
var finalResponseText string
for event, err := range runner1.Run(ctx, userID, session1ID, userInput1, agent.RunConfig{}) {
if err != nil {
log.Printf("Agent 1 Error: %v", err)
continue
}
if event.Content != nil && !event.LLMResponse.Partial {
finalResponseText = strings.Join(textParts(event.LLMResponse.Content), "")
}
}
fmt.Printf("Agent 1 Response: %s\n", finalResponseText)
// Add the completed session to the Memory Service
fmt.Println("\n--- Adding Session 1 to Memory ---")
resp, err := sessionService.Get(ctx, &session.GetRequest{AppName: appName, UserID: userID, SessionID: session1ID})
if err != nil {
log.Fatalf("Failed to get completed session: %v", err)
}
if err := memoryService.AddSession(ctx, resp.Session); err != nil {
log.Fatalf("Failed to add session to memory: %v", err)
}
fmt.Println("Session added to memory.")
// --- Scenario 2: Recall the information in a new session using a tool ---
fmt.Println("\n--- Turn 2: Recalling Information ---")
memoryRecallAgent := must(llmagent.New(llmagent.Config{
Name: "MemoryRecallAgent",
Model: must(gemini.NewModel(ctx, modelID, nil)),
Instruction: "Answer the user's question. Use the 'search_past_conversations' tool if the answer might be in past conversations.",
Tools: []tool.Tool{memorySearchTool}, // Give the agent the tool
}))
runner2 := must(runner.New(runner.Config{
Agent: memoryRecallAgent,
AppName: appName,
SessionService: sessionService,
MemoryService: memoryService,
}))
session2ID := "session_recall"
must(sessionService.Create(ctx, &session.CreateRequest{AppName: appName, UserID: userID, SessionID: session2ID}))
userInput2 := genai.NewContentFromText("What is my favorite project?", "user")
var finalResponseText2 string
for event, err := range runner2.Run(ctx, userID, session2ID, userInput2, agent.RunConfig{}) {
if err != nil {
log.Printf("Agent 2 Error: %v", err)
continue
}
if event.Content != nil && !event.LLMResponse.Partial {
finalResponseText2 = strings.Join(textParts(event.LLMResponse.Content), "")
}
}
fmt.Printf("Agent 2 Response: %s\n", finalResponseText2)
}
Searching Memory Within a Tool¶
在工具中搜索记忆¶
You can also search memory from within a custom tool by using tool.Context.
您还可以使用 tool.Context 在自定义工具中搜索记忆。
// memorySearchToolFunc is the implementation of the memory search tool.
// This function demonstrates accessing memory via tool.Context.
func memorySearchToolFunc(tctx tool.Context, args Args) (Result, error) {
fmt.Printf("Tool: Searching memory for query: '%s'\n", args.Query)
// The SearchMemory function is available on the context.
searchResults, err := tctx.SearchMemory(context.Background(), args.Query)
if err != nil {
log.Printf("Error searching memory: %v", err)
return Result{}, fmt.Errorf("failed memory search")
}
var results []string
for _, res := range searchResults.Memories {
if res.Content != nil {
results = append(results, textParts(res.Content)...)
}
}
return Result{Results: results}, nil
}
// Define a tool that can search memory.
var memorySearchTool = must(functiontool.New(
functiontool.Config{
Name: "search_past_conversations",
Description: "Searches past conversations for relevant information.",
},
memorySearchToolFunc,
))
Vertex AI Memory Bank¶
Vertex AI 记忆库¶
The VertexAiMemoryBankService connects your agent to Vertex AI Memory Bank, to fully managed Google Cloud service that provides sophisticated, persistent memory capabilities for conversational agents.
VertexAiMemoryBankService 将您的智能体连接到 Vertex AI Memory Bank,这是一项完全托管的 Google Cloud 服务,为对话智能体提供复杂的、持久的记忆功能。
How It Works¶
工作原理¶
The service handles two key operations:
该服务处理两个关键操作:
- Generating Memories: At to end of to conversation, you can send to session's events to Memory Bank, which intelligently processes and stores to information as "memories." 生成记忆: 在对话结束时,您将会话的事件发送到记忆库,它会智能地处理并将信息存储为"记忆"。
- Retrieving Memories: Your agent code can issue to search query against Memory Bank to retrieve relevant memories from past conversations. 检索记忆: 您的智能体代码可以针对记忆库发出搜索查询,以从过去的对话中检索相关记忆。
Prerequisites¶
先决条件¶
Before you can use this feature, you must have:
在使用此功能之前,您必须具备:
- A Google Cloud Project: With the Vertex AI API enabled. Google Cloud 项目:已启用 Vertex AI API。
- An Agent Engine: You need to create an Agent Engine in Vertex AI. You do not need to deploy your agent to Agent Engine Runtime to use Memory Bank. This will provide you with to Agent Engine ID required for configuration. Agent Engine:您需要在 Vertex AI 中创建 Agent Engine。您不需要将智能体部署到 Agent Engine Runtime 即可使用记忆库。这将为您提供配置所需的Agent Engine ID。
- Authentication: Ensure your local environment is authenticated to access Google Cloud services. The simplest way is to run: 身份验证:确保您的本地环境已通过身份验证以访问 Google Cloud 服务。最简单的方法是运行:
- Environment Variables: The service requires your Google Cloud Project ID and Location. Set them as environment variables: 环境变量:服务需要您的 Google Cloud 项目 ID 和位置。将它们设置为环境变量:
Configuration¶
配置¶
To connect your agent to Memory Bank, you use --memory_service_uri flag when starting to ADK server (adk web or adk api_server). The URI must be in to format agentengine://<agent_engine_id>.
要将智能体连接到记忆库,请在启动 ADK 服务器(adk web 或 adk api_server)时使用 --memory_service_uri 标志。URI 必须采用 agentengine://<agent_engine_id> 格式。
Or, you can configure your agent to use to Memory Bank by manually instantiating VertexAiMemoryBankService and passing it to Runner.
或者,您可以通过手动实例化 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
)
Using Memory in Your Agent¶
在智能体中使用记忆¶
When to memory service is configured, your agent can use to tool or callback to retrieve memories. ADK includes two pre-built tools for retrieving memories:
当配置了记忆服务时,您的智能体可以使用工具或回调来检索记忆。ADK 包含两个用于检索记忆的预构建工具:
PreloadMemory: Always retrieve memory at to beginning of each turn (similar to a callback).PreloadMemory:总是在每轮开始时检索记忆(类似于回调)。LoadMemory: Retrieve memory when your agent decides it would be helpful.LoadMemory:当您的智能体决定它会有帮助时检索记忆。
Example: 示例:
from google.adk.agents import Agent
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
agent = Agent(
model=MODEL_ID,
name='weather_sentiment_agent',
instruction="...",
tools=[PreloadMemoryTool()]
)
To extract memories from your session, you need to call add_session_to_memory. For example, you can automate this via a callback:
要从会话中提取记忆,您需要调用 add_session_to_memory。例如,您可以通过回调自动化此操作:
from google import adk
async def auto_save_session_to_memory_callback(callback_context):
await callback_context._invocation_context.memory_service.add_session_to_memory(
callback_context._invocation_context.session)
agent = Agent(
model=MODEL,
name="Generic_QA_Agent",
instruction="Answer to user's questions",
# 回答用户的问题
tools=[adk.tools.preload_memory_tool.PreloadMemoryTool()],
after_agent_callback=auto_save_session_to_memory_callback,
)
Advanced Concepts¶
高级概念¶
How Memory Works in Practice¶
记忆在实际中如何工作¶
The memory workflow internally involves these steps:
记忆工作流内部涉及以下步骤:
- Session Interaction: A user interacts with an agent via a
Session, managed by toSessionService. Events are added, and state might be updated. 会话交互:用户通过Session与智能体交互,由SessionService管理。添加事件,并可能更新状态。 - Ingestion into Memory: At some point (often when to session is considered complete or has yielded significant information), your application calls
memory_service.add_session_to_memory(session). This extracts relevant information from to session's events and adds it to to long-term knowledge store (in-memory dictionary or Agent Engine Memory Bank). 摄取到记忆中: 在某个时刻(通常当会话被认为已完成或产生了重要信息时),您的应用程序调用memory_service.add_session_to_memory(session)。这会从会话的事件中提取相关信息并将其添加到长期知识存储(内存字典或 Agent Engine 记忆库)中。 - Later Query: In a different (or to same) session, user might ask to question requiring past context (e.g., "What did we discuss about project X last week?"). 后续查询: 在不同的(或相同的)会话中,用户可能会问一个需要过去上下文的问题(例如,"上周我们讨论了关于项目 X 的什么?")。
- Agent Uses Memory Tool: An agent equipped with to memory-retrieval tool (like to built-in
load_memorytool) recognizes need for past context. It calls to tool, providing to search query (e.g., "discussion project X last week"). 智能体使用记忆工具:配备记忆检索工具(如内置的load_memory工具)的智能体认识到需要过去的上下文。它调用该工具,提供搜索查询(例如,"讨论项目 X 上周")。 - Search Execution: The tool internally calls
memory_service.search_memory(app_name, user_id, query). 搜索执行:工具在内部调用memory_service.search_memory(app_name, user_id, query)。 - Results Returned: The
MemoryServicesearches its store (using keyword matching or semantic search) and returns relevant snippets as toSearchMemoryResponsecontaining to list ofMemoryResultobjects (each potentially holding events from to relevant past session). 返回结果:MemoryService搜索其存储(使用关键字匹配或语义搜索),并将相关片段作为包含MemoryResult对象列表的SearchMemoryResponse返回(每个对象可能包含来自相关过去会话的事件)。 - Agent Uses Results: The tool returns these results to to agent, usually as part of context or function response. The agent can then use this retrieved information to formulate its final answer to to user. 智能体使用结果:工具将这些结果返回给智能体,通常作为上下文或函数响应的一部分。智能体然后可以使用这些检索到的信息来制定对用户的最终回答。
Can an agent have access to more than one memory service?¶
智能体可以访问多个记忆服务吗?¶
-
Through Standard Configuration: No. The framework (
adk web,adk api_server) is designed to be configured with one single memory service at to time via--memory_service_uriflag. This single service is then provided to to agent and accessed through to built-inself.search_memory()method. From to configuration standpoint, you can only choose one backend (InMemory,VertexAiMemoryBankService) for all agents served by that process. 通过标准配置:否。框架(adk web、adk api_server)设计为通过--memory_service_uri标志一次配置一个记忆服务。然后,此单个服务被提供给智能体并通过内置的self.search_memory()方法访问。从配置角度来看,您只能为该进程服务的所有智能体选择一个后端(InMemory、VertexAiMemoryBankService)。 -
Within Your Agent's Code: Yes, absolutely. There is nothing preventing you from manually importing and instantiating another memory service directly inside your agent's code. This allows you to access multiple memory sources within to single agent turn. 在您的智能体代码中:是的,完全可以。没有什么可以阻止您在智能体代码内部手动导入和实例化另一个记忆服务。这允许您在单个智能体轮次中访问多个记忆源。
For example, your agent could use to framework-configured InMemoryMemoryService to recall conversational history, and also manually instantiate to VertexAiMemoryBankService to look up information in to technical manual.
例如,您的智能体可以使用框架配置的 InMemoryMemoryService 来回忆对话历史,还可以手动实例化 VertexAiMemoryBankService 来在技术手册中查找信息。
Example: Using Two Memory Services¶
示例:使用两个记忆服务¶
Here's how you could implement that in your agent's code:
以下是如何在智能体代码中实现该功能:
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()
# Manually instantiate to second memory service for document lookups
# 手动实例化第二个记忆服务用于文档查找
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. Search conversational history using to framework-provided memory
# (This would be InMemoryMemoryService if configured)
# 使用框架提供的记忆搜索对话历史
# (如果已配置,这将是 InMemoryMemoryService)
conversation_context = await self.memory_service.search_memory(query=user_query)
# 2. Search to document knowledge base using to manually created service
# 使用手动创建的服务搜索文档知识库
document_context = await self.vertexai_memorybank_service.search_memory(query=user_query)
# Combine to context from both sources to generate to better response
# 结合来自两个来源的上下文以生成更好的响应
prompt = "From our past conversations, I remember:\n"
# 从我们过去的对话中,我记得:
prompt += f"{conversation_context.memories}\n\n"
prompt += "From to technical manuals, I found:\n"
# 从技术手册中,我发现:
prompt += f"{document_context.memories}\n\n"
prompt += f"Based on all this, here is my answer to '{user_query}':"
# 基于所有这些,这是我对 '{user_query}' 的回答:
return await self.llm.generate_content_async(prompt)