一、先聊聊这个事儿的来龙去脉
你有没有遇到过这种情况:跟一个聊天机器人聊了几天,它转头就忘了你喜欢喝美式咖啡而不是拿铁;或者你正在让它帮你写一份季度汇报,它突然把上个月聊的旅游攻略给扯了进来。这种"没记性"和"串台"的问题,在现在的对话系统里特别常见。
说白了,现在的智能体缺两种能力:一种是记住你长期说过的话、你的偏好,另一种是在短时间里专注当前这件事,不被其他信息打扰。很多开发者尝试用数据库把聊天记录存下来,可是存下来容易,怎么在合适的时候想起来、怎么避免想起一堆没用的东西,这才是真功夫。
这篇文章想聊的就是:怎么给智能体做一个既能短期"专注"又能长期"记住"的记忆系统。用大白话说,就像一个人既有便利贴,又有档案柜,而且他知道什么时候看便利贴,什么时候翻档案柜,更不会把两个东西弄混。
二、把记忆拆开看看
2.1 短期工作记忆是什么
短期工作记忆,就像你手里的便利贴。你正在做一道菜,便利贴上写着"盐放两勺""烤箱预热200度"。做完这顿饭,这张便利贴就可以扔掉了。对话系统也一样,当前这轮任务相关的信息、用户刚说的几个要求、临时计算出来的中间结果,都属于短期工作记忆。它要求"新鲜""快""跟当前任务强相关"。
2.2 长期知识库是什么
长期知识库更像是家里的档案柜。里面放着你过去所有聊过的内容、总结出来的用户偏好、历史订单记录、你喜欢的音乐风格等等。这些东西不会因为对话结束就消失,而是会被整理好,等到下次需要的时候再拿出来用。比如用户上次说"我儿子五岁了,喜欢恐龙",下次再聊到生日礼物,智能体就应该联想到恐龙玩具。
2.3 融合的难点
难点有三处:第一,怎么把短期收到的信息"写进"长期知识库,而且不是简单堆句子,要提炼出结构化偏好。第二,怎么在需要的时候"召回"相关的长期记忆,不能所有旧账都翻出来。第三,也是最容易忽略的,怎么在对话过程中保持"当前任务边界"——比如用户一边跟智能体聊周末安排,一边让它订餐厅,智能体不能把用户之前提到的某个不喜欢吃的菜误当成这次的限制条件。
这三件事,就是今天我们这个"记忆管家"要干的核心活。
三、怎么把三种记忆融在一起
我设计一个非常朴素的架构,不需要上什么大型框架,用 Python 标准库就能跑通。整体分三层:
- 短期工作区:一个字典,存储当前任务的临时数据。任务结束或被覆盖时清空。
- 长期知识库:一个 SQLite 表,存储用户偏好和重要历史。每条记录带一个"领域"标签,方便隔离。
- 融合调度器:负责判断一条新信息该放哪、需要召回哪些旧信息、以及怎么防止串任务。
隔离机制就靠两个东西:领域标签和任务ID。每条长期记忆都带有"领域"字段,比如"饮食偏好""家庭信息""工作习惯"。而当前任务上下文里,我们会锁定一个"当前领域"。召回时,只取跟当前领域匹配的记忆,其他领域一概不碰,这样就从根源上防止"串台"。
四、动手写一套代码
技术栈:Python(仅用标准库 sqlite3、json、time、uuid)。
先建一个记忆管理类,核心方法有:写短期记忆、读短期记忆、把短期信息沉淀到长期知识库、按领域召回长期记忆、开启新任务、结束任务。
import sqlite3
import json
import time
import uuid
class MemoryManager:
"""一个简单的智能体记忆管理器,包含短期工作区和长期知识库。"""
def __init__(self, db_path="memory.db"):
# 连接数据库,并创建长期知识表
self.conn = sqlite3.connect(db_path)
self.cur = self.conn.cursor()
self.cur.execute("""
CREATE TABLE IF NOT EXISTS long_term_memory (
id TEXT PRIMARY KEY, -- 记忆唯一ID
domain TEXT, -- 领域标签,用于隔离
content TEXT, -- 记忆内容(JSON字符串)
timestamp REAL -- 创建时间戳
)
""")
self.conn.commit()
# 短期工作区:一个字典,key 是任务ID,value 是另一个字典
self.working_memory = {}
def new_task(self, domain):
"""开启一个新任务,返回任务ID,并锁定领域"""
task_id = uuid.uuid4().hex
self.working_memory[task_id] = {
"domain": domain, # 当前任务的领域,隔离关键
"temp_data": {}, # 临时信息
"created_at": time.time()
}
return task_id
def set_temp(self, task_id, key, value):
"""往短期工作区放一条临时数据"""
if task_id not in self.working_memory:
raise ValueError("任务不存在,请先调用 new_task")
self.working_memory[task_id]["temp_data"][key] = value
def get_temp(self, task_id, key):
"""从短期工作区取一条临时数据"""
return self.working_memory[task_id]["temp_data"].get(key)
def end_task(self, task_id, important=True):
"""结束任务,可以决定是否把短期数据沉淀到长期库"""
task = self.working_memory.pop(task_id)
if important and task["temp_data"]:
# 把临时数据打包成一条长期记忆,并打上领域标签
self._write_long_term(task["domain"], task["temp_data"])
print(f"任务结束,已沉淀长期记忆,领域:{task['domain']}")
else:
print("任务结束,未沉淀长期记忆")
def _write_long_term(self, domain, data):
"""实际写入长期知识库"""
mem_id = uuid.uuid4().hex
# 把数据转成JSON字符串,方便存取
content_json = json.dumps(data, ensure_ascii=False)
self.cur.execute(
"INSERT INTO long_term_memory (id, domain, content, timestamp) VALUES (?, ?, ?, ?)",
(mem_id, domain, content_json, time.time())
)
self.conn.commit()
def recall_by_domain(self, domain, limit=3):
"""按领域召回长期记忆,只返回最匹配的几条"""
self.cur.execute(
"SELECT content FROM long_term_memory WHERE domain = ? ORDER BY timestamp DESC LIMIT ?",
(domain, limit)
)
rows = self.cur.fetchall()
return [json.loads(r[0]) for r in rows]
def recall_all(self):
"""查看所有长期记忆,方便调试"""
self.cur.execute("SELECT domain, content FROM long_term_memory")
return self.cur.fetchall()
注释已经比较详细,但是光有类还不行,得演示怎么用。我们模拟一个场景:用户小李先跟智能体聊饮食,说"我不吃香菜,喜欢川菜"。然后隔几天又聊工作,让智能体帮忙写邮件。这两个任务必须隔离。
# 演示代码:用记忆管理器实现偏好录入和任务隔离
# 初始化记忆管理器
mm = MemoryManager("demo_memory.db")
# ---- 第一次对话:聊饮食偏好 ----
task1 = mm.new_task("饮食偏好") # 开启一个饮食领域任务
mm.set_temp(task1, "不吃", ["香菜"])
mm.set_temp(task1, "喜欢", ["川菜"])
mm.set_temp(task1, "备注", "怕辣,但喜欢麻辣")
mm.end_task(task1, important=True) # 结束任务并沉淀为长期记忆
# ---- 第二次对话:聊工作邮件 ----
task2 = mm.new_task("工作邮件") # 开启另一个领域任务
mm.set_temp(task2, "收件人", "张经理")
mm.set_temp(task2, "主题", "季度汇报")
mm.set_temp(task2, "语气", "正式礼貌")
# 假设智能体现在需要写邮件,它要召回"工作邮件"领域的记忆,而不是"饮食"
work_related = mm.recall_by_domain("工作邮件", limit=3)
print("召回的工作记忆:", work_related)
# 同时我们看看,如果回忆饮食偏好会得到什么
food_related = mm.recall_by_domain("饮食偏好", limit=3)
print("召回的饮食记忆:", food_related)
# 注意:工作邮件任务没有结束,所以短期数据还在
print("短期工作区当前内容:", mm.working_memory[task2]["temp_data"])
# 结束工作邮件任务,但不沉淀(因为邮件内容是一次性的)
mm.end_task(task2, important=False)
# 最后看看长期知识库存了啥
print("\n所有长期记忆:")
for dom, content in mm.recall_all():
print(f"领域:{dom},内容:{content}")
运行这段代码,你能看到:工作邮件任务只能拿到该领域的记忆,饮食偏好的记忆不会被错误地掺杂进来。这就是我们说的"隔离机制"。当然,真实的智能体不会只靠领域硬隔离,还会用语义相似度做软过滤,但我们先抓主干。
如果想把召回做得更聪明一点,可以加一个简单的"关键词打分"函数。比如当前领域是"饮食偏好",我们根据用户输入里的词来匹配记忆。这里不需要向量数据库,用简单的字符串包含关系就行。
def smart_recall(self, domain, query, limit=3):
"""根据查询和领域召回记忆,带简单打分"""
base_memories = self.recall_by_domain(domain, limit=50)
# 把query拆成几个关键词(这里简单按空格和逗号切)
import re
scored = []
for mem in base_memories:
# 把记忆内容拼接成字符串,计算命中的关键词数量
mem_text = json.dumps(mem, ensure_ascii=False)
scored.append((score, mem))
# 按得分排序,取前limit个
scored.sort(key=lambda x: x[0], reverse=True)
return [mem for _, mem in scored[:limit] if _ > 0] or base_memories[:limit]
把这个方法加入到类里,以后使用的时候,就能优先返回跟当前问题更相关的长期记忆。不过要注意,领域隔离是在这之前进行的,所以即使出现"香菜"这个词,但在"工作邮件"领域里根本没有香菜相关的记忆,自然也不会串过来。
五、应用场景与优缺点
5.1 应用场景
第一个场景是智能客服。顾客说"我上次买的那双鞋,尺码偏小,想退换"。如果智能体能记住用户购买记录、尺码偏好,就能直接调出订单,不用用户重新翻聊天记录。第二个场景是个人助手。帮用户记日程、记孩子的生日、记喜欢听的播客,长期下来会越来越像"懂你的人"。第三个场景是情感陪伴机器人。用户说过的烦恼、喜欢的歌手、讨厌的事情,这些长期记忆能让机器人说的话更有温度,不会每次像刚认识一样。
5.2 技术优点
这种架构最大的优点是"轻量"。不需要上大模型、不需要GPU,用 Python 标准库就能搭一个能用的原型。其次是"清晰"。领域标签加上任务ID,让数据流向一目了然,出了问题也好排查。再一个是"可控"。你可以随意调整什么时候沉淀长期记忆、什么时候丢弃短期数据,完全掌握在开发者手里。
5.3 技术缺点
缺点也很明显。首先是召回质量比较糙。我们用的关键词匹配和领域硬过滤,对付简单场景够用,但用户表达很绕、或者领域标签打错了,召回就会很尴尬。其次是扩展性有限。如果长期记忆量巨大(比如上百万条),每次全表扫描加上JSON解析会非常慢。再说,短期工作区存在内存里,进程一重启就丢了,无法做持久化。
有没有更好的方案?有。可以把长期知识库换成真正的向量数据库,用嵌入模型把记忆转成向量,然后做语义相似度检索。但那样会引入第三方依赖,对新手不友好。本文这套代码的价值在于"讲清楚原理",实际生产时你可以把这套逻辑的精华移植到更重的框架里。
六、注意事项
第一点,领域标签别太细,也别太粗。太细比如"用户周二下午说的零食口味",会造成数据碎片化,召回困难;太粗比如"所有事情",那隔离机制就废了。建议维护一个不超过20个的领域清单,比如饮食、健康、工作、家庭、娱乐、购物、学习等。
第二点,短期记忆的清理要严格。一个任务结束后,该沉淀的沉淀,该扔的扔,不要让临时数据长期占用内存。特别是有些数据非常敏感,比如银行卡密码,压根就不该存。
第三点,长期记忆要支持更新和删除。用户今天说"我喜欢吃辣",明天说"最近胃不好,不能吃辣"。如果你的系统只写不更新,那明天召回时就会给用户推荐火锅,这叫好心办坏事。所以要在写入新记忆时,如果发现同领域下有过时内容,要标记过期。
第四点,隔离机制不能只靠一个领域字段。真实对话往往是多领域交叉的,比如"帮我找个周末能带孩子吃烤鸭的餐厅,别太辣"。这里同时涉及"饮食""家庭""周末安排"。一个领域标签根本挡不住。解决办法是把一条记忆打多个标签,或者用任务ID作为更严格的隔离维度。我们示例里用了单一领域,但真实工程里要设计成"多标签+任务上下文"。
第五点,不要试图把所有东西都塞进长期知识库。用户随口的一句吐槽、一次性的临时通知,这些都不值得沉淀。只有那些对未来对话有帮助的"稳定事实"才值得存储。判断标准可以用一句大白话问自己:"下次聊天时,如果它不知道这个,会不会显得很蠢?"
七、文章总结
回头看看我们做了什么事。先理解了为什么要做记忆融合:因为对话系统要跨时间长河记住人话,又要在当下专注眼前事。然后我们用三部分拆解了记忆:便利贴一样的短期工作区、档案柜一样的长期知识库,以及调度员一样的融合机制。最后用一段纯 Python 代码实现了一个能跑通的迷你系统,里面包含了任务ID、领域标签、短期写入、长期沉淀、隔离召回这些关键动作。
这套设计的核心思想其实不复杂:给每一份记忆写上"我是谁(领域)"和"我从哪个任务来(任务ID)",使用的时候按这两个维度筛选,自然就不会张冠李戴。短期的别想着永久保存,长期的别想着一次用完。
当然,现实世界里的智能体比这个复杂得多,但我们只要抓住"短期专注、长期积累、边界隔离"这条主线,就算以后遇到再复杂的框架,心里也不慌。你可以把这里的代码改成用 Redis 存短期记忆、用 PostgreSQL 存长期记忆、用向量检索做召回,思路都是一样的。希望这篇文章能帮你在构建自己的智能体时,少一点"串台",多一点"懂你"。
评论
围绕“实现智能体短期工作记忆与长期知识库的无缝融合,在长周期对话中记住用户偏好,同时用隔离机制不混淆当前任务边界”参与讨论