纲要记忆系统的核心挑战多轮对话与上下文窗口限制多用户会话隔离技术选型为什么选择Redis架构设计基础组件ConversationBufferMemoryRedisChatMessageHistory智能总结自动摘要链防止上下文溢出会话隔离基于session_id的 Key 管理完整实现项目结构记忆管理器MemoryManager类总结链设计保持人格统一读写流程与自动触发逻辑可运行示例总结与相关度说明记忆系统要解决什么问题任何聊天机器人都面临两个核心问题上下文窗口有限大模型一次能处理的 token 数量有上限。对话轮次增多历史记录不断累积迟早会撑爆窗口。多用户隔离张三和李四的对话不能混在一起否则智能体会“串台”把别人的事当成你的事。一个健壮的记忆系统必须在“记住足够上下文”和“不超出窗口限制”之间找到平衡同时保证每个用户的记忆是独立的。技术选型为什么用 Redis记忆存储有很多选择但Redis在对话场景下有天然优势读写极快内存存储适合高频率的聊天消息存取。数据结构匹配对话记录是一条一条追加的与Redis的 List 结构天然吻合。键值隔离通过不同的 Key 可以轻松实现多用户会话隔离。LangChain 官方提供了RedisChatMessageHistory组件可以将对话消息以标准格式直接存入Redis省去了手动序列化和解析的麻烦。架构设计历史记录否是用户发送消息Agent 执行器get_memory 读取历史Redis条数 80?返回原始历史调用总结链压缩清除旧记录保存摘要到 RedisLLM 生成回复set_memory 存储新消息核心思路每次读取时检查当前会话的消息条数。若超过阈值例如 80 条触发总结链将历史压缩为一段摘要清空旧记录后只保留摘要。这样上下文窗口永远不会无限膨胀同时保留了关键信息。完整实现项目结构memory_demo/ ├── config.py ├── memory_manager.py ├── main.py └── .env依赖安装pipinstalllangchain langchain-openai langchain-community redis python-dotenv配置文件config.py# config.pyimportosfromdotenvimportload_dotenv load_dotenv()classConfig:OPENAI_API_KEYos.getenv(OPENAI_API_KEY)OPENAI_BASE_URLos.getenv(OPENAI_BASE_URL,https://api.openai.com/v1)REDIS_URLos.getenv(REDIS_URL,redis://localhost:6379/0)MAX_HISTORY_LENGTHint(os.getenv(MAX_HISTORY_LENGTH,80))记忆管理器memory_manager.py这是记忆系统的核心封装了读取、写入和自动总结的全部逻辑。# memory_manager.pyfromtypingimportListfromlangchain.memoryimportConversationBufferMemoryfromlangchain_community.chat_message_historiesimportRedisChatMessageHistoryfromlangchain_openaiimportChatOpenAIfromlangchain.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromconfigimportConfigclassMemoryManager:带自动摘要的对话记忆管理器def__init__(self,system_prompt:str):self.memory_keychat_historyself.llmChatOpenAI(modelgpt-3.5-turbo,temperature0,openai_api_keyConfig.OPENAI_API_KEY,base_urlConfig.OPENAI_BASE_URL,)# 构建总结链summary_template(system_prompt\n\n以下是你和用户之间的对话记录请用第一人称「我」对其进行摘要提取关键信息保留用户的核心诉求和你的处理结果。\n对话记录:\n{chat_history}\n\n摘要:)self.summary_promptChatPromptTemplate.from_template(summary_template)self.summary_chainself.summary_prompt|self.llm|StrOutputParser()def_get_history(self,session_id:str)-RedisChatMessageHistory:获取指定会话的 Redis 历史记录对象returnRedisChatMessageHistory(session_idsession_id,urlConfig.REDIS_URL)defget_memory(self,session_id:str)-ConversationBufferMemory:读取会话记忆超过阈值时自动压缩historyself._get_history(session_id)messageshistory.messages# 如果消息数量超过阈值触发总结iflen(messages)Config.MAX_HISTORY_LENGTH:# 将所有消息格式化为文本chat_textformsginmessages:role用户ifmsg.typehumanelse助手chat_textf{role}:{msg.content}\n# 调用总结链生成摘要summaryself.summary_chain.invoke({chat_history:chat_text})# 清空 Redis 中的旧记录history.clear()# 将摘要作为新的系统级消息存入history.add_user_message(以下为历史对话摘要)history.add_ai_message(summary)# 返回标准的 ConversationBufferMemoryreturnConversationBufferMemory(memory_keyself.memory_key,chat_memoryhistory,return_messagesTrue,)defsave_memory(self,session_id:str,user_input:str,ai_output:str)-None:保存一轮对话到记忆historyself._get_history(session_id)history.add_user_message(user_input)history.add_ai_message(ai_output)演示主程序main.py模拟多轮对话观察记忆读取和自动总结的行为。# main.pyfromlangchain_openaiimportChatOpenAIfromlangchain.agentsimportAgentExecutor,create_tool_calling_agentfromlangchain.promptsimportChatPromptTemplate,MessagesPlaceholderfromlangchain.toolsimporttoolfromconfigimportConfigfrommemory_managerimportMemoryManager# 定义一个简单的工具供测试tooldefget_weather(city:str)-str:查询城市天气returnf{city}今天晴天25°Ctools[get_weather]defmain():system_prompt你是小浪助手友好且专业地回答用户问题。memory_managerMemoryManager(system_promptsystem_prompt)llmChatOpenAI(modelgpt-3.5-turbo,temperature0,openai_api_keyConfig.OPENAI_API_KEY,base_urlConfig.OPENAI_BASE_URL,)# 模拟两个不同用户的会话foruser_idin[user_001,user_002]:print(f\n 会话{user_id})memorymemory_manager.get_memory(user_id)promptChatPromptTemplate.from_messages([(system,system_prompt),MessagesPlaceholder(variable_namechat_history),(human,{input}),MessagesPlaceholder(variable_nameagent_scratchpad),])agentcreate_tool_calling_agent(llm,tools,prompt)executorAgentExecutor(agentagent,toolstools,memorymemory,verboseFalse,handle_parsing_errorsTrue)# 第一轮resexecutor.invoke({input:f你好我叫{user_id}请记住我的名字。})print(助手:,res[output])memory_manager.save_memory(user_id,f你好我叫{user_id}请记住我的名字。,res[output])# 第二轮resexecutor.invoke({input:我叫什么名字})print(助手:,res[output])memory_manager.save_memory(user_id,我叫什么名字,res[output])# 检查记忆是否隔离print(f---{user_id}的 Redis 消息数: ---)historymemory_manager._get_history(user_id)print(len(history.messages))if__name____main__:main()运行说明确保本地Redis服务已启动默认localhost:6379。在.env中填入OPENAI_API_KEY。执行python main.py观察两个用户的记忆完全独立互不干扰。关键设计总结链中的人格保持在构造总结提示词时刻意将原始的system_prompt拼接在最前面。这样总结出来的摘要仍然以“我是小浪助手”的第一人称视角叙述避免了摘要内容变成冷冰冰的第三方记述从而在后续对话中保持人格统一。许多开发者容易忽略这一点导致压缩后的记忆与原始角色设定产生割裂。总结本文实现了一个基于Redis的智能体记忆系统核心特性包括多用户会话隔离、自动触发摘要压缩、以及人格统一的第一人称总结。这套方案可直接嵌入任何 LangChain 单智能体项目有效解决长对话上下文溢出的痛点。本文覆盖了 Redis 记忆存储、会话隔离、总结链设计、自动压缩触发逻辑等全部技术细节并提供了示例代码。
