AI Agent Harness Engineering 记忆机制解析：短期记忆、长期记忆与上下文管理

关键词：AI Agent, Harness Engineering, 短期记忆, 长期记忆, 上下文管理, 向量数据库, 检索增强生成

摘要：本文深入浅出地解析了AI Agent（人工智能代理）的核心记忆机制，包括短期记忆、长期记忆与上下文管理三大模块。我们将通过生活中的生动比喻、数学模型、算法流程图与Python代码实战，带你一步步理解AI Agent如何像人类一样“记住”信息、学习知识并流畅交互。文章还涵盖了实际应用场景、工具资源推荐以及未来发展趋势，帮助你构建更智能、更人性化的AI Agent系统。

---

背景介绍

目的和范围

你有没有想过，为什么Siri、小爱同学这类语音助手有时候会“忘事”？比如你刚问完“今天天气怎么样”，接着说“那明天呢”，它却一脸懵地问“明天什么？”——这就是因为它们的记忆机制不够完善。

AI Agent是一种能自主感知环境、做出决策并执行动作的人工智能系统，而记忆机制就是AI Agent的“大脑硬盘”和“ sticky note（便签）”，决定了它能不能记住你的喜好、理解你的上下文、从过去的经验中学习。

本文的目的就是：

1. 用通俗易懂的方式讲清楚AI Agent的三种核心记忆类型（短期记忆、长期记忆、上下文管理）是什么；
2. 解释它们的工作原理、算法实现和数学模型；
3. 通过一个完整的Python项目实战，带你从零搭建一个有记忆的AI Agent；
4. 分享实际应用场景、工具推荐和未来趋势。

本文的范围聚焦于**Harness Engineering（AI Agent harness 工程，即AI Agent的框架与工程实现）**中的记忆模块，不会涉及太多深度学习的底层原理（比如transformer的注意力机制），而是侧重于如何工程化地实现和应用这些记忆机制。

预期读者

• AI爱好者与初学者：想了解AI Agent是怎么“记住”东西的；
• Python开发者：想动手实现一个有记忆的AI Agent；
• 产品经理与架构师：想设计基于AI Agent的产品，需要了解记忆机制的能力与边界；
• NLP工程师：想深入学习检索增强生成（RAG）、向量数据库等记忆相关技术。

不管你是刚接触AI的小白，还是有一定经验的开发者，本文都能让你有所收获——我们会像给小学生讲故事一样，把复杂的技术讲得明明白白！

文档结构概述

本文的结构就像搭积木一样，从基础概念到实战项目，一步步递进：

1. 背景介绍：为什么AI Agent需要记忆，本文要讲什么；
2. 核心概念与联系：用生活比喻解释三种记忆类型，分析它们的关系，画出架构图；
3. 核心算法原理：讲解实现三种记忆的算法，用Python代码演示；
4. 数学模型：用公式解释记忆背后的数学原理（比如相似度计算、token限制）；
5. 项目实战：从零搭建一个有记忆的AI聊天助手，包含完整代码；
6. 实际应用场景：看看记忆机制在客服、教育、编程等领域的应用；
7. 工具与资源推荐：推荐好用的向量数据库、框架和学习资源；
8. 未来发展趋势与挑战：聊聊记忆机制的未来和需要解决的问题；
9. 总结与思考题：回顾重点，给你留几个小问题练练手；
10. 附录与扩展阅读：常见问题解答和进一步学习的资料。

术语表

在开始之前，我们先把一些“生词”解释清楚，就像课前预习一样！

核心术语定义

1. AI Agent：一种能自主感知环境、存储记忆、做出决策并执行动作的人工智能系统——简单说就是“有脑子、能行动的AI”，比如AutoGPT、你的私人AI助手。
2. Harness Engineering：AI Agent的“框架工程”，指的是如何设计、实现和优化AI Agent的核心模块（记忆、推理、行动等），让AI Agent稳定、高效地工作。
3. 短期记忆（Short-Term Memory, STM）：AI Agent的“便签纸”，用来临时存储最近的信息（比如最近5轮对话），容量小、保存时间短。
4. 长期记忆（Long-Term Memory, LTM）：AI Agent的“文件柜”，用来永久存储重要信息（比如你的喜好、过去的对话总结、学习到的知识），容量大、保存时间长。
5. 上下文管理（Context Management）：AI Agent的“当前笔记本页面”，用来管理当前交互的上下文（比如把短期记忆和检索到的长期记忆组合成一个完整的prompt），确保AI能理解当前的对话。
6. 向量数据库（Vector Database）：一种专门用来存储和检索向量（即“数字表示的信息”）的数据库，是长期记忆的核心存储工具——比如Chroma、FAISS、Pinecone。
7. 检索增强生成（Retrieval-Augmented Generation, RAG）：一种让AI先从长期记忆中检索相关信息，再生成回答的技术，能让AI的回答更准确、更有依据。

