第8章：内存管理 Memory Management

有效的内存管理对于智能体保留信息至关重要。智能体需要不同类型的内存，就像人类一样，以实现高效运作。本章深入探讨内存管理，特别是智能体的即时（短期）和持久（长期）内存需求。

在智能体系统中，内存指的是智能体保留并利用过去的交互、观察和学习经验的能力。这种能力使智能体能够做出明智的决策、保持对话上下文并随着时间推移不断改进。智能体内存通常分为两种主要类型：

短期内存（上下文内存）： 类似于工作内存，它存储当前正在处理或最近访问的信息。对于使用大型语言模型（LLM）的智能体来说，短期内存主要存在于上下文窗口中。上下文窗口包含最近的消息、智能体回复、工具使用结果以及当前交互中的智能体反思，这些信息共同影响LLM的后续响应和行为。上下文窗口的容量有限，限制了智能体可以直接访问的近期信息量。有效的短期内存管理需要在有限的空间内保留最相关的信息，这可能通过总结较早的对话片段或强调关键细节来实现。具有“长上下文”窗口的模型的出现仅仅是扩展了短期内存的大小，使得在单次交互中可以存储更多信息。然而，这种上下文仍然是短暂的，会在会话结束后丢失，并且每次处理所有信息可能成本高昂且效率低下。因此，智能体需要其他类型的内存来实现真正的持久性，能够从过去的交互中回忆信息并构建持久的知识库。
长期内存（持久内存）： 这充当智能体需要在多个交互、任务或较长时间内保留的信息的存储库，类似于长期知识库。数据通常存储在智能体的即时处理环境之外，通常使用数据库、知识图谱或向量数据库。在向量数据库中，信息被转换为数值向量并存储，从而使智能体能够基于语义相似性而不是精确的关键词匹配来检索数据，这一过程称为语义搜索。当智能体需要长期内存中的信息时，它会查询外部存储，检索相关数据，并将其整合到短期上下文中以供即时使用，从而将先前的知识与当前交互结合起来。

实际应用与使用场景

内存管理对于智能体跟踪信息并随着时间推移智能执行任务至关重要。这对于智能体超越基本的问答能力尤为重要。应用包括：

聊天机器人与会话式人工智能： 维持对话流畅性依赖于短期记忆。聊天机器人需要记住用户之前的输入，以提供连贯的回应。长期记忆使聊天机器人能够回忆用户偏好、过去的问题或之前的讨论，从而提供个性化和持续的互动。
任务导向型智能体： 管理多步骤任务的智能体需要短期记忆来跟踪之前的步骤、当前进度以及总体目标。这些信息可能存储在任务的上下文或临时存储中。长期记忆对于访问不在当前上下文中的特定用户相关数据至关重要。
个性化体验： 提供定制化互动的智能体利用长期记忆存储和检索用户偏好、过去行为以及个人信息。这使得智能体能够调整其响应和建议。
学习与改进： 智能体可以通过从过去的互动中学习来优化其性能。成功的策略、错误以及新信息存储在长期记忆中，促进未来的适应性。强化学习智能体通过这种方式存储学习到的策略或知识。
信息检索（RAG）： 专门用于回答问题的智能体访问知识库，即它们的长期记忆，通常通过检索增强生成（RAG）实现。智能体检索相关文档或数据以提供信息支持其响应。
自主系统： 机器人或自动驾驶汽车需要记忆地图、路线、物体位置以及学习到的行为。这包括用于即时环境的短期记忆和用于一般环境知识的长期记忆。

记忆使智能体能够维护历史记录、学习、个性化互动以及管理复杂的时间相关问题。

实践代码：Google Agent Developer Kit (ADK) 中的记忆管理

Google Agent Developer Kit (ADK) 提供了一种结构化的方法来管理上下文和记忆，包括用于实际应用的组件。对 ADK 的会话、状态和记忆的深入理解对于构建需要保留信息的智能体至关重要。

就像人类互动一样，智能体需要能够回忆之前的交流，以进行连贯自然的对话。ADK 通过三个核心概念及其相关服务简化了上下文管理。

与智能体的每次互动都可以被视为一个独特的对话线程。智能体可能需要访问先前互动中的数据。ADK 将其结构化如下：

