007、记忆模块（一）：短期记忆与会话上下文管理

上周调试一个对话型Agent用户连续问了三个问题“当前温度多少”“湿度呢”“把空调调到26度”结果Agent在第三个请求直接报错“未找到温度参数”。明明第一次对话已经获取了温度值为什么第三次请求时“失忆”了这就是短期记忆管理没做好——今天咱们就拆解这个最基础又最容易出问题的模块。会话上下文是什么简单说就是对话过程中产生的临时数据池。比如你问“今天天气如何”Agent调用天气API返回“28度晴”这个结果不仅要回复给你还得暂存起来因为接下来你很可能问“那需要带伞吗”。如果没有这个暂存区每次对话都是孤立事件Agent就变成了金鱼——只有7秒记忆。实际项目中这个“暂存区”通常是个数据结构classSessionContext:def__init__(self,session_id):self.session_idsession_id self.messages[]# 存放对话历史self.temp_data{}# 存放临时变量self.created_attime.time()self.last_activetime.time()注意那个last_active字段——很多新手忘了加结果内存里堆满僵尸会话服务器重启才清理。短期记忆的典型实现看个真实案例我们团队早期版本这样设计# 反面教材全局列表存所有会话all_sessions[]# 这里踩过大坑内存泄漏预警defhandle_request(user_id,query):sessionfind_session(user_id)session.messages.append({role:user,content:query})# 业务逻辑处理responseprocess(query,session.messages)session.messages.append({role:assistant,content:response})returnresponse问题在哪all_sessions只增不减部署三天后内存爆了。更糟的是没有消息数量限制某个话痨用户传了1000条历史记录导致后续API调用超时。改进后的方案fromcollectionsimportOrderedDictimporttimeclassSessionManager:def__init__(self,max_sessions10000,ttl1800):self.sessionsOrderedDict()# 用OrderedDict实现LRUself.max_sessionsmax_sessions self.ttlttl# 30分钟无活动自动清理defget_session(self,session_id):# 清理过期会话每次访问时触发避免单独起线程self._cleanup()ifsession_idnotinself.sessions:iflen(self.sessions)self.max_sessions:# LRU淘汰删除最久未使用的self.sessions.popitem(lastFalse)self.sessions[session_id]{messages:[],created:time.time(),updated:time.time()}sessionself.sessions[session_id]session[updated]time.time()# 刷新活跃时间self.sessions.move_to_end(session_id)# 移到最新位置returnsessiondef_cleanup(self):nowtime.time()expired[]forsid,datainself.sessions.items():ifnow-data[updated]self.ttl:expired.append(sid)else:break# OrderedDict按时间排序遇到未过期就可停止forsidinexpired:delself.sessions[sid]这个实现有几个关键点用OrderedDict实现LRU缓存淘汰机制惰性清理访问时触发避免独立清理线程的复杂度双时间戳created/updated方便监控和调试上下文窗口的管理技巧现在大模型都有限制上下文长度比如4K、16K tokens不能无脑存全部历史。常见策略是滑动窗口deftrim_context(messages,max_tokens4000):保持总token数不超过限制优先保留最近对话total_tokenscalculate_tokens(messages)whiletotal_tokensmax_tokensandlen(messages)1:# 从最旧的对话开始删但永远保留最后一条系统提示removedmessages.pop(1)iflen(messages)2elsemessages.pop(0)total_tokens-calculate_tokens([removed])returnmessages注意那个pop(1)而不是pop(0)——索引0通常是系统指令比如“你是个助手…”这个要尽量保留。我们吃过亏把系统提示删了结果Agent开始用奇怪的人格回答问题。实际调试中的坑序列化问题把会话存Redis时直接pickle.dumps()结果版本升级后反序列化失败。现在都用JSON存基本类型复杂对象另做处理。时间戳同步多实例部署时如果服务器时间不同步TTL机制会混乱。要么用Redis统一时间要么用逻辑时间戳自增ID。上下文污染用户问“忘记刚才说的重新开始”这时候要能清空会话记忆。我们现在的做法是加个特殊指令检测ifquery.strip()/clear:session[messages][session[messages][0]]# 只留系统提示return会话已重置个人经验建议短期记忆模块别追求完美——它本质上就是个带缓存的键值存储。前期用最简单的字典超时机制就行等用户量上来再考虑Redis分布式存储。关键指标要监控会话数量、内存占用、平均会话长度。我们设置过报警如果平均会话长度突然从10条跳到50条很可能是有用户在用脚本刷接口。实际项目中80%的“记忆问题”不是算法缺陷而是基础机制没做好没设会话上限、没清理过期数据、没处理异常中断。先把这些基础打牢再考虑向量数据库那些高级功能。下次我们聊长期记忆——怎么让Agent记住“用户喜欢喝美式咖啡”这种跨会话信息。你会发现有了可靠的短期记忆模块做基础长期记忆的实现会简单很多。