相关概念解释

1. Embedding（嵌入）：把文字、图片等信息转换成一串数字（向量）的过程——比如把“我喜欢喝咖啡”转换成[0.12, 0.56, -0.34, …]这样的向量，让计算机能“理解”信息的含义。
2. Cosine Similarity（余弦相似度）：一种计算两个向量相似程度的方法——数值越接近1，说明两个信息越相关；越接近-1，说明越不相关。
3. Token（词元）：AI处理文本的最小单位——比如“我喜欢喝咖啡”可能被分成“我”、“喜欢”、“喝”、“咖啡”4个token，AI的上下文窗口（context window）就是用token数量来限制的。
4. Prompt（提示词）：给AI的输入文本——比如“你是一个私人助手，请帮我订咖啡”就是一个prompt。

缩略词列表

缩略词	全称	中文含义
AI	Artificial Intelligence	人工智能
STM	Short-Term Memory	短期记忆
LTM	Long-Term Memory	长期记忆
RAG	Retrieval-Augmented Generation	检索增强生成
NLP	Natural Language Processing	自然语言处理
LLM	Large Language Model	大语言模型
DB	Database	数据库

---

核心概念与联系

故事引入：你的AI私人助理“小月”

让我们先从一个生活中的小故事开始，理解AI Agent的记忆机制是怎么工作的。

假设你有一个超级智能的私人AI助理，叫“小月”：

1. 周一早上：你跟小月说：“小月，我每天早上都要喝一杯冰美式，不加糖不加奶。”小月把这句话记在了她的“文件柜”（长期记忆）里。
2. 周二早上：你刚起床，说：“小月，帮我订杯咖啡。”小月马上就从“文件柜”里找出你的喜好，订了一杯冰美式——这就是长期记忆的作用。
3. 周二下午：你跟小月聊天：“今天天气真热啊，我想去买个西瓜。”小月把这句话记在了她的“便签纸”（短期记忆）上。
4. 过了5分钟：你又说：“对了，别忘了帮我拿快递。”小月说：“好的，我会帮你拿快递——另外，你刚才说想去买西瓜，需要我帮你查附近的水果店吗？”——这就是短期记忆的作用，小月记住了你最近说的话。
5. 周三早上：你问小月：“我昨天说想喝什么来着？”小月先看了看“便签纸”（短期记忆，可能已经清空了），然后从“文件柜”里找出了你的咖啡喜好，还从“文件柜”里找到了昨天下午的对话总结——这就是上下文管理的作用，小月把不同来源的记忆组合起来，给你一个准确的回答。

你看，小月的工作方式是不是和人类很像？人类的大脑也有短期记忆（比如你刚记住的一个电话号码）、长期记忆（比如你的生日）和上下文管理（比如你跟朋友聊天时，能记住刚才聊了什么）。AI Agent的记忆机制就是模仿人类大脑的记忆系统设计的！

核心概念解释（像给小学生讲故事一样）

好了，故事讲完了，现在我们来正式解释三个核心概念——我们会用生活中的比喻，让你一听就懂！

核心概念一：短期记忆（STM）——你的“便签纸”

什么是短期记忆？
短期记忆就像你书桌上的那张黄色便签纸：

• 容量小：便签纸只有那么大，只能写几句话——AI的短期记忆也一样，只能存储最近的几轮对话或者少量信息（比如最近10轮对话，或者1000个token）。
• 保存时间短：便签纸用完就会扔掉，或者被新的内容覆盖——AI的短期记忆也一样，过了一段时间（比如对话结束），或者内容太多，旧的信息就会被删掉。
• 读写速度快：便签纸就在你眼前，拿起来就能写、就能看——AI的短期记忆也一样，存取信息非常快，不需要去“文件柜”里翻找。

短期记忆有什么用？
短期记忆的主要作用是保持当前交互的连贯性——比如你跟AI聊天时，刚说“我喜欢看科幻电影”，接着说“推荐几部”，AI需要用短期记忆记住你刚才说的“科幻电影”，才能给你推荐正确的内容。

