Context Engineering_ Sessions & Memory

致谢

简介

• 上下文工程： 在 LLM 的上下文窗口内动态组装和管理信息的过程，以实现有状态的智能 Agent。

• 会话： 与 Agent 进行完整对话的容器，保存着对话的时间顺序历史和 Agent 的工作记忆。

• 记忆： 实现长期持久性的机制，跨多个会话捕获并整合关键信息，从而为 LLM Agent 提供连续且个性化的体验。

上下文工程

• 指导推理的上下文定义了 Agent 的基本推理模式和可用动作，支配着其行为：
• 系统指令： 定义 Agent 人设、能力和约束的高级指令。
• 工具定义： Agent 可用于与外部世界交互的 API 或函数的架构。
• 少样本示例： 通过上下文学习指导模型推理过程的精选示例。

• 证据与事实数据是 Agent 进行推理的实质性数据，包括预先存在的知识和为特定任务动态检索的信息；它是 Agent 回复的“证据”：
• 长期记忆： 跨多个会话收集的关于用户或主题的持久化知识。
• 外部知识： 从数据库或文档中检索的信息，通常使用检索增强生成 (RAG)¹。
• 工具输出： 工具返回的数据或结果。
• 子 Agent 输出： 被委派特定子任务的专用 Agent 返回的结论或结果。

• 工件： 与用户或会话相关的非文本数据（例如文件、图像）。

• 即时对话信息将 Agent 建立在当前的交互中，定义即时任务：
• 对话历史： 当前交互的逐轮记录。
• 状态 / 草稿本： Agent 用于其即时推理过程的临时、进行中的信息或计算。
• 用户的提示： 需要解决的即时查询。

1. 获取上下文： Agent 首先检索上下文——例如用户记忆、RAG 文档和最近的对话事件。对于动态上下文检索，Agent 将使用用户查询和其他元数据来识别要检索的信息。

2. 准备上下文： Agent 框架动态构建用于 LLM 调用的完整提示。虽然单个 API 调用可能是异步的，但准备上下文是一个阻塞的、“热路径（hot-path）”过程。在上下文准备好之前，Agent 无法继续。

3. 调用 LLM 和工具： Agent 迭代调用 LLM 和任何必要的工具，直到生成给用户的最终响应。工具和模型输出被追加到上下文中。

1. 上传上下文： 在该轮次中收集的新信息被上传到持久存储。这通常是一个“后台”过程，允许 Agent 完成执行，而记忆整合或其他后处理异步发生。

会话

# Python
contents = [
  {
  "role": "user",
  "parts": [ {"text": "What is the capital of France?"} ]
  }, {
  "role": "model",
  "parts": [ {"text": "The capital of France is Paris."} ]
  }
]
response = client.models.generate_content(
  model="gemini-2.5-flash",
  contents=contents
)

框架和模型之间的差异

多 Agent 系统的会话

# Python
from google.adk.agents import LlmAgent
# 子 Agent 可以访问 Session 并向其写入事件。
sub_agent_1 = LlmAgent(...)
# 可选地，子 Agent 可以将最终响应文本（或结构化输出）保存到指定的状态键。
sub_agent_2 = LlmAgent(
  ...,
  output_key="...")

# 父 Agent。
root_agent = LlmAgent(
  ...,
  sub_agents=[sub_agent_1, sub_agent_2]
)

跨多个 Agent 框架的互操作性

会话的生产环境考量

管理长上下文对话：权衡与优化

1. 上下文窗口限制： 每个 LLM 都有一次可以处理的最大文本量（上下文窗口）。如果对话历史超过此限制，API 调用将失败。

2. API 成本： 大多数 LLM 提供商根据您发送和接收的 Token 数量收费。较短的历史记录意味着每轮更少的 Token 和更低的成本。

1. 延迟（速度）： 向模型发送更多文本需要更长的时间来处理，导致用户的响应时间变慢。压缩使 Agent 感觉快速且响应灵敏。

2. 质量： 随着 Token 数量的增加，由于上下文中的额外噪音和自回归错误，性能可能会变差。

• 保留最近 N 轮： 这是最简单的策略。Agent 只保留对话的最近 N 轮（“滑动窗口”）并丢弃所有更旧的内容。