会话（Session）： 一个单独的聊天线程，记录该特定互动的消息和动作（事件），同时存储与该对话相关的临时数据（状态）。
状态（State，session.state）： 存储在会话中的数据，仅包含与当前活动聊天线程相关的信息。
记忆（Memory）： 一个可搜索的信息库，来源于各种过去的聊天或外部资源，用于检索超出当前对话范围的数据。

ADK 提供专门的服务来管理构建复杂的、有状态的、上下文感知智能体所必需的关键组件。SessionService 通过处理会话对象的启动、记录和终止来管理聊天线程，而 MemoryService 负责长期知识（记忆）的存储和检索。

SessionService 和 MemoryService 都提供多种配置选项，允许用户根据应用需求选择存储方法。内存选项可用于测试目的，但数据不会在重启后持久化。为了实现持久存储和可扩展性，ADK 还支持数据库和基于云的服务。

会话：跟踪每次聊天

ADK 中的 Session 对象旨在跟踪和管理单个聊天线程。在与智能体开始对话时，SessionService 会生成一个 Session 对象，表示为 google.adk.sessions.Session。该对象封装了与特定对话线程相关的所有数据，包括唯一标识符（id、app_name、user_id）、按时间顺序记录的事件（Event 对象）、用于存储会话特定临时数据的存储区（state），以及表示最后更新时间的时间戳（last_update_time）。开发者通常通过 SessionService 间接与 Session 对象进行交互。SessionService 负责管理对话会话的生命周期，包括启动新会话、恢复之前的会话、记录会话活动（包括状态更新）、识别活动会话以及管理会话数据的删除。ADK 提供了多种 SessionService 实现，它们使用不同的存储机制来保存会话历史记录和临时数据，例如 InMemorySessionService，它适用于测试但无法在应用程序重启后提供数据持久性。

## 示例：使用 InMemorySessionService
## 适用于本地开发和测试，不需要在应用程序重启后保持数据持久性。
from google.adk.sessions import InMemorySessionService

session_service = InMemorySessionService()

如果需要可靠地保存到您管理的数据库，可以使用 DatabaseSessionService。

## 示例：使用 DatabaseSessionService
## 适用于需要持久存储的生产环境或开发环境。
## 需要配置数据库 URL（例如，SQLite、PostgreSQL 等）。
## 依赖项：pip install google-adk[sqlalchemy] 和数据库驱动（例如 PostgreSQL 的 psycopg2）
from google.adk.sessions import DatabaseSessionService

## 使用本地 SQLite 文件的示例：
db_url = "sqlite:///./my_agent_data.db"
session_service = DatabaseSessionService(db_url=db_url)

此外，还有 VertexAiSessionService，它使用 Google Cloud 的 Vertex AI 基础设施，适用于可扩展的生产环境。

## 示例：使用 VertexAiSessionService
## 适用于利用 Google Cloud Platform 的 Vertex AI 基础设施进行会话管理的可扩展生产环境。
## 依赖项：pip install google-adk[vertexai] 和 GCP 设置/认证
from google.adk.sessions import VertexAiSessionService

PROJECT_ID = "your-gcp-project-id" # 替换为您的 GCP 项目 ID
LOCATION = "us-central1" # 替换为您选择的 GCP 区域

## 使用此服务时，app_name 应与推理引擎的 ID 或名称对应
REASONING_ENGINE_APP_NAME = "projects/your-gcp-project-id/locations/us-central1/reasoningEngines/your-engine-id" # 替换为您的推理引擎资源名称

session_service = VertexAiSessionService(project=PROJECT_ID, location=LOCATION)

## 使用此服务时，将 REASONING_ENGINE_APP_NAME 传递给服务方法：
## session_service.create_session(app_name=REASONING_ENGINE_APP_NAME, ...)
## session_service.get_session(app_name=REASONING_ENGINE_APP_NAME, ...)
## session_service.append_event(session, event, app_name=REASONING_ENGINE_APP_NAME)
## session_service.delete_session(app_name=REASONING_ENGINE_APP_NAME, ...)

选择适当的 SessionService 至关重要，因为它决定了智能体的交互历史和临时数据的存储方式及其持久性。