生活中的例子：
你去超市买东西，列了一个购物清单写在便签纸上：“牛奶、面包、鸡蛋”——这就是你的短期记忆。等你买完东西，便签纸就可以扔掉了——这就是短期记忆的“临时性”。

---

核心概念二：长期记忆（LTM）——你的“文件柜”

什么是长期记忆？
长期记忆就像你家里的那个大文件柜：

• 容量大：文件柜可以放很多文件夹、很多文件——AI的长期记忆也一样，可以存储成千上万条信息（比如你的所有对话记录、学习到的所有知识）。
• 保存时间长：文件柜里的文件可以放几年、几十年——AI的长期记忆也一样，只要你不删掉，信息就会一直存在。
• 读写速度慢：找文件柜里的文件需要翻找，不如便签纸快——AI的长期记忆也一样，存取信息需要一定的时间（比如从向量数据库里检索信息）。

长期记忆有什么用？
长期记忆的主要作用是存储重要信息和知识——比如你的喜好、过去的对话总结、学习到的专业知识、产品文档等等。有了长期记忆，AI就能“个性化”地为你服务（比如记住你的咖啡口味），还能“学习”新的知识（比如把你的产品文档存进去，回答用户的问题）。

生活中的例子：
你把你的出生证明、学历证书、银行卡密码（当然要加密！）都放在文件柜里——这些都是你的长期记忆，需要长期保存，随时可能用到。

---

核心概念三：上下文管理——你的“当前笔记本页面”

什么是上下文管理？
上下文管理就像你正在写的笔记本的当前页面：

• 你会把便签纸上的内容（短期记忆）抄到当前页面上；
• 你会从文件柜里找出相关的文件（长期记忆），把重点内容也抄到当前页面上；
• 然后你看着当前页面，思考接下来要写什么——这就是AI的上下文管理：把短期记忆、检索到的长期记忆、系统提示词组合成一个完整的“上下文”（即prompt），让AI能理解当前的情况，生成合适的回答。

上下文管理有什么用？
上下文管理的主要作用是整合记忆，让AI理解当前的场景——因为AI的“脑子”（大语言模型）有一个“上下文窗口”的限制（比如GPT-4的上下文窗口是8k或者32k token），不能把所有的记忆都放进去，所以需要上下文管理来“挑选”最相关的信息，组合成一个有效的prompt。

生活中的例子：
你在写一篇关于“咖啡历史”的论文：

1. 你先看了看你的便签纸（短期记忆），上面写着“今天要写咖啡的起源”；
2. 你从文件柜里找出了《咖啡通史》这本书（长期记忆），翻到了“起源”那一章；
3. 你把便签纸的内容和书里的重点都抄到了笔记本的当前页面上；
4. 然后你看着当前页面，开始写论文——这就是上下文管理的过程！

---

核心概念之间的关系

现在我们知道了三个核心概念是什么，接下来我们来看看它们是怎么“合作”的——就像一个团队一样，每个成员都有自己的分工，一起完成任务！

概念一和概念二的关系：短期记忆是长期记忆的“缓冲垫”

你有没有发现，你记东西的时候，总是先记在脑子里（短期记忆），然后过了一段时间，重要的东西才会被“存”到长期记忆里？AI Agent的记忆也是一样的！

关系描述：
短期记忆是长期记忆的“缓冲垫”和“筛选器”：

1. 所有的新信息（比如用户的对话、环境的变化）都会先进入短期记忆；
2. 过了一段时间，或者短期记忆满了，我们会把短期记忆里的重要信息（比如用户的喜好、关键的对话内容）“总结”一下，存到长期记忆里；
3. 不重要的信息（比如用户说的“嗯”、“哦”这类语气词）就会被删掉，不占用长期记忆的空间。

生活中的例子：
你今天跟朋友聊了很多话：

1. 先都记在你的脑子里（短期记忆）；
2. 晚上睡觉前，你会把今天的重要事情（比如朋友说她下周要过生日）记在你的日记里（长期记忆）；
3. 那些不重要的闲聊（比如“今天天气真好”）就会被你忘掉，不占用你的“长期记忆空间”。

---

概念二和概念三的关系：长期记忆是上下文管理的“知识库”

上下文管理需要“素材”才能组合出有效的prompt，而长期记忆就是这个“素材库”！

关系描述：
长期记忆是上下文管理的“知识库”和“数据源”：

