AI Agent Harness Engineering 记忆管理：短期缓存+长期RAG融合，解决上下文丢失问题

Golang编程笔记

254人浏览 · 2026-06-10 01:11:10

Golang编程笔记 · 2026-06-10 01:11:10 发布

AI Agent Harness Engineering 记忆管理：短期缓存+长期RAG融合，解决上下文丢失问题

1. 引入与连接

1.1 从一个令人困扰的场景开始

想象一下，你正在与一个AI助手进行一场深入的技术讨论。你们已经聊了20分钟，探讨了复杂的系统架构设计，你刚刚分享了一个关键的创新点，希望AI能基于前面的讨论给出深入分析。然而，AI的回应却让你大失所望：它似乎完全忘记了你们刚才讨论的内容，给出的建议与之前的对话毫无关联。

这种"健忘症"在今天的AI应用中太常见了。无论是智能客服、编程助手还是个人AI伙伴，我们都或多或少经历过这种令人沮丧的上下文丢失问题。当对话变得复杂、交互次数增多时，AI似乎就像一个注意力短暂的学生，无法记住之前讨论的关键信息。

那么，为什么会出现这种情况？我们能否构建一个拥有更好"记忆力"的AI系统？这正是本文要探讨的核心问题——AI Agent的记忆管理系统，特别是如何通过短期缓存与长期RAG（检索增强生成）的融合，有效解决上下文丢失问题。

1.2 与你的知识建立连接

如果你曾经使用过任何现代AI应用，你已经对这个问题有了直观的体验。但让我们从更技术的角度来建立连接：

如果你熟悉计算机体系结构，你会发现这就像CPU的缓存层次结构——L1、L2缓存速度快但容量小，内存和硬盘容量大但速度慢。AI的记忆系统也面临类似的权衡。
如果你有机器学习背景，你知道Transformer模型的上下文窗口是有限的（如GPT-4的8K或32K token），这就是为什么长对话会导致信息丢失的物理限制。
如果你从事过软件开发，你可能使用过Redis等缓存系统和数据库的组合来优化应用性能——这与我们将要讨论的短期缓存+长期RAG架构有着惊人的相似之处。

在这篇文章中，我们将借鉴这些领域的思想，构建一个高效的AI Agent记忆管理系统。

1.3 为什么这很重要：学习价值与应用场景

解决AI的记忆问题不仅仅是为了让对话更流畅，它实际上开启了一系列新的可能性：

持久化的个人助手：一个能够记住你所有偏好、历史对话和专业知识的AI助手，可以提供真正个性化的服务。
连续的问题解决：对于复杂任务，如软件开发、研究分析或战略规划，AI需要能够跟踪长时间线的进展和决策。
知识积累与进化：能够从交互中持续学习和积累知识的AI系统，会随着时间变得越来越智能和有用。
专业领域应用：在医疗、法律、金融等专业领域，AI需要能够记住大量的历史案例、专业知识和上下文信息，才能提供有价值的辅助。

我们不仅要解决当前的问题，还要为下一代更智能的AI应用奠定基础。

1.4 我们的学习路径概览

在接下来的内容中，我们将按照以下路径逐步深入：

概念地图：首先建立整体认知框架，了解AI Agent记忆系统的关键概念和组成部分。
基础理解：从直观角度理解短期记忆和长期记忆的作用，以及它们各自的局限性。
层层深入：逐步探索记忆系统的工作原理、技术细节和实现机制。
多维透视：从历史、实践、批判和未来多个角度审视这个问题。
实践转化：提供具体的实现方案、代码示例和最佳实践。
整合提升：总结核心观点，构建完整的知识体系，并指出进一步学习的方向。

让我们开始这段探索之旅，一起构建更智能、更有"记忆力"的AI系统。

2. 概念地图：建立整体认知框架

在深入技术细节之前，让我们先构建一个整体的概念地图，了解AI Agent记忆系统的核心概念、它们之间的关系以及在更大的AI生态系统中的位置。

2.1 核心概念与关键术语

首先，让我们定义一些在本文中会反复使用的关键术语：

