AI Agent 从入门到封神:24 讲打造你的超级智能体~系列文章13:Agent的记忆系统:短期记忆、长期记忆、向量记忆,到底怎么用?
·
Agent的记忆系统:短期记忆、长期记忆、向量记忆,到底怎么用? 💾
导读:你有没有遇到过这种情况——Agent聊着聊着就"忘了"之前说过的话?或者每次对话都从零开始,完全不记得你的偏好?这就是记忆系统没做好!今天我们把Agent的记忆系统彻底讲透,让你的Agent从"金鱼记忆"变成"过目不忘"!🐟→🧠
一、为什么记忆对Agent如此重要?🤔
1.1 没有记忆的Agent有多蠢
| 场景 | 无记忆 | 有记忆 |
|---|---|---|
| 第二次对话 | “你是谁?” | “小明你好!上次我们讨论了项目方案…” |
| 用户偏好 | 每次都要重新问 | “还是按你喜欢的简洁风格来?” |
| 错误纠正 | 同样的错犯两次 | “上次这个问题我处理得不好,这次我换个方法” |
| 长任务 | 做到一半忘了前面 | 完整记住整个任务的进度 |
💡 没有记忆的Agent就像一条金鱼——每次都是"初次见面"。
1.2 记忆的三种类型
二、短期记忆: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 向量记忆原理图
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 协同架构图
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期,记忆与规划篇进行中…
作者:高炉炼铁智能化技术研究者,专注钢铁冶金与人工智能 交叉领域。
👍 如果觉得有帮助,请点赞、收藏、转发!
版权归作者所有,未经许可请勿抄袭,套用,商用(或其它具有利益性行为)。
🔔 关注专栏,不错过后续精彩内容
更多推荐

所有评论(0)