• 基于 Token 的截断： 在将历史记录发送给模型之前，Agent 计算消息中的 Token，从最近的开始向后推算。它在不超过预定义 Token 限制（例如 4000 个 Token）的情况下包含尽可能多的消息。所有更旧的内容都被直接切断。

• 递归摘要： 对话的较旧部分被 AI 生成的摘要替换。随着对话的增长，Agent 定期使用另一个 LLM 调用来总结最旧的消息。此摘要随后被用作历史的浓缩形式，通常作为前缀添加到较新的逐字消息之前。

# Python
from google.adk.apps import App
from google.adk.plugins.context_filter_plugin import ContextFilterPlugin
app = App(
  name='hello_world_app',
  root_agent=agent,
  plugins=[
    # 保留最近 10 轮和最新的用户查询。
    ContextFilterPlugin(num_invocations_to_keep=10),
  ],
)

• 基于计数的触发器（即 Token 大小或轮次计数阈值）：一旦对话超过某个预定义的阈值，就会压缩对话。这种方法对于管理上下文长度通常是“足够好”的。

• 基于时间的触发器： 压缩不是由对话的大小触发，而是由缺乏活动触发。如果用户停止交互一段时间（例如 15 或 30 分钟），系统可以在后台运行压缩作业。

• 基于事件的触发器（即语义/任务完成）：当 Agent 检测到特定任务、子目标或对话主题已结束时，决定触发压缩。

# Python
from google.adk.apps import App
from google.adk.apps.app import EventsCompactionConfig
app = App(
  name='hello_world_app',
  root_agent=agent,
  events_compaction_config=EventsCompactionConfig(
    compaction_interval=5,
    overlap_size=1,
  ),
)

记忆

• 个性化： 最常见的用例是记住用户偏好、事实和过去的交互，以定制未来的回复。例如，记住用户最喜欢的运动队或他们在飞机上的首选座位，可以创造更有帮助和个性化的体验。

• 上下文窗口管理： 随着对话变长，完整的历史记录可能会超过 LLM 的上下文窗口。记忆系统可以通过创建摘要或提取关键事实来压缩此历史记录，在保留上下文的同时，无需在每一轮中发送数千个 Token。这降低了成本和延迟。

• 数据挖掘和洞察： 通过分析许多用户的存储记忆（以聚合的、保护隐私的方式），您可以从噪音中提取洞察。例如，一个零售聊天机器人可能会发现许多用户都在询问特定产品的退货政策，从而标记潜在问题。

• Agent 自我改进和适应： Agent 通过创建关于自身表现的程序性记忆从以前的运行中学习——记录哪些策略、工具或推理路径导致了成功的结果。这使得 Agent 能够建立有效解决方案的剧本，使其能够随着时间的推移适应并改进其解决问题的能力。

1. 用户： 为记忆提供原始源数据。在某些系统中，用户可能直接提供记忆（即通过表单）。

2. Agent（开发者逻辑）： 配置如何决定记住什么和何时记住，编排对记忆管理器的调用。在简单的架构中，开发者可以实现这样的逻辑：记忆总是被检索，并且总是被触发生成。在更高级的架构中，开发者可以实现记忆作为工具（memory-as-a-tool），其中 Agent（通过 LLM）决定何时应该检索或生成记忆。

1. Agent 框架（例如 ADK, LangGraph）： 提供用于记忆交互的结构和工具。框架充当管道。它定义了开发者的逻辑如何访问对话历史并与记忆管理器交互，但它本身并不管理长期存储。它还定义了如何将检索到的记忆填充到上下文窗口中。

2. 会话存储（即 Agent Engine Sessions, Spanner, Redis）： 存储会话的逐轮对话。原始对话将被摄入到记忆管理器中以生成记忆。

3. 记忆管理器（例如 Agent Engine Memory Bank, Mem0, Zep）： 处理记忆的存储、检索和压缩。存储和检索记忆的机制取决于使用的提供商。这是专门的服务或组件，它获取由 Agent 识别的潜在记忆并处理其整个生命周期。
• 提取从源数据中提炼关键信息。
• 整合策划记忆以合并重复的实体。
• 存储将记忆持久化到持久数据库。
• 检索获取相关记忆以为新的交互提供上下文。

