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论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。