前面四篇分别讲解了会话级、任务级、长期级、知识级记忆的核心实现和实操要点，但生产级智能体的核心的是“五大模块协同工作”（新增：记忆清理与优化模块），而非单独使用某一种记忆。
 

很多开发者的误区是“单独开发某一种记忆，忽略模块间的协同”，导致智能体出现“对话不连贯、无法自主推进任务、专业度不足、记忆冗余、Token消耗过高”等问题。比如，只使用会话记忆，智能体重启后遗忘用户偏好；只使用知识记忆，无法实现个性化交互。

本章将整合五大记忆模块（会话+任务+长期+知识+记忆清理），讲解模块协同逻辑、工业级落地完整流程，附可直接复用的工程化代码，解决生产环境中的核心痛点（Token控制、记忆精度、持久化、可维护性）。

五大模块的核心协同逻辑是“分层存储、按需检索、动态优化”，各模块的定位和协同关系如下：

1. 会话级记忆：底层基础，负责当前对话上下文衔接，为所有模块提供“即时交互语境”；
2. 任务级记忆：核心驱动，负责自主推进任务，调用其他模块的记忆（如从知识记忆获取专业知识，从长期记忆获取用户偏好）；
3. 长期级记忆：个性化支撑，负责跨会话、跨任务复用用户相关关键信息，为任务推进和知识调用提供个性化适配；
4. 知识级记忆：专业支撑，负责提供行业知识、外部文档、结构化关联知识，确保任务推进的专业性和准确性；
5. 记忆清理与优化模块：保障性能，负责清理过期记忆、冗余记忆，控制Token消耗，优化检索精度，避免记忆膨胀。

协同流程：用户发起指令 → 会话记忆记录当前语境 → 任务记忆规划任务步骤 → 按需检索长期记忆（用户偏好）和知识记忆（专业知识） → 完成任务/生成回复 → 记忆清理模块清理冗余信息 → 长期记忆/知识记忆更新关键内容。