AI Agent (智能代理)：一个能够感知环境、做出决策并采取行动的自主系统。在本文中，我们主要关注与人类进行交互的对话式AI Agent。
上下文窗口 (Context Window)：Transformer模型一次能够处理的最大token数量。这是模型"注意力"能够覆盖的范围。
短期记忆 (Short-term Memory)：也叫工作记忆，是AI Agent在当前交互中能够即时访问的信息，类似于人类的短期记忆。
长期记忆 (Long-term Memory)：AI Agent能够长期存储和检索的信息，类似于人类的长期记忆。
缓存 (Caching)：一种存储机制，用于临时保存频繁访问的数据，以便更快地检索。
RAG (Retrieval-Augmented Generation，检索增强生成)：一种将信息检索与文本生成相结合的技术，用于增强AI模型的知识和上下文感知能力。
向量数据库 (Vector Database)：一种专门用于存储和检索高维向量的数据库，常用于语义相似性搜索。
嵌入 (Embedding)：将文本等数据转换为高维向量的过程，这些向量能够捕捉数据的语义含义。
记忆检索 (Memory Retrieval)：从存储系统中查找和获取相关记忆的过程。
记忆更新 (Memory Update)：将新信息添加到记忆系统中的过程。

这些概念构成了我们讨论的基础，接下来我们将看到它们如何相互关联，形成一个完整的记忆管理系统。

2.2 概念间的层次与关系

AI Agent的记忆系统不是一个单一的组件，而是一个多层次、相互关联的复杂系统。让我们从不同维度来理解这些概念间的关系：

2.2.1 时间维度：从短期到长期

从时间维度看，记忆可以按其持续时间和访问频率组织成一个层次结构：

即时上下文：当前正在处理的信息，直接在模型的上下文窗口中。
短期缓存：最近的交互历史，存储在高速缓存系统中。
工作记忆：当前任务相关的关键信息，可能从短期缓存或长期记忆中提取。
长期记忆：所有过去的交互和学到的知识，存储在持久化存储系统中。

这个层次结构类似于计算机的存储层次：寄存器 → L1缓存 → L2缓存 → 内存 → 硬盘。每一层都有不同的容量、速度和持久性。

2.2.2 功能维度：从存储到应用

从功能维度看，记忆系统包括以下几个关键环节：

感知与编码：将输入信息转换为适合存储的形式。
存储与组织：将编码后的信息存储在适当的位置，并建立索引。
检索与激活：根据当前上下文，找到并激活相关的记忆。
整合与生成：将检索到的记忆与当前输入整合，生成响应。
学习与更新：根据交互结果更新记忆系统。

这些功能形成了一个完整的循环，确保记忆系统能够持续学习和适应。

2.2.3 技术维度：从基础组件到完整系统

从技术实现的角度，记忆系统由以下组件构成：

嵌入模型：将文本转换为向量表示。
向量数据库：存储和检索向量表示的记忆。
缓存系统：提供高速访问的短期存储。
检索引擎：实现各种检索策略的算法组件。
记忆管理器：协调各个组件，实现整体记忆功能。

这些技术组件需要精心设计和集成，才能构建一个高效、可靠的记忆系统。

2.3 学科定位与边界

AI Agent的记忆管理是一个跨学科领域，它融合了以下多个学科的思想和技术：

认知科学：借鉴人类记忆的模型和理论，如Atkinson-Shiffrin记忆模型。
自然语言处理：提供文本处理、语义理解和生成的技术。
信息检索：贡献了高效的索引和检索算法。
数据库系统：提供了数据存储、管理和查询的基础设施。
机器学习：特别是深度学习，提供了嵌入模型和表示学习的技术。
系统设计：提供了构建复杂软件系统的方法论。

同时，我们也需要明确这个领域的边界：

它不是关于"意识"或"主观体验"的哲学讨论，而是关于工程实现的技术探讨。
它不追求模拟人类记忆的所有复杂性，而是专注于解决实际应用中的上下文丢失问题。
它是AI系统的一个组件，而不是完整的AI系统本身，需要与其他组件如规划、推理等协同工作。

2.4 概念图谱

为了更直观地展示这些概念之间的关系，让我们构建一个概念图谱：

这个概念图谱展示了记忆系统在AI Agent中的位置，其内部组成，支撑技术，以及核心功能。它为我们后续的讨论提供了一个清晰的路线图。

在接下来的章节中，我们将深入探讨这个图谱中的每个组件，特别是如何通过短期缓存和长期RAG的融合来解决上下文丢失问题。

3. 基础理解：建立直观认识

现在我们已经有了整体概念框架，让我们从最直观的角度来理解AI Agent的记忆问题，以及短期缓存和长期RAG如何协同工作来解决这个问题。

3.1 核心概念的生活化解释

让我们先把技术放在一边，用一些日常生活中的类比来理解这些概念。

3.1.1 AI的"注意力"与"记忆"：一场鸡尾酒会对话

想象你在一个热闹的鸡尾酒会上，正在与几个人同时交谈。你的大脑需要：

关注当前说话的人（处理当前输入）
记住刚才你们讨论的内容（短期记忆）
回忆起你对这个话题已有的知识（长期记忆）
可能还需要记住房间里其他人在说什么，以防有人叫你的名字（更广泛的上下文）

