《Agentic Design Patterns：构建智能系统的实战指南》- 第八章内存管理

书籍名称：Agentic Design Patterns: A Hands-On Guide to Building Intelligent Systems
本书作者：Antonio Gulli
链接地址：https://docs.google.com/document/d/18vvNESEwHnVUREzIipuaDNCnNAREGqEfy9MQYC9wb4o
内容摘要：本文是对《智能体设计模式》第八章的翻译。此章节介绍了智能体系统的内存管理机制。
翻译：煤矿工厂

摘要

高效的内存管理对于智能体保留信息至关重要。与人类一样，智能体需要不同类型的内存才能高效运作。本章深入探讨内存管理，重点关注智能体的短期和长期内存需求。

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

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

实际应用与用例

内存管理对于智能体跟踪信息并随时间推移表现出智能至关重要。这对于智能体超越基本的问答能力是必不可少的。应用包括：

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

内存使智能体能够维护历史记录、学习、个性化交互，并管理复杂、与时间相关的问题。

动手实践代码：Google 智能体开发套件 (ADK) 中的内存管理

Google 智能体开发套件 (ADK) 提供了一种结构化的方法来管理上下文和内存，包括用于实际应用的组件。深入理解 ADK 的 Session、State 和 Memory 对于构建需要信息保留能力的智能体至关重要。就像在人类互动中一样，智能体需要能够回忆起之前的交流，以进行连贯自然的对话。ADK 通过三个核心概念及其相关服务简化了上下文管理。与智能体的每次交互都可以被视为一个独特的对话线程，智能体可能需要访问来自早期交互的数据。ADK 将此结构化如下：

Session： 一个独立的聊天线程，记录该特定交互的消息和动作 (Events)，同时存储与该对话相关的临时数据 (State)。
State (session.state)： 存储在 Session 内的数据，包含仅与当前活动聊天线程相关的信息。
Memory： 一个可搜索的信息存储库，信息源自过去的各种聊天或外部来源，作为超越即时对话的数据检索资源。

ADK 提供了专门的服务来管理构建复杂、有状态和具备上下文感知能力的智能体所必需的关键组件。SessionService 通过处理聊天线程（Session 对象）的启动、记录和终止来对其进行管理，而 MemoryService 则负责长期知识 (Memory) 的存储和检索。SessionService 和 MemoryService 都提供了多种配置选项，允许用户根据应用需求选择存储方法。对于测试目的，提供了内存中的选项，但这些数据在应用重启后不会持久化。对于持久化存储和可扩展性，ADK 还支持数据库和基于云的服务。

Session：跟踪每次聊天

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，它使用 Vertex AI 基础设施在 Google Cloud 上实现可扩展的生产部署。

# 示例：使用 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 应对应于
# Reasoning Engine 的 ID 或名称
REASONING_ENGINE_APP_NAME = "projects/your-gcp-project-id/locations/us-central1/reasoningEngines/your-engine-id"# 替换为你的 Reasoning Engine 资源名称

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 如何通过管理特定于会话的历史和临时数据来维持连续性。

State：会话的草稿板

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

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

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

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

user： 前缀将数据与一个用户 ID 关联，跨所有会话。
app： 前缀指定在应用程序的所有用户之间共享的数据。
temp： 前缀表示数据仅在当前处理轮次中有效，并且不会被持久存储。

智能体通过单个 session.state 字典访问所有状态数据。SessionService 负责数据的检索、合并和持久化。当通过 session_service.append_event() 向会话历史中添加 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-1.5-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"Initial state: {session.state}")

# --- 运行 Agent ---
user_message = Content(parts=[Part(text="Hello")])
print("\n--- Running the agent ---")
for event in runner.run(
    user_id=user_id,
    session_id=session_id,
    new_message=user_message
):
if event.is_final_response():
     print("Agent responded.")
# --- 检查更新后的 State ---
# 在 runner 完成处理所有事件 *之后* 正确检查 state。
updated_session = session_service.get_session(app_name, user_id, session_id)
print(f"\nState after agent run: {updated_session.state}")

在幕后，Runner 会检测到你的 output_key，并在调用 append_event 时自动创建带有 state_delta 的必要操作。