每次消息交换都涉及一个循环过程：接收到消息后，Runner 使用 SessionService 检索或建立一个 Session，智能体通过 Session 的上下文（状态和历史交互）处理消息，生成响应并可能更新状态。然后，Runner 将其封装为一个 Event，session_service.append_event 方法记录新的事件并在存储中更新状态。Session 随后等待下一条消息。理想情况下，在交互结束时会使用 delete_session 方法终止会话。这个过程展示了 SessionService 如何通过管理特定 Session 的历史记录和临时数据来保持连续性。

状态：Session 的工作记事本

在 ADK 中，每个 Session（表示一个聊天线程）都包含一个状态组件，类似于智能体在特定对话期间的临时工作记忆。虽然 session.events 记录整个聊天历史，session.state 则存储并更新与当前聊天相关的动态数据点。

从本质上讲，session.state 作为一个字典运行，以键值对的形式存储数据。其核心功能是使智能体能够保留和管理对连贯对话至关重要的细节，例如用户偏好、任务进度、增量数据收集或影响后续智能体行为的条件标志。

状态的结构由字符串键和可序列化的 Python 类型值组成，包括字符串、数字、布尔值、列表以及包含这些基本类型的字典。状态是动态的，在整个对话中不断演变。这些变化的持久性取决于配置的 SessionService。

可以通过使用键前缀来定义数据范围和持久性来组织状态。没有前缀的键是特定于会话的。

user: 前缀将数据与用户 ID 关联，适用于所有会话。
app: 前缀指定数据在应用程序的所有用户之间共享。
temp: 前缀表示数据仅对当前处理周期有效，并且不会被持久存储。

智能体通过单一的 session.state 字典访问所有状态数据。SessionService 负责数据的检索、合并和持久化。状态应在通过 session_service.append_event() 将事件添加到会话历史记录时更新。这确保了准确跟踪、在持久服务中正确保存以及安全处理状态变化。

简单方法：使用 output_key（用于智能体文本回复）： 如果您只想将智能体的最终文本响应直接保存到状态，这是最简单的方法。当您设置 LlmAgent 时，只需指定您想要使用的 output_key。Runner 会识别这一点，并在追加事件时自动创建必要的操作以将响应保存到状态中。以下是通过 output_key 更新状态的代码示例：

## 从 Google Agent Developer Kit (ADK) 导入必要的类
from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService, Session
from google.adk.runners import Runner
from google.genai.types import Content, Part

## 定义一个带有 output_key 的 LlmAgent
greeting_agent = LlmAgent(
   name="Greeter",
   model="gemini-2.0-flash",
   instruction="Generate a short, friendly greeting.",
   output_key="last_greeting"
)

## --- 设置 Runner 和 Session ---
app_name, user_id, session_id = "state_app", "user1", "session1"
session_service = InMemorySessionService()
runner = Runner(
   agent=greeting_agent,
   app_name=app_name,
   session_service=session_service
)
session = session_service.create_session(
   app_name=app_name,
   user_id=user_id,
   session_id=session_id
)
print(f"初始状态: {session.state}")

## --- 运行智能体 ---
user_message = Content(parts=[Part(text="Hello")])
print("\n--- 运行智能体 ---")
for event in runner.run(
   user_id=user_id,
   session_id=session_id,
   new_message=user_message
):
   if event.is_final_response():
     print("智能体已响应。")

## --- 检查更新后的状态 ---
## 在 runner 完成处理所有事件后正确检查状态。
updated_session = session_service.get_session(app_name, user_id, session_id)
print(f"\n智能体运行后的状态: {updated_session.state}")

在后台，Runner 会查看你的 output_key 并自动创建必要的操作，同时在调用 append_event 时生成 state_delta。

标准方式：使用 EventActions.state_delta（用于更复杂的更新）： 当需要处理更复杂的情况时，例如同时更新多个键、保存非文本内容、针对特定范围（如 user: 或 app:）进行操作，或者进行与智能体最终文本回复无关的更新时，你需要手动构建一个状态更改字典（state_delta），并将其包含在你要追加的事件的 EventActions 中。以下是一个示例：