现在，想象你的注意力有一个严格的限制：你只能专注于最近2-3分钟内的对话内容。任何超过这个时间范围的信息，你就会完全忘记。这就是今天大多数AI模型的处境——它们的"注意力"被限制在一个固定大小的上下文窗口内。

当对话变长时，AI就像一个只有短暂注意力的鸡尾酒会参与者，无法记住之前讨论的关键内容。这就是我们要解决的上下文丢失问题。

3.1.2 短期记忆就像你的办公桌，长期记忆就像你的书房

让我们继续用办公场景来类比：

上下文窗口就像你手里正在拿着和处理的几张纸。你可以立即访问和处理这些信息，但数量非常有限。
短期缓存就像你的办公桌。你可以把当前项目相关的文件、笔记和参考资料放在桌上，方便快速取用。桌面空间有限，所以你只能放最相关的材料。
长期RAG记忆就像你的书房或图书馆。这里可以存储大量的书籍、文件和笔记，但检索需要一些时间和 effort。

当你工作时，你会：

首先看手里的纸张（上下文窗口）
如果需要更多相关信息，从办公桌上拿（短期缓存）
如果还需要更深入或更早的信息，去书房查找（长期RAG）
完成工作后，把重要的结果整理好，放回书房，同时更新办公桌上的材料（记忆更新）

这个类比很好地捕捉了我们记忆系统的工作原理：不同的存储层次有不同的容量和访问速度，我们需要智能地管理信息在这些层次之间的流动。

3.1.3 RAG就像给AI配了一个研究助理

你可能听说过RAG（检索增强生成）这个术语，让我们用一个简单的类比来理解它：

想象你是一个作家，正在写一本关于某个历史事件的书。你的大脑里有一些关于这个事件的基本知识，但不够详细和准确。你可以：

自己凭空想象所有细节（这就是没有RAG的生成模型）
或者，雇一个研究助理，当你需要某个具体信息时，让他去图书馆查资料，然后把相关信息带给你（这就是RAG）

在这个类比中：

你就是AI的生成模型
研究助理就是检索系统
图书馆就是我们的知识库/向量数据库

RAG不是让AI"记住"所有东西，而是给它提供了一个高效的"查阅"机制，让它在需要时能够找到相关信息。

3.2 简化模型与类比

现在让我们把这些类比结合起来，构建一个简化的记忆系统模型。

3.2.1 记忆的三层楼模型

让我们想象一个三层楼的建筑，每一层代表记忆系统的一个层次：

一楼（前台大厅）：上下文窗口
- 这是访客（当前输入）首先到达的地方
- 空间有限，只能容纳少数人（少量token）
- 接待员（模型）可以立即与这里的所有人互动
二楼（会议室）：短期缓存
- 这是刚刚离开一楼但可能很快需要再次交流的人等待的地方
- 空间比一楼大，但仍然有限
- 接待员可以快速召唤这里的人到一楼
三楼（档案室）：长期RAG记忆
- 这是存储所有过去访客记录和重要文件的地方
- 空间几乎无限
- 但需要通过索引和检索系统才能找到需要的信息
- 找到后，可以把相关信息送到一楼或二楼

这个模型展示了信息如何在不同层次之间流动，以及我们如何根据需要访问不同层次的记忆。

3.2.2 信息流动的水管模型

另一个有用的类比是将记忆系统想象成一个水管系统：

输入是流入系统的水（新信息）
过滤器决定哪些水应该直接送到哪里（记忆分类）
小水箱是短期缓存，存储最近的水（最近的交互）
大水库是长期记忆，存储所有的水（所有历史信息）
水泵是检索系统，根据需要从水库中抽水（检索相关记忆）
混合器将来自不同来源的水混合在一起（整合不同层次的记忆）
输出是最终流出系统的水（生成的响应）

这个模型帮助我们理解记忆系统的动态性和信息的流动路径。

3.3 直观示例与案例

让我们通过一个具体的例子来看看这个系统是如何工作的。

3.3.1 一个编程助手的对话

假设你正在使用一个AI编程助手开发一个电商网站，下面是你们的对话：

你：我需要创建一个Python类来管理购物车，帮我设计一下。

AI：好的，我来帮你设计一个购物车类。让我先创建一个基本结构…（提供了一个基本的ShoppingCart类）

你：很好，现在我需要添加一个方法来计算总价格，包括税费。假设税率是8%。

AI：没问题，我来添加一个计算总价的方法…（更新了类，添加了calculate_total方法，硬编码了8%的税率）

你：实际上，税率应该是可配置的，因为我们的电商网站要服务不同地区的用户。

