一、核心理论教学笔记

1. Agent记忆体系核心概念
   1.1 为什么需要独立记忆模块
   前几天多轮对话仅简单拼接消息列表，存在致命缺陷：

对话轮次上涨，Token持续累积，快速触达模型上下文上限
无关闲聊、重复内容占用大量上下文，干扰模型推理与工具调用
服务重启、多请求隔离场景，会话历史丢失，无法跨轮持久保存
复杂规划任务无法复用过往对话信息，每次提问都要重复交代背景
1.2 分层记忆架构（行业标准三层设计）
短期记忆（即时上下文）
存储最近N轮原始对话消息，直接塞进LLM请求，保留完整细节；
容量有限，超出阈值自动压缩/转移至长期记忆。
中期记忆（对话摘要）
对过期历史对话批量生成精简摘要，替代原始长消息，大幅降低Token消耗；
保留核心业务信息，丢弃闲聊、冗余表述。
长期记忆（向量记忆库）
将全部历史对话向量化存入向量库；用户提问时召回相关历史片段，实现跨会话、跨天数记忆检索；
适合长期客户咨询、多轮复杂业务场景。
1.3 三种记忆压缩/裁剪方案对比
方案 原理 优点 缺点 适用场景
滑动窗口裁剪 保留system+最近N轮，删除最早对话 实现最简单、无额外LLM调用、性能高 丢失早期历史信息 短会话、简单闲聊、轻量Agent
对话摘要压缩 触发Token阈值时，LLM将旧历史生成一段摘要 保留核心语义，Token压缩比极高 消耗额外调用，摘要存在轻微信息丢失 中长多轮对话、工具调用场景
向量检索记忆 历史对话向量化存储，只召回相关片段 理论无存储上限，精准匹配关联历史 依赖Embedding与向量库，架构较重 长期业务会话、长周期客户对话
1.4 Redis会话持久化设计
本地内存会话仅适用于单进程演示，线上必须分布式存储：

会话唯一key：chat:session:{session_id}
存储结构：列表存储完整消息，Hash存储会话基础信息（创建时间、用户ID、记忆阈值）
过期策略：设置7天过期，自动清理无效会话，释放存储
读写逻辑：每次对话先拉取历史，执行记忆裁剪/压缩，再写入更新
1.5 记忆与规划Agent联动流程
用户提问 → Redis读取会话历史 → 记忆模块自动裁剪/摘要压缩 → 拼接当前提问构建消息列表 → Plan-Solve任务规划 → 调度执行RAG/工具 → 保存本轮问答至Redis记忆

2. 记忆模块关键规则
   System人设消息永久保留，不参与裁剪、摘要
   工具调用、tool返回结果属于关键业务信息，优先保留，不轻易压缩
   摘要生成temperature=0，保证信息完整、无多余发散内容
   向量记忆仅存储用户+assistant对话，过滤工具原始返回减少向量冗余
3. 今日新增技术点
   Redis异步读写（aioredis），适配全异步代码体系
   自动阈值判断：设置安全Token阈值，超过自动触发摘要压缩
   混合记忆策略：滑动窗口+摘要双重兜底，兼顾性能与信息完整性
   会话隔离：多用户独立session_id，互不干扰历史记忆
   二、今日学习重点
   三层分层记忆架构原理与落地取舍
   实现异步Redis会话存储，完成对话持久化读写
   封装自动记忆压缩工具：滑动窗口裁剪+对话摘要生成
   改造Day4规划Agent，接入完整记忆链路
   调试Token阈值、摘要压缩逻辑，解决超长对话超限报错
   三、今日难点 & 解决方案
   难点1：多轮对话持续累积，频繁触发Token超限报错
   解决方案：

实时预估总Token，设置安全阈值，提前拦截超限
双重压缩策略：先滑动窗口裁剪，剩余文本过长再生成摘要替换旧历史
精简tool工具返回内容，过滤换行、重复描述
难点2：摘要压缩丢失关键业务参数，后续工具调用出错
解决方案：

摘要Prompt强制要求保留数字、订单、计算结果、工具参数等关键信息
工具调用记录不参与摘要，永久保留在短期记忆
压缩后保留最少2轮原始对话，避免完全依赖摘要导致信息缺失
难点3：Redis并发读写会话历史，消息错乱丢失
解决方案：

