前几天多轮对话仅简单拼接消息列表，存在致命缺陷：
| 方案 | 原理 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| 滑动窗口裁剪 | 保留system+最近N轮，删除最早对话 | 实现最简单、无额外LLM调用、性能高 | 丢失早期历史信息 | 短会话、简单闲聊、轻量Agent |
| 对话摘要压缩 | 触发Token阈值时，LLM将旧历史生成一段摘要 | 保留核心语义，Token压缩比极高 | 消耗额外调用，摘要存在轻微信息丢失 | 中长多轮对话、工具调用场景 |
| 向量检索记忆 | 历史对话向量化存储，只召回相关片段 | 理论无存储上限，精准匹配关联历史 | 依赖Embedding与向量库，架构较重 | 长期业务会话、长周期客户对话 |
本地内存会话仅适用于单进程演示，线上必须分布式存储：
chat:session:{session_id}用户提问 → Redis读取会话历史 → 记忆模块自动裁剪/摘要压缩 → 拼接当前提问构建消息列表 → Plan-Solve任务规划 → 调度执行RAG/工具 → 保存本轮问答至Redis记忆
解决方案：
解决方案：
解决方案：
解决方案：
pip install aioredis
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
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())
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)
memory_store.py测试消息读写、过期策略token_safe_threshold阈值，触发摘要压缩，查看历史摘要内容session_id测试会话隔离，验证用户历史互不干扰滑动窗口 vs 摘要压缩
滑动窗口直接删除早期对话，性能高、无额外LLM消耗，但丢失历史；适合短会话。
摘要压缩用LLM浓缩旧历史，保留核心信息，Token压缩效率高；适合中长多轮业务对话，额外消耗一次模型调用。
Redis持久化必要性
内存会话仅单进程有效，服务重启、多实例部署会丢失历史；Redis分布式存储，多请求、多服务共享会话，支持过期自动清理，实现用户会话隔离。
Token超限带来的问题
接口返回400参数错误；输入自动截断，对话失忆；工具调用参数缺失、推理逻辑断裂；模型输出提前终止。
摘要temperature=0原因
摘要要求客观完整保留关键信息，不能自由发散、删减内容；0温度保证输出稳定、无多余编造。
Day5完成Agent记忆持久化模块，解决多轮对话Token爆炸、会话丢失两大核心工程痛点。分层记忆架构是工业级Agent标配，Redis分布式会话、滑动窗口+摘要双重压缩是面试高频考点。
当前已完整搭建：异步LLM客户端、SSE流式、Function Calling、RAG知识库、Plan-Solve任务规划、分层持久化记忆全链路基础Agent，具备完整业务落地能力。下一课将学习ReAct动态反思智能体、工具网关统一封装。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。