记忆的类型

信息的类型

组织模式

存储架构

创建机制

记忆范围

多模态记忆

# Python
from google.genai import types
client = vertexai.Client(project=..., location=...)
response = client.agent_engines.memories.generate(
  name=agent_engine_name,
  direct_contents_source={
    "events": [
      {
        "content": types.Content(
          role="user",
          parts=[
            types.Part.from_text(
              "This is context about the multimodal input."            ),
            types.Part.from_bytes(
              data=CONTENT_AS_BYTES,
              mime_type=MIME_TYPE
            ),
            types.Part.from_uri(
              file_uri="file/path/to/content",
              mime_type=MIME_TYPE
            )
          ]
        )
      }
    ]
  },
  scope={"user_id": user_id}
)

记忆生成：提取与整合

1. 摄入： 该过程始于客户端向记忆管理器提供原始数据源（通常是对话历史）。

2. 提取与过滤： 记忆管理器使用 LLM 从源数据中提取有意义的内容。关键在于此 LLM 不会提取所有内容；它只捕获符合预定义主题定义的信息。如果摄入的数据不包含与这些主题匹配的信息，则不会创建记忆。

3. 整合： 这是最复杂的阶段，记忆管理器在此处理冲突解决和去重。它执行“自我编辑”过程，使用 LLM 将新提取的信息与现有记忆进行比较。为了确保用户的知识库保持连贯、准确，并根据新信息随时间演变，管理器可以决定：
• 合并新见解到现有记忆中。
• 删除现有的记忆，如果它现在已失效。
• 创建全新的记忆，如果主题是新颖的。

4. 存储： 最后，新的或更新的记忆被持久化到持久存储层（如向量数据库或知识图谱），以便可以在未来的交互中检索。

# Python
from google.cloud import vertexai
client = vertexai.Client(project=..., location=...)
client.agent_engines.memories.generate(
  name="projects/.../locations/...reasoningEngines/...",
  scope={"user_id": "123"},
  direct_contents_source={
    "events": [...]
  },
  config={
    # 在后台运行记忆生成。
    "wait_for_completion": False  }
)

深入探讨：记忆提取

# Python
from google.genai.types import Content, Part
# 详见 https://cloud.google.com/agent-builder/agent-engine/memory-bank/set-up
memory_bank_config = {
  "customization_configs": [{
    "memory_topics": [
      { "managed_memory_topic": {"managed_topic_enum": "USER_PERSONAL_INFO" }},
      {
        "custom_memory_topic": {
          "label": "business_feedback",
          "description": """Specific user feedback about their experience at the coffeeshop. This includes opinions on drinks, food, pastries, ambiance, staff friendliness,service speed, cleanliness, and any suggestions for improvement."""        }
      }
    ],
    "generate_memories_examples": {
      "conversationSource": {
        "events": [
          {
            "content": Content(
              role="model",
              parts=[Part(text="Welcome back to The Daily Grind! We'd love to hearyour feedback on your visit.")])          },{
            "content": Content(
              role="user",
              parts=[Part(text= "Hey. The drip coffee was a bit lukewarm today, whichwas a bummer. Also, the music was way too loud, I could barely hear my friend.")])          }]
      },
      "generatedMemories": [
        {"fact": "The user reported that the drip coffee was lukewarm."},
        {"fact": "The user felt the music in the shop was too loud."}
      ]
    }
  }]
}
agent_engine = client.agent_engines.create(
  config={
    "context_spec": {"memory_bank_config": memory_bank_config }
  }
)

深入探讨：记忆整合

• 更新： 用新的或修正的信息修改现有记忆。

• 创建： 如果新见解是全新的且与现有记忆无关，则创建一个新记忆。

• 删除 / 失效： 如果新信息使旧记忆完全不相关或不正确，则删除或使其失效。

记忆溯源

• 引导数据： 从内部系统（如 CRM）预加载的信息。这种高信任度数据可用于初始化用户记忆以解决冷启动问题，即向 Agent 从未交互过的用户提供个性化体验的挑战。