import time
from google.adk.tools.tool_context import ToolContext
from google.adk.sessions import InMemorySessionService

## --- 定义推荐的基于工具的方法 ---
def log_user_login(tool_context: ToolContext) -> dict:
   """
   在用户登录事件时更新会话状态。
   此工具封装了与用户登录相关的所有状态更改。
   参数:
       tool_context: 由 ADK 自动提供，允许访问会话状态。
   返回:
       一个字典，确认操作成功。
   """
   # 通过提供的上下文直接访问状态。
   state = tool_context.state

   # 获取当前值或默认值，然后更新状态。
   # 这种方式更清晰，并且将逻辑集中在一起。
   login_count = state.get("user:login_count", 0) + 1
   state["user:login_count"] = login_count
   state["task_status"] = "active"
   state["user:last_login_ts"] = time.time()
   state["temp:validation_needed"] = True

   print("状态已从 `log_user_login` 工具中更新。")

   return {
       "status": "success",
       "message": f"用户登录已记录。总登录次数: {login_count}."
   }

## --- 使用演示 ---
## 在实际应用中，LLM 智能体会决定调用此工具。
## 这里，我们模拟直接调用以进行演示。

## 1. 设置
session_service = InMemorySessionService()
app_name, user_id, session_id = "state_app_tool", "user3", "session3"
session = session_service.create_session(
   app_name=app_name,
   user_id=user_id,
   session_id=session_id,
   state={"user:login_count": 0, "task_status": "idle"}
)
print(f"初始状态: {session.state}")

## 2. 模拟工具调用（在实际应用中，ADK Runner 会执行此操作）
## 我们手动创建一个 ToolContext，仅用于此独立示例。
from google.adk.tools.tool_context import InvocationContext
mock_context = ToolContext(
   invocation_context=InvocationContext(
       app_name=app_name, user_id=user_id, session_id=session_id,
       session=session, session_service=session_service
   )
)

## 3. 执行工具
log_user_login(mock_context)

## 4. 检查更新后的状态
updated_session = session_service.get_session(app_name, user_id, session_id)
print(f"工具执行后的状态: {updated_session.state}")

## 预期输出将显示与“之前”情况相同的状态更改，
## 但代码组织显著更清晰和更健壮。

这段代码展示了一种基于工具的方法，用于管理应用程序中的用户会话状态。它定义了一个名为 log_user_login 的函数，该函数作为一个工具，用于在用户登录时更新会话状态。

该函数接收一个由 ADK 提供的 ToolContext 对象，用于访问和修改会话的状态字典。在工具内部，它会增加 user:login_count 的值，将 task_status 设置为 "active"，记录 user:last_login_ts (时间戳)，并添加一个临时标志 temp:validation_needed。

代码的演示部分模拟了该工具的使用方式。它设置了一个内存中的会话服务，并创建了一个具有预定义状态的初始会话。然后手动创建一个 ToolContext，以模拟 ADK Runner 执行工具的环境。接着，使用这个模拟的上下文调用 log_user_login 函数。最后，代码再次检索会话，以展示工具执行后状态的更新情况。其目标是展示将状态更改封装在工具中如何使代码比直接在工具外部操作状态更加简洁和有条理。

需要注意的是，强烈建议避免在检索会话后直接修改 `session.state` 字典，因为这种操作会绕过标准的事件处理机制。这种直接修改不会记录在会话的事件历史中，可能不会被选定的 `SessionService` 持久化，可能导致并发问题，并且不会更新诸如时间戳等重要元数据。推荐的会话状态更新方法是使用 `LlmAgent` 的 `output_key` 参数（专用于智能体的最终文本响应），或者在通过 `session_service.append_event()` 附加事件时，将状态更改包含在 `EventActions.state_delta` 中。`session.state` 应主要用于读取现有数据。

总结来说，在设计状态时，应保持简单，使用基本数据类型，为键命名清晰并正确使用前缀，避免深度嵌套，并始终通过 append_event 过程更新状态。

内存：使用 MemoryService 管理长期知识

在智能体系统中，Session 组件维护当前聊天历史记录（事件）和特定于单个对话的临时数据（状态）。然而，为了使智能体能够在多个交互中保留信息或访问外部数据，长期知识管理是必要的。这可以通过 MemoryService 来实现。

