08 记忆与检索

我们的智能体已经拥有许多能力，例如调用 Tools 或使用 Skills，但它仍然缺少一个关键能力：记忆。如果智能体无法记住先前的交互内容，也无法从历史经验中学习，那么在连续对话或复杂任务中，其表现将受到极大限制。 智能体要帮助干好活，就得跟人一样，具备记忆功能，如果一个Agent什么记忆能力也没有，刚处理完的事情就忘记了，那智能体肯定就不能办好事情。我们希望智能体就像一个秘书一样，能够逐步学习，具备好记忆。

在本章中，我们将介绍 Agent 的两大核心能力：记忆系统（Memory System）与检索增强生成（Retrieval-Augmented Generation, RAG）。

一、记忆系统

在构建智能体的记忆系统之前，我们先从认知科学角度理解人类如何处理与存储信息。人类记忆[1]是一个多层级认知系统，不仅能够存储信息，还能根据重要性、时间与上下文对信息进行分类整理。根据认知心理学研究，人类记忆主要分为两大类型：短期记忆和长期记忆。然而，现代研究更倾向于将短期记忆理解为工作记忆，因此认知心理学常以工作记忆框架研究记忆系统。

1. 工作记忆

工作记忆（Working Memory）[2]是一种容量有限的认知系统，用于短暂保存处理信息。其更准确的定义是：在存在干扰的情境下，个体仍能暂时保持并处理信息，以支持各种日常认知活动。1956 年，米勒最早对短期记忆能力进行定量研究，提出“神奇的数字：7±2”。他观察到，年轻人的记忆广度大约为 7 个单位（阿拉伯数字、字母、单词或其他单位）[3]。后续研究发现，这一7±2容量并非单指短期记忆，而是工作记忆与长期记忆共同作用的结果。打个比方，假设有6个字母“C,A,T,D,O,G”，如果没有长期记忆参与，那么工作记忆中只能记住6个单独的字母，但是长期记忆告诉你这构成了两个单词，所以变成“CAT”和“DOG“只占两个单位的记忆容量了。

关于工作记忆已有多种理论模型，其中广泛接受的一种是巴德利与希奇提出的巴德利工作记忆模型[4]（Baddeley's model of working memory）。该模型同时考虑视觉与听觉刺激，并以长期记忆为参照，由中央执行系统（Central Executive）对各类信息进行整合与处理。

2. 长期记忆

长期记忆（Long-term Memory）是能够持续数天到数年的记忆，通常可分为三个主要子类别：

• 程序性记忆（Procedural Memory）是指对特定动作或技能执行方式的记忆。此类记忆通常在潜意识层面被激活，或仅需极少意识投入。程序性记忆包含与特定任务、程序或惯例相关的刺激-反应（Stimulus-Response）式信息，并通过习惯化关联被激活。当一个人在特定情境下几乎“自动地”作出反应时，便是在运用程序性知识。典型例子是驾驶汽车。
• 语义记忆（Semantic Memory）指个体所拥有的百科式知识。例如，对上海东方明珠外观的认识，或对同班同学姓名的记忆，都是语义记忆的体现。语义记忆的提取难度从轻松到费力不等，这取决于多种因素，包括但不限于：信息编码时间先后、信息之间关联数量、被提取频率，以及意义层次（即信息在最初学习时被处理的深层程度）。
• 情景记忆（Episodic Memory）是对个人生活事件的记忆，并且能够被明确陈述。它包括所有具有时间特征的记忆，例如上次吃饭的时间，或得知某项重大事件时所处的位置。情景记忆通常需要较高程度的意识加工，因为其形成往往需要整合语义记忆与时间信息，以构建完整的事件记忆。

人类的学习和经验的形成都离不开长期记忆的构建。

3. 为什么 Agent 需要记忆

人类智能的一个重要特征，是能够记住过去经历并从中学习，再将经验迁移到新的情境中。同样，真正的智能体也需要具备记忆能力，才能完成更复杂的任务。

由于 Agent 的核心“大脑”即大模型是无状态的，因此无论模型多么强大，如果不进行记忆存储，每一次用户请求（或 API 调用）都只是独立且无关联的计算。这会带来许多问题，例如：

