Skip to content
🔗 分享本题
查看我的学习进度 →

Agent 工作记忆、短期记忆和长期记忆的分层读写检索与遗忘机制

🧠 记忆锚点:工作记忆管任务,短期保会话,长期跨会话;检索与遗忘同样重要。

💡 答案要点

记忆系统 = Agent的"大脑存储",决定能否长期陪伴用户

记忆分类

类型存储周期容量实现用途
短期记忆单次会话小(受Context Window限制)对话历史保持对话连贯
长期记忆跨会话永久大(无限)向量数据库用户偏好/历史事实
工作记忆任务期间临时变量中间计算结果

短期记忆实现

方案1: 滑动窗口

python
class ShortTermMemory:
    def __init__(self, max_turns=5):
        self.messages = []
        self.max_turns = max_turns

    def add(self, role, content):
        self.messages.append({"role": role, "content": content})

        # 保留最近N轮对话
        if len(self.messages) > self.max_turns * 2:  # *2因为每轮有user+assistant
            self.messages = self.messages[-(self.max_turns * 2):]

    def get_context(self):
        return self.messages

# 问题: 丢失早期信息

方案2: 分层管理(推荐⭐)

展开 Python 代码示例(36 行)
python
class HierarchicalMemory:
    def __init__(self):
        self.recent = []      # 最近3轮完整保留
        self.summary = ""     # 历史摘要

    def add_turn(self, user_msg, ai_msg):
        self.recent.append({"user": user_msg, "ai": ai_msg})

        # 超过3轮,摘要最早的
        if len(self.recent) > 3:
            old = self.recent.pop(0)

            # 用LLM摘要
            summary_chunk = llm.summarize(
                f"用户: {old['user']}\n助手: {old['ai']}"
            )
            self.summary += summary_chunk + "\n"

    def get_context(self):
        context = []

        # 历史摘要
        if self.summary:
            context.append({"role": "system", "content": f"历史: {self.summary}"})

        # 最近3轮完整对话
        for turn in self.recent:
            context.append({"role": "user", "content": turn["user"]})
            context.append({"role": "assistant", "content": turn["ai"]})

        return context

# 效果:
# 10轮对话,token消耗: 摘要(200) + 最近3轮(1500) = 1700 tokens
# vs 全保留: 5000 tokens
# 节省: 66%

长期记忆实现

核心: 向量数据库

展开 Python 代码示例(52 行)
python
from langchain.vectorstores import Qdrant
from langchain.embeddings import OpenAIEmbeddings

class LongTermMemory:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings()
        self.vectordb = Qdrant(
            collection_name="user_memory",
            embedding_function=self.embeddings
        )

    def remember(self, key, value, metadata=None):
        """存储长期记忆"""
        self.vectordb.add_texts(
            texts=[value],
            metadatas=[{
                "key": key,
                "timestamp": time.time(),
                **(metadata or {})
            }]
        )

    def recall(self, query, k=5):
        """检索相关记忆"""
        results = self.vectordb.similarity_search(query, k=k)
        return [doc.page_content for doc in results]

    def forget(self, key):
        """删除记忆"""
        self.vectordb.delete(filter={"key": key})

# 使用
memory = LongTermMemory()

# 存储用户偏好
memory.remember(
    key="food_preference",
    value="用户喜欢吃川菜,不吃香菜",
    metadata={"category": "preference"}
)

# 存储历史事实
memory.remember(
    key="birthday",
    value="用户生日是1990年5月20日",
    metadata={"category": "fact"}
)

# 3个月后,检索记忆
query = "用户吃什么?"
memories = memory.recall(query, k=3)
# 返回: ["用户喜欢吃川菜,不吃香菜", ...]

混合记忆策略

问题: 如何决定什么存长期,什么存短期?

