智能体的记忆(Memory)系统

智能体的记忆系统是其实现持续学习和情境理解的核心组件。下面我将全面介绍智能体记忆的主要类型、技术原理和实现方式。

一、记忆系统的基本分类

1. 按存储时长分类

2. 按内容类型分类

• ​​情景记忆​​：具体事件和经历

• ​​语义记忆​​：事实和知识

• ​​程序记忆​​：技能和操作流程

二、主流记忆类型详解

1. 对话记忆 (Conversation Memory)

from langchain.memory import ConversationBufferMemory

# 基本对话记忆实现
memory = ConversationBufferMemory()
memory.save_context({"input": "你好"}, {"output": "你好！我是AI助手"})
memory.save_context({"input": "今天天气如何"}, {"output": "请告诉我您所在的城市"})

print(memory.load_memory_variables({}))
# 输出: {'history': 'Human: 你好\nAI: 你好！我是AI助手\nHuman: 今天天气如何\nAI: 请告诉我您所在的城市'}

​​变体类型​​：

• ConversationBufferWindowMemory：仅保留最近N轮对话
• ConversationSummaryMemory：存储对话摘要而非全文

• ConversationKGMemory：基于知识图谱存储对话实体关系

2. 向量记忆 (Vector Memory)

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings

# 创建向量记忆存储
embeddings = OpenAIEmbeddings()
texts = ["苹果是水果", "华为是手机品牌", "Python是编程语言"]
vector_memory = FAISS.from_texts(texts, embeddings)

# 相似性检索
result = vector_memory.similarity_search("科技公司", k=1)
print(result[0].page_content)  # 输出: 华为是手机品牌

​​适用场景​​：

• 大规模知识检索

• 模糊记忆查询

• 跨模态记忆

3. 外部记忆 (External Memory)

from langchain.memory import RedisChatMessageHistory

# 连接Redis数据库存储记忆
message_history = RedisChatMessageHistory(
    session_id="user123",
    url="redis://localhost:6379/0"
)

message_history.add_user_message("我喜欢科幻电影")
message_history.add_ai_message("已记录您的偏好")

# 后续会话可以读取
print(message_history.messages)

​​常用外部存储​​：

• SQL数据库（PostgreSQL/MySQL）

• 文档数据库（MongoDB）

• 键值存储（Redis）

• 文件系统

4. 情景记忆 (Episodic Memory)

from datetime import datetime
from pydantic import BaseModel

class MemoryItem(BaseModel):
    content: str
    timestamp: datetime
    importance: float

class EpisodicMemory:
    def __init__(self):
        self.memories = []
    
    def add(self, content: str, importance: float = 0.5):
        self.memories.append(
            MemoryItem(
                content=content,
                timestamp=datetime.now(),
                importance=importance
            )
        )
    
    def recall_recent(self, n: int = 3):
        return sorted(self.memories, 
                    key=lambda x: x.timestamp,
                    reverse=True)[:n]

# 使用示例
ep_memory = EpisodicMemory()
ep_memory.add("用户喜欢蓝色", importance=0.7)
ep_memory.add("用户是程序员", importance=0.9)
print([m.content for m in ep_memory.recall_recent()])

5. 程序记忆 (Procedural Memory)

class SkillLibrary:
    def __init__(self):
        self.skills = {
            "math": {
                "description": "数学计算能力",
                "function": self._math_processor
            },
            "translation": {
                "description": "中英翻译能力",
                "function": self._translate
            }
        }
    
    def _math_processor(self, expression: str) -> float:
        try:
            return eval(expression)
        except:
            return None
    
    def _translate(self, text: str) -> str:
        # 实际实现会调用翻译API
        return f"Translated: {text}"
    
    def execute(self, skill_name: str, input_data: str):
        if skill_name in self.skills:
            return self.skills[skill_name]["function"](input_data)
        return None

# 使用示例
skill_memory = SkillLibrary()
print(skill_memory.execute("math", "3*(4+5)"))  # 输出: 27