• 用户输入： 这包括显式提供的数据（例如通过表单，这是高信任度的）或从对话中隐式提取的信息（通常可信度较低）。

• 工具输出： 从外部工具调用返回的数据。通常不建议从工具输出生成记忆，因为这些记忆往往脆弱且陈旧，使这种来源类型更适合短期缓存。

在记忆管理期间考量记忆谱系

• 基于时间的衰减： 记忆的重要性会随着时间的推移而降低。关于两年前会议的记忆可能不如上周的记忆相关。

• 低置信度： 从弱推断创建且从未被其他来源佐证的记忆可能会被修剪。

• 不相关性： 随着 Agent 对用户获得更复杂的理解，它可能会确定某些旧的、琐碎的记忆不再与用户的当前目标相关。

在推理期间考量记忆谱系

触发记忆生成

• 会话完成： 在多轮会话结束时触发生成。

• 轮次节奏： 在特定数量的轮次后运行该过程（例如，每 5 轮）。

• 实时： 在每一轮之后生成记忆。

• 显式命令： 根据直接的用户命令激活该过程（例如，“记住这个”）。

记忆作为工具

# Python
from google.adk.agents import LlmAgent
from google.adk.memory import VertexAiMemoryBankService
from google.adk.runners import Runner
from google.adk.tools import ToolContext

def generate_memories(tool_context: ToolContext):
  """Triggers memory generation to remember the session."""
  # 选项 1：使用 ADK 记忆服务从完整的对话历史中提取记忆。
  tool_context._invocation_context.memory_service.add_session_to_memory(
    session)
  # 选项 2：从最后的对话轮次中提取记忆。
  client.agent_engines.memories.generate(
    name="projects/.../locations/...reasoningEngines/...",
    direct_contents_source={
      "events": [
        {"content": tool_context._invocation_context.user_content}
      ]
    },
    scope={
      "user_id": tool_context._invocation_context.user_id,
      "app_name": tool_context._invocation_context.app_name
    },
    # 在后台生成记忆
    config={"wait_for_completion": False}
  )
  return {"status": "success"}

agent = LlmAgent(
  ...,
  tools=[generate_memories]
)

runner = Runner(
  agent=agent,
  app_name=APP_NAME,
  session_service=session_service,
  memory_service=VertexAiMemoryBankService(
    agent_engine_id=AGENT_ENGINE_ID,
    project=PROJECT,
    location=LOCATION
  )
)

# Python

def extract_memories(query: str, tool_context: ToolContext):
  """Triggers memory generation to remember information.

  Args:
    query: Meaningful information that should be persisted about the user.
  """
  client.agent_engines.memories.generate(
    name="projects/.../locations/...reasoningEngines/...",
    # 有意义的信息已经从对话中提取出来了，所以我们
    # 只是想把它与同一用户的现有记忆进行整合。
    direct_memories_source={
      "direct_memories": [{"fact": query}]
    },
    scope={
      "user_id": tool_context._invocation_context.user_id,
      "app_name": tool_context._invocation_context.app_name
    },
    config={"wait_for_completion": False}
  )
  return {"status": "success"}

agent = LlmAgent(
  ...,
  tools=[extract_memories]
)

后台操作与阻塞操作

记忆检索

• 相关性（语义相似性）： 这个记忆与当前对话在概念上的相关程度如何？

• 新近度（基于时间）： 这个记忆是多久前创建的？

• 重要性（显著性）： 这个记忆总体上有多关键？与相关性不同，记忆的“重要性”可能在生成时定义。

检索时机

# Python
# 选项 1：使用内置的 PreloadMemoryTool，它每轮都通过相似性搜索检索记忆。
agent = LlmAgent(
  ...,
  tools=[adk.tools.preload_memory_tool.PreloadMemoryTool()]
)

