Agent的记忆系统:短期记忆、长期记忆、向量记忆,到底怎么用? 💾

导读:你有没有遇到过这种情况——Agent聊着聊着就"忘了"之前说过的话?或者每次对话都从零开始,完全不记得你的偏好?这就是记忆系统没做好!今天我们把Agent的记忆系统彻底讲透,让你的Agent从"金鱼记忆"变成"过目不忘"!🐟→🧠


一、为什么记忆对Agent如此重要?🤔

1.1 没有记忆的Agent有多蠢

场景 无记忆 有记忆
第二次对话 “你是谁?” “小明你好!上次我们讨论了项目方案…”
用户偏好 每次都要重新问 “还是按你喜欢的简洁风格来?”
错误纠正 同样的错犯两次 “上次这个问题我处理得不好,这次我换个方法”
长任务 做到一半忘了前面 完整记住整个任务的进度

💡 没有记忆的Agent就像一条金鱼——每次都是"初次见面"。

1.2 记忆的三种类型

💾 Agent记忆系统

类比:工作台

类比:笔记本

类比:图书馆

📌 短期记忆
Short-term Memory

📚 长期记忆
Long-term Memory

🧬 向量记忆
Vector Memory

当前对话的消息列表
用完即弃

用户偏好、历史经验
持久存储

语义索引的知识库
按需检索


二、短期记忆:Agent的"工作台" 📌

2.1 什么是短期记忆?

短期记忆就是当前对话的消息列表。它让Agent记住"我们刚才聊了什么"。

特征 说明
📍 存储位置 内存中(消息列表)
生命周期 单次会话
📏 容量限制 受模型上下文窗口限制
🔄 更新方式 每轮对话自动追加

2.2 滑动窗口机制

对话越来越长,不可能把所有消息都发给模型(太贵且超出上下文限制)。解决方案是滑动窗口

from typing import Annotated
from langgraph.graph import MessagesState
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage

# 只保留最近20轮对话(40条消息)
MAX_MESSAGES = 40

def _windowed_messages(old, new):
    """滑动窗口:只保留最近的消息"""
    all_messages = add_messages(old, new)
    return all_messages[-MAX_MESSAGES:]  # 只保留最后N条

class AgentState(MessagesState):
    messages: Annotated[list[AnyMessage], _windowed_messages]

2.3 滑动窗口可视化

对话历史(共100条消息):
[msg1] [msg2] ... [msg58] [msg59] [msg60] ... [msg99] [msg100]
                       ↑                              ↑
                   窗口起点                          窗口终点
                   
发送给模型的:[msg61] [msg62] ... [msg100] (最近40条)

2.4 不同窗口策略对比

策略 原理 优点 缺点 适用场景
📏 固定窗口 只保留最近N条 简单高效 可能丢失重要上下文 日常对话
🎯 摘要压缩 旧消息压缩为摘要 保留关键信息 摘要可能丢失细节 长对话
🧠 重要性筛选 按重要性保留消息 精准保留 需要额外判断逻辑 专业场景

三、长期记忆:Agent的"笔记本" 📚

3.1 什么是长期记忆?

长期记忆存储跨会话的信息——用户偏好、历史经验、学到的知识。

特征 说明
📍 存储位置 数据库/文件
生命周期 永久(直到主动删除)
📏 容量限制 几乎无限
🔄 更新方式 Agent主动存储

3.2 长期记忆的实现

import json
import os

class LongTermMemory:
    """长期记忆管理器"""
    
    def __init__(self, storage_path="memory/long_term.json"):
        self.storage_path = storage_path
        self.memory = self._load()
    
    def _load(self):
        """从文件加载记忆"""
        if os.path.exists(self.storage_path):
            with open(self.storage_path, 'r') as f:
                return json.load(f)
        return {"user_preferences": {}, "experiences": [], "facts": {}}
    
    def _save(self):
        """保存记忆到文件"""
        os.makedirs(os.path.dirname(self.storage_path), exist_ok=True)
        with open(self.storage_path, 'w') as f:
            json.dump(self.memory, f, ensure_ascii=False, indent=2)
    
    def remember_fact(self, key: str, value: str):
        """记住一个事实"""
        self.memory["facts"][key] = value
        self._save()
    
    def recall_fact(self, key: str) -> str:
        """回忆一个事实"""
        return self.memory["facts"].get(key, "我不记得这个")
    
    def remember_preference(self, user_id: str, pref: dict):
        """记住用户偏好"""
        self.memory["user_preferences"][user_id] = pref
        self._save()
    
    def remember_experience(self, experience: str):
        """记住一条经验"""
        self.memory["experiences"].append(experience)
        # 只保留最近100条经验
        self.memory["experiences"] = self.memory["experiences"][-100:]
        self._save()
    
    def get_all_memories(self) -> dict:
        """获取所有记忆"""
        return self.memory