AI：明白了，让我修改一下，使税率可配置…（进一步更新了类）

你：现在我们还需要考虑折扣，用户可以应用折扣码。让我们添加这个功能。

现在，如果没有良好的记忆管理，AI可能会：

忘记之前我们讨论的购物车类的结构
忘记我们刚刚决定让税率可配置
重新发明轮子，而不是在现有代码基础上构建

但有了我们的记忆系统：

上下文窗口中包含了你最后的请求（添加折扣功能）和AI的上一个响应。
短期缓存中存储了整个对话历史，特别是关于ShoppingCart类的所有设计决策。
如果需要，长期RAG还可以检索过去类似项目的代码模式、最佳实践等。

AI可以利用这些信息，在之前的基础上连贯地添加折扣功能，而不会丢失任何上下文。

3.3.2 一个个人助理的例子

让我们再看一个个人助理的例子：

你（1月15日）：我对坚果过敏，请记住这一点。

AI：好的，我会记住你对坚果过敏。

你（1月20日）：帮我订一下明天中午的餐厅，要在市中心。

AI：好的，我来帮你找市中心的餐厅。（推荐了几家，但没有考虑过敏信息）

你（有点恼火）：我告诉过你我对坚果过敏！

没有记忆管理的AI可能真的忘记了过敏信息。但有了我们的系统：

短期缓存可能还存储着最近几天的交互，但如果1月15日的对话已经过去了几天，可能已经不在缓存中了。
但是，长期RAG会存储这条重要信息，并在你提到"餐厅"或"食物"时，自动检索出过敏信息。
记忆系统会将这条信息添加到上下文中，确保AI在推荐餐厅时考虑到你的过敏情况。

这个例子展示了长期记忆的重要性，特别是对于那些不常使用但关键时刻不能忘记的信息。

3.4 常见误解澄清

在我们深入探讨之前，让我们澄清一些关于AI记忆的常见误解：

误解1：“更大的上下文窗口就能解决所有问题”

虽然更大的上下文窗口确实有帮助（比如GPT-4的32K或100K token版本），但它不是万能药：

即使是100K token，也只能容纳大约7-8万字的文本，对于长期交互或大量知识仍然不够。
更大的上下文窗口意味着更高的计算成本和延迟。
模型的"注意力"并不是均匀分布的，它可能更关注最近的token，而忽略较早的token。

这就像仅仅通过增大办公桌来解决存储问题——最终你还是需要一个书房。

误解2：“RAG就是把所有东西都存起来，需要的时候全部拿出来”

有效的RAG不仅仅是存储和检索，它需要：

智能地分块和索引信息，而不是简单地存储整个文档。
评估相关性，只检索最相关的信息，而不是所有相关信息。
组织和呈现检索到的信息，使其对生成模型最有用。

这就像一个好的研究助理，不仅仅是给你一堆书，而是帮你找到最相关的章节，甚至总结关键点。

误解3：“记忆系统会让AI’知道’所有事情”

记忆系统增强了AI的能力，但它并没有赋予AI真正的"理解"或"意识"：

AI仍然只是模式匹配机器，它不"知道"它在说什么。
记忆系统可能会检索到错误或不相关的信息，导致AI产生错误的响应。
没有完美的检索系统，总会有相关信息被遗漏的情况。

理解这些局限性有助于我们设计更健壮的系统，并合理设定用户期望。

通过这些生活化的类比、简化模型和直观示例，你应该已经对AI Agent的记忆管理有了基本的直观理解。在接下来的章节中，我们将深入探讨技术细节，了解如何实际构建这样一个系统。

4. 层层深入：逐步增加复杂度

在建立了直观理解之后，现在让我们逐步深入，探索记忆系统的技术细节、工作原理和实现机制。我们将从基本原理开始，逐步增加复杂度，直到掌握底层逻辑和高级应用。

4.1 第一层：基本原理与运作机制

让我们从最基本的原理开始，了解短期缓存和长期RAG是如何工作的，以及它们如何协同解决上下文丢失问题。

4.1.1 短期缓存的基本原理

短期缓存是我们记忆系统的第一层，它的主要目标是快速访问最近的交互历史。

什么是短期缓存？

短期缓存可以看作是对话历史的一个滑动窗口，但比模型的原生上下文窗口更大，同时更智能地管理内容。它的核心思想是：

保留最近的交互，因为它们最可能与当前任务相关。
提供比原生上下文窗口更大的容量，但仍然保持快速访问。
可以应用一些智能策略来决定保留什么、丢弃什么。

短期缓存的基本数据结构

最简单的短期缓存可以用一个先进先出（FIFO）队列实现：