以下代码整合五大记忆模块，实现“用户个性化需求+自主任务推进+专业知识支撑+性能优化”，适配生产环境，包含完整的持久化、记忆清理、多模块协同逻辑。

 from langchain.memory import ( ConversationBufferWindowMemory, VectorStoreRetrieverMemory, CombinedMemory, ConversationTokenBufferMemory ) from langchain_community.vectorstores import Chroma from langchain_community.graphs import Neo4jGraph from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain.chains import RetrievalQA, GraphCypherQAChain, AgentExecutor, create_openai_tools_agent from langchain.prompts import ChatPromptTemplate from langchain.tools import Tool from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.document_loaders import TextLoader import time # ------------------- 1. 初始化基础组件（大模型、Embedding、文本分割器）------------------- llm = ChatOpenAI(model="gpt-3.5-turbo", api_key="你的API密钥") embeddings = OpenAIEmbeddings(api_key="你的API密钥") text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) # ------------------- 2. 五大记忆模块实现 ------------------- # 2.1 会话级记忆（滑动窗口+Token控制，平衡连贯与性能） short_term_memory = ConversationTokenBufferMemory( llm=llm, max_token_limit=1000, # 精准控制Token消耗 return_messages=True ) # 2.2 长期级记忆（向量检索，持久化存储用户偏好） long_term_vectorstore = Chroma( embedding_function=embeddings, persist_directory="./long_term_memory_db" ) long_term_memory = VectorStoreRetrieverMemory( retriever=long_term_vectorstore.as_retriever(search_kwargs={"k": 2}) ) # 2.3 知识级记忆（RAG+知识图谱，专业知识支撑） # 2.3.1 RAG记忆（外部文档：养老政策、操作规范） loader = TextLoader("养老行业知识库.txt") documents = loader.load() splits = text_splitter.split_documents(documents) knowledge_vectorstore = Chroma.from_documents( documents=splits, embedding=embeddings, persist_directory="./knowledge_memory_db" ) rag_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="map_reduce", retriever=knowledge_vectorstore.as_retriever(search_kwargs={"k": 2}) ) # 2.3.2 知识图谱记忆（结构化关联知识） graph = Neo4jGraph( url="bolt://localhost:7687", username="neo4j", password="你的数据库密码" ) # 初始化知识图谱（若已存在可跳过） graph.query(""" CREATE (a:养老智能体 {name: "养老智能体"}), (b:核心功能 {name: "照护提醒"}), (c:核心功能 {name: "用药提醒"}), (d:核心功能 {name: "健康数据监测"}), (e:依赖设备 {name: "传感器"}), (f:依赖设备 {name: "血压计"}), (g:用户需求 {name: "语音控制"}), (h:政策要求 {name: "操作便捷性"}), (a)-[:包含]->(b), (a)-[:包含]->(c), (a)-[:包含]->(d), (b)-[:依赖]->(e), (d)-[:依赖]->(f), (a)-[:满足]->(g), (a)-[:符合]->(h) """) graph_chain = GraphCypherQAChain.from_llm(llm=llm, graph=graph, verbose=False) # 2.4 任务级记忆（自定义TaskMemory类，自主推进任务） class TaskMemory: def __init__(self, task_name: str): self.task_name = task_name self.steps = [] self.current_step = 0 self.results = [] self.tools_used = [] self.status = "pending" self.exceptions = [] def add_step(self, step: str): self.steps.append(step) def update_step(self, step_index: int, new_step: str): if 0 <= step_index < len(self.steps): self.steps[step_index] = new_step else: raise IndexError("步骤索引超出范围") def add_result(self, result: str): self.results.append(result) def add_tool_usage(self, tool_name: str, tool_params: dict, tool_result: str): self.tools_used.append({ "tool_name": tool_name, "tool_params": tool_params, "tool_result": tool_result }) def update_status(self, status: str): valid_status = ["pending", "processing", "completed", "failed"] if status in valid_status: self.status = status else: raise ValueError(f"非法状态值，仅支持{valid_status}") def add_exception(self, exception_msg: str, handle_method: str): self.exceptions.append({ "exception_msg": exception_msg, "handle_method": handle_method }) def to_prompt(self) -> str: steps_str = [f"{i+1}. {step}" for i, step in enumerate(self.steps)] if self.steps else "无" results_str = [f"步骤{i+1}：{result}" for i, result in enumerate(self.results)] if self.results else "无" tools_str = "" if self.tools_used: for idx, tool in enumerate(self.tools_used, 1): tools_str += f"{idx}. 工具：{tool['tool_name']}，参数：{tool['tool_params']}，结果：{tool['tool_result']} " else: tools_str = "无" exceptions_str = "" if self.exceptions: for idx, exc in enumerate(self.exceptions, 1): exceptions_str += f"{idx}. 错误：{exc['exception_msg']}，处理：{exc['handle_method']} " else: exceptions_str = "无" return f""" 【任务记忆】 任务名称：{self.task_name} 任务步骤：{steps_str} 当前步骤：{self.current_step + 1}. 执行结果：{results_str} 工具调用：{tools_str} 任务状态：{self.status} 异常记录：{exceptions_str} 请结合会话记忆、长期记忆、知识记忆，自主推进任务，确保专业、贴合用户需求。 """ # 2.5 记忆清理与优化模块（定期清理冗余、过期记忆） class MemoryCleaner: def __init__(self, long_term_vectorstore, knowledge_vectorstore): self.long_term_db = long_term_vectorstore self.knowledge_db = knowledge_vectorstore def clean_short_term_memory(self, memory, max_tokens=1000): """清理会话记忆，确保不超过Token限制""" current_tokens = self._count_tokens(memory.load_memory_variables({})["history"]) if current_tokens > max_tokens: # 裁剪会话记忆，保留最近3轮 memory.k = 3 memory.load_memory_variables({}) print(f"会话记忆已清理，当前Token数：{self._count_tokens(memory.load_memory_variables({})['history'])}") def clean_long_term_memory(self, expired_days=30): """清理长期记忆中过期的内容（如30天未访问的记忆）""" # 模拟过期记忆清理（实际可结合记忆的访问时间戳实现） all_memories = self.long_term_db.get()["documents"] if len(all_memories) > 100: # 当记忆数量超过100条，清理前50条过期内容 self.long_term_db.delete(ids=list(range(50))) self.long_term_db.persist() print("长期记忆已清理过期内容") def clean_knowledge_memory(self): """清理知识记忆中过期的文档（如失效政策）""" # 模拟清理，实际可根据文档更新时间筛选 expired_docs = ["养老行业旧政策.txt"] for doc in expired_docs: self.knowledge_db.delete(documents=[doc]) self.knowledge_db.persist() print("知识记忆已清理过期文档") def _count_tokens(self, text): """统计Token数量（简化版，实际可使用tiktoken库）""" return len(text.split()) # ------------------- 3. 模块协同与工具定义 ------------------- # 3.1 定义工具（任务推进中调用，关联知识记忆） def knowledge_retrieval_tool(query: str) -> str: """知识检索工具：调用RAG和知识图谱，获取专业知识""" rag_response = rag_chain.invoke(query) graph_response = graph_chain.invoke(query) return f"RAG知识：{rag_response['result']} 知识图谱关联：{graph_response['result']}" def memory_clean_tool() -> str: """记忆清理工具：清理冗余、过期记忆""" cleaner = MemoryCleaner(long_term_vectorstore, knowledge_vectorstore) cleaner.clean_short_term_memory(short_term_memory) cleaner.clean_long_term_memory() cleaner.clean_knowledge_memory() return "所有记忆模块已清理优化，Token消耗和检索精度已达标" # 注册工具 tools = [ Tool( name="知识检索工具", func=knowledge_retrieval_tool, description="用于获取养老行业专业知识、政策、设备关联等信息，参数为查询语句" ), Tool( name="记忆清理工具", func=memory_clean_tool, description="用于清理冗余、过期记忆，优化Token消耗和检索精度，无需参数" ) ] # 3.2 组合记忆（会话+长期） combined_memory = CombinedMemory(memories=[short_term_memory, long_term_memory]) # 3.3 定义Prompt模板（整合所有记忆模块） prompt = ChatPromptTemplate.from_messages([ ("system", "你是专业的养老智能体开发助手，需结合会话记忆、长期记忆、任务记忆、知识记忆，自主推进任务，确保回复个性化、专业、连贯。"), ("placeholder", "{chat_history}"), # 组合记忆（会话+长期） ("placeholder", "{task_memory}"), # 任务记忆 ("human", "{input}"), ("placeholder", "{agent_scratchpad}") # Agent推理过程 ]) # ------------------- 4. 生产级智能体启动与测试 ------------------- if __name__ == "__main__": # 初始化任务记忆、记忆清理器、Agent task_memory = TaskMemory(task_name="开发社区养老服务中心AI智能体") cleaner = MemoryCleaner(long_term_vectorstore, knowledge_vectorstore) agent = create_openai_tools_agent(llm, tools, prompt) agent_executor = AgentExecutor( agent=agent, memory=combined_memory, verbose=True, extra_context={"task_memory": task_memory.to_prompt()} # 注入任务记忆 ) # 步骤1：存入长期记忆（用户个性化需求） long_term_memory.save_context( {"input": "我是社区养老服务中心的，需要开发适配老年用户的智能体，核心需求是照护提醒、用药提醒、健康数据监测，要求操作简单、支持语音控制，符合行业政策"}, {"output": "已记录你的需求和场景，后续将优先适配"} ) # 步骤2：添加任务步骤，启动任务 task_memory.add_step("需求分析：结合用户需求和行业知识，梳理详细需求清单") task_memory.add_step("架构设计：设计五大记忆模块和核心功能的架构") task_memory.add_step("代码开发：实现核心功能和记忆模块的集成") task_memory.add_step("测试优化：验证功能和记忆模块协同效果，清理冗余记忆") task_memory.update_status("processing") # 步骤3：启动智能体，自主推进任务 print("=== 生产级养老智能体启动，开始自主推进任务 === ") # 第一次交互：推进需求分析步骤 agent_executor.invoke({ "input": "开始推进任务，先完成需求分析步骤，结合我的需求和行业知识，梳理详细的需求清单", "task_memory": task_memory.to_prompt() }) # 模拟任务推进：完成需求分析，记录结果，推进到架构设计 task_memory.add_result("需求分析完成：核心需求包括照护提醒（定时、重复、异常预警）、用药提醒（剂量、禁忌、记录）、健康数据监测（实时采集、异常预警、报告生成）；用户需求：语音控制、字体放大、操作简单；政策要求：符合老年用户操作标准，可申请研发补贴；依赖设备：传感器、血压计、电子病历系统。") task_memory.current_step = 1 task_memory.add_tool_usage( tool_name="知识检索工具", tool_params={"query": "养老智能体需求分析的行业标准和政策要求"}, tool_result="已获取相关知识，需求分析符合行业规范" ) # 第二次交互：推进架构设计步骤，调用记忆清理工具 print(" === 推进到架构设计步骤 === ") agent_executor.invoke({ "input": "推进到架构设计步骤，设计五大记忆模块和核心功能的架构，同时调用记忆清理工具优化性能", "task_memory": task_memory.to_prompt() }) # 步骤4：任务总结，更新长期记忆 task_memory.update_status("completed") long_term_memory.save_context( {"input": "养老智能体开发任务总结：已完成需求分析和架构设计，核心功能明确，记忆模块协同正常，性能达标"}, {"output": "已保存任务总结，后续开发可直接复用"} ) print(" === 任务推进完成，所有记忆模块协同正常 ===") print("长期记忆已更新，知识记忆可复用，会话记忆已清理优化")