## 示例：使用 InMemoryMemoryService
## 适用于不需要在应用程序重启后保留数据的本地开发和测试。
## 应用程序停止时，内存内容将丢失。
from google.adk.memory import InMemoryMemoryService

memory_service = InMemoryMemoryService()

Session 和 State 可以被概念化为单个聊天会话的短期记忆，而由 MemoryService 管理的长期知识则充当一个持久且可搜索的存储库。该存储库可能包含多个过去交互或外部来源的信息。MemoryService 由 BaseMemoryService 接口定义，建立了管理这种可搜索的长期知识的标准。其主要功能包括添加信息（通过从会话中提取内容并使用 `add_session_to_memory` 方法存储），以及检索信息（允许智能体查询存储并使用 `search_memory` 方法接收相关数据）。

ADK 提供了多种实现方式来创建长期知识存储。InMemoryMemoryService 提供了适合测试用途的临时存储解决方案，但数据不会在应用程序重启后保留。在生产环境中，通常使用 VertexAiRagMemoryService。该服务利用 Google Cloud 的检索增强生成（RAG）服务，支持可扩展、持久化和语义搜索功能（另请参阅第 14 章关于 RAG 的内容）。

## 示例：使用 VertexAiRagMemoryService
## 适用于在 GCP 上进行可扩展的生产环境，利用 Vertex AI RAG（检索增强生成）
## 提供持久化、可搜索的内存。
## 要求：pip install google-adk[vertexai]，GCP 设置/认证，以及 Vertex AI RAG Corpus。
from google.adk.memory import VertexAiRagMemoryService

## Vertex AI RAG Corpus 的资源名称
RAG_CORPUS_RESOURCE_NAME = "projects/your-gcp-project-id/locations/us-central1/ragCorpora/your-corpus-id" # 替换为您的 Corpus 资源名称

## 可选的检索行为配置
SIMILARITY_TOP_K = 5 # 检索的前 K 个结果数量
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
)

## 使用此服务时，像 add_session_to_memory 和 search_memory 这样的方法
## 将与指定的 Vertex AI RAG Corpus 交互。

实操代码：LangChain 和 LangGraph 中的内存管理

在 LangChain 和 LangGraph 中，内存是创建智能且自然的对话应用程序的关键组件。它使 AI 智能体能够记住过去交互的信息，从反馈中学习，并根据用户偏好进行调整。LangChain 的内存功能通过引用存储的历史记录来丰富当前提示，并记录最新的交流以供将来使用。随着智能体处理更复杂的任务，这种能力对于提高效率和用户满意度至关重要。

短期内存： 短期内存是线程范围的，意味着它跟踪单个会话或线程中的对话。它提供即时上下文，但完整的历史记录可能会对 LLM 的上下文窗口造成挑战，从而可能导致错误或性能下降。LangGraph 将短期内存作为智能体状态的一部分进行管理，并通过检查点机制进行持久化，从而允许随时恢复线程。

长期内存： 长期内存存储用户特定或应用级别的数据，跨会话共享，并且可以在任何会话中随时调用。LangGraph 提供存储功能，用于保存和调用长期内存，使智能体能够无限期地保留知识。

LangChain 提供了多种工具来管理对话历史记录，从手动控制到在链中自动集成。

ChatMessageHistory：手动内存管理。 如果需要在正式链之外直接简单地控制对话历史记录，ChatMessageHistory 类是理想选择。它允许手动跟踪对话交换。

from langchain.memory import ChatMessageHistory

## 初始化历史记录对象
history = ChatMessageHistory()

## 添加用户和 AI 消息
history.add_user_message("我下周要去纽约。")
history.add_ai_message("太棒了！那是一个很棒的城市。")

## 访问消息列表
print(history.messages)

ConversationBufferMemory：用于链的自动内存管理。 如果需要将内存直接集成到链中，ConversationBufferMemory 是一个常见的选择。它保存对话的缓冲区，并将其提供给您的提示。其行为可以通过两个关键参数进行自定义：