• 上下文丢失：Agent 无法记住你与它上一次对话的内容。
• 个性化缺失：Agent 无法记住用户偏好与习惯，难以动态适应用户需求。
• 学习能力受限：Agent 无法从既往问题中吸取教训，下次遇到类似问题仍可能重复犯错。
• 一致性问题：Agent 可能给出前后矛盾的回答，难以保证长期一致性。

我们经常说Agent记忆会有问题，大模型的记忆很难，可是人的记忆就不会出问题么，人的记忆就是准的么？

大家看看这个图片，回顾一下哪个是正确的，这个时候你还觉得自己的记忆是准确的么？据统计，人类日常记忆的平均准确率仅 62%。

不要说记忆了，研究现场看到的都不一定准确的，例如下面这个图片，大家觉得A和B哪个颜色更深？答案可能是出乎意料的，A和B颜色是一样的。可见即使是人，即使是现场看到，也都会判断错误，也都会产生幻觉：）那么幻觉是否可能就是神经网络带来的本来问题？

4. 记忆系统的设计

借鉴人类记忆机制，我们也可以用类似思路构建 Agent 的记忆系统。例如，可将记忆分为两大类：工作记忆与长期记忆；其中长期记忆又可分为情景记忆与语义记忆。对当前 Agent 记忆系统建模而言，程序性记忆可暂不纳入重点。

（1）工作记忆

工作记忆主要用于存储当前对话的上下文信息。为了确保高速访问与响应，并保持信息“近期性”，我们可以限制其容量（例如 20 条），并设置 TTL（Time To Live）机制。系统还可按固定时间间隔触发“记忆整合”（将工作记忆压缩或转化为长期记忆）或“记忆遗忘”（删除不重要或过时的信息）。

这个龙虾里面就是作为每次交互的上下文，所以龙虾中让人诟病的一个地方就是交互使用的token太多了，因为需要携带太多的上下文信息，也就是所谓的工作记忆信息。

（2）情景记忆

情景记忆主要负责长期存储具体交互事件与智能体学习经历，重点在于保持事件完整性与时间序列关系。情景记忆设计应保留必要上下文信息，并附带时间戳等元数据，以支持 Agent 回顾过往经历。

这个是龙虾里面的几个初始化配置文件，表明龙虾的身份，对应的一些特点等。

（3）语义记忆

语义记忆主要存储更抽象的知识、概念与规则。例如，通过对话获得的用户偏好、需长期遵守的指令或领域知识点。语义记忆设计应强调信息的结构化与关联性，以便智能体在需要时快速检索和应用，甚至结合知识图谱进行复杂关系推理。

这个有什么案例么？

记忆系统既要“存”也要“取”，因此通常需要借助向量数据库（Vector Database）进行记忆存储，并通过向量检索技术实现高效召回。这一部分将结合检索增强生成（RAG）展开说明。

二、检索增强生成

除了容易遗忘对话历史，LLM 的另一个核心局限在于其知识是静态且有限的。这些知识完全来自训练数据，因此会带来一系列问题：

• 知识时效性：大模型训练数据存在截止日期，因此无法了解该时间点之后的新事件与新知识。
• 专业领域覆盖不足：训练数据虽然庞大，但仍无法覆盖所有领域，尤其是专业或小众领域。
• 知识准确性风险：训练数据可能包含错误或过时信息；通过检索验证，可降低模型幻觉。

为了解决这些问题，我们可以引入检索增强生成（Retrieval-Augmented Generation, RAG）技术。 RAG 的核心思想是将外部知识库与生成模型结合：在生成回答前，先检索相关信息，再将检索结果作为上下文输入模型，从而提高回答的准确性与时效性。

RAG 的实现通常包括两个核心环节： （1）数据准备阶段。系统通过数据提取、文本分割与向量化，将外部知识库中的文本转换为向量表示，并存储至向量数据库，以供后续检索。 （2）检索与生成阶段。当用户提出查询时，系统从数据库检索相关信息，将其注入 Prompt，最终驱动大语言模型生成答案。

此外，还可以采用一些高级检索策略进一步提升 RAG 性能，例如多查询扩展（MQE）和假设文档嵌入（HyDE）。

1. 数据准备阶段

数据准备阶段的核心任务，是将外部知识库文本转换为向量表示并存储到向量数据库。例如，可先将 PDF、Word、Excel 等格式文档统一转为 Markdown；再对文本进行分割（如按标题层级、段落、句子）；随后使用预训练语言模型（如 BERT、RoBERTa）生成向量表示（Embedding）；最后将向量写入向量数据库（如 Pinecone、Weaviate）。

