Agent记忆的三层架构 Agent需要像人类一样管理不同类型的记忆：
工作记忆：当前对话的上下文（短期） 情景记忆：过去交互的具体经历（中期） 语义记忆：从交互中提炼的知识（长期） 工作记忆：对话窗口管理 from collections import deque class WorkingMemory: def __init__(self, max_messages=20, max_tokens=4096): self.messages = deque(maxlen=max_messages) self.max_tokens = max_tokens self.token_counter = 0 def add(self, role, content): tokens = len(content) // 4 self.messages.append({"role": role, "content": content, "tokens": tokens}) self.token_counter += tokens # 超出token限制时移除最早的消息 while self.token_counter > self.max_tokens and len(self.messages) > 2: old = self.messages.popleft() self.token_counter -= old["tokens"] def get_context(self): return list(self.messages) def summarize_old_context(self): """当记忆过长时，摘要旧对话""" if len(self.messages) < 10: return old_messages = list(self.messages)[:len(self.messages)//2] summary = await self.summarize(old_messages) # 用摘要替换旧消息 for _ in range(len(old_messages)): self.messages.popleft() self.messages.appendleft({"role": "system", "content": f"之前的对话摘要：{summary}"}) 情景记忆：交互历史存储 class EpisodicMemory: def __init__(self, vector_store, llm): self.store = vector_store # 向量数据库 self.llm = llm async def save_interaction(self, user_id, user_message, assistant_response, metadata=None): """保存一次交互记录""" interaction = { "user_id": user_id, "user_message": user_message, "assistant_response": assistant_response, "timestamp": datetime.now().isoformat(), "metadata": metadata or {}, } # 生成摘要用于检索 summary = await self.llm.generate( f"用一句话概括这次交互：\n用户：{user_message}\n助手：{assistant_response}" ) # 向量化并存储 embedding = await self.llm.embed(summary) await self.store.add( id=generate_uuid(), embedding=embedding, document=json.dumps(interaction, ensure_ascii=False), metadata={"user_id": user_id, "summary": summary} ) async def recall(self, user_id, query, top_k=5): """检索相关的历史交互""" query_embedding = await self.llm.embed(query) results = await self.store.search( query_embedding, filter={"user_id": user_id}, top_k=top_k ) return [json.loads(r.document) for r in results] 语义记忆：知识图谱 class SemanticMemory: def __init__(self, graph_store): self.graph = graph_store # 图数据库（如Neo4j） async def learn(self, user_id, fact): """从交互中提取并存储知识""" # 使用LLM提取结构化知识 extracted = await self.extract_facts(fact) for fact_data in extracted: await self.graph.add_triple( subject=fact_data["subject"], predicate=fact_data["predicate"], object=fact_data["object"], metadata={"user_id": user_id, "source": "interaction"} ) async def query(self, user_id, entity): """查询关于某实体的知识""" triples = await self.graph.query( "MATCH (s)-[p]->(o) WHERE s.name = $entity RETURN s, p, o", {"entity": entity} ) knowledge = [] for triple in triples: knowledge.append(f"{triple['s']} {triple['p']} {triple['o']}") return knowledge async def extract_facts(self, text): """从文本中提取三元组""" prompt = f"""从以下文本中提取知识三元组（主语-谓语-宾语）： 文本：{text} 输出JSON数组： [{{"subject": "...", "predicate": "...", "object": "..."}}]""" result = await self.llm.generate(prompt) return json.loads(result) 记忆整合 class AgentMemorySystem: def __init__(self, working, episodic, semantic): self.working = working # 工作记忆 self.episodic = episodic # 情景记忆 self.semantic = semantic # 语义记忆 async def build_context(self, user_id, current_message): """构建完整的记忆上下文""" context_parts = [] # 1. 工作记忆（当前对话） working_ctx = self.working.get_context() context_parts.append({"type": "working", "messages": working_ctx}) # 2. 情景记忆（相关历史交互） episodic_results = await self.episodic.recall( user_id, current_message, top_k=3 ) if episodic_results: context_parts.append({ "type": "episodic", "memories": [r["summary"] for r in episodic_results] }) # 3. 语义记忆（相关知识） entities = await self.extract_entities(current_message) for entity in entities: knowledge = await self.semantic.query(user_id, entity) if knowledge: context_parts.append({ "type": "semantic", "entity": entity, "facts": knowledge }) return self.format_context(context_parts) def format_context(self, parts): """格式化记忆上下文""" context = "" for part in parts: if part["type"] == "working": context += "## 当前对话\n" for msg in part["messages"]: context += f"{msg['role']}: {msg['content']}\n" elif part["type"] == "episodic": context += "\n## 相关历史\n" for mem in part["memories"]: context += f"- {mem}\n" elif part["type"] == "semantic": context += f"\n## 关于{part['entity']}的知识\n" for fact in part["facts"]: context += f"- {fact}\n" return context 遗忘机制 class ForgettingMechanism: def __init__(self, decay_rate=0.01): self.decay_rate = decay_rate async def decay(self, memory_store): """时间衰减：降低旧记忆的重要性""" now = datetime.now() memories = await memory_store.get_all() for mem in memories: age_days = (now - mem["timestamp"]).days importance = mem.get("importance", 1.0) importance *= (1 - self.decay_rate) ** age_days if importance < 0.1: await memory_store.delete(mem["id"]) else: await memory_store.update(mem["id"], importance=importance) async def consolidate(self, memory_store): """记忆整合：将频繁出现的情景记忆转为语义记忆""" # 找出高频出现的事实 memories = await memory_store.get_all(min_importance=0.5) # 聚类相似记忆 clusters = self.cluster_memories(memories) for cluster in clusters: if len(cluster) >= 3: # 出现3次以上的模式转为知识 summary = await self.summarize_cluster(cluster) await self.semantic_memory.learn(summary) # 降低原始记忆的重要性 for mem in cluster: await memory_store.update(mem["id"], importance=mem["importance"] * 0.3) 存储选择 记忆类型 推荐存储 特点 工作记忆 内存（Redis） 快速读写，无需持久化 情景记忆 向量数据库 语义检索，按相似度召回 语义记忆 图数据库 关系查询，知识推理 结语 Agent记忆系统是构建长期智能助手的核心基础设施。工作记忆处理当前对话，情景记忆保存交互历史，语义记忆积累结构化知识。三层记忆的协同工作加上遗忘和整合机制，让Agent像人类一样"记得住该记的，忘得掉该忘的"。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
...