使用Redis列表lrange/lpush原子操作读写消息
单会话串行处理，同一session并发请求加简易锁
每次对话完成后一次性全量覆盖写入，避免增量追加错乱
难点4：向量记忆召回大量无关历史，干扰当前对话
解决方案：

设置相似度阈值，过滤低相关历史片段
历史对话分块存储，每条对话单独向量，精准召回单轮内容
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: str
content: str

异步Redis持久化会话存储

class RedisChatMemory:
def init(self, redis_url=“redis://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_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_client
self.token_safe_threshold = 1800 # Token安全阈值
self.keep_raw_round = 3 # 保留最近3轮原始对话
self.summary_prompt = “”"
请将以下历史对话精简为一段摘要，保留所有数字、计算结果、工具参数、业务关键信息，删除无关闲聊，输出纯摘要文本，不要额外解释。
历史对话：
{history_text}
“”"

# 统计消息总token
def calc_total_token(self, messages: List[Dict]) -> int:
    total = 0
    for m in messages:
        total += estimate_text_token(m.get("content", ""))
    return total

# 滑动窗口裁剪
def slide_trim(self, messages: List[Dict]) -> List[Dict]:
    # 永久保留system消息
    system_msg = None
    other_msg = []
    for m in messages:
        if m["role"] == "system":
            system_msg = m
        else:
            other_msg.append(m)
    # 保留最近N轮
    new_other = other_msg[-self.keep_raw_round:]
    if system_msg:
        return [system_msg] + new_other
    return new_other

# 旧历史生成摘要替换
async def compress_to_summary(self, messages: List[Dict]) -> List[Dict]:
    system_msg = None
    history_raw = []
    recent_raw = []
    for m in messages:
        if m["role"] == "system":
            system_msg = m
        else:
            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 AsyncLLMClientV2

class MemoryAgent:
def init(self, model_name=“qwen-turbo”):
self.model_name = model_name
self.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}] + history
    full_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:]  # 剔除临时system
    await 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 MemoryAgent

app = 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 result

if name == “main”:
import uvicorn
uvicorn.run(“main_memory:app”, reload=True)
五、今日必做练习任务
本地启动Redis，填入API Key，运行memory_store.py测试消息读写、过期策略
连续多轮提问拉长对话，观察自动滑动窗口裁剪逻辑
调高token_safe_threshold阈值，触发摘要压缩，查看历史摘要内容
使用不同session_id测试会话隔离，验证用户历史互不干扰
修改摘要Prompt，观察摘要丢失/保留关键数字、计算参数的差异
启动FastAPI，通过docs接口传入自定义session_id，持续对话查看压缩后的历史记录
六、今日配套面试题（Agent记忆模块高频）
基础问答
Agent三层分层记忆分别是什么？各自作用与存储形式？
滑动窗口裁剪和对话摘要压缩的区别，分别适合什么场景？
为什么要用Redis存储会话记忆，不能只用内存列表？
多轮对话Token超限会带来哪些问题？
生成对话摘要时temperature为什么设置为0？
工程实操题
摘要压缩丢失计算参数、业务关键信息，如何优化Prompt？
同一用户并发请求读写Redis会话，如何避免消息错乱？
如何设计混合记忆策略，同时兼顾性能与历史信息完整性？
会话过期如何处理，过期后用户重新对话如何恢复记忆？
拓展思考题（高阶Agent）
长期向量记忆如何与短期摘要记忆联动使用？
如何实现记忆自动遗忘（不重要历史自动丢弃）？
生产环境海量用户会话，Redis内存占用过高如何优化？
多轮对话过长，除摘要、滑动窗口外还有哪些高级优化方案？
面试题标准答案
基础问答
三层分层记忆
短期记忆：最近N轮原始对话，直接传入LLM，Redis列表存储；
中期记忆：旧对话压缩摘要，替代原始长消息，减少Token；
长期记忆：全部对话向量化存入向量库，按需召回关联历史。
滑动窗口 vs 摘要压缩
滑动窗口直接删除早期对话，性能高、无额外LLM消耗，但丢失历史；适合短会话。
摘要压缩用LLM浓缩旧历史，保留核心信息，Token压缩效率高；适合中长多轮业务对话，额外消耗一次模型调用。

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

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

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

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