# 选项 2：使用自定义回调以更好地控制记忆的检索方式。
def retrieve_memories_callback(callback_context, llm_request):
  user_id = callback_context._invocation_context.user_id
  app_name = callback_context._invocation_context.app_name
  response = client.agent_engines.memories.retrieve(
    name="projects/.../locations/...reasoningEngines/...",
    scope={
      "user_id": user_id,
      "app_name": app_name
    }
  )
  memories = [f"* {memory.memory.fact}" for memory in list(response)]
  if not memories:
    # 没有记忆要添加到系统指令中。
    return
  # 将格式化的记忆追加到系统指令中
  llm_request.config.system_instruction += "\nHere is information that you have about the user:\n"
  llm_request.config.system_instruction += "\n".join(memories)

agent = LlmAgent(
  ...,
  before_model_callback=retrieve_memories_callback,
)

# Python
# 选项 1：使用内置的 LoadMemory。
agent = LlmAgent(
  ...,
  tools=[adk.tools.load_memory_tool.LoadMemoryTool()],
)

# 选项 2：使用自定义工具，您可以在其中描述可能可用的信息类型。
def load_memory(query: str, tool_context: ToolContext):
  """Retrieves memories for the user.

  The following types of information may be stored for the user:
  * User preferences, like the user's favorite foods.
  ...
  """
  # 使用相似性搜索检索记忆。
  response = tool_context.search_memory(query)
  return response.memories

agent = LlmAgent(
  ...,
  tools=[load_memory],
)

基于记忆的推理

系统指令中的记忆

# Python
from jinja2 import Template

template = Template(""" system_instructions }<MEMORIES>Here is some information about the user:{% for retrieved_memory in data %}*  retrieved_memory.memory.fact {% endfor %}</MEMORIES>""")

prompt = template.render(
  system_instructions=system_instructions,
  data=retrieved_memories
)

对话历史中的记忆

# Python

def load_memory(query: str, tool_context: ToolContext):
  """Loads memories into the conversation history..."""
  response = tool_context.search_memory(query)
  return response.memories

agent = LlmAgent(
  ...,
  tools=[load_memory],
)

程序性记忆

1. 提取： 程序性提取需要专门的提示，旨在从成功的交互中提炼出可重用的策略或“剧本”，而不仅仅是捕获事实或有意义的信息。

2. 整合： 陈述性整合合并相关事实（“什么”），而程序性整合策划工作流程本身（“如何”）。这是一个主动的逻辑管理过程，专注于将新的成功方法与现有的“最佳实践”集成，修补已知计划中的缺陷步骤，并修剪过时或无效的程序。

3. 检索： 目标不是检索数据来回答问题，而是检索一个指导 Agent 如何执行复杂任务的计划。因此，程序性记忆可能具有与陈述性记忆不同的数据模式。

测试与评估

• 精确率 (Precision)： 在 Agent 创建的所有记忆中，有多少比例是准确且相关的？高精确率可以防止“过度热切”的记忆系统用不相关的噪音污染知识库。

• 召回率 (Recall)： 在它应该从源中记住的所有相关事实中，它捕获了多少比例？高召回率确保 Agent 不会错过关键信息。

• F1 分数 (F1-Score)： 精确率和召回率的调和平均值，提供单一、平衡的质量度量。

• Recall@K： 当需要记忆时，是否在前 ‘K’ 个检索结果中找到了正确的那个？这是检索系统准确性的主要衡量标准。

• 延迟： 检索位于 Agent 响应的“热路径”上。整个检索过程必须在严格的延迟预算内执行（例如，低于 200 毫秒），以避免降低用户体验。

记忆的生产环境考量

1. Agent 推送数据： 在相关事件（例如会话结束）之后，Agent 应用程序对记忆管理器进行非阻塞 API 调用，“推送”原始源数据（如对话记录）以进行处理。

2. 记忆管理器在后台处理： 记忆管理器服务立即确认请求，并将生成任务放入其自己的内部托管队列中。然后，它全权负责异步繁重的工作：进行必要的 LLM 调用以提取、整合和格式化记忆。管理器可能会延迟处理事件，直到经过一段时间的不活动。

3. 记忆被持久化： 服务将最终记忆——可能是新条目或对现有条目的更新——写入专用的、持久的数据库。对于托管记忆管理器，存储是内置的。

4. Agent 检索记忆： 当主 Agent 应用程序需要为新的用户交互检索上下文时，它可以直接查询此记忆存储。

隐私与安全风险

结论