1. Token控制技巧（核心痛点解决）

• 会话记忆：使用ConversationTokenBufferMemory，精准控制max_token_limit（结合大模型上下文窗口，如GPT-3.5设为1000-2000）；
• 长期/知识记忆：控制检索数量（k=2-3），避免检索过多无关记忆导致Prompt冗余；
• 定时清理：通过MemoryCleaner模块，定期清理会话记忆、过期长期记忆、失效知识，避免Token消耗过高。

2. 记忆精度优化

• 长期记忆：只存用户偏好、任务总结等关键信息，不存无关闲聊；
• 知识记忆：定期更新知识库/知识图谱，删除过期内容，新增最新行业知识；
• 检索优化：向量数据库选择适配场景（中小规模用Chroma，大规模用Pinecone/FAISS），调整检索参数（k值、相似度阈值）。

3. 持久化与可维护性

• 所有向量数据库（长期、知识）均设置persist_directory，确保重启应用后记忆不丢失；
• 建立记忆更新机制：任务完成后自动更新长期记忆，定期手动更新知识记忆；
• 日志记录：为每个记忆模块添加操作日志，便于排查记忆检索、存储、清理中的问题。

4. 模块协同优化

• 任务记忆优先调用知识记忆获取专业支撑，再调用长期记忆适配用户个性化需求；
• 会话记忆为所有模块提供即时语境，避免记忆脱节；
• 记忆清理模块定期运行（如每天凌晨），不影响智能体正常运行。

大模型智能体的记忆体系，核心是“分层存储、按需检索、协同工作”，五大模块各司其职、相互支撑，共同实现智能体的“连贯交互、自主推进、个性化、专业化”：

落地建议：新手可从“会话+长期+RAG知识”三个基础模块入手，快速验证智能体功能；中大型项目需整合五大模块，重点优化Token控制、记忆精度和协同逻辑，确保智能体适配生产环境，实现“真正的智能”。