标准方式：使用 EventActions.state_delta（用于更复杂的更新）：
当需要执行更复杂的操作时——比如一次更新多个键、保存非文本内容、针对特定作用域（如 user: 或 app:）进行更新，或者进行与智能体最终文本回复无关的更新——你需要手动构建一个包含状态变化的字典（即 state_delta），并将其包含在你正在附加的 Event 的 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:
"""
  在用户登录事件时更新会话状态。
  此工具封装了与用户登录相关的所有状态更改。
  
  Args:
    tool_context: 由 ADK 自动提供，用于访问会话状态。
  
  Returns:
    一个确认操作成功的字典。
  """
# 通过提供的上下文直接访问状态。
  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("State updated from within the `log_user_login` tool.")
      return {
          "status": "success",
          "message": f"User login tracked. Total logins: {login_count}."
      }
# --- 使用示例 ---

# 在实际应用中，一个 LLM Agent 会决定调用此工具。
# 这里，我们为了演示目的模拟一次直接调用。

# 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"Initial state: {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"State after tool execution: {updated_session.state}")

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

此代码演示了一种基于工具的方法来管理应用程序中的用户会话状态。它定义了一个函数 log_user_login，该函数充当一个工具（Tool），负责在用户登录时更新会话状态。该函数接收一个由 ADK 提供的 ToolContext 对象，以访问和修改会话的 state 字典。在工具内部，它增加了一个 user:login_count，将 task_status 设置为 "active"，记录user:last_login_ts（时间戳），并添加一个临时标志 temp:validation_needed。

代码的演示部分模拟了如何使用此工具。它设置了一个内存中的 SessionService，并创建了一个带有预定义状态的初始会话。然后手动创建一个 ToolContext 来模拟 ADK Runner 执行该工具的环境，使用这个模拟上下文调用 log_user_login 函数。最后，代码再次检索会话以显示状态已通过工具的执行得到更新。其目的是展示将状态更改封装在工具中如何使代码比在工具外部直接操作状态更清晰、更有条理。

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

总而言之，在设计你的 state 时，应保持简洁，使用基本数据类型，为键取清晰的名称并正确使用前缀，避免深度嵌套，并始终通过 append_event 流程来更新状态。

Memory：用 MemoryService 实现长期知识

在智能体系统中，Session 组件维护当前聊天历史（events）和特定于单个对话的临时数据（state）的记录。

然而，为了让智能体能够跨多个交互保留信息或访问外部数据，长期的知识管理是必要的。这通过 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# 要检索的 top 结果数量
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 中，Memory 是创建智能且自然对话应用的核心组件。它允许 AI 智能体记住过去交互的信息，从反馈中学习，并适应用户偏好。LangChain 的内存机制通过引用历史记录来丰富当前提示，并在每次交互后记录最新内容以供后续使用。随着智能体承担更复杂的任务，这种能力对效率提升和用户体验优化都至关重要。

短期记忆（Short-term Memory）： 这是线程范围的内存，用于跟踪单个会话或线程内正在进行的对话。它提供即时上下文，但由于 LLM 的上下文窗口限制，完整的历史记录可能导致上下文溢出或性能下降。LangGraph 将短期记忆作为智能体状态 (Agent State) 的一部分进行管理，并通过检查点机制 (checkpointer) 实现持久化，使对话线程能够在任何时间点恢复执行。

长期记忆（Long-term Memory）： 用于存储跨会话的用户特定或应用级数据，在多个线程之间共享。它保存在自定义的 “命名空间 (namespaces)” 中，可在任何会话中随时调用。LangGraph 提供了专门的存储库 (store) 来保存和检索长期记忆，使智能体能够长期保留知识和经验。

LangChain 提供了多种用于管理对话历史的工具，从手动控制到在 链 (chains) 内的自动集成。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 (默认值)，它返回一个格式化的单个字符串，这对于标准 LLM 是理想的。
- 如果为 True，它返回一个消息对象列表，这是聊天模型 (Chat Models) 的推荐格式。

from langchain.memory import ConversationBufferMemory

# 初始化内存
memory = ConversationBufferMemory()

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

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

将此内存集成到 LLMChain 中，可以使模型访问对话历史并提供与上下文相关的响应。

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

# 1. 定义 LLM 和提示
llm = OpenAI(temperature=0)
template = """你是一个乐于助人的旅行代理。

Previous conversation:
{history}

New question: {question}
Response:"""
prompt = PromptTemplate.from_template(template)

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

# 3. 构建链
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)