展开 Python 代码示例(102 行)
python
class HybridMemoryManager:
    def __init__(self):
        self.short_term = HierarchicalMemory()
        self.long_term = LongTermMemory()

    def add_interaction(self, user_msg, ai_msg):
        # 1. 短期记忆:直接存
        self.short_term.add_turn(user_msg, ai_msg)

        # 2. 判断是否值得长期存储
        if self.is_important(user_msg, ai_msg):
            # 提取关键信息
            key_info = self.extract_key_info(user_msg, ai_msg)

            # 存入长期记忆
            self.long_term.remember(
                key=f"conv_{time.time()}",
                value=key_info
            )

    def is_important(self, user_msg, ai_msg):
        """判断是否重要"""
        # 规则1: 包含用户偏好
        if any(kw in user_msg.lower() for kw in ["我喜欢", "我不喜欢", "我的"]):
            return True

        # 规则2: 包含事实信息
        if any(kw in user_msg for kw in ["是", "叫", "生日", "地址"]):
            return True

        # 规则3: 用LLM判断
        prompt = f"""
        判断以下对话是否包含值得长期记忆的信息(用户偏好/事实/重要决策):

        用户: {user_msg}
        助手: {ai_msg}

        回答: 是/否
        """
        decision = llm.generate(prompt).strip()
        return decision == "是"

    def extract_key_info(self, user_msg, ai_msg):
        """提取关键信息"""
        prompt = f"""
        从对话中提取值得长期记忆的关键信息:

        用户: {user_msg}
        助手: {ai_msg}

        关键信息(一句话):
        """
        return llm.generate(prompt).strip()

    def get_full_context(self, current_query):
        """获取完整上下文"""
        # 1. 短期记忆
        short = self.short_term.get_context()

        # 2. 检索相关长期记忆
        long = self.long_term.recall(current_query, k=3)

        # 3. 合并
        context = []

        # 长期记忆作为背景
        if long:
            context.append({
                "role": "system",
                "content": f"用户背景信息:\n" + "\n".join(long)
            })

        # 短期记忆作为对话历史
        context.extend(short)

        return context

# 实战示例
manager = HybridMemoryManager()

# 第1次对话
manager.add_interaction(
    user_msg="我叫张三,是个程序员",
    ai_msg="你好张三!很高兴认识你。"
)
# → 长期记忆存储: "用户叫张三,职业是程序员"

# 第2次对话
manager.add_interaction(
    user_msg="今天天气真好",
    ai_msg="是啊,适合出去走走。"
)
# → 仅短期记忆,不重要

# 1个月后,第100次对话
context = manager.get_full_context("给我推荐适合程序员的书")
# context包含:
# - 长期记忆: "用户叫张三,职业是程序员"
# - 短期记忆: 最近3轮对话

response = llm.generate(context + [{"role": "user", "content": "给我推荐适合程序员的书"}])
# LLM能结合长期记忆(职业)给出个性化推荐

记忆索引优化

问题: 向量检索不准,检索到无关记忆

优化: 混合索引

展开 Python 代码示例(64 行)
python
class EnhancedMemory:
    def __init__(self):
        self.vectordb = Qdrant(...)
        self.metadata_index = {}  # 元数据索引

    def remember(self, key, value, category, tags):
        # 1. 向量存储
        self.vectordb.add_texts(
            texts=[value],
            metadatas=[{
                "key": key,
                "category": category,
                "tags": tags,
                "timestamp": time.time()
            }]
        )

        # 2. 元数据索引
        self.metadata_index[key] = {
            "category": category,
            "tags": tags
        }

    def recall(self, query, category=None, tags=None, k=5):
        # 构建过滤条件
        filter_dict = {}
        if category:
            filter_dict["category"] = category
        if tags:
            filter_dict["tags"] = {"$in": tags}

        # 向量检索+元数据过滤
        results = self.vectordb.similarity_search(
            query,
            k=k,
            filter=filter_dict
        )

        return results

# 使用
memory = EnhancedMemory()

memory.remember(
    key="food1",
    value="用户喜欢吃川菜",
    category="preference",
    tags=["food", "cuisine"]
)

memory.remember(
    key="work1",
    value="用户在字节跳动工作",
    category="fact",
    tags=["job", "company"]
)

# 只检索食物偏好
food_memories = memory.recall(
    query="吃什么",
    category="preference",
    tags=["food"]
)
# 不会检索到工作信息

面试话术:

示例表达(仅在能用本人经历或可复现实验佐证时使用): "Agent记忆分短期和长期。短期用分层管理,最近3轮保留原文+历史摘要,节省66% token。长期用向量数据库,存用户偏好和事实,用LLM判断重要性决定是否存储。检索时混合向量+元数据过滤,避免无关记忆。实测3个月后,Agent还记得用户职业,推荐更个性化。"