1. 当用户问了一个问题，上下文管理会先“理解”用户的问题（比如用embedding把问题转换成向量）；
2. 然后去长期记忆里“检索”最相关的信息（比如用余弦相似度找最相似的向量）；
3. 把检索到的信息放到上下文里，让AI能根据这些信息生成回答——这就是RAG（检索增强生成）的核心思想！

生活中的例子：
你老师问你：“咖啡的起源地是哪里？”

1. 你先“理解”了这个问题；
2. 然后从你的“文件柜”（长期记忆）里找出了《咖啡通史》这本书；
3. 翻到“起源”那一章，找到“埃塞俄比亚”这个答案；
4. 然后把这个答案告诉老师——这就是长期记忆和上下文管理的合作过程！

---

概念一和概念三的关系：短期记忆是上下文管理的“当前场景”

上下文管理不仅需要长期记忆的“知识”，还需要短期记忆的“当前场景”——比如最近的对话内容！

关系描述：
短期记忆是上下文管理的“当前场景记录”：

1. 上下文管理会把短期记忆里的最近几轮对话（比如最近5轮）直接放到上下文里；
2. 这样AI就能知道“刚才聊了什么”，保持对话的连贯性；
3. 如果短期记忆里的内容太多，超过了上下文窗口的限制，上下文管理会“压缩”短期记忆（比如只保留最近3轮，或者总结一下最近的对话）。

生活中的例子：
你跟朋友聊天：

• 你：“我昨天去看了《流浪地球2》，太好看了！”
• 朋友：“是吗？讲的什么内容？”
• 你：“讲的是人类为了躲避太阳氦闪，带着地球一起流浪的故事……”
  在这个过程中，你的短期记忆记住了刚才的对话，所以你能接着朋友的问题回答——这就是短期记忆和上下文管理的合作！

---

概念核心属性维度对比

为了让你更清楚地看到三个概念的区别，我们做了一个对比表格：

对比维度	短期记忆（STM）	长期记忆（LTM）	上下文管理
核心比喻	便签纸	文件柜	当前笔记本页面
容量	小（比如1000token）	大（比如无限）	中等（受上下文窗口限制，比如8k token）
保存时间	短（对话结束或覆盖）	长（永久）	临时（仅当前交互）
读写速度	快	慢	中等
主要作用	保持当前交互连贯性	存储重要信息和知识	整合记忆，生成有效prompt
存储内容	最近的对话、临时信息	用户喜好、知识、对话总结	系统提示词+短期记忆+检索到的长期记忆
典型实现方式	队列（FIFO）、滑动窗口	向量数据库、关系型数据库	Prompt压缩、RAG

---

概念联系的ER实体关系图

我们用Mermaid画一个ER（实体-关系）图，更直观地展示三个核心概念和AI Agent、用户交互之间的关系：

这个ER图告诉我们：

1. 一个AI Agent有一个短期记忆、一个长期记忆，使用一个上下文管理器；
2. 短期记忆存储用户交互，长期记忆归档用户交互、存储知识；
3. 上下文管理器从短期记忆读取内容，从长期记忆检索内容，然后喂给大语言模型（LLM）；
4. 用户交互触发AI Agent的工作。

---

交互关系图（Mermaid流程图）

接下来，我们用Mermaid画一个交互流程图，展示AI Agent的记忆机制是怎么工作的——从用户输入到AI输出的完整过程！

这个流程图展示了AI Agent处理用户输入的完整步骤：

1. 用户输入问题，先存入短期记忆（如果满了就删掉最早的）；
2. 上下文管理器启动，把短期记忆的最近对话、检索到的长期记忆、系统提示词组合成prompt；
3. 如果prompt太长，就压缩一下；
4. 把prompt喂给LLM，生成回答；
5. 把回答存入短期记忆，如果需要的话，总结一下存入长期记忆；
6. 最后把回答返回给用户。

是不是很清晰？接下来我们来看看每个步骤的核心算法原理！

---

核心算法原理 & 具体操作步骤

现在我们来“拆解”AI Agent记忆机制的核心算法——我们会用Python代码来演示每个算法的实现，让你能动手试试！

算法一：短期记忆的实现——FIFO队列与滑动窗口

短期记忆的核心需求是“存储最近的信息，容量有限，旧的信息会被覆盖”——最适合的实现方式是FIFO队列（先进先出队列）或者滑动窗口。

什么是FIFO队列？

FIFO（First In First Out）队列就像排队买奶茶：

• 先到的人先买，买完就走；
• 如果队伍太长（超过容量），最早来的人就会被“请出”队伍。

什么是滑动窗口？

