AI Agent 30天速成｜Day5 笔记

AI Agent 全日制30天速成｜Day5 教学笔记

今日总学习目标

1. 区分短期记忆/长期记忆，掌握Agent分层记忆架构设计
2. 实现滑动窗口、摘要压缩、向量记忆三种主流记忆方案
3. 基于Day4规划Agent改造，接入持久化对话记忆（Redis存储会话）
4. 解决超长多轮对话Token爆炸、历史遗忘、上下文冗余问题
   每日时长分配（全天8h）

• 理论笔记阅读+理解：2.5h
• 代码编写调试：4h
• 复盘+面试背诵：1.5h

一、核心理论教学笔记

1. Agent记忆体系核心概念

1.1 为什么需要独立记忆模块

前几天多轮对话仅简单拼接消息列表，存在致命缺陷：

1. 对话轮次上涨，Token持续累积，快速触达模型上下文上限
2. 无关闲聊、重复内容占用大量上下文，干扰模型推理与工具调用
3. 服务重启、多请求隔离场景，会话历史丢失，无法跨轮持久保存
4. 复杂规划任务无法复用过往对话信息，每次提问都要重复交代背景

1.2 分层记忆架构（行业标准三层设计）

1. 短期记忆（即时上下文）
   存储最近N轮原始对话消息，直接塞进LLM请求，保留完整细节；
   容量有限，超出阈值自动压缩/转移至长期记忆。
2. 中期记忆（对话摘要）
   对过期历史对话批量生成精简摘要，替代原始长消息，大幅降低Token消耗；
   保留核心业务信息，丢弃闲聊、冗余表述。
3. 长期记忆（向量记忆库）
   将全部历史对话向量化存入向量库；用户提问时召回相关历史片段，实现跨会话、跨天数记忆检索；
   适合长期客户咨询、多轮复杂业务场景。

1.3 三种记忆压缩/裁剪方案对比

方案	原理	优点	缺点	适用场景
滑动窗口裁剪	保留system+最近N轮，删除最早对话	实现最简单、无额外LLM调用、性能高	丢失早期历史信息	短会话、简单闲聊、轻量Agent
对话摘要压缩	触发Token阈值时，LLM将旧历史生成一段摘要	保留核心语义，Token压缩比极高	消耗额外调用，摘要存在轻微信息丢失	中长多轮对话、工具调用场景
向量检索记忆	历史对话向量化存储，只召回相关片段	理论无存储上限，精准匹配关联历史	依赖Embedding与向量库，架构较重	长期业务会话、长周期客户对话

1.4 Redis会话持久化设计

本地内存会话仅适用于单进程演示，线上必须分布式存储：

1. 会话唯一key：chat:session:{session_id}
2. 存储结构：列表存储完整消息，Hash存储会话基础信息（创建时间、用户ID、记忆阈值）
3. 过期策略：设置7天过期，自动清理无效会话，释放存储
4. 读写逻辑：每次对话先拉取历史，执行记忆裁剪/压缩，再写入更新

1.5 记忆与规划Agent联动流程

用户提问 → Redis读取会话历史 → 记忆模块自动裁剪/摘要压缩 → 拼接当前提问构建消息列表 → Plan-Solve任务规划 → 调度执行RAG/工具 → 保存本轮问答至Redis记忆

2. 记忆模块关键规则

1. System人设消息永久保留，不参与裁剪、摘要
2. 工具调用、tool返回结果属于关键业务信息，优先保留，不轻易压缩
3. 摘要生成temperature=0，保证信息完整、无多余发散内容
4. 向量记忆仅存储用户+assistant对话，过滤工具原始返回减少向量冗余

3. 今日新增技术点

1. Redis异步读写（aioredis），适配全异步代码体系
2. 自动阈值判断：设置安全Token阈值，超过自动触发摘要压缩
3. 混合记忆策略：滑动窗口+摘要双重兜底，兼顾性能与信息完整性
4. 会话隔离：多用户独立session_id，互不干扰历史记忆

二、今日学习重点

1. 三层分层记忆架构原理与落地取舍
2. 实现异步Redis会话存储，完成对话持久化读写
3. 封装自动记忆压缩工具：滑动窗口裁剪+对话摘要生成
4. 改造Day4规划Agent，接入完整记忆链路
5. 调试Token阈值、摘要压缩逻辑，解决超长对话超限报错