class SimpleShortTermCache:
    def __init__(self, max_size=10):
        self.max_size = max_size
        self.cache = []
    
    def add(self, item):
        self.cache.append(item)
        if len(self.cache) > self.max_size:
            self.cache.pop(0)  # 移除最旧的项
    
    def get_all(self):
        return self.cache

这个简单实现有几个问题：

它只考虑了时间因素，没有考虑重要性。
它没有对内容进行任何处理或总结。
它没有与长期记忆交互。

尽管如此，它展示了短期缓存的基本思想。在后面的章节中，我们将改进这个实现。

为什么需要短期缓存？

你可能会问：既然模型已经有了上下文窗口，为什么还需要额外的短期缓存？

成本效率：将所有内容都放在模型的上下文窗口中可能非常昂贵。短期缓存可以让我们智能地选择最相关的内容放入上下文窗口。
灵活性：我们可以在短期缓存中应用各种策略（如总结、重要性排序），而这些策略在模型的原生上下文窗口中不容易实现。
桥接作用：短期缓存可以作为原生上下文窗口和长期记忆之间的桥梁，管理信息在两者之间的流动。

4.1.2 长期RAG的基本原理

长期RAG是我们记忆系统的第二层，它的主要目标是持久化存储和智能检索大量信息。

什么是RAG？

RAG（检索增强生成）是一种结合信息检索和文本生成的技术。它的基本流程是：

索引阶段：将文档分割成小块，转换为向量表示，存储在向量数据库中。
检索阶段：当用户提出问题时，将问题也转换为向量，在向量数据库中搜索最相似的文档块。
生成阶段：将检索到的文档块作为上下文，与用户的问题一起提供给生成模型，生成最终答案。

这个流程可以用以下简化代码表示：

class SimpleRAG:
    def __init__(self, embedder, vector_db, generator):
        self.embedder = embedder
        self.vector_db = vector_db
        self.generator = generator
    
    def index(self, documents):
        # 将文档分割成块
        chunks = self._split_documents(documents)
        # 生成向量表示
        vectors = self.embedder.embed(chunks)
        # 存储到向量数据库
        self.vector_db.store(chunks, vectors)
    
    def query(self, question):
        # 生成问题的向量表示
        question_vector = self.embedder.embed(question)
        # 检索相关文档块
        relevant_chunks = self.vector_db.search(question_vector, top_k=5)
        # 构建提示
        prompt = self._build_prompt(question, relevant_chunks)
        # 生成答案
        answer = self.generator.generate(prompt)
        return answer

这个简化版本省略了很多细节，但展示了RAG的核心流程。

向量表示与相似度搜索

RAG的核心是向量表示和相似度搜索。让我们更详细地了解这两个概念：

向量表示（嵌入）：

嵌入模型将文本转换为高维向量（通常是几百或几千维）。
这些向量的神奇之处在于：语义相似的文本在向量空间中也更接近。
例如，"猫"和"猫咪"的向量会很接近，而"猫"和"汽车"的向量会相距较远。

相似度搜索：

有了向量表示，我们可以通过计算向量之间的距离来衡量文本的语义相似度。
常用的距离度量包括余弦相似度、欧氏距离等。
向量数据库专门优化了这种相似度搜索，即使在数十亿向量的规模下也能快速返回结果。

用数学公式表示，两个向量 $a⃗\vec{a}$ 和 $b⃗\vec{b}$ 之间的余弦相似度为：

$\text{cosine similarity}(\vec{a}, \vec{b}) = \frac{\vec{a} \cdot \vec{b}}{\|\vec{a}\| \|\vec{b}\|}$

其中 $a⃗⋅b⃗\vec{a} \cdot \vec{b}$ 是向量点积， $∥a⃗∥\|\vec{a}\|$ 是向量 $a⃗\vec{a}$ 的模长。

为什么需要长期RAG？

长期RAG解决了以下几个关键问题：

知识截止日期：训练模型有知识截止日期，无法获取之后的新信息。RAG可以通过更新索引来添加新知识。
领域特定知识：通用模型可能缺乏特定领域的专业知识，RAG可以用领域特定文档来增强模型。
可解释性：RAG可以引用来源，让我们知道模型的答案来自哪里，增加可信度。
成本效益：与微调相比，RAG通常更便宜、更灵活，不需要重新训练模型。

4.1.3 两者如何协同工作

现在我们了解了短期缓存和长期RAG的基本原理，让我们看看它们如何协同工作，形成一个完整的记忆系统。

记忆系统的基本工作流程

一个完整的记忆系统工作流程如下：