滑动窗口就像一个固定大小的“窗口”，在对话历史上“滑动”：

• 只保留窗口里的内容（比如最近5轮对话）；
• 新的内容进来，窗口就向右滑动，旧的内容就被移出窗口。

Python代码实现：FIFO队列版本的短期记忆

我们用Python的collections.deque（双端队列）来实现FIFO队列——deque的popleft()方法可以快速删除最早的元素，时间复杂度是O(1)。

首先，我们定义一个ShortTermMemory类：

from collections import deque
from typing import List, Dict

class ShortTermMemory:
    def __init__(self, max_tokens: int = 1000, max_turns: int = 10):
        """
        初始化短期记忆
        :param max_tokens: 最大token数（可选，用于更精确的容量控制）
        :param max_turns: 最大对话轮数（比如最近10轮）
        """
        self.max_turns = max_turns
        self.max_tokens = max_tokens
        # 用deque存储对话历史，每个元素是一个字典：{"role": "user", "content": "..."}
        self.memory: deque = deque(maxlen=max_turns)
        # 当前总token数（简化版，实际项目中可以用LLM的tokenizer计算）
        self.current_tokens: int = 0

    def add(self, role: str, content: str) -> None:
        """
        添加一条对话到短期记忆
        :param role: 角色（"user"或"assistant"）
        :param content: 内容
        """
        # 简化版token计算：假设每个中文字符是1个token，每个英文单词是1个token
        # 实际项目中应该用LLM的tokenizer，比如tiktoken（OpenAI的tokenizer）
        tokens = self._count_tokens(content)
        
        # 如果添加后超过max_tokens，就删除最早的对话，直到有足够空间
        while self.current_tokens + tokens > self.max_tokens and len(self.memory) > 0:
            removed = self.memory.popleft()
            self.current_tokens -= self._count_tokens(removed["content"])
        
        # 添加新的对话
        self.memory.append({"role": role, "content": content})
        self.current_tokens += tokens

    def get(self) -> List[Dict[str, str]]:
        """
        获取短期记忆中的所有对话
        :return: 对话列表
        """
        return list(self.memory)

    def clear(self) -> None:
        """
        清空短期记忆
        """
        self.memory.clear()
        self.current_tokens = 0

    def _count_tokens(self, content: str) -> int:
        """
        简化版token计数函数
        :param content: 文本内容
        :return: token数
        """
        # 中文字符：每个字算1个token
        chinese_chars = len([c for c in content if '\u4e00' <= c <= '\u9fff'])
        # 英文单词：按空格分割，每个单词算1个token
        english_words = len(content.split())
        # 总token数（简化处理）
        return chinese_chars + english_words

好，我们来测试一下这个短期记忆类：

# 初始化短期记忆：最多5轮对话，最多50个token
stm = ShortTermMemory(max_turns=5, max_tokens=50)

# 添加几轮对话
stm.add("user", "我喜欢喝咖啡")
stm.add("assistant", "好的，我记住了")
stm.add("user", "今天天气真热")
stm.add("assistant", "是啊，要不要来杯冰的？")
stm.add("user", "好的，帮我订一杯冰美式")
stm.add("assistant", "没问题，马上为你订")  # 这是第6轮，会把第1轮挤掉

# 打印短期记忆
print("短期记忆内容：")
for turn in stm.get():
    print(f"{turn['role']}: {turn['content']}")

# 输出：
# 短期记忆内容：
# assistant: 好的，我记住了
# user: 今天天气真热
# assistant: 是啊，要不要来杯冰的？
# user: 好的，帮我订一杯冰美式
# assistant: 没问题，马上为你订

你看，第6轮对话加进去后，第1轮（“我喜欢喝咖啡”）就被挤掉了——这就是FIFO队列的作用！

---

算法二：长期记忆的实现——向量数据库与嵌入检索

长期记忆的核心需求是“存储大量信息，能快速检索相关内容”——最适合的实现方式是向量数据库（Vector Database）+ 嵌入（Embedding）+ 相似度检索。

什么是嵌入（Embedding）？

嵌入就是把“人类能看懂的文字”转换成“计算机能看懂的数字向量”——比如把“我喜欢喝咖啡”转换成[0.12, 0.56, -0.34, 0.78, …]这样的一串数字。

为什么要这么做？因为计算机不会“理解”文字的含义，但它会“计算”数字的相似程度——两个向量越相似，对应的文字含义就越接近！

什么是余弦相似度（Cosine Similarity）？

余弦相似度是计算两个向量相似程度的常用方法——它的取值范围是[-1, 1]：