memory_key: 一个字符串，用于指定在提示中保存聊天历史记录的变量名称。默认值为 "history"。
return_messages: 一个布尔值，用于决定历史记录的格式。
- 如果为 False（默认值），则返回一个格式化的单一字符串，这种格式适合标准的 LLMs。
- 如果为 True，则返回消息对象列表，这种格式是推荐用于聊天模型的。

from langchain.memory import ConversationBufferMemory

## 初始化 memory
memory = ConversationBufferMemory()

## 保存一次对话记录
memory.save_context({"input": "天气怎么样？"}, {"output": "今天是晴天。"})

## 以字符串形式加载 memory
print(memory.load_memory_variables({}))

将此 memory 集成到 LLMChain 中，可以让模型访问对话历史记录并提供上下文相关的响应。

from langchain_openai import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory

## 1. 定义 LLM 和 Prompt
llm = OpenAI(temperature=0)
template = """你是一位乐于助人的旅行智能体。
之前的对话：{history}
新问题：{question}
回复："""
prompt = PromptTemplate.from_template(template)

## 2. 配置 Memory
## memory_key "history" 与提示中的变量匹配
memory = ConversationBufferMemory(memory_key="history")

## 3. 构建 Chain
conversation = LLMChain(llm=llm, prompt=prompt, memory=memory)

## 4. 运行对话
response = conversation.predict(question="我想订一张机票。")
print(response)
response = conversation.predict(question="顺便说一下，我叫 Sam。")
print(response)
response = conversation.predict(question="你还记得我的名字吗？")
print(response)

为了在聊天模型中提高效果，建议通过设置 `return_messages=True` 使用结构化的消息对象列表。

from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import (
   ChatPromptTemplate,
   MessagesPlaceholder,
   SystemMessagePromptTemplate,
   HumanMessagePromptTemplate,
)

## 1. 定义聊天模型和 Prompt
llm = ChatOpenAI()
prompt = ChatPromptTemplate(
   messages=[
       SystemMessagePromptTemplate.from_template("你是一位友好的助手。"),
       MessagesPlaceholder(variable_name="chat_history"),
       HumanMessagePromptTemplate.from_template("{question}")
   ]
)

## 2. 配置 Memory
## 对于聊天模型，return_messages=True 是必需的
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

## 3. 构建 Chain
conversation = LLMChain(llm=llm, prompt=prompt, memory=memory)

## 4. 运行对话
response = conversation.predict(question="你好，我是 Jane。")
print(response)
response = conversation.predict(question="你还记得我的名字吗？")
print(response)

长期记忆的类型：长期记忆使系统能够在不同的对话中保留信息，从而提供更深层次的上下文和个性化服务。它可以分为三种类型，类似于人类记忆：

语义记忆：记住事实： 这涉及保留具体的事实和概念，例如用户偏好或领域知识。语义记忆用于支持智能体的响应，从而实现更个性化和相关的交互。这些信息可以作为一个持续更新的用户“档案”（一个 JSON 文档）或作为一组单独的事实文档的“集合”来管理。
情景记忆：记住经历： 这涉及回忆过去的事件或行为。对于人工智能体来说，情景记忆通常用于记住如何完成任务。在实践中，它通常通过少量示例提示（few-shot example prompting）来实现，智能体通过过去成功的交互序列学习如何正确执行任务。
程序记忆：记住规则： 这是关于如何执行任务的记忆——智能体的核心指令和行为，通常包含在其系统提示中。智能体通常会修改自己的提示以适应和改进。一种有效的技术是“反思”（Reflection），即智能体通过当前的指令和最近的交互进行提示，然后被要求优化自己的指令。

以下是伪代码示例，展示智能体如何使用反思来更新存储在 LangGraph BaseStore 中的程序记忆：

## 更新智能体指令的节点
def update_instructions(state: State, store: BaseStore):
   namespace = ("instructions",)
   # 从存储中获取当前指令
   current_instructions = store.search(namespace)[0]

   # 创建一个提示，要求 LLM 对对话进行反思
   # 并生成新的、改进的指令
   prompt = prompt_template.format(
       instructions=current_instructions.value["instructions"],
       conversation=state["messages"]
   )

   # 从 LLM 获取新的指令
   output = llm.invoke(prompt)
   new_instructions = output['new_instructions']

   # 将更新后的指令保存回存储
   store.put(("agent_instructions",), "agent_a", {"instructions": new_instructions})