为了提高与聊天模型 (chat models) 的效果，建议通过设置 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. 定义聊天模型和提示
llm = ChatOpenAI()
prompt = ChatPromptTemplate(
    messages=[
        SystemMessagePromptTemplate.from_template("你是一个友好的助手。"),
        MessagesPlaceholder(variable_name="chat_history"),
        HumanMessagePromptTemplate.from_template("{question}")
    ]
)

# 2. 配置内存
# return_messages=True 对聊天模型至关重要
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)


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

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

长期记忆的类型：长期记忆使系统能够在不同对话之间保留信息，提供更深层次的上下文和个性化体验。它通常可分为三种类型，类似于人类的记忆结构：

语义记忆（Semantic Memory）——记住事实：
用于保留特定的事实与概念，例如用户偏好或领域知识。语义记忆为智能体提供知识基础，使其能够生成更个性化、更相关的回答。
在实现上，这些信息可以以动态更新的用户“个人资料”（一个 JSON 文档）或独立事实文档的集合形式进行管理。
情景记忆（Episodic Memory）——记住经历：
负责回忆过去的事件或行为。对于 AI 智能体而言，情景记忆通常用于记住如何完成某个任务。
实践中，常通过少样本示例提示（few-shot example prompting）来实现，使智能体能从过去成功的交互序列中学习，以更准确地执行任务。
程序记忆（Procedural Memory）——记住规则：
表示智能体如何执行任务的记忆，即其核心指令与行为准则，通常体现在系统提示（system prompt）中。
智能体可以自我修改提示以适应与改进自身行为。一种常见且有效的方法是 “反思（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 文档存储在一个存储库 (store) 中。每个记忆都在一个自定义的命名空间（像一个文件夹）和一个唯一的键（像一个文件名）下进行组织。这种分层结构便于信息的组织和检索。以下代码演示了如何使用 InMemoryStore 来存放 (put)、获取 (get) 和搜索 (search) 记忆。

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": [
            "User likes short, direct language",
            "User only speaks English & 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 等定义的范围进行组织，同时会智能更新以整合新数据并解决潜在矛盾。在开始新会话时，智能体可通过完整数据召回或使用嵌入相似性搜索（embedding-based retrieval）来检索相关记忆。这一机制使智能体能够跨会话保持连续性，并基于回忆的信息生成更具上下文感和个性化的响应。

智能体的 Runner 与 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): 当智能体需要做的不仅仅是回答一个问题时，请使用此模式。对于那些必须在整个对话维持上下文、跟踪多步骤任务进度或通过回忆用户偏好和历史来个性化交互的智能体来说，这是必不可少的。每当期望智能体根据过去的成功、失败或新获取的信息进行学习或适应时，都应实现内存管理。

可视化摘要

《Agentic Design Patterns：构建智能系统的实战指南》- 第八章内存管理 — 图1：内存管理设计模式

本章重点

快速回顾一下关于内存管理的要点：

内存对于智能体跟踪事物、学习和个性化交互非常重要。
对话式 AI依赖于用于单个聊天中即时上下文的短期记忆和用于跨多个会话的持久知识的长期记忆。
短期记忆（即时信息）是临时的，通常受到 LLM 上下文窗口 或框架传递上下文方式的限制。
长期记忆（能持久保存的信息）通过使用像向量数据库（Vector DB）这样的外部存储，跨不同聊天保存信息，并通过搜索来访问。
像 ADK 这样的框架有特定的部分来管理内存，如 Session（聊天线程）、State（临时聊天数据） 和 MemoryService（可搜索的长期知识）。
ADK 的 SessionService 处理聊天会话的整个生命周期，包括其历史（events）和临时数据（state）。
ADK 的 session.state 是一个用于临时聊天数据的字典（dict）。前缀（user:、app:、temp:）告诉你数据属于哪里以及它是否会持久存在。
在 ADK 中，你应该通过使用 EventActions.state_delta 或 output_key 在添加事件时更新状态，而不是直接更改状态字典。
ADK 的 MemoryService 用于将信息放入长期存储并让智能体进行搜索，通常使用工具（tools）实现。
LangChain 提供了像 ConversationBufferMemory 这样的实用工具，可以自动将单个对话的历史注入到提示中，使智能体能够回忆起即时上下文。
LangGraph 通过使用存储库（repositories）来保存和检索语义事实、情景体验，甚至跨不同用户会话的可更新程序规则，从而实现高级长期记忆。
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