• 1：两个向量完全相同（含义完全一样）；
• 0：两个向量完全不相关（含义完全不一样）；
• -1：两个向量完全相反（含义完全相反）。

余弦相似度的公式是：
$$\text{cosine similarity}(A,B)=\frac{A\cdot B}{\Vert A\Vert \times \Vert B\Vert }$$
其中：

• $A\cdot B$ 是向量A和向量B的点积；
• $\Vert A\Vert$ 是向量A的模长（即向量的长度）；
• $\Vert B\Vert$ 是向量B的模长。

什么是向量数据库？

向量数据库是专门用来存储向量和检索相似向量的数据库——它能在几百万、几千万条向量中，快速找到最相似的Top K条向量（比如Top 5）。

常用的向量数据库有：

1. Chroma：开源、轻量级，适合本地开发和小项目；
2. FAISS：Facebook开源的向量检索库，速度极快，适合大规模数据；
3. Pinecone：云服务，不需要自己搭建，适合生产环境；
4. Weaviate：开源，支持多种数据类型，适合知识图谱。

今天我们用Chroma来实现长期记忆——因为它开源、轻量级，安装和使用都很简单！

Python代码实现：基于Chroma的长期记忆

首先，我们需要安装Chroma和OpenAI的SDK（用来生成Embedding，你也可以用其他Embedding模型，比如Sentence-Transformers）：

pip install chromadb openai python-dotenv

然后，我们定义一个LongTermMemory类：

import chromadb
from chromadb.utils import embedding_functions
from typing import List, Dict
from dotenv import load_dotenv
import os

# 加载环境变量（需要在.env文件里放你的OPENAI_API_KEY）
load_dotenv()

class LongTermMemory:
    def __init__(self, collection_name: str = "ai_agent_ltm", persist_directory: str = "./chroma_db"):
        """
        初始化长期记忆
        :param collection_name: Chroma集合的名称（类似数据库的表）
        :param persist_directory: Chroma数据持久化的目录（存储在本地）
        """
        # 初始化Chroma客户端，数据持久化到本地
        self.client = chromadb.PersistentClient(path=persist_directory)
        
        # 初始化Embedding函数：用OpenAI的text-embedding-3-small模型
        # 你也可以用Sentence-Transformers的模型（免费，不需要API Key）
        self.embedding_function = embedding_functions.OpenAIEmbeddingFunction(
            api_key=os.getenv("OPENAI_API_KEY"),
            model_name="text-embedding-3-small"
        )
        
        # 获取或创建Chroma集合
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            embedding_function=self.embedding_function,
            metadata={"description": "AI Agent的长期记忆"}
        )

    def add(self, content: str, metadata: Dict = None, id: str = None) -> None:
        """
        添加一条信息到长期记忆
        :param content: 信息内容（文本）
        :param metadata: 元数据（比如来源、时间戳等，可选）
        :param id: 信息的唯一ID（可选，Chroma会自动生成）
        """
        # 如果没有提供metadata，就用空字典
        if metadata is None:
            metadata = {}
        # 添加时间戳到metadata
        from datetime import datetime
        metadata["timestamp"] = datetime.now().isoformat()
        
        # 添加到Chroma集合
        self.collection.add(
            documents=[content],
            metadatas=[metadata],
            ids=[id] if id else None
        )

    def retrieve(self, query: str, top_k: int = 5) -> List[Dict]:
        """
        从长期记忆中检索最相关的Top K条信息
        :param query: 查询文本（比如用户的问题）
        :param top_k: 返回的结果数量
        :return: 相关信息列表，每个元素包含content、metadata、distance
        """
        # 检索
        results = self.collection.query(
            query_texts=[query],
            n_results=top_k
        )
        
        # 整理结果
        retrieved = []
        for i in range(len(results["documents"][0])):
            retrieved.append({
                "content": results["documents"][0][i],
                "metadata": results["metadatas"][0][i],
                "distance": results["distances"][0][i]  # 距离越小，越相似
            })
        return retrieved

    def delete(self, id: str) -> None:
        """
        删除长期记忆中的一条信息
        :param id: 信息的唯一ID
        """
        self.collection.delete(ids=[id])

    def clear(self) -> None:
        """
        清空长期记忆（删除整个集合）
        """
        self.client.delete_collection(self.collection.name)
        # 重新创建集合
        self.collection = self.client.get_or_create_collection(
            name=self.collection.name,
            embedding_function=self.embedding_function,
            metadata={"description": "AI Agent的长期记忆"}
        )