## 使用指令生成响应的节点
def call_model(state: State, store: BaseStore):
   namespace = ("agent_instructions", )
   # 从存储中检索最新的指令
   instructions = store.get(namespace, key="agent_a")[0]

   # 使用检索到的指令格式化提示
   prompt = prompt_template.format(instructions=instructions.value["instructions"])
   # ... 应用逻辑继续

LangGraph 将长期记忆以 JSON 文档的形式存储在存储中。每个记忆都以自定义命名空间（类似文件夹）和唯一键（类似文件名）进行组织。这种层次结构便于信息的组织和检索。以下代码展示了如何使用 InMemoryStore 来存储、获取和搜索记忆。

from langgraph.store.memory import InMemoryStore

## 一个真实嵌入函数的占位符
def embed(texts: list[str]) -> list[list[float]]:
   # 在实际应用中，请使用合适的嵌入模型
   return [[1.0, 2.0] for _ in texts]

## 初始化一个内存存储。在生产环境中，请使用基于数据库的存储。
store = InMemoryStore(index={"embed": embed, "dims": 2})

## 为特定用户和应用上下文定义一个命名空间
user_id = "my-user"
application_context = "chitchat"
namespace = (user_id, application_context)

## 1. 将记忆存入存储
store.put(
   namespace,
   "a-memory",  # 此记忆的键
   {
       "rules": [
           "用户喜欢简短直接的语言",
           "用户只使用英语和 Python",
       ],
       "my-key": "my-value",
   },
)

## 2. 根据命名空间和键获取记忆
item = store.get(namespace, "a-memory")
print("Retrieved Item:", item)

## 3. 在命名空间内搜索记忆，按内容过滤并根据与查询的向量相似度排序
items = store.search(
   namespace,
   filter={"my-key": "my-value"},
   query="language preferences"
)
print("Search Results:", items)

Vertex Memory Bank

Memory Bank 是 Vertex AI Agent Engine 中的一个托管服务，为智能体提供持久的长期记忆功能。该服务使用 Gemini 模型异步分析对话历史，以提取关键事实和用户偏好。

这些信息被持久存储，并按照定义的范围（如用户 ID）进行组织，同时智能更新以整合新数据并解决矛盾。在开始新会话时，智能体通过完整数据回忆或使用嵌入向量的相似性搜索来检索相关记忆。此过程使智能体能够在会话之间保持连续性，并基于回忆的信息个性化响应。

智能体的运行器与 VertexAiMemoryBankService 交互，该服务首先被初始化。此服务负责自动存储智能体对话期间生成的记忆。每个记忆都带有唯一的 USER_ID 和 APP_NAME 标签，以确保未来能够准确检索。

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
)

session = await session_service.get_session(
   app_name=app_name,
   user_id="USER_ID",
   session_id=session.id
)

await memory_service.add_session_to_memory(session)

Memory Bank 与 Google ADK 无缝集成，提供即开即用的体验。对于其他智能体框架的用户，例如 LangGraph 和 CrewAI，Memory Bank 也通过直接 API 调用提供支持。在线代码示例展示了这些集成，供感兴趣的读者参考。

概览

定义（What）

智能体系统需要记住过去交互中的信息，以执行复杂任务并提供连贯的体验。如果没有记忆机制，智能体将是无状态的，无法维护对话上下文、从经验中学习或为用户提供个性化响应。这会从根本上限制它们只能进行简单的单次交互，无法处理多步骤流程或不断变化的用户需求。核心问题在于如何有效管理单次对话中的即时临时信息以及长期积累的大量持久知识。

设计意图（Why）

标准化解决方案是实现一个区分短期存储和长期存储的双组件记忆系统。短期上下文记忆在 LLM 的上下文窗口内保存最近的交互数据，以维持对话流。对于需要持久保存的信息，长期记忆解决方案通常使用外部数据库（通常是向量存储）进行高效的语义检索。像 Google ADK 这样的智能体框架提供了管理这些记忆的特定组件，例如用于对话线程的 Session 和用于临时数据的 State。专用的 MemoryService 用于连接长期知识库，使智能体能够检索并将相关的过去信息整合到当前上下文中。

