一、核心理论教学笔记

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: 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)

五、今日必做练习任务

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？