# 使用示例
ltm = LongTermMemory()
ltm.remember_fact("用户职业", "程序员")
ltm.remember_preference("user_001", {"language": "中文", "style": "简洁"})
ltm.remember_experience("上次SQL查询超时,需要加LIMIT")

3.3 长期记忆在Agent中的使用

from langchain.tools import tool

# 将长期记忆封装为工具,让Agent可以主动存取
@tool
def remember(key: str, value: str) -> str:
    """记住一条信息到长期记忆。当用户说'记住这个'或提供了重要信息时使用。
    
    Args:
        key: 记忆的键(简短描述)
        value: 记忆的内容
    """
    ltm = LongTermMemory()
    ltm.remember_fact(key, value)
    return f"✅ 已记住:{key} = {value}"

@tool
def recall(query: str) -> str:
    """从长期记忆中检索信息。当需要回忆之前的对话或用户信息时使用。
    
    Args:
        query: 要检索的内容描述
    """
    ltm = LongTermMemory()
    facts = ltm.get_all_memories()["facts"]
    # 简单的关键词匹配
    results = {k: v for k, v in facts.items() if query.lower() in k.lower() or query.lower() in v.lower()}
    if results:
        return f"📚 找到相关记忆:\n" + "\n".join([f"- {k}: {v}" for k, v in results.items()])
    return "没有找到相关记忆"

四、向量记忆:Agent的"图书馆" 🧬

4.1 什么是向量记忆?

向量记忆使用向量数据库存储信息,通过语义相似度检索。它让Agent能够"模糊回忆"——不需要精确的关键词,只要语义相近就能找到。

特征 说明
📍 存储方式 文本→向量→向量数据库
🔍 检索方式 语义相似度匹配
💡 核心优势 “模糊记忆”——意思接近就能找到

4.2 向量记忆原理图

文本信息

Embedding模型

向量表示
0.12, -0.34, 0.56...

向量数据库

查询

Embedding模型

查询向量

相似度搜索

最相关的记忆

4.3 向量记忆实现

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document

class VectorMemory:
    """向量记忆系统"""
    
    def __init__(self, collection_name="agent_memory"):
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = Chroma(
            collection_name=collection_name,
            embedding_function=self.embeddings,
            persist_directory="./chroma_db"
        )
    
    def store(self, text: str, metadata: dict = None):
        """存储一条记忆"""
        doc = Document(
            page_content=text,
            metadata=metadata or {}
        )
        self.vectorstore.add_documents([doc])
    
    def recall(self, query: str, k: int = 5) -> list:
        """根据查询检索相关记忆"""
        results = self.vectorstore.similarity_search(query, k=k)
        return [{"content": doc.page_content, "metadata": doc.metadata} 
                for doc in results]
    
    def recall_with_score(self, query: str, k: int = 5) -> list:
        """检索记忆并返回相似度分数"""
        results = self.vectorstore.similarity_search_with_score(query, k=k)
        return [{"content": doc.page_content, "score": score, "metadata": doc.metadata}
                for doc, score in results]

# 使用示例
vm = VectorMemory()

# 存储记忆
vm.store("用户喜欢简洁的回答风格", {"type": "preference", "timestamp": "2025-06-15"})
vm.store("项目使用Python + FastAPI技术栈", {"type": "project_info"})
vm.store("上次部署遇到了Docker内存不足的问题", {"type": "experience"})

# 检索记忆
results = vm.recall("用户偏好")
# 会找到"用户喜欢简洁的回答风格"(语义最接近)

五、三种记忆的协同工作 🎯

5.1 协同架构图

当前对话上下文

用户偏好/事实

相关经验/知识

用户输入

Agent思考

需要什么记忆?

📌 短期记忆

📚 长期记忆

🧬 向量记忆

综合记忆