接收输入：用户的问题或消息到达系统。
查询短期缓存：首先检查短期缓存，获取最近的交互历史。
查询长期记忆：同时，使用当前输入和短期缓存中的相关信息作为查询，检索长期记忆。
整合上下文：将当前输入、短期缓存中的相关内容和长期记忆中的相关信息整合在一起。
生成响应：将整合后的上下文提供给生成模型，生成响应。
更新记忆：将新的交互添加到短期缓存，同时考虑是否需要更新长期记忆。

这个流程可以用以下流程图表示：

简单的协同示例

让我们通过一个简单的例子来看看这个流程如何工作：

假设你正在与一个AI助手讨论旅行计划：

你：我正在计划去日本的旅行，你能推荐一些东京的景点吗？

系统处理：

短期缓存是空的，所以没有最近的历史。
用"日本旅行"、"东京景点"等关键词检索长期记忆，找到关于东京景点的信息。
生成关于东京景点的推荐。
将这次交互添加到短期缓存。
考虑是否需要将你的旅行计划添加到长期记忆（在这个例子中，可能先不添加）。

AI：当然！东京有很多很棒的景点，包括浅草寺、东京塔、涩谷十字路口、上野公园等等。你对什么类型的景点特别感兴趣？

你：我对历史文化景点更感兴趣，而且我只有两天时间在东京。

系统处理：

从短期缓存中获取上一次交互（推荐东京景点，询问兴趣）。
用"东京历史文化景点"、"两天东京行程"等关键词检索长期记忆。
整合上下文：用户计划日本旅行，对历史文化景点感兴趣，只有两天时间。
生成一个针对历史文化景点的两天行程建议。
将这次交互添加到短期缓存。
现在可能需要将你的旅行偏好（历史文化景点，两天行程）添加到长期记忆。

这个例子展示了短期缓存和长期RAG如何协同工作，提供连贯、上下文感知的响应。

4.2 第二层：细节、例外与特殊情况

现在我们了解了基本原理，让我们深入探讨一些更复杂的细节、例外情况和特殊场景。

4.2.1 短期缓存的高级策略

简单的FIFO队列对于短期缓存来说往往不够，我们需要更智能的策略。

内容总结策略

随着对话变长，即使是短期缓存也可能变得太大，无法全部放入模型的上下文窗口。一个解决方案是对旧的交互进行总结：

class SummarizingCache:
    def __init__(self, max_items=10, summarizer=None):
        self.max_items = max_items
        self.summarizer = summarizer
        self.cache = []
        self.summaries = []
    
    def add(self, item):
        self.cache.append(item)
        # 当缓存超过大小时，总结最早的一部分
        if len(self.cache) > self.max_items and self.summarizer:
            # 总结前半部分
            to_summarize = self.cache[:len(self.cache)//2]
            summary = self.summarizer.summarize(to_summarize)
            self.summaries.append(summary)
            # 保留后半部分
            self.cache = self.cache[len(self.cache)//2:]
    
    def get_context(self):
        # 组合总结和当前缓存
        context = []
        if self.summaries:
            context.append("对话摘要:\n" + "\n".join(self.summaries))
        context.extend(self.cache)
        return context

这种渐进式总结策略允许我们保留更多的历史信息，同时控制上下文窗口的大小。

重要性评分策略

另一个策略是为每个交互分配重要性分数，保留重要的，丢弃不重要的：

class ImportanceScoredCache:
    def __init__(self, max_size=10, scorer=None):
        self.max_size = max_size
        self.scorer = scorer
        self.cache = []  # 存储元组 (item, score, timestamp)
    
    def add(self, item):
        score = self.scorer.score(item) if self.scorer else 1.0
        timestamp = time.time()
        self.cache.append((item, score, timestamp))
        
        # 如果超过最大大小，删除得分最低的
        if len(self.cache) > self.max_size:
            # 综合考虑得分和时间衰减
            def decay_score(item):
                it_score, it_time = item[1], item[2]
                # 时间衰减因子：每小时衰减一半
                hours_passed = (time.time() - it_time) / 3600
                decay_factor = 0.5 ** hours_passed
                return it_score * decay_factor
            
            # 按衰减后的分数排序
            self.cache.sort(key=decay_score, reverse=True)
            # 保留前max_size个
            self.cache = self.cache[:self.max_size]

重要性评分可以考虑多个因素：

用户明确标记为重要的内容
包含实体（如人名、地名、产品名）的内容
用户反复提及的内容
情感强烈的内容

对话线程分离

如果对话涉及多个不同的主题，一个更好的策略是将对话分离成不同的线程：

class ThreadedCache:
    def __init__(self, max_threads=5, max_per_thread=10):
        self.max_threads = max_threads
        self.max_per_thread = max_per_thread
        self.threads = {}  # thread_id -> list of items
        self.current_thread = None
        self.thread_timestamps = {}  # thread_id -> last_access_time
    
    def add(self, item, thread_id=None):
        # 如果没有指定线程，尝试分类到现有线程或创建新线程
        if not thread_id:
            thread_id = self._classify_to_thread(item)
        
        if thread_id not in self.threads:
            # 如果超过最大线程数，删除最久未使用的
            if len(self.threads) >= self.max_threads:
                oldest_thread = min(self.thread_timestamps, key=self.thread_timestamps.get)
                del self.threads[oldest_thread]
                del self.thread_timestamps[oldest_thread]
            
            self.threads[thread_id] = []
        
        # 添加到线程
        self.threads[thread_id].append(item)
        if len(self.threads[thread_id]) > self.max_per_thread:
            self.threads[thread_id].pop(0)
        
        # 更新时间戳
        self.thread_timestamps[thread_id] = time.time()
        self.current_thread = thread_id
    
    def get_context(self):
        # 返回当前线程的内容，以及其他线程的简要摘要
        context = []
        if self.current_thread and self.current_thread in self.threads:
            context.append("当前话题:\n" + "\n".join(self.threads[self.current_thread]))
        
        other_threads = [tid for tid in self.threads if tid != self.current_thread]
        if other_threads:
            context.append("\n其他相关话题:")
            for tid in other_threads[:3]:  # 只显示最多3个其他话题
                if self.threads[tid]:
                    # 只显示每个线程的最后一条消息
                    context.append(f"- {self.threads[tid][-1][:50]}...")
        
        return "\n".join(context)

这种多线程方法对于复杂的多主题对话特别有用，它可以帮助AI保持不同话题的上下文分离，同时仍然能够利用其他话题的相关信息。

4.2.2 长期RAG的高级技术

基本的RAG系统对于简单场景可能足够，但对于更复杂的应用，我们需要一些高级技术。

分块策略

如何将文档分割成块是RAG系统性能的关键因素。不好的分块策略可能会导致：

重要信息被切断
检索到不完整的上下文
噪音过多

让我们看看几种不同的分块策略：

固定大小分块：

def fixed_size_chunking(text, chunk_size=1000, overlap=100):
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)
        start = end - overlap  # 重叠一部分，避免信息被切断
    return chunks

这是最简单的方法，但可能会在句子或段落中间切断。

语义感知分块：

import nltk
from nltk.tokenize import sent_tokenize

nltk.download('punkt')

def semantic_chunking(text, max_chunk_size=1000):
    sentences = sent_tokenize(text)
    chunks = []
    current_chunk = []
    current_size = 0
    
    for sentence in sentences:
        sentence_size = len(sentence)
        
        # 如果当前句子超过最大块大小，需要特殊处理
        if sentence_size > max_chunk_size:
            # 如果当前有内容，先保存
            if current_chunk:
                chunks.append(" ".join(current_chunk))
                current_chunk = []
                current_size = 0
            
            # 对长句子进行子分块
            words = sentence.split()
            sub_chunk = []
            sub_size = 0
            for word in words:
                if sub_size + len(word) + 1 > max_chunk_size:  # +1 for space
                    chunks.append(" ".join(sub_chunk))
                    sub_chunk = [word]
                    sub_size = len(word)
                else:
                    sub_chunk.append(word)
                    sub_size += len(word) + 1
            
            if sub_chunk:
                chunks.append(" ".join(sub_chunk))
        
        # 如果添加当前句子会超过最大块大小
        elif current_size + sentence_size + 1 > max_chunk_size:  # +1 for space
            chunks.append(" ".join(current_chunk))
            current_chunk = [sentence]
            current_size = sentence_size
        
        # 否则添加到当前块
        else:
            current_chunk.append(sentence)
            current_size += sentence_size + 1  # +1 for space
    
    # 添加最后一个块
    if current_chunk:
        chunks.append(" ".join(current_chunk))
    
    return chunks

这种方法尝试在语义边界（如句子、段落）处分割，保持语义完整性。

递归分块：
更高级的方法是递归地将文档分块，创建一个层次结构：

第一层：整个文档
第二层：章节
第三层：段落
第四层：句子

在检索时，可以在不同层次上搜索，找到最相关的信息粒度。

混合检索策略

仅仅依赖向量相似度搜索有时是不够的，我们可以使用混合检索策略，结合多种搜索方法：

class HybridRetriever:
    def __init__(self, vector_db, keyword_db, reranker=None):
        self.vector_db = vector_db
        self.keyword_db = keyword_db
        self.reranker = reranker
    
    def retrieve(self, query, top_k=10, vector_weight=0.5, keyword_weight=0.5):
        # 向量检索
        vector_results = self.vector_db.search(query, top_k=top_k*2)
        # 关键词检索
        keyword_results = self.keyword_db.search(query, top_k=top_k*2)
        
        # 合并结果，去除重复
        all_results = {}
        for doc, score in vector_results:
            all_results[doc.id] = {'doc': doc, 'vector_score': score, 'keyword_score': 0}
        
        for doc, score in keyword_results:
            if doc.id in all_results:
                all_results[doc.id]['keyword_score'] = score
            else:
                all_results[doc.id] = {'doc': doc, 'vector_score': 0, 'keyword_score': score}
        
        # 计算混合分数
        for doc_id, result in all_results.items():
            # 归一化分数（简化处理）
            norm_vector = result['vector_score'] / max(r[0] for r in vector_results) if vector_results else 0
            norm_keyword = result['keyword_score'] / max(r[0] for r in keyword_results) if keyword_results else 0
            
            result['hybrid_score'] = (vector_weight * norm_vector + 
                                      keyword_weight * norm_keyword)
        
        # 排序
        sorted_results = sorted(all_results.values(), 
                               key=lambda x: x['hybrid_score'], 
                               reverse=True)
        
        # 如果有重排序器，使用它重新排序
        if self.reranker:
            top_candidates = [r['doc'] for r in sorted_results[:top_k*2]]
            reranked = self.reranker.rerank(query, top_candidates)
            return reranked[:top_k]
        
        return [(r['doc'], r['hybrid_score']) for r in sorted_results[:top_k]]

混合检索结合了：

向量搜索：捕获语义相似性
关键词搜索：精确匹配重要术语
重排序：使用更精细的模型对初始搜索结果重新排序

查询扩展与转换

有时用户的查询不够明确或不够优化，我们可以对查询进行扩展和转换：

class QueryProcessor:
    def __init__(self, llm):
        self.llm = llm
    
    def expand_query(self, original_query):
        # 使用LLM生成多个相关查询
        prompt = f"""原始查询: {original_query}

请生成3-5个与原始查询相关的搜索查询，从不同角度表达相同的信息需求。
只返回查询，每行一个，不要编号。"""
        
        response = self.llm.generate(prompt)
        expanded_queries = [line.strip() for line in response.split('\n') if line.strip()]
        expanded_queries.insert(0, original_query)  # 包含原始查询
        
        return expanded_queries
    
    def decompose_complex_query(self, complex_query):
        # 分解复杂查询为多个子查询
        prompt = f"""复杂查询: {complex_query}

这个查询是否可以分解为多个更简单的子查询？如果可以，请将其分解为2-4个子查询。
每个子查询应该独立且能覆盖原查询的一部分。
只返回子查询，每行一个，不要编号。如果不能分解，只返回原查询。"""
        
        response = self.llm.generate(prompt)
        subqueries = [line.strip() for line in response.split('\n') if line.strip()]
        
        return subqueries if subqueries else [complex_query]
    
    def transform_query(self, query, context=None):
        # 根据上下文转换查询，使其更明确
        if not context:
            return query
        
        prompt = f"""对话历史:
{context}

当前用户查询: {query}

请根据对话历史，将用户的当前查询转换为更明确、更完整的查询。
如果用户使用了代词，请替换为具体的实体。如果查询依赖于上下文信息，请将这些信息包含在转换后的查询中。
只返回转换后的查询。"""
        
        return self.llm.generate(prompt)

查询处理可以显著提高检索质量，特别是对于复杂或依赖上下文的查询。

4.2.3 处理特殊情况

让我们探讨一些特殊情况和如何处理它们。

处理冲突信息

当记忆系统中存在冲突信息时，我们需要一种策略来解决这些冲突：

class ConflictResolver:
    def __init__(self, reliability_scorer=None):
        self.reliability_scorer = reliability_scorer
    
    def resolve_conflicts(self, documents, query=None):
        # 首先，识别冲突的信息
        conflicting_groups = self._identify_conflicts(documents)
        
        resolved_docs = []
        for group in conflicting_groups:
            if len(group) == 1:
                # 没有冲突，直接添加
                resolved_docs.append(group[0])
            else:
                # 有冲突，需要解决
                resolved = self._resolve_group(group, query)
                resolved_docs.append(resolved)
        
        return resolved_docs
    
    def _identify_conflicts(self, documents):
        # 简化实现：按主题分组
        # 实际应用中可能需要更复杂的冲突检测
        topics = {}
        for doc in documents:
            topic = doc.metadata.get('topic', 'default')
            if topic not in topics:
                topics[topic] = []
            topics[topic].append(doc)
        return list(topics.values())
    
    def _resolve_group(self, group, query