三、今日难点 & 解决方案

难点1：多轮对话持续累积，频繁触发Token超限报错

解决方案：

1. 实时预估总Token，设置安全阈值，提前拦截超限
2. 双重压缩策略：先滑动窗口裁剪，剩余文本过长再生成摘要替换旧历史
3. 精简tool工具返回内容，过滤换行、重复描述

难点2：摘要压缩丢失关键业务参数，后续工具调用出错

解决方案：

1. 摘要Prompt强制要求保留数字、订单、计算结果、工具参数等关键信息
2. 工具调用记录不参与摘要，永久保留在短期记忆
3. 压缩后保留最少2轮原始对话，避免完全依赖摘要导致信息缺失

难点3：Redis并发读写会话历史，消息错乱丢失

解决方案：

1. 使用Redis列表lrange/lpush原子操作读写消息
2. 单会话串行处理，同一session并发请求加简易锁
3. 每次对话完成后一次性全量覆盖写入，避免增量追加错乱

难点4：向量记忆召回大量无关历史，干扰当前对话

解决方案：

1. 设置相似度阈值，过滤低相关历史片段
2. 历史对话分块存储，每条对话单独向量，精准召回单轮内容
3. Prompt区分「当前对话短期历史」和「长期关联记忆」，权重区分

四、完整练习代码（基于Day1~Day4扩展）

新增依赖安装，遇到Windows环境下的redis连接断开可参考https://www.cnblogs.com/ylxin/p/20669550

pip install aioredis

1. 记忆模块 memory_store.py

