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: 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 = """
请将以下历史对话精简为一段摘要，保留所有数字、计算结果、工具参数、业务关键信息，删除无关闲聊，输出纯摘要文本，不要额外解释。
历史对话：