接下来，我们需要创建一个.env文件，放你的OpenAI API Key：

OPENAI_API_KEY=你的OpenAI API Key

然后，我们来测试一下这个长期记忆类：

# 初始化长期记忆
ltm = LongTermMemory()

# 添加几条信息到长期记忆
ltm.add(
    content="用户喜欢喝冰美式，不加糖不加奶",
    metadata={"source": "对话历史", "user_id": "user_001"}
)
ltm.add(
    content="用户的生日是1990年1月1日",
    metadata={"source": "用户档案", "user_id": "user_001"}
)
ltm.add(
    content="用户住在北京市朝阳区",
    metadata={"source": "对话历史", "user_id": "user_001"}
)
ltm.add(
    content="用户喜欢看科幻电影，比如《流浪地球》系列",
    metadata={"source": "对话历史", "user_id": "user_001"}
)

# 测试检索：问“用户喜欢喝什么？”
print("检索结果（用户喜欢喝什么？）：")
results = ltm.retrieve(query="用户喜欢喝什么？", top_k=2)
for result in results:
    print(f"内容：{result['content']}")
    print(f"元数据：{result['metadata']}")
    print(f"距离：{result['distance']:.4f}")
    print("---")

# 输出：
# 检索结果（用户喜欢喝什么？）：
# 内容：用户喜欢喝冰美式，不加糖不加奶
# 元数据：{'source': '对话历史', 'user_id': 'user_001', 'timestamp': '2024-05-20T12:34:56.789012'}
# 距离：0.1234
# ---
# 内容：用户喜欢看科幻电影，比如《流浪地球》系列
# 元数据：{'source': '对话历史', 'user_id': 'user_001', 'timestamp': '2024-05-20T12:34:56.789012'}
# 距离：0.5678
# ---

你看，第一条结果是最相关的（距离最小）——这就是向量检索的作用！

---

算法三：上下文管理的实现——Prompt组合与压缩

上下文管理的核心需求是“把短期记忆、检索到的长期记忆、系统提示词组合成一个有效的prompt，并且不超过LLM的上下文窗口限制”。

什么是Prompt组合？

Prompt组合就是把几个部分的内容“拼”在一起，形成一个完整的prompt——典型的结构是：

系统提示词（System Prompt）：你是一个什么角色，你要做什么
----------
检索到的长期记忆（Retrieved Long-Term Memory）：相关的知识和信息
----------
短期记忆（Short-Term Memory）：最近的对话历史
----------
用户当前的问题（User's Current Question）：用户现在问的问题

什么是Prompt压缩？

如果组合后的prompt超过了LLM的上下文窗口限制（比如GPT-3.5-turbo是4k token，GPT-4是8k/32k token），我们就需要“压缩”prompt——常见的压缩方法有：

1. 减少短期记忆的轮数：比如从最近10轮改成最近5轮；
2. 精简检索到的长期记忆：比如只保留最相关的Top 3条，而不是Top 5；
3. 总结短期记忆：用LLM把最近的对话总结成一段简短的文字，而不是保留完整的对话；
4. 截断内容：把太长的内容截断（但要注意不要截断关键信息）。

Python代码实现：上下文管理器

我们定义一个ContextManager类：

from typing import List, Dict
from .short_term_memory import ShortTermMemory
from .long_term_memory import LongTermMemory