import 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: strcontent: str# 异步Redis持久化会话存储
class RedisChatMemory:def __init__(self, redis_url="redis://127.0.0.1:6379"):self.redis_url = redis_urlself.redis: Optional[aioredis.Redis] = Noneself.expire_second = 7 * 24 * 3600  # 7天过期async def connect(self):self.redis = aioredis.from_url(self.redis_url, decode_responses=True)async def close(self):if self.redis:await self.redis.close()def _get_session_key(self, session_id: str):return f"chat: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_clientself.token_safe_threshold = 1800  # Token安全阈值self.keep_raw_round = 3  # 保留最近3轮原始对话self.summary_prompt = """
请将以下历史对话精简为一段摘要，保留所有数字、计算结果、工具参数、业务关键信息，删除无关闲聊，输出纯摘要文本，不要额外解释。
历史对话：
{history_text}
"""# 统计消息总tokendef calc_total_token(self, messages: List[Dict]) -> int:total = 0for m in messages:total += estimate_text_token(m.get("content", ""))return total# 滑动窗口裁剪def slide_trim(self, messages: List[Dict]) -> List[Dict]:# 永久保留system消息system_msg = Noneother_msg = []for m in messages:if m["role"] == "system":system_msg = melse:other_msg.append(m)# 保留最近N轮new_other = other_msg[-self.keep_raw_round:]if system_msg:return [system_msg] + new_otherreturn new_other# 旧历史生成摘要替换async def compress_to_summary(self, messages: List[Dict]) -> List[Dict]:system_msg = Nonehistory_raw = []recent_raw = []for m in messages:if m["role"] == "system":system_msg = melse:history_raw.append(m)# 分割旧历史与最近原始对话if len(history_raw) > self.keep_raw_round:old_part = history_raw[:-self.keep_raw_round]recent_raw = history_raw[-self.keep_raw_round:]else:return messages# 拼接旧历史文本生成摘要old_text = "\n".join([f"{m['role']}:{m['content']}" for m in old_part])prompt = self.summary_prompt.format(history_text=old_text)summary = await self.llm.chat([{"role": "user", "content": prompt}], temperature=0.0)# 构建新消息列表：system + 摘要 + 最近原始对话new_messages = []if system_msg:new_messages.append(system_msg)new_messages.append({"role": "system", "content": f"【历史对话摘要】{summary}"})new_messages.extend(recent_raw)return new_messages# 入口：自动判断并执行压缩async def auto_compress(self, messages: List[Dict]) -> List[Dict]:total_token = self.calc_total_token(messages)if total_token < self.token_safe_threshold:return messages# 第一步滑动裁剪trim_msg = self.slide_trim(messages)if self.calc_total_token(trim_msg) < self.token_safe_threshold:return trim_msg# 第二步摘要压缩兜底summary_msg = await self.compress_to_summary(messages)return summary_msg

2. 接入记忆的完整规划Agent memory_agent.py

import asyncio
from memory_store import RedisChatMemory, MemoryCompressor
from task_planner import TaskPlanner
from llm_client_v2 import AsyncLLMClientV2class MemoryAgent:def __init__(self, model_name="qwen-turbo"):self.model_name = model_nameself.llm = AsyncLLMClientV2(model_name)self.memory_store = RedisChatMemory()self.compressor = MemoryCompressor(self.llm)self.planner = TaskPlanner(model_name)self.system_base = "你是具备任务规划、知识库检索、计算器工具的智能助手，基于历史对话连贯回答用户问题。"async def init(self):await self.memory_store.connect()async def close(self):await self.memory_store.close()async def chat(self, session_id: str, user_query: str):# 1. 读取历史会话history = await self.memory_store.load_history(session_id)# 2. 拼接system+历史+当前提问full_messages = [{"role": "system", "content": self.system_base}] + historyfull_messages.append({"role": "user", "content": user_query})# 3. 自动压缩超长对话compressed_msg = await self.compressor.auto_compress(full_messages)# 4. 执行规划Agent完整流程plan_result = await self.planner.run_full_agent(user_query)final_ans = plan_result["final_answer"]# 5. 保存本轮问答到记忆await self.memory_store.append_message(session_id, "user", user_query)await self.memory_store.append_message(session_id, "assistant", final_ans)# 6. 压缩后覆盖会话，防止持续膨胀compressed_only_history = compressed_msg[1:]  # 剔除临时systemawait self.memory_store.override_history(session_id, compressed_only_history)return {"session_id": session_id,"user_input": user_query,"final_answer": final_ans,"task_info": plan_result,"history_after_compress": compressed_only_history}# 测试入口
async def test_memory_agent():agent = MemoryAgent()await agent.init()session_id = "test_session_001"# 多轮连续提问，模拟超长对话q1 = "计算 10*5"res1 = await agent.chat(session_id, q1)print(f"Q:{q1}\nA:{res1['final_answer']}\n")q2 = "再算上20，总和是多少，同时介绍RAG是什么"res2 = await agent.chat(session_id, q2)print(f"Q:{q2}\nA:{res2['final_answer']}\n")q3 = "结合前面所有计算结果，给我一段总结"res3 = await agent.chat(session_id, q3)print(f"Q:{q3}\nA:{res3['final_answer']}\n")await agent.close()if __name__ == "__main__":asyncio.run(test_memory_agent())

3. FastAPI会话记忆接口 main_memory.py

from fastapi import FastAPI, Query
import asyncio
from memory_agent import MemoryAgentapp = FastAPI(title="Day5 分层记忆持久化Agent")
agent = MemoryAgent("qwen-turbo")@app.on_event("startup")
async def startup():await agent.init()@app.on_event("shutdown")
async def shutdown():await agent.close()@app.get("/agent/memory_chat")
async def memory_chat(session_id: str = Query(..., description="用户唯一会话ID"),prompt: str = Query(..., description="用户提问")
):result = await agent.chat(session_id, prompt)return resultif __name__ == "__main__":import uvicornuvicorn.run("main_memory:app", reload=True)

五、今日必做练习任务

1. 本地启动Redis，填入API Key，运行memory_store.py测试消息读写、过期策略
2. 连续多轮提问拉长对话，观察自动滑动窗口裁剪逻辑
3. 调高token_safe_threshold阈值，触发摘要压缩，查看历史摘要内容
4. 使用不同session_id测试会话隔离，验证用户历史互不干扰
5. 修改摘要Prompt，观察摘要丢失/保留关键数字、计算参数的差异
6. 启动FastAPI，通过docs接口传入自定义session_id，持续对话查看压缩后的历史记录

六、今日配套面试题（Agent记忆模块高频）

基础问答

1. Agent三层分层记忆分别是什么？各自作用与存储形式？
2. 滑动窗口裁剪和对话摘要压缩的区别，分别适合什么场景？
3. 为什么要用Redis存储会话记忆，不能只用内存列表？
4. 多轮对话Token超限会带来哪些问题？
5. 生成对话摘要时temperature为什么设置为0？

工程实操题

1. 摘要压缩丢失计算参数、业务关键信息，如何优化Prompt？
2. 同一用户并发请求读写Redis会话，如何避免消息错乱？
3. 如何设计混合记忆策略，同时兼顾性能与历史信息完整性？
4. 会话过期如何处理，过期后用户重新对话如何恢复记忆？

拓展思考题（高阶Agent）

1. 长期向量记忆如何与短期摘要记忆联动使用？
2. 如何实现记忆自动遗忘（不重要历史自动丢弃）？
3. 生产环境海量用户会话，Redis内存占用过高如何优化？
4. 多轮对话过长，除摘要、滑动窗口外还有哪些高级优化方案？

面试题标准答案

基础问答

1. 三层分层记忆

• 短期记忆：最近N轮原始对话，直接传入LLM，Redis列表存储；
• 中期记忆：旧对话压缩摘要，替代原始长消息，减少Token；
• 长期记忆：全部对话向量化存入向量库，按需召回关联历史。

2. 滑动窗口 vs 摘要压缩
   滑动窗口直接删除早期对话，性能高、无额外LLM消耗，但丢失历史；适合短会话。
   摘要压缩用LLM浓缩旧历史，保留核心信息，Token压缩效率高；适合中长多轮业务对话，额外消耗一次模型调用。

3. Redis持久化必要性
   内存会话仅单进程有效，服务重启、多实例部署会丢失历史；Redis分布式存储，多请求、多服务共享会话，支持过期自动清理，实现用户会话隔离。

4. Token超限带来的问题
   接口返回400参数错误；输入自动截断，对话失忆；工具调用参数缺失、推理逻辑断裂；模型输出提前终止。

5. 摘要temperature=0原因
   摘要要求客观完整保留关键信息，不能自由发散、删减内容；0温度保证输出稳定、无多余编造。

工程实操题

1. 优化摘要Prompt
   明确要求保留数字、计算式、工具返回参数、业务关键词；禁止省略数据；增加少样本示例，展示完整保留关键信息的摘要模板。
2. Redis并发读写防错乱
   使用Redis原子List操作；单会话加分布式锁；每次对话完成后全量覆盖写入历史，避免增量追加冲突。
3. 混合记忆策略
   先滑动窗口裁剪做第一层轻量化压缩，Token仍超标再执行摘要压缩；永久保留system、最近3轮原始对话，旧历史用摘要替代，兼顾性能与信息完整。
4. 会话过期处理
   Redis设置过期时间；会话过期后读取不到历史，自动开启全新对话；可增加长期向量记忆兜底，召回该用户历史对话片段恢复上下文。

拓展思考题

1. 向量记忆+摘要联动
   短期：滑动窗口+摘要作为当前上下文；用户提问时向量库召回跨会话长期相关历史，拼接进Prompt补充背景，实现长短记忆互补。
2. 自动遗忘机制
   给每条对话打分（信息权重、时效性），低权重闲聊、过期信息优先在压缩时丢弃；摘要只保留高价值业务内容。
3. Redis内存优化
   设置会话过期自动淘汰；拆分冷热数据，冷会话写入磁盘；压缩消息JSON字符串；采用Redis集群分片存储海量会话。
4. 超长对话高级优化
   分层记忆、向量检索召回相关历史、对话分段摘要、滑动窗口、冗余文本清洗、结构化历史存储替代自然语言。

学习总结

Day5完成Agent记忆持久化模块，解决多轮对话Token爆炸、会话丢失两大核心工程痛点。分层记忆架构是工业级Agent标配，Redis分布式会话、滑动窗口+摘要双重压缩是面试高频考点。
当前已完整搭建：异步LLM客户端、SSE流式、Function Calling、RAG知识库、Plan-Solve任务规划、分层持久化记忆全链路基础Agent，具备完整业务落地能力。下一课将学习ReAct动态反思智能体、工具网关统一封装。