分块策略非常丰富。除了按段落、句子分割，还可以按文本结构分割（如标题、子标题），或按语义分割（如利用文本相似度将内容划分为不同主题块）。也可采用父子块策略：检索时使用子块提高精确性，生成时回填父块保证上下文完整性。

2. 检索与生成阶段

检索与生成阶段是 RAG 的核心环节，主要负责根据用户查询从向量数据库检索相关信息，并将其与用户输入共同作为上下文提供给生成模型。该过程通常包括以下步骤：

1. 用户查询预处理：对用户的查询进行预处理，例如提取关键词、重写查询等，以便于后续的向量化和检索。
2. 向量化：使用预训练的语言模型将用户查询转换为向量表示。
3. 向量检索：在向量数据库中找到与用户查询向量相似的向量，并返回对应的文本信息。
4. 上下文构建：将检索到的信息与用户查询一起构建成一个新的上下文输入给生成模型。
5. 生成回答：使用生成模型根据新的上下文生成回答。

通过引入 RAG，智能体可以有效弥补大模型的知识局限，输出更准确、更具时效性的回答，同时提升回答的可信度与可解释性。

3. 高级检索策略

RAG 系统的检索能力是其核心竞争力。在实际应用中，用户查询表述与文档内容常存在用词差异，导致相关文档未被召回。为解决这一问题，可以采用两种额外的高级检索策略：多查询扩展（MQE）和假设文档嵌入（HyDE）。

（1）多查询扩展（MQE）

多查询扩展（Multi-Query Expansion）是一种通过生成语义等价、表达多样的查询来提升检索召回率的技术。其核心是：同一个问题可以有多种表述，而不同表述可能匹配到不同文档。例如，“如何学习 Python”可扩展为“Python 入门教程”“Python 学习方法”“Python 编程指南”等。通过并行执行这些扩展查询并合并结果，系统可覆盖更广泛的相关文档，避免因措辞差异遗漏关键信息。

MQE 的优势在于能够自动理解查询的多种潜在含义，尤其对模糊查询或专业术语查询效果显著。系统通常使用 LLM 生成扩展查询，以保证扩展的多样性与语义相关性。

（2）假设文档嵌入（HyDE）

假设文档嵌入（Hypothetical Document Embeddings，HyDE）是一种创新检索技术，其核心思想是“用答案找答案”。传统方法通常用问题直接匹配文档，但问题与答案在语义空间中的分布往往存在差异: 问题常为疑问句，而文档多为陈述句。HyDE 先让 LLM 生成一段“假设答案”，再用该答案检索真实文档，从而缩小查询与文档之间的语义鸿沟。

这种方法的优势在于，假设答案与真实答案在语义空间中更接近，因此更容易匹配到相关文档。即使假设答案并不完全正确，其中包含的关键术语、概念和表达风格也能有效引导检索系统定位正确文档。尤其在专业领域查询中，HyDE 能生成包含领域术语的假设文档，显著提升检索精度。

（3）扩展检索框架

将 MQE 与 HyDE 整合到统一扩展检索框架后，可根据具体场景灵活启用策略：需要高召回率时可同时启用两者；性能敏感场景可仅使用基础检索或启用单一策略。

扩展检索的核心机制是“扩展-检索-合并”三步流程。首先，系统基于原始查询生成多个扩展查询（包括 MQE 生成的多样化查询和 HyDE 生成的假设文档）；然后，对每个扩展查询并行执行向量检索，构建候选文档池；最后，通过去重与分数排序合并结果，返回最相关的 Top-k 文档。该设计通过扩大候选池提升召回上限，并通过智能去重避免重复内容。

在实际应用中，这三种策略组合使用通常效果最佳。MQE 擅长解决用词多样性问题，HyDE 擅长跨越查询与文档的语义鸿沟，统一框架则保障结果质量与多样性。

三、向量数据库的使用

向量数据库的核心价值，在于把“不可直接计算语义距离的文本”转换为“可在高维空间中计算相似度的向量”，从而让检索系统具备可扩展、可量化的语义召回能力。

1. 从文本到向量

文本向量化，本质上就是 Embedding 生成。系统先将文档分块，再使用向量模型把每个文本块映射为定长向量，最后连同元数据一起写入向量数据库。实践中有两个关键点：