生成回答

值得记住吗?

存入长期/向量记忆

仅保留在短期记忆

5.2 完整代码:三层记忆Agent

import os
import json
from typing import Annotated
from langchain.agents import create_react_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langgraph.graph import MessagesState
from langgraph.graph.message import add_messages
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import AnyMessage

# ===== 短期记忆:滑动窗口 =====
MAX_MESSAGES = 40
def _windowed_messages(old, new):
    return add_messages(old, new)[-MAX_MESSAGES:]

class AgentState(MessagesState):
    messages: Annotated[list[AnyMessage], _windowed_messages]

# ===== 长期记忆工具 =====
MEMORY_FILE = "memory/agent_memory.json"

def _load_memory():
    if os.path.exists(MEMORY_FILE):
        with open(MEMORY_FILE, 'r') as f:
            return json.load(f)
    return {}

def _save_memory(data):
    os.makedirs(os.path.dirname(MEMORY_FILE), exist_ok=True)
    with open(MEMORY_FILE, 'w') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

@tool
def save_memory(key: str, value: str) -> str:
    """保存信息到长期记忆。当用户说'记住这个'或提供了重要的个人信息时使用。
    Args:
        key: 信息类别,如'用户姓名'、'项目信息'
        value: 具体内容
    """
    memory = _load_memory()
    memory[key] = value
    _save_memory(memory)
    return f"已记住:{key} = {value}"

@tool
def load_memory(key: str) -> str:
    """从长期记忆中检索信息。
    Args:
        key: 要检索的信息类别
    """
    memory = _load_memory()
    if key in memory:
        return f"{key}: {memory[key]}"
    # 模糊搜索
    results = {k: v for k, v in memory.items() if key.lower() in k.lower()}
    if results:
        return "\n".join([f"{k}: {v}" for k, v in results.items()])
    return "没有找到相关记忆"

# ===== 构建Agent =====
tools = [save_memory, load_memory]

agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=tools,
    prompt="""你是一个有记忆的AI助手。
    
记忆系统说明:
- 短期记忆:自动保持当前对话上下文
- 长期记忆:使用save_memory保存重要信息,使用load_memory检索

当用户提供个人信息或重要事项时,主动使用save_memory保存。
当需要回忆之前的信息时,使用load_memory检索。""",
    checkpointer=MemorySaver(),  # 短期记忆
    state_schema=AgentState,
)

六、记忆系统的性能与成本 📊

6.1 三种记忆的成本对比

记忆类型 存储成本 检索速度 Token消耗 实现难度
📌 短期记忆 低(内存) 极快 高(全部发给模型)
📚 长期记忆 低(文件/DB) 低(按需加载) ⭐⭐
🧬 向量记忆 中(向量DB) 低(只检索相关部分) ⭐⭐⭐

6.2 Token消耗优化策略

策略 效果 实现方式
📏 滑动窗口 Token减少60% 只保留最近N条消息
📝 摘要压缩 Token减少80% 旧消息压缩为一段摘要
🔍 按需检索 Token减少90% 只检索与当前问题相关的记忆
🗑️ 去重 Token减少20% 删除重复的消息

七、本期小结 📝

记忆类型 类比 存储内容 生命周期 实现方式
📌 短期记忆 工作台 当前对话 单次会话 消息列表+滑动窗口
📚 长期记忆 笔记本 用户偏好、事实 永久 JSON文件/数据库
🧬 向量记忆 图书馆 经验、知识 永久 向量数据库

🔥 记忆系统是Agent从"工具"进化为"伙伴"的关键。 没有记忆的Agent只是一个函数调用,有了记忆的Agent才是一个真正的"智能体"。


📢 下期预告:《RAG+Agent王炸组合:让你的智能体拥有"过目不忘"的知识库》—— 向量记忆进阶版!用RAG让Agent拥有一个专属知识库!📚


📌 三连走起!有记忆的Agent,才是真正的智能体! 💪

📚 专栏第13/24期,记忆与规划篇进行中…

作者:高炉炼铁智能化技术研究者,专注钢铁冶金与人工智能 交叉领域。

👍 如果觉得有帮助,请点赞、收藏、转发!
版权归作者所有,未经许可请勿抄袭,套用,商用(或其它具有利益性行为)
🔔 关注专栏,不错过后续精彩内容

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