AI Agent工程实战系列 · 第03篇 / 共10篇
四层记忆架构、滚动摘要、语义压缩——附完整实现
以及为什么"把所有对话塞进context window"是最危险的懒办法
从一个客服Agent的崩溃说起
我们有一个客服Agent，上线初期表现不错。用了两个月之后，用户开始投诉：
“我已经说了三遍我的订单号了，它还在问我。”
“它记得我上周说的事，但不记得我五分钟前刚说的话。”
“我跟它聊了半小时，聊到后面它开始胡说八道。”
三个投诉，三个完全不同的记忆问题。
第一个是短期记忆失效——对话轮次多了之后，早期内容被截断。第二个是记忆层次混乱——长期记忆和短期记忆没有分开管理。第三个是上下文窗口溢出——Token超限后模型开始产生幻觉。
这三个问题，根因都是同一个：我们没有设计记忆系统，只是在无脑地把所有对话追加进一个列表。
# 99%的教程示例，也是99%的生产事故根源messages=[]whileTrue:user_input=get_user_input()messages.append({"role":"user","content":user_input})response=llm.invoke(messages)# 第50轮时Token炸了messages.append({"role":"assistant","content":response})今天把记忆系统从零到生产的设计全部拆开。
为什么记忆对Agent比对普通对话Bot更关键
普通聊天Bot的记忆需求：记住当前对话就够了。
Agent的记忆需求复杂得多：
Agent需要记住的东西： ① 当前对话（短期） 用户刚说了什么，我刚做了什么 ② 任务执行状态（工作记忆） 当前任务进行到第几步了，哪些工具调用过了， 哪些中间结果需要传递给下一步 ③ 用户画像（长期） 这个用户的偏好、历史行为、之前反馈过的问题 ④ 知识库（语义记忆） 领域知识、工具使用说明、业务规则 不是每次对话带进来，而是按需检索 这四类记忆，生命周期、存储方式、检索方式完全不同。 用同一个列表管理，是根本性的架构错误。四层记忆架构
四层各司其职，Agent在每次推理时，从不同层取不同类型的信息组装进context。
Layer 1：短期记忆——滑动窗口 + 智能压缩
最朴素的实现：固定窗口截断
fromcollectionsimportdequefromtypingimportList,DictclassSlidingWindowMemory:""" 滑动窗口短期记忆 只保留最近N轮对话，超出则丢弃最早的 优点：简单，零成本 缺点：早期重要信息会被丢弃 """def__init__(self,max_turns:int=20):self.max_turns=max_turns# 每个元素是一轮对话（user + assistant各一条）self.turns:deque=deque(maxlen=max_turns)defadd_turn(self,user_msg:str,assistant_msg:str):self.turns.append({"user":user_msg,"assistant":assistant_msg})defget_messages(self)->List[Dict]:"""展开为LLM可用的messages格式"""messages=[]forturninself.turns:messages.append({"role":"user","content":turn["user"]})messages.append({"role":"assistant","content":turn["assistant"]})returnmessagesdeftoken_count(self)->int:"""粗略估算当前记忆的Token数"""total_chars=sum(len(t["user"])+len(t["assistant"])fortinself.turns)returntotal_chars//4# 粗略估算：4个字符≈1个Token固定窗口的问题：用户在第1轮说了"我的订单号是ORDER-001"，第25轮之后这条信息被丢弃了。但第26轮用户问"我那个订单发货了吗"，Agent已经不知道订单号是什么了。
更好的实现：基于Token预算的动态窗口
importtiktokenfromtypingimportList,Dict,Optional,TupleclassTokenBudgetMemory:""" 基于Token预算的短期记忆 不按轮次截断，而是按Token数截断 并且在截断前先尝试压缩，而不是直接丢弃 """def__init__(self,max_tokens:int=4000,compression_threshold:float=0.8,model_name:str="gpt-4"):self.max_tokens=max_tokens self.compression_threshold=compression_threshold self.messages:List[Dict]=[]try:self.encoder=tiktoken.encoding_for_model(model_name)exceptKeyError:self.encoder=tiktoken.get_encoding("cl100k_base")# 摘要缓存：被压缩的早期对话的摘要self.summary:Optional[str]=Nonedefcount_tokens(self,messages:List[Dict])->int:"""精确计算messages列表的Token数"""total=0formsginmessages:# 每条消息有固定的格式开销（约4 Token）total+=4total+=len(self.encoder.encode(msg.get("content","")))returntotaldefadd_message(self,role:str,content:str,llm=None):""" 添加一条消息，自动管理Token预算 当Token接近上限时： 1. 先尝试压缩早期消息（用LLM生成摘要） 2. 压缩后还超，则丢弃最早的消息 """self.messages.append({"role":role,"content":content})current_tokens=self.count_tokens(self.messages)# 超过阈值时触发压缩ifcurrent_tokens>self.max_tokens*self.compression_threshold:ifllmandlen(self.messages)>6:self._compress_early_messages(llm)else:self._truncate_early_messages()def_compress_early_messages(self,llm):""" 用LLM压缩早期消息：把前1/3的对话变成一段摘要 保留摘要 + 后2/3的原始消息 """n_to_compress=max(4,len(self.messages)//3)messages_to_compress=self.messages[:n_to_compress]# 构建压缩Promptconversation_text="\n".join([f"{m['role'].upper()}:{m['content']}"forminmessages_to_compress])compress_prompt=f"""请将以下对话压缩成一段简洁的摘要。 要求： 1. 保留所有关键信息（用户身份、重要数据、已完成的操作、未解决的问题） 2. 使用第三人称描述（"用户提到..."、"助手已完成..."） 3. 摘要长度不超过200字 对话内容：{conversation_text}摘要："""summary_response=llm.invoke([{"role":"user","content":compress_prompt}])new_summary=summary_response.content# 合并已有摘要和新摘要ifself.summary:self.summary=f"{self.summary}\n\n{new_summary}"else:self.summary=new_summary# 用摘要替换被压缩的消息self.messages=self.messages[n_to_compress:]print(f"记忆压缩：