class ContextManager:
    def __init__(
        self,
        system_prompt: str,
        llm_max_context_tokens: int = 4096,
        short_term_memory: ShortTermMemory = None,
        long_term_memory: LongTermMemory = None
    ):
        """
        初始化上下文管理器
        :param system_prompt: 系统提示词
        :param llm_max_context_tokens: LLM的最大上下文窗口token数
        :param short_term_memory: 短期记忆对象
        :param long_term_memory: 长期记忆对象
        """
        self.system_prompt = system_prompt
        self.llm_max_context_tokens = llm_max_context_tokens
        self.stm = short_term_memory
        self.ltm = long_term_memory

    def build_prompt(self, user_query: str, top_k_ltm: int = 5) -> List[Dict[str, str]]:
        """
        构建完整的prompt（OpenAI的Chat Completions格式）
        :param user_query: 用户当前的问题
        :param top_k_ltm: 从长期记忆中检索的Top K条结果
        :return: OpenAI格式的prompt列表
        """
        # 第一步：先从长期记忆中检索相关内容
        retrieved_ltm = []
        if self.ltm:
            retrieved_ltm = self.ltm.retrieve(user_query, top_k=top_k_ltm)
        
        # 第二步：获取短期记忆
        stm_history = []
        if self.stm:
            stm_history = self.stm.get()
        
        # 第三步：构建初始的prompt
        prompt = []
        
        # 3.1 添加系统提示词
        prompt.append({"role": "system", "content": self.system_prompt})
        
        # 3.2 添加检索到的长期记忆（用assistant的角色，或者system的角色）
        if retrieved_ltm:
            ltm_content = "以下是相关的历史信息和知识：\n"
            for i, item in enumerate(retrieved_ltm, 1):
                ltm_content += f"{i}. {item['content']}\n"
            prompt.append({"role": "system", "content": ltm_content})
        
        # 3.3 添加短期记忆（对话历史）
        prompt.extend(stm_history)
        
        # 3.4 添加用户当前的问题
        prompt.append({"role": "user", "content": user_query})
        
        # 第四步：计算prompt的token数，如果超过限制就压缩
        prompt_tokens = self._count_prompt_tokens(prompt)
        while prompt_tokens > self.llm_max_context_tokens and len(stm_history) > 0:
            # 压缩策略：删除短期记忆中最早的一轮对话
            stm_history.pop(0)
            # 重新构建prompt
            prompt = [{"role": "system", "content": self.system_prompt}]
            if retrieved_ltm:
                prompt.append({"role": "system", "content": ltm_content})
            prompt.extend(stm_history)
            prompt.append({"role": "user", "content": user_query})
            # 重新计算token数
            prompt_tokens = self._count_prompt_tokens(prompt)
        
        # 如果还是超过限制，就减少长期记忆的检索数量
        while prompt_tokens > self.llm_max_context_tokens and len(retrieved_ltm) > 1:
            retrieved_ltm.pop()
            ltm_content = "以下是相关的历史信息和知识：\n"
            for i, item in enumerate(retrieved_ltm, 1):
                ltm_content += f"{i}. {item['content']}\n"
            prompt = [{"role": "system", "content": self.system_prompt}]
            prompt.append({"role": "system", "content": ltm_content})
            prompt.extend(stm_history)
            prompt.append({"role": "user", "content": user_query})
            prompt_tokens = self._count_prompt_tokens(prompt)
        
        return prompt

    def _count_prompt_tokens(self, prompt: List[Dict[str, str]]) -> int:
        """
        简化版prompt token计数函数
        :param prompt: OpenAI格式的prompt列表
        :return: token数
        """
        total = 0
        for message in prompt:
            # 简化版：每个消息的role和content都算token
            total += self._count_tokens(message["role"])
            total += self._count_tokens(message["content"])
            # 每个消息之间的分隔符也算几个token（简化处理）
            total += 3
        return total

    def _count_tokens(self, content: str) -> int:
        """
        简化版token计数函数（和ShortTermMemory里的一样）
        """
        chinese_chars = len([c for c in content if '\u4e00' <= c <= '\u9fff'])
        english_words = len(content.split())
        return chinese_chars + english_words

好，我们来测试一下这个上下文管理器：

from short_term_memory import ShortTermMemory
from long_term_memory import LongTermMemory
from context_manager import ContextManager

# 初始化三个模块
stm = ShortTermMemory(max_turns=5, max_tokens=500)
ltm = LongTermMemory()
system_prompt = "你是一个贴心的私人AI助手，名叫小月。你会根据用户的历史信息和当前问题，给出友好、准确的回答。"
context_manager = ContextManager(
    system_prompt=system_prompt,
    llm_max_context_tokens=4096,
    short_term_memory=stm,
    long_term_memory=ltm
)

# 先往长期记忆里加点信息
ltm.add("用户喜欢喝冰美式，不加糖不加奶")
ltm.add("用户的生日是1990年1月1日")

# 往短期记忆里加点对话
stm.add("user", "你好，我是小明")
stm.add("assistant", "你好小明，我是小月，很高兴为你服务！")

# 构建prompt
user_query = "你知道我喜欢喝什么吗？"
prompt = context_manager.build_prompt(user_query=user_query, top_k_ltm=3)

# 打印prompt
print("构建的Prompt：")
for message in prompt:
    print(f"{message['role']}: {message['content']}")
    print("---")

输出的prompt应该是这样的：

构建的Prompt：
system: 你是一个贴心的私人AI助手，名叫小月。你会根据用户的历史信息和当前问题，给出友好、准确的回答。
---
system: