AI Agent 全日制30天速成Day5 教学笔记今日总学习目标区分短期记忆/长期记忆掌握Agent分层记忆架构设计实现滑动窗口、摘要压缩、向量记忆三种主流记忆方案基于Day4规划Agent改造接入持久化对话记忆Redis存储会话解决超长多轮对话Token爆炸、历史遗忘、上下文冗余问题每日时长分配全天8h理论笔记阅读理解2.5h代码编写调试4h复盘面试背诵1.5h一、核心理论教学笔记1. Agent记忆体系核心概念1.1 为什么需要独立记忆模块前几天多轮对话仅简单拼接消息列表存在致命缺陷对话轮次上涨Token持续累积快速触达模型上下文上限无关闲聊、重复内容占用大量上下文干扰模型推理与工具调用服务重启、多请求隔离场景会话历史丢失无法跨轮持久保存复杂规划任务无法复用过往对话信息每次提问都要重复交代背景1.2 分层记忆架构行业标准三层设计短期记忆即时上下文存储最近N轮原始对话消息直接塞进LLM请求保留完整细节容量有限超出阈值自动压缩/转移至长期记忆。中期记忆对话摘要对过期历史对话批量生成精简摘要替代原始长消息大幅降低Token消耗保留核心业务信息丢弃闲聊、冗余表述。长期记忆向量记忆库将全部历史对话向量化存入向量库用户提问时召回相关历史片段实现跨会话、跨天数记忆检索适合长期客户咨询、多轮复杂业务场景。1.3 三种记忆压缩/裁剪方案对比方案原理优点缺点适用场景滑动窗口裁剪保留system最近N轮删除最早对话实现最简单、无额外LLM调用、性能高丢失早期历史信息短会话、简单闲聊、轻量Agent对话摘要压缩触发Token阈值时LLM将旧历史生成一段摘要保留核心语义Token压缩比极高消耗额外调用摘要存在轻微信息丢失中长多轮对话、工具调用场景向量检索记忆历史对话向量化存储只召回相关片段理论无存储上限精准匹配关联历史依赖Embedding与向量库架构较重长期业务会话、长周期客户对话1.4 Redis会话持久化设计本地内存会话仅适用于单进程演示线上必须分布式存储会话唯一keychat:session:{session_id}存储结构列表存储完整消息Hash存储会话基础信息创建时间、用户ID、记忆阈值过期策略设置7天过期自动清理无效会话释放存储读写逻辑每次对话先拉取历史执行记忆裁剪/压缩再写入更新1.5 记忆与规划Agent联动流程用户提问 → Redis读取会话历史 → 记忆模块自动裁剪/摘要压缩 → 拼接当前提问构建消息列表 → Plan-Solve任务规划 → 调度执行RAG/工具 → 保存本轮问答至Redis记忆2. 记忆模块关键规则System人设消息永久保留不参与裁剪、摘要工具调用、tool返回结果属于关键业务信息优先保留不轻易压缩摘要生成temperature0保证信息完整、无多余发散内容向量记忆仅存储用户assistant对话过滤工具原始返回减少向量冗余3. 今日新增技术点Redis异步读写aioredis适配全异步代码体系自动阈值判断设置安全Token阈值超过自动触发摘要压缩混合记忆策略滑动窗口摘要双重兜底兼顾性能与信息完整性会话隔离多用户独立session_id互不干扰历史记忆二、今日学习重点三层分层记忆架构原理与落地取舍实现异步Redis会话存储完成对话持久化读写封装自动记忆压缩工具滑动窗口裁剪对话摘要生成改造Day4规划Agent接入完整记忆链路调试Token阈值、摘要压缩逻辑解决超长对话超限报错三、今日难点 解决方案难点1多轮对话持续累积频繁触发Token超限报错解决方案实时预估总Token设置安全阈值提前拦截超限双重压缩策略先滑动窗口裁剪剩余文本过长再生成摘要替换旧历史精简tool工具返回内容过滤换行、重复描述难点2摘要压缩丢失关键业务参数后续工具调用出错解决方案摘要Prompt强制要求保留数字、订单、计算结果、工具参数等关键信息工具调用记录不参与摘要永久保留在短期记忆压缩后保留最少2轮原始对话避免完全依赖摘要导致信息缺失难点3Redis并发读写会话历史消息错乱丢失解决方案使用Redis列表lrange/lpush原子操作读写消息单会话串行处理同一session并发请求加简易锁每次对话完成后一次性全量覆盖写入避免增量追加错乱难点4向量记忆召回大量无关历史干扰当前对话解决方案设置相似度阈值过滤低相关历史片段历史对话分块存储每条对话单独向量精准召回单轮内容Prompt区分「当前对话短期历史」和「长期关联记忆」权重区分四、完整练习代码基于Day1~Day4扩展新增依赖安装遇到Windows环境下的redis连接断开可参考https://www.cnblogs.com/ylxin/p/20669550pip install aioredis1. 记忆模块 memory_store.pyimport asyncio import re import json import aioredis from pydantic import BaseModel from typing import List, Dict, Optional # 复用前序代码 from llm_client_v2 import AsyncLLMClientV2 from rag_store import split_text_chunk # Token简易估算 def estimate_text_token(text: str) - int: return len(text) * 2 # 会话消息结构 class ChatMessage(BaseModel): role: str content: str # 异步Redis持久化会话存储 class RedisChatMemory: def __init__(self, redis_urlredis://127.0.0.1:6379): self.redis_url redis_url self.redis: Optional[aioredis.Redis] None self.expire_second 7 * 24 * 3600 # 7天过期 async def connect(self): self.redis aioredis.from_url(self.redis_url, decode_responsesTrue) async def close(self): if self.redis: await self.redis.close() def _get_session_key(self, session_id: str): return fchat:session:{session_id} # 读取会话全部历史 async def load_history(self, session_id: str) - List[Dict]: key self._get_session_key(session_id) raw_list await self.redis.lrange(key, 0, -1) messages [] for item in raw_list: messages.append(json.loads(item)) return messages # 追加单条消息并持久化 async def append_message(self, session_id: str, role: str, content: str): key self._get_session_key(session_id) msg json.dumps({role: role, content: content}) await self.redis.rpush(key, msg) await self.redis.expire(key, self.expire_second) # 全量覆盖更新消息列表裁剪/压缩后使用 async def override_history(self, session_id: str, messages: List[Dict]): key self._get_session_key(session_id) await self.redis.delete(key) for msg in messages: await self.redis.rpush(key, json.dumps(msg)) await self.redis.expire(key, self.expire_second) # 记忆压缩管理器滑动窗口摘要压缩 class MemoryCompressor: def __init__(self, llm_client: AsyncLLMClientV2): self.llm llm_client self.token_safe_threshold 1800 # Token安全阈值 self.keep_raw_round 3 # 保留最近3轮原始对话 self.summary_prompt 请将以下历史对话精简为一段摘要保留所有数字、计算结果、工具参数、业务关键信息删除无关闲聊输出纯摘要文本不要额外解释。 历史对话