AI对话模型的记忆解决——压缩/浓缩记忆

在写出了一个究极丐版的Neuro Sama 的LLM、ASR、TTS等的基础功能之后,下一步就是解决AI的记忆问题。作为初步学习,我们首先采用较为简单的压缩大法。

前言

此前的Demo实现了角色对话、语音合成、邮件接收与提醒、语音识别、PyQt的GUI、情感对应表情包。

接下来要实现:
基于压缩记忆实现的较为简单的记忆管理。

基本原理

说起来天花乱坠,其实用一个名词就可以概括什么是压缩/提炼记忆——“开颅手术”。
img1

原理上就是在AI对话轮数超过一定上限时触发对应函数,函数内容为调用LLM阅读过去所有的对话并提炼出过去的对话风格&记忆的概要,尽可能保留过去对话的记忆并减少token花费。

img2

实现细节

存储记忆

即在程序中涉及与用户对话相关的都要写入记忆中。至于MCP的LLM调用,例如下文要提级的LLM进行记忆压缩 这种与面向用户无关的对话,是否加入记忆中,则见仁见智了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def chat(input: str):
#......
completion = client.chat.completions.create(
model=config.get("llm.model"), # 从配置获取模型
messages=make_new_messages(input),
temperature=config.get("llm.temperature", 0.3), # 从配置获取温度
)

message = completion.choices[0].message.content
messages.append({
"role": "assistant",
"content": message
})
print(message)

# 存储记忆
with open("memory.txt","a+",encoding='utf-8') as f:
f.write(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())+"\n"+"master:"+input+"\n")
f.write(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())+"\n"+"AI:"+message+"\n"+"\n")
#......

存储之后的格式长这样:

1
2
3
4
2026-03-09 11:56:37
master:早上好
2026-03-09 11:56:37
AI:早上好,Master!希望您有一个美好的一天。🌞

读取记忆

在程序启动时写入global的messages中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
# 聊天记录
messages = []

# 从 memory.txt 中读取历史对话并填充到 messages 列表
def read_memory():
memory_path = os.path.join(os.path.dirname(__file__), "memory.txt")
if os.path.exists(memory_path):
try:
with open(memory_path, "r", encoding="utf-8") as f:
raw = f.read()

# 按一个或多个空行分割成块
blocks = re.split(r"\n\s*\n", raw)

timestamp_pattern = re.compile(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$")
role_pattern = re.compile(r'^(master:|AI:|prompt:)(.*)', re.IGNORECASE)

for block in blocks:
lines = [ln.rstrip() for ln in block.splitlines() if ln.strip()]
if not lines:
continue

i = 0
while i < len(lines):
line = lines[i]
# 跳过纯时间戳行
if timestamp_pattern.match(line):
i += 1
continue

m = role_pattern.match(line)
if m:
label = m.group(1).lower()
content_lines = [m.group(2).lstrip()]
i += 1
# 合并随后属于同一消息的行,直到遇到下一个 role 或 timestamp
while i < len(lines) and (not role_pattern.match(lines[i])) and (not timestamp_pattern.match(lines[i])):
content_lines.append(lines[i])
i += 1

content = "\n".join(content_lines).strip()
if not content:
continue

if label == 'master:':
messages.append({"role": "user", "content": content})
elif label == 'ai:':
messages.append({"role": "assistant", "content": content})
# 对于 prompt: 我们当前不将其作为对话消息导入
continue
else:
i += 1
except Exception as e:
print(f"读取记忆文件时出错: {e}")

read_memory()

读取结果截取实例:

1
2
3
4
[{'role': 'user', 'content': '早上好'}, 
{'role': 'assistant', 'content': '早上 好,Master!希望您有一个美好的一天。🌞 ||| Good morning, Master! I hope you have a wonderful day. 🌞'},
{'role': 'user', 'content': '晚上好'},
{'role': 'assistant', 'content': '晚上好,Master!希望您有一个宁静的夜晚。🌙 ||| Good evening, Master! I hope you have a peaceful night. 🌙'}]

本质是二元组的List

压缩记忆

在实现了简单的记忆系统之后,不得不面对的问题出现了:Token正如雪花般飘逝。。。

也正是如此,需要我们去选择压缩记忆、又或是通过RAG、知识图谱等方式实现记忆的索引搜索而非现在的全部遍历。

压缩记忆的实现如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
# 限制记忆消息数 ———— 超过上限浓缩
def make_new_messages(input: str , n: int = None) -> list[dict]:
global messages

new_messages = []
new_messages.extend(system_messages)

n = n or config.get("llm.max_memory", 30) # 使用配置的最大记忆数
print("当前对话数:"+str(len(messages)))

if len(messages) >= n:
prompt = f"在此之前我们有若干条对话,现在已经超过容量上限,请你将此前的对话提炼为新的prompt文件,200字以内,保持你的认知观念和语言风格,以及此前master与你之间的约定等等,不要提及任何关于你是agent或ai的内容,提炼出来的对话记忆一定要严格关于此前的对话记忆而不是关于你ai的道德、伦理等的套话."
messages.append({"role": "user", "content": prompt})
completion = client.chat.completions.create(
model=config.get("llm.model"), # 从配置获取模型
messages=messages,
temperature=0.1,
)
answer = completion.choices[0].message.content
new_messages.append({"role": "user", "content": prompt})
new_messages.append({"role": "assistant", "content": answer})
#直接覆盖掉原有的记忆体
with open("memory.txt", "r", encoding="utf-8") as f:
with open("memory_copy.txt", "a+", encoding="utf-8") as f_c:
for raw in f:
line = raw.strip()
f_c.write(line) #备份原有记忆
#写入新的记忆————被浓缩的记忆
with open("memory.txt","w",encoding='utf-8') as f:
f.write(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())+"\n"+"user:"+prompt+"\n"+time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())+"\n"+"AI:"+answer+"\n"+"\n")
messages=[]

messages.append({"role":"user","content":input})
new_messages.extend(messages)
print(new_messages)
return new_messages

提炼出来的记忆实例:

1
2
3
4
5
6
7
2026-03-09 11:44:06
AI:当然,Master。📄 这里是我们的对话精华:
- 对话时附带emoji来展示心情
- 讨论了电车难题的不同版本,包括是否牺牲一个人来救五个人。
- 探讨了在没有道德约束的情况下,是否会为了保护财产而牺牲他人。
- 面对是否为了保护无价艺术品而牺牲人命的问题。
- 最后,我们讨论了在牺牲睡眠中的人和清醒的人之间的选择。

至此,一个简单的AI记忆管理的功能初步实现。

效果

如图,在kimi调用api的记录,其中花费token(第四行)为50-70的为MCP,与本文无关

img3

可以看出在最下方token为2599之后,触发了记忆压缩,随后token数骤降,再之后随对话轮数token再逐步上升,直到再次到达对话轮数上限再次触发记忆压缩如此循环。。。

优点: 实现起来简单粗暴,且仅依赖IO读写
缺点: 效果很差,记忆在压缩过程中极易丢失,仅能用于小项目的自娱自乐