使用原则（Rule of Thumb）

当智能体需要执行超过回答单个问题的任务时，使用这种模式是必要的。对于需要在整个对话中保持上下文、跟踪多步骤任务进度或通过回忆用户偏好和历史记录来个性化交互的智能体来说，记忆管理至关重要。每当智能体需要根据过去的成功、失败或新获取的信息进行学习或适应时，都应实施记忆管理。

图解（Visual Summary）

图1：记忆管理设计模式

关键要点

快速回顾有关记忆管理的主要内容：

记忆对智能体而言至关重要，它帮助智能体跟踪信息、学习以及个性化交互。
会话式人工智能依赖短期记忆来处理单次对话中的即时上下文，同时依赖长期记忆来存储跨多个会话的持久知识。
短期记忆（即时信息）是临时的，通常受限于LLM的上下文窗口或框架传递上下文的方式。
长期记忆（持久信息）通过使用外部存储（如向量数据库）保存跨不同对话的信息，并通过搜索进行访问。
像ADK这样的框架有特定的组件，例如Session（对话线程）、State（临时对话数据）和MemoryService（可搜索的长期知识）来管理记忆。
ADK的SessionService负责管理整个对话会话的生命周期，包括其历史记录（事件）和临时数据（状态）。
ADK的session.state是一个用于存储临时对话数据的字典。前缀（如user:、app:、temp:）表明数据的归属以及是否是持久数据。
在ADK中，更新状态时应通过使用EventActions.state_delta或output_key来添加事件，而不是直接修改状态字典。
ADK的MemoryService用于将信息存储到长期存储中，并允许智能体通过工具进行搜索。
LangChain提供了实用工具，例如ConversationBufferMemory，可以自动将单次对话的历史注入到提示中，使智能体能够回忆起即时上下文。
LangGraph通过使用存储来保存和检索语义事实、情节体验，甚至是可更新的程序规则，支持高级的长期记忆功能，适用于不同用户会话。
Memory Bank是一种托管服务，通过自动提取、存储和回忆用户特定信息，为智能体提供持久的长期记忆，从而在Google的ADK、LangGraph和CrewAI等框架中实现个性化、连续的对话。

结论

本章深入探讨了智能体系统中记忆管理的重要作用，展示了短期记忆和长期记忆之间的区别。我们讨论了这些记忆类型的设置方式以及它们在构建能够记住信息的智能体中的应用。详细介绍了Google ADK如何通过Session、State和MemoryService等组件来处理记忆。既然我们已经了解了智能体如何记住信息，无论是短期还是长期记忆，接下来我们将探讨它们如何学习和适应。下一章节“学习与适应”将讨论智能体如何根据新的经验或数据改变其思维、行为或知识。

参考文献

ADK Memory, https://google.github.io/adk-docs/sessions/memory/
LangGraph Memory, https://langchain-ai.github.io/langgraph/concepts/memory/
Vertex AI Agent Engine Memory Bank, https://cloud.google.com/blog/products/ai-machine-learning/vertex-ai-memory-bank-in-public-preview

第8章：记忆管理

第8章：内存管理 Memory Management

实际应用与使用场景

实践代码：Google Agent Developer Kit (ADK) 中的记忆管理

会话：跟踪每次聊天

状态：Session 的工作记事本

内存：使用 MemoryService 管理长期知识

实操代码：LangChain 和 LangGraph 中的内存管理

Vertex Memory Bank

概览

定义（What）

设计意图（Why）

使用原则（Rule of Thumb）

图解（Visual Summary）

关键要点

结论

参考文献

results matching ""

No results matching ""

第8章：内存管理 Memory Management

实际应用与使用场景

实践代码：Google Agent Developer Kit (ADK) 中的记忆管理

会话：跟踪每次聊天

状态：Session 的工作记事本

内存：使用 MemoryService 管理长期知识

实操代码：LangChain 和 LangGraph 中的内存管理

Vertex Memory Bank

概览

定义（What）

设计意图（Why）

使用原则（Rule of Thumb）

图解 （Visual Summary）

关键要点

结论

参考文献

results matching ""

No results matching ""

图解（Visual Summary）