1. 查询向量与文档向量应尽量使用同一模型（或同一家族模型），以减少向量空间不一致带来的召回偏差。
2. 入库时应保留必要元数据（如来源、时间、标签、段落位置），便于后续按条件过滤与结果追溯。

2. 向量相似度

将文本映射为向量后，检索的本质就是计算“查询向量”与“文档向量”之间的距离或相似度，并返回最接近的 Top-k 结果。常见指标包括余弦相似度（Cosine Similarity）、点积（Dot Product）和欧氏距离（Euclidean Distance）。

余弦相似度关注向量夹角，衡量“方向是否一致”，因此在文本语义检索中最常见；点积同时受方向和长度影响，在某些模型与索引配置下性能更优；欧氏距离衡量几何空间中的直线距离，适合特定向量分布。不同指标并无绝对优劣，关键在于与 Embedding 模型和索引方式匹配。

3. 召回与重排序

相似度分数并不总能直接代表“可用于回答”的质量。工程上常通过“向量召回 + 重排序（Rerank）”提升结果质量：先用向量检索召回候选文档，再用更强的排序模型进行二次打分，最终将更相关、更可用的片段提供给大模型生成答案。

4. 向量数据库的使用示例

下面给出一个轻量示例：使用 Chroma 作为本地向量数据库，完成“入库 + 检索”的操作。

# 依赖：pip install chromadb openai
import os
import chromadb
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# 1) 初始化本地向量库
chroma = chromadb.PersistentClient(path="./chroma_data")
collection = chroma.get_or_create_collection(name="agent_memory_demo")

# 2) 准备文本数据（真实项目中通常来自分块后的文档）
docs = [
    "RAG 会先检索外部知识，再把检索结果注入 Prompt。",
    "工作记忆强调近期性，通常会限制容量并设置 TTL。",
    "HyDE 通过先生成假设答案来缩小查询和文档的语义鸿沟。",
]

# 3) 生成向量并写入数据库
def embed(text: str) -> list[float]:
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=text,
    )
    return resp.data[0].embedding

embeddings = [embed(d) for d in docs]
ids = [f"doc_{i}" for i in range(len(docs))]
metadatas = [{"source": "chapter08"} for _ in docs]

collection.add(
    ids=ids,
    documents=docs,
    embeddings=embeddings,
    metadatas=metadatas,
)

# 4) 查询：把问题向量化后做相似度检索
query = "为什么要给 Agent 设置工作记忆 TTL？"
query_embedding = embed(query)

result = collection.query(
    query_embeddings=[query_embedding],
    n_results=2,
    where={"source": "chapter08"},
)

print("检索结果：")
for i, text in enumerate(result["documents"][0], start=1):
    print(f"{i}. {text}")

这个示例对应了 RAG 的关键步骤：文本向量化、向量入库、相似度检索与结果返回。若要进一步增强能力，可在此基础上增加分块策略优化、召回后重排序、高级检索策略等。

四、记忆系统与检索工具

记忆系统和检索增强系统都可以以工具的形式提供给 Agent 使用。Agent 可以自主调用记忆工具来存储与管理智能体的记忆内容，或调用检索工具从外部知识库中检索相关信息。

1. 记忆系统的实现

（0）记忆条目

我们首先需要一个统一的记忆条目，用来保存 Agent 的每条记忆。

@dataclass
class MemoryItem:
    """统一的记忆条目结构，供三类记忆模块共用。"""

    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    content: str
    importance: float = 0.5
    session_id: str
    tags: list[str] = field(default_factory=list)
    timestamp: float = field(
        default_factory=lambda: datetime.now(timezone.utc).timestamp()
    )

在每一条记忆中，我们需要保存id用以区分不同的记忆，需要保存content记忆的内容与importance重要性；我们可以保存session_id，用来记录 Agent 的记忆产生于哪一次会话当中，保存tags来标注某条记忆以便分类筛选，以及一个元数据timestamp用以记录记忆创建的时间。

（1）工作记忆

添加:工作记忆的目标是“快”和“新”。因此实现时采用内存字典保存数据，并在每次添加后执行两件事：

1. 清理过期记忆（TTL）
2. 超容量时仅保留最新的若干条

这样做的直觉很简单：工作记忆只服务当前会话窗口，不追求长期完整保存。

def add(self, memory: MemoryItem) -> None:
    # 1) 写入当前记忆
    self._items[memory.id] = memory
    # 2) 清理过期记忆
    self.cleanup_expired()
    # 3) 控制容量
    self._enforce_capacity()

def cleanup_expired(self) -> int:
    now = datetime.now(timezone.utc).timestamp()
    removed_ids = [
        mid for mid, m in self._items.items()
        if now - m.timestamp > self.ttl_seconds
    ]
    for mid in removed_ids:
        self._items.pop(mid, None)
    return len(removed_ids)

def _enforce_capacity(self) -> None:
    if len(self._items) <= self.capacity:
        return

    # 超过容量时，仅保留最近写入的记忆。
    sorted_items = sorted(
        self._items.values(),
        key=lambda m: m.timestamp,
        reverse=True,
    )
    keep = {m.id for m in sorted_items[: self.capacity]}
    self._items = {m.id: m for m in sorted_items if m.id in keep}

查询

工作记忆查询使用轻量的文本相似度（TF-IDF 风格），再叠加近期性与重要性：

• 相关性：问题与记忆内容的词项匹配程度
• 近期性：越新得分越高（指数衰减）
• 重要性：人工或后续模型打分

综合得分示例：

def query(self, query_text: str, top_k: int = 5, tags: list[str] | None = None):
    # ... 预处理：过滤标签、计算 query 向量、准备统计量
    for mem in candidates:
        rel = self._tfidf_similarity(query_vec, mem_vec, doc_freq, len(candidates))
        rec = recency_score(now, mem.timestamp, half_life_seconds=max(300, self.ttl_seconds / 2))
        imp = normalize_importance(mem.importance)
        total = 0.5 * rel + 0.35 * rec + 0.15 * imp
        scored.append(ScoredMemory(memory=mem, score=total, ...))

    scored.sort(key=lambda x: x.score, reverse=True)
    return scored[:top_k]


def recency_score(now_ts: float, memory_ts: float, half_life_seconds: float) -> float:
    age = max(0.0, now_ts - memory_ts)
    if half_life_seconds <= 0:
        return 1.0
    return math.exp(-math.log(2) * age / half_life_seconds)

这部分实现体现了“短期记忆优先近期信息”的原则，因此近期性权重较高。

（2）情景记忆

情景记忆用于存储“发生过什么”。相比工作记忆，它更关注事件的可回溯性和长期保存。

添加

情景记忆添加时除了保存原始文本，还会同步生成向量并持久化到本地文件。

def add(self, memory: MemoryItem) -> None:
    self._items[memory.id] = memory
    self._vectors[memory.id] = self.embed(memory.content)
    self._save()

def embed(self, text: str) -> list[float]:
    # 暂时用哈希向量代替真实 embedding 模型
    return hash_embedding(text, dim=96)

查询

查询时按标签过滤后，融合相关性、近期性、重要性打分并返回 Top-k：

1. 持久化：使用本地 JSON 文件存储条目和向量
2. 向量检索：使用模拟 embedding 计算语义相关性
3. 打分策略：相关性 + 近期性 + 重要性，但近期性半衰期更长

def query(self, query_text: str, top_k: int = 5, tags: list[str] | None = None):
    # ... 计算 query_vec
    for mem in self._items.values():
        # ... 标签过滤
        rel = cosine_similarity(query_vec, self._vectors.get(mem.id, []))
        rec = recency_score(now, mem.timestamp, half_life_seconds=7 * 24 * 3600)
        imp = normalize_importance(mem.importance)
        total = 0.5 * rel + 0.3 * rec + 0.2 * imp
        # ... 追加评分结果
    # ... 排序并返回 top_k

def _save(self) -> None:
    self.storage_path.parent.mkdir(parents=True, exist_ok=True)
    data = {
        "items": [m.to_dict() for m in self._items.values()],
        "vectors": self._vectors,
    }
    self.storage_path.write_text(
        json.dumps(data, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

情景记忆的时间衰减速度比工作记忆慢，原因是“过去事件”本身就有长期价值。

（3）语义记忆

语义记忆用于保存“可复用知识”，例如偏好、规则、事实。它不是一次事件，而是可迁移的知识片段。

添加

语义记忆的添加流程与情景记忆类似：写入条目、生成语义向量、执行持久化。

def add(self, memory: MemoryItem) -> None:
    self._items[memory.id] = memory
    self._vectors[memory.id] = self.embed(memory.content)
    self._save()

def embed(self, text: str) -> list[float]:
    # 语义记忆使用更高维度，增强语义区分度
    return hash_embedding(text, dim=128)

查询

语义记忆更强调“相关性”，并在向量相似度上叠加关键词重叠奖励：

1. 同样支持持久化与向量检索
2. 检索时除了向量相似度，再加入少量关键词重叠奖励
3. 提供高级语义检索占位函数，便于后续接入更强方案

def query(self, query_text: str, top_k: int = 5, tags: list[str] | None = None):
    # ... 生成 query_vec 与 query_terms
    for mem in self._items.values():
        # ... 标签过滤
        rel = cosine_similarity(query_vec, memory_vec)
        overlap = len(query_terms & mem_terms) / max(1, len(query_terms))
        rel = min(1.0, 0.85 * rel + 0.15 * overlap)
        total = 0.65 * rel + 0.25 * imp + 0.10 * rec
        # ... 追加评分结果
    # ... 排序并返回 top_k

def advanced_semantic_search_placeholder(self, query_text: str) -> list[str]:
    """
    占位函数：后续可扩展 cross-encoder 重排、知识图谱扩展、混合检索等。
    """
    _ = query_text
    return []

语义记忆中“相关性”权重最高，因为它的核心任务是知识匹配，而不是时间回放。语义记忆可以采用更高级的检索方法，如引入“实体-关系”模型等，有兴趣的同学可以了解相关内容。

（4）统一记忆管理器

如果每次都直接操作三类记忆模块，接口会很分散。为此实现一个统一管理器，对外提供一致的 CRUD 与查询能力。

核心接口：

1. add_memory：按记忆类型写入
2. update_memory / delete_memory / get_memory：按类型路由
3. query_memory / query_all：支持单类或多类查询
4. run_maintenance：统一执行维护任务

维护逻辑目前包含：

• 工作记忆过期清理
• 重要性重平衡（占位实现）
• 低重要性长期记忆遗忘

其中“重要性重平衡”已预留大模型接入点，目前使用占位函数，避免引入额外依赖，便于先理解架构。

def query_all(self, query_text: str, top_k: int = 5, tags: list[str] | None = None):
    return {
        MemoryType.WORKING: self.working_memory.query(query_text, top_k=top_k, tags=tags),
        MemoryType.EPISODIC: self.episodic_memory.query(query_text, top_k=top_k, tags=tags),
        MemoryType.SEMANTIC: self.semantic_memory.query(query_text, top_k=top_k, tags=tags),
    }

def run_maintenance(self) -> dict[str, int]:
    # ... 可继续扩展更多维护动作
    removed_working = self.working_memory.cleanup_expired()
    updated_importance = self.rebalance_priorities()
    return {
        "removed_working": removed_working,
        "updated_importance": updated_importance,
    }

实际每条记忆的重要性计算可以通过询问大模型的方式，或者使用特别训练的模型，有兴趣的同学可以自行探索。

2. 检索工具（RAG）的实现

RAG 管理器的目标是两件事：这个部分不是前面已经讲过了么？

1. 建库：把文件/目录内容切块并向量化
2. 检索：根据查询召回相关片段

（1）添加（建库）

实现中提供两个入口：

• add_file：处理单个文件
• add_directory：递归处理目录

每个文本块保存以下信息：

• chunk_id：块唯一标识
• source：来源文件
• content：块内容
• tags：标签
• timestamp 与 metadata：辅助过滤和追踪

分块策略采用“按段落优先 + 超长硬切分”的简化方案。

def add_directory(self, dir_path: str, tags: list[str] | None = None, metadata: dict | None = None):
    # ... 遍历目录并复用 add_file
    ...

def _split_text(self, text: str) -> list[str]:
    raw_parts = [p.strip() for p in text.split("\n\n") if p.strip()]
    chunks: list[str] = []
    current = ""

    # 文本分块
    for part in raw_parts:
        if len(current) + len(part) + 2 <= self.chunk_size:
            current = f"{current}\n\n{part}".strip()
        else:
            if current:
                chunks.append(current)
            if len(part) <= self.chunk_size:
                current = part
            else:
                for i in range(0, len(part), self.chunk_size):
                    chunks.append(part[i : i + self.chunk_size])
                current = ""

    if current:
        chunks.append(current)
    return chunks

def add_file(self, file_path: str, tags: list[str] | None = None, metadata: dict | None = None):
    # ... 读取文件
    chunks = self._split_text(text)
    for index, chunk_text in enumerate(chunks):
        # ... 为每个 chunk 生成 id、向量、元数据并入库
        ...
    return len(chunks)

（2）查询（检索）

基础检索采用“稠密相似度 + 稀疏重叠”的混合打分：

def query(
    self,
    query_text: str,
    top_k: int = 5,
    tags: list[str] | None = None,
    use_mqe: bool = False,
    use_hyde: bool = False,
    use_rerank: bool = False,
):
    expanded_queries = [query_text]

    # 启用高级检索策略
    if use_mqe:
        expanded_queries.extend(self.multi_query_expansion_placeholder(query_text))

    if use_hyde:
        hypothetical = self.hyde_placeholder(query_text)
        if hypothetical:
            expanded_queries.append(hypothetical)

    # 去重
    merged = self._merge_deduplicate([
        self._retrieve_once(q, top_k=top_k, tags=tags)
        for q in expanded_queries
    ])
    merged.sort(key=lambda r: r.score, reverse=True)
    final_results = merged[:top_k]
    return self.rerank_placeholder(query_text, final_results) if use_rerank else final_results

def _retrieve_once(self, query_text: str, top_k: int, tags: list[str] | None):
    # ... 计算 query_vec 与 query_terms
    for chunk in self._chunks.values():
        # ... 标签过滤
        # 叠加向量相似度与重叠性
        dense_score = cosine_similarity(query_vec, self._vectors.get(chunk.id, []))
        sparse_overlap = len(query_terms & chunk_terms) / max(1, len(query_terms))
        score = 0.8 * dense_score + 0.2 * sparse_overlap
        # ... 构建 RetrievalResult

    candidates.sort(key=lambda r: r.score, reverse=True)
    return candidates[:top_k]

这种设计能兼顾语义匹配与关键词命中，更容易观察检索行为。

（3）高级策略占位：MQE 与 HyDE

RAG 管理器预留了三个扩展接口：

1. multi_query_expansion_placeholder：多查询扩展（MQE）
2. hyde_placeholder：假设文档嵌入（HyDE）
3. rerank_placeholder：二阶段重排序

实际执行顺序是：扩展 -> 并行检索 -> 合并去重 -> 可选重排。

去重逻辑采用“同一 chunk 保留最高分”策略，避免不同扩展查询召回同一内容造成结果冗余。

@staticmethod
def _merge_deduplicate(result_lists: list[list[RetrievalResult]]) -> list[RetrievalResult]:
    best_by_chunk: dict[str, RetrievalResult] = {}
    for one_list in result_lists:
        for item in one_list:
            existing = best_by_chunk.get(item.chunk_id)
            if existing is None or item.score > existing.score:
                best_by_chunk[item.chunk_id] = item
    return list(best_by_chunk.values())

3. 小结

到这里我们已经完成了一个记忆与检索系统原型：

1. 三类记忆各自有独立检索与打分机制
2. 统一管理器提供一致接口与维护能力
3. RAG 支持建库、查询和高级策略扩展

后续如果要走向实际应用，优先加入三部分：真实embedding、向量数据库、重排序模型。

记忆体的开发没有止境，现在的记忆体仍然有很大的空间，龙虾的记忆和上下文管理仍然也有问题，所以token的消耗量也非常大，这里面如何优化，请大家再好好思考一下。

五、作业练习

本章作业位于：

minimal_agents/hw/chapter-8/memory/

这一组作业对应本章讲到的两部分核心内容：

• 记忆系统：工作记忆、情节记忆、语义记忆
• 检索系统：RAG 的分块、召回、去重与排序

目录中已经准备好了带 TODO 的代码框架与评测脚本，读者不需要从零搭建工程，可以直接围绕关键函数补全实现。作业的重点不是重新写一个完整项目，而是把本章介绍过的记忆管理与检索逻辑真正落到代码里。

建议按下面的顺序完成：

1. 先阅读 README.md，了解作业目标、文件结构与评分方式；
2. 先完成 src/memory/ 下的几类记忆模块；
3. 再完成 src/rag/manager.py 中的检索部分；
4. 最后运行 tests/test_grading.py 检查结果。

如果希望把这一章的内容真正学扎实，这份作业是很重要的一步。因为只有自己动手补过一次记忆与检索流程，才会更清楚这些模块在智能体系统中到底承担什么职责。