写在前面
当你和一个 AI Agent 对话时，你是否想过：它真的"记住"我们之前聊过什么了吗？
在单轮对话中，LLM 只能看到当前输入的上下文。但在真实的应用场景中——无论是智能客服、个人助手还是研究工具——Agent 需要记住用户的历史偏好、对话上下文、积累的知识，才能提供真正个性化的服务。
本文将深入剖析 Agent 记忆系统的设计原理，实现从工作记忆（Working Memory）→ 短期记忆（Short-Term Memory）→ 长期记忆（Long-Term Memory） 的完整三层架构，并提供可运行的代码实现。
一、为什么 Agent 需要记忆系统？
1.1 人类的记忆启示
让我们先看看人类是如何组织记忆的：
┌─────────────────────────────────────────────────────────────────────┐
│ 人类记忆三层模型 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 长期记忆 (LTM) │ │
│ │ ┌─────────────────────────────────────────────────────┐ │ │
│ │ │ 语义记忆：概念、知识、事实 │ │ │
│ │ │ 程序记忆：技能、习惯、自动行为 │ │ │
│ │ │ 情景记忆：个人经历、特定事件 │ │ │
│ │ └─────────────────────────────────────────────────────┘ │ │
│ │ 容量：无限 持久性：数十年 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ ▲ │
│ │ 遗忘/巩固 │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 短期记忆 (STM) │ │
│ │ ┌─────────────────────────────────────────────────────┐ │ │
│ │ │ 工作记忆：当前任务的临时信息 │ │ │
│ │ │ 会话记忆：当前会话的对话历史 │ │ │
│ │ └─────────────────────────────────────────────────────┘ │ │
│ │ 容量：7±2 项 持续时间：秒到分钟 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ ▲ │
│ │ 选择性编码 │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 工作记忆 (WM) │ │
│ │ ┌─────────────────────────────────────────────────────┐ │ │
│ │ │ 当前注意焦点：正在处理的信息 │ │ │
│ │ └─────────────────────────────────────────────────────┘ │ │
│ │ 容量：4 项 持续时间：秒级 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘1.2 Agent 记忆系统的必要性
Agent 同样需要类似的记忆层次结构：
没有记忆系统的 Agent：
用户：这个药每天吃几次？
Agent：抱歉，我不知道"这个药"是什么。
用户：就是刚才说的那个
Agent：抱歉，我无法关联之前的对话。有记忆系统的 Agent：
用户：这个药每天吃几次？
Agent：根据处方，这个药每天吃3次，饭后服用。
用户：就是刚才说的那个
Agent：根据之前的对话，这是阿司匹林，用于缓解头痛。
您提到您有轻微头痛，建议按说明书服用。1.3 记忆系统的核心挑战
构建 Agent 记忆系统面临三大挑战：
容量限制：LLM 的上下文窗口是有限的（即使是最新的模型也有 100K-1M token 的限制）
检索效率：如何在海量记忆中找到最相关的部分？
信息衰减：如何让重要的记忆保持"新鲜"，不重要的记忆自然遗忘？
二、Agent 记忆的三层模型
2.1 架构全景图
┌─────────────────────────────────────────────────────────────────────┐
│ Agent 记忆系统三层架构 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 用户输入 │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 工作记忆层 (Working Memory) │ │
│ │ ┌─────────────────────────────────────────────────────┐ │ │
│ │ │ • 当前轮次的用户输入 │ │ │
│ │ │ • 当前轮次的 Agent 输出 │ │ │
│ │ │ • 任务状态/进度 │ │ │
│ │ └─────────────────────────────────────────────────────┘ │ │
│ │ 生命周期：单轮对话 │ │
│ │ 存储：内存（Python 变量） │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ │ 编码/压缩 │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 短期记忆层 (Short-Term Memory) │ │
│ │ ┌─────────────────────────────────────────────────────┐ │ │
│ │ │ • 当前会话的完整对话历史 │ │ │
│ │ │ • 会话级用户偏好 │ │ │
│ │ │ • 任务相关的中间结果 │ │ │
│ │ └─────────────────────────────────────────────────────┘ │ │
│ │ 生命周期：单次会话（可配置 TTL） │ │
│ │ 存储：Redis / 内存 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ │ 总结/提取 │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 长期记忆层 (Long-Term Memory) │ │
│ │ ┌─────────────────────────────────────────────────────┐ │ │
│ │ │ • 用户永久偏好（语言风格、关注领域） │ │ │
│ │ │ • 重要交互记录 │ │ │
│ │ │ • 领域知识/技能 │ │ │
│ │ └─────────────────────────────────────────────────────┘ │ │
│ │ 生命周期：永久 │ │
│ │ 存储：向量数据库 + 知识图谱 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘2.2 各层职责与特点
三、工作记忆实现：消息列表管理
3.1 工作记忆的核心概念
工作记忆是 Agent 处理的"前台"，存储当前任务相关的信息。在 LangGraph 中，这通常通过 messages 字段实现。
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langgraph.graph import StateGraph, START, END
class WorkingMemoryState(TypedDict):
"""
工作记忆状态定义
工作记忆的核心是 messages 列表，
使用 add_messages reducer 自动处理消息的追加和更新
"""
# 消息历史，使用 add_messages 自动合并
messages: Annotated[list, add_messages]
# 当前任务状态
current_task: str
task_status: str # pending/in_progress/completed
# 上下文摘要（用于快速检索）
context_summary: str
def chat_node(state: WorkingMemoryState) -> dict:
"""
聊天节点
每次调用都会将新的消息追加到 messages 中
"""
from langchain_openai import ChatOpenAI
import os
llm = ChatOpenAI(model="gpt-4", api_key=os.getenv("OPENAI_API_KEY"))
# 调用 LLM，传入消息历史
response = llm.invoke(state["messages"])
# 返回更新：add_messages 会自动将新消息追加到列表
return {
"messages": [response],
"task_status": "in_progress"
}
# 构建简单的聊天工作流
workflow = StateGraph(WorkingMemoryState)
workflow.add_node("chat", chat_node)
workflow.add_edge(START, "chat")
workflow.add_edge("chat", END)
app = workflow.compile()3.2 消息的智能管理
工作记忆不仅仅是追加消息，还需要智能管理：
class MessageManager:
"""
消息管理器
负责工作记忆的智能管理：
1. 消息去重
2. 长度控制
3. 角色聚合
"""
MAX_MESSAGES = 20 # 最大消息数
MAX_TOTAL_TOKENS = 8000 # 最大 token 数估算
@staticmethod
def trim_messages(messages: list, max_messages: int = None) -> list:
"""
修剪消息列表
策略：
1. 保留最近 N 条消息
2. 如果过长，从中间开始删除
"""
if max_messages is None:
max_messages = MessageManager.MAX_MESSAGES
if len(messages) <= max_messages:
return messages
# 策略：保留系统消息、最近的消息，以及中间的一些历史
system_messages = [m for m in messages if hasattr(m, 'type') and m.type == 'system']
other_messages = [m for m in messages if hasattr(m, 'type') and m.type != 'system']
# 保留最近的消息
recent = other_messages[-(max_messages - len(system_messages)):]
return system_messages + recent
@staticmethod
def summarize_old_messages(messages: list, llm) -> list:
"""
对旧消息进行摘要压缩
当消息过长时，将早期对话压缩成摘要
"""
if len(messages) <= 10:
return messages
# 分离需要摘要的消息和保留的消息
to_summarize = messages[:-5] # 保留最近 5 条
recent = messages[-5:]
# 生成摘要
summarize_prompt = f"""
请将以下对话历史压缩成简短的摘要：
{' '.join([m.content if hasattr(m, 'content') else str(m) for m in to_summarize])}
摘要要求：
1. 保留关键信息
2. 压缩到 200 字以内
3. 提取用户偏好和重要事实
"""
summary = llm.invoke(summarize_prompt)
# 创建摘要消息
summary_message = type('SummaryMessage', (), {
'type': 'system',
'content': f'[对话摘要] {summary.content}'
})()
return [summary_message] + recent四、短期记忆实现：Redis + 对话摘要
4.1 短期记忆的存储架构
┌─────────────────────────────────────────────────────────────────────┐
│ 短期记忆存储架构 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ Agent │ │
│ └──────┬──────┘ │
│ │ │
│ │ 存储/读取 │
│ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Redis │◀───▶│ 会话 │◀───▶│ 摘要 │ │
│ │ 主存储 │ │ 元信息 │ │ 历史 │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
│ Key 结构： │
│ - session:{session_id}:messages # 消息列表 (List) │
│ - session:{session_id}:meta # 会话元信息 (Hash) │
│ - session:{session_id}:summary # 对话摘要 (String) │
│ │
└─────────────────────────────────────────────────────────────────────┘4.2 Redis 短期记忆实现
import redis
import json
import time
from typing import Optional, TypedDict, list
from dataclasses import dataclass, asdict
from langchain_openai import ChatOpenAI
@dataclass
class SessionMeta:
"""会话元信息"""
session_id: str
user_id: str
created_at: float
last_active: float
message_count: int = 0
status: str = "active" # active/archived/expired
class ShortTermMemory:
"""
短期记忆实现
使用 Redis 存储会话级记忆：
1. 消息历史：完整的对话记录
2. 会话摘要：压缩后的关键信息
3. 元信息：会话状态和统计
"""
# 默认 TTL：7 天
DEFAULT_TTL = 7 * 24 * 60 * 60
def __init__(
self,
redis_url: str = "redis://localhost:6379/0",
llm: Optional[ChatOpenAI] = None,
ttl: int = DEFAULT_TTL
):
"""
初始化短期记忆
Args:
redis_url: Redis 连接地址
llm: 用于生成摘要的 LLM
ttl: 会话过期时间（秒）
"""
self.redis = redis.from_url(redis_url, decode_responses=True)
self.llm = llm
self.ttl = ttl
# 摘要触发阈值：消息数量超过此值时生成摘要
self.summarize_threshold = 20
def _message_key(self, session_id: str) -> str:
"""生成消息存储的 Redis key"""
return f"session:{session_id}:messages"
def _meta_key(self, session_id: str) -> str:
"""生成元信息存储的 Redis key"""
return f"session:{session_id}:meta"
def _summary_key(self, session_id: str) -> str:
"""生成摘要存储的 Redis key"""
return f"session:{session_id}:summary"
def create_session(self, session_id: str, user_id: str) -> SessionMeta:
"""
创建新会话
Args:
session_id: 会话 ID
user_id: 用户 ID
Returns:
会话元信息
"""
now = time.time()
meta = SessionMeta(
session_id=session_id,
user_id=user_id,
created_at=now,
last_active=now,
message_count=0,
status="active"
)
# 存储元信息
self.redis.hset(self._meta_key(session_id), mapping={
"session_id": meta.session_id,
"user_id": meta.user_id,
"created_at": str(meta.created_at),
"last_active": str(meta.last_active),
"message_count": str(meta.message_count),
"status": meta.status
})
# 设置过期时间
self.redis.expire(self._meta_key(session_id), self.ttl)
return meta
def add_message(
self,
session_id: str,
role: str,
content: str,
metadata: Optional[dict] = None
) -> dict:
"""
添加消息到会话
Args:
session_id: 会话 ID
role: 角色（user/assistant/system）
content: 消息内容
metadata: 附加元数据
Returns:
添加的消息
"""
now = time.time()
message = {
"role": role,
"content": content,
"timestamp": now,
"metadata": metadata or {}
}
# 存储消息
self.redis.rpush(
self._message_key(session_id),
json.dumps(message, ensure_ascii=False)
)
# 更新元信息
self.redis.hincrby(self._meta_key(session_id), "message_count", 1)
self.redis.hset(self._meta_key(session_id), "last_active", str(now))
# 设置过期时间
self.redis.expire(self._message_key(session_id), self.ttl)
# 检查是否需要生成摘要
meta = self.get_meta(session_id)
if meta and meta.message_count >= self.summarize_threshold:
self._maybe_summarize(session_id)
return message
def get_messages(
self,
session_id: str,
limit: Optional[int] = None,
include_summary: bool = True
) -> list:
"""
获取会话消息
Args:
session_id: 会话 ID
limit: 最多返回的消息数
include_summary: 是否在开头包含摘要
Returns:
消息列表
"""
messages = []
# 如果包含摘要，先获取摘要
if include_summary:
summary = self.get_summary(session_id)
if summary:
messages.append({
"role": "system",
"content": f"[历史摘要] {summary}"
})
# 获取消息
raw_messages = self.redis.lrange(self._message_key(session_id), 0, -1)
for msg_str in raw_messages:
messages.append(json.loads(msg_str))
# 如果限制了数量，只返回最近的
if limit:
messages = messages[-limit:]
return messages
def get_summary(self, session_id: str) -> Optional[str]:
"""
获取会话摘要
Args:
session_id: 会话 ID
Returns:
摘要内容，如果没有摘要则返回 None
"""
summary = self.redis.get(self._summary_key(session_id))
return summary if summary else None
def _maybe_summarize(self, session_id: str):
"""
条件生成摘要
当消息积累到一定量时，生成摘要以压缩历史
"""
if not self.llm:
return
# 获取最近的 N 条消息进行摘要
messages = self.get_messages(session_id, limit=30, include_summary=False)
if not messages:
return
# 构建摘要提示
content_parts = [f"{m['role']}: {m['content']}" for m in messages]
content_str = "\n".join(content_parts)
summarize_prompt = f"""
请将以下对话历史压缩成简短的摘要，保留关键信息和用户偏好：
{content_str}
摘要要求：
1. 保留关键事实和决定
2. 提取用户偏好和特点
3. 压缩到 300 字以内
4. 使用第三人称客观描述
"""
try:
response = self.llm.invoke(summarize_prompt)
summary = response.content
# 存储摘要
self.redis.set(self._summary_key(session_id), summary, ex=self.ttl)
# 从消息列表中删除已摘要的旧消息（保留最近 5 条）
trim_count = len(messages) - 5
if trim_count > 0:
self.redis.ltrim(self._message_key(session_id), -5, -1)
except Exception as e:
print(f"摘要生成失败: {e}")
def get_meta(self, session_id: str) -> Optional[SessionMeta]:
"""
获取会话元信息
Args:
session_id: 会话 ID
Returns:
会话元信息
"""
raw_meta = self.redis.hgetall(self._meta_key(session_id))
if not raw_meta:
return None
return SessionMeta(
session_id=raw_meta["session_id"],
user_id=raw_meta["user_id"],
created_at=float(raw_meta["created_at"]),
last_active=float(raw_meta["last_active"]),
message_count=int(raw_meta["message_count"]),
status=raw_meta["status"]
)
def delete_session(self, session_id: str):
"""
删除会话（清除所有记忆）
Args:
session_id: 会话 ID
"""
self.redis.delete(self._message_key(session_id))
self.redis.delete(self._meta_key(session_id))
self.redis.delete(self._summary_key(session_id))4.3 短期记忆使用示例
# === 短期记忆使用示例 ===
from langchain_openai import ChatOpenAI
# 初始化
llm = ChatOpenAI(model="gpt-4")
memory = ShortTermMemory(redis_url="redis://localhost:6379/0", llm=llm)
# 创建新会话
session = memory.create_session(
session_id="sess_001",
user_id="user_123"
)
print(f"创建会话: {session.session_id}")
# 添加消息
memory.add_message("sess_001", "user", "我叫张三，是一名软件工程师")
memory.add_message("sess_001", "assistant", "你好张三！很高兴认识你。请问有什么我可以帮你的？")
memory.add_message("sess_001", "user", "我想了解一些关于 AI Agent 的知识")
# 获取消息历史
messages = memory.get_messages("sess_001", include_summary=True)
print("\n消息历史:")
for msg in messages:
print(f" {msg['role']}: {msg['content'][:50]}...")
# 获取元信息
meta = memory.get_meta("sess_001")
print(f"\n会话统计: {meta.message_count} 条消息")五、长期记忆实现：向量数据库 + 知识图谱
5.1 长期记忆的存储架构
┌─────────────────────────────────────────────────────────────────────┐
│ 长期记忆存储架构 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 长期记忆层 │ │
│ │ │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ 向量数据库 │ │ 知识图谱 │ │ │
│ │ │ (Vector DB) │ │ (Knowledge Graph)│ │ │
│ │ │ │ │ │ │ │
│ │ │ • 语义记忆 │ │ • 实体关系 │ │ │
│ │ │ • 经验记忆 │ │ • 概念层次 │ │ │
│ │ │ • 偏好向量 │ │ • 事实三元组 │ │ │
│ │ └────────┬────────┘ └────────┬────────┘ │ │
│ │ │ │ │ │
│ │ └──────────┬───────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌─────────────────┐ │ │
│ │ │ 联合检索引擎 │ │ │
│ │ └─────────────────┘ │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ 数据来源： │
│ • 用户偏好（显式提供） │
│ • 对话总结（从短期记忆提取） │
│ • 外部知识（文档、网页） │
│ • Agent 经验（成功/失败的案例） │
│ │
└─────────────────────────────────────────────────────────────────────┘5.2 向量数据库实现
from typing import Optional, list
from dataclasses import dataclass
from datetime import datetime
import numpy as np
@dataclass
class MemoryEntry:
"""记忆条目"""
id: str
content: str
embedding: Optional[np.ndarray] = None
memory_type: str = "experience" # experience/preference/fact/skill
importance: float = 0.5 # 重要性评分 0-1
created_at: float = None
last_accessed: float = None
access_count: int = 0
metadata: dict = None
def __post_init__(self):
import time
now = time.time()
if self.created_at is None:
self.created_at = now
if self.last_accessed is None:
self.last_accessed = now
if self.metadata is None:
self.metadata = {}
class LongTermMemory:
"""
长期记忆实现
使用向量数据库存储记忆：
1. 支持语义检索
2. 基于重要性和时间衰减的记忆评分
3. 自动遗忘低价值记忆
"""
# 记忆过期时间（秒）：30 天
MEMORY_EXPIRY = 30 * 24 * 60 * 60
# 最小重要性阈值：低于此值的记忆会被遗忘
MIN_IMPORTANCE = 0.3
def __init__(
self,
vector_store=None, # 可以是 Chroma/Pinecone/Milvus 等
embedding_model=None,
llm: Optional[ChatOpenAI] = None
):
"""
初始化长期记忆
Args:
vector_store: 向量数据库客户端
embedding_model: 嵌入模型
llm: 用于处理复杂记忆的 LLM
"""
# 如果没有提供向量存储，使用内存实现
self.vector_store = vector_store or InMemoryVectorStore()
self.embedding_model = embedding_model
self.llm = llm
# 内存索引，用于快速查找
self._memory_index: dict[str, MemoryEntry] = {}
def store_memory(
self,
content: str,
memory_type: str = "experience",
importance: float = 0.5,
metadata: Optional[dict] = None
) -> str:
"""
存储新记忆
Args:
content: 记忆内容
memory_type: 记忆类型
importance: 重要性评分
metadata: 附加元数据
Returns:
记忆 ID
"""
import uuid
import time
memory_id = str(uuid.uuid4())
# 生成嵌入向量
embedding = None
if self.embedding_model:
embedding = self.embedding_model.embed(content)
# 创建记忆条目
entry = MemoryEntry(
id=memory_id,
content=content,
embedding=embedding,
memory_type=memory_type,
importance=importance,
created_at=time.time(),
last_accessed=time.time(),
access_count=0,
metadata=metadata or {}
)
# 存储到向量数据库
self.vector_store.add(
id=memory_id,
embedding=embedding,
document=content,
metadata={
"memory_type": memory_type,
"importance": importance,
"created_at": entry.created_at
}
)
# 更新内存索引
self._memory_index[memory_id] = entry
return memory_id
def retrieve(
self,
query: str,
top_k: int = 5,
memory_types: Optional[list[str]] = None,
min_importance: float = None
) -> list[MemoryEntry]:
"""
检索相关记忆
Args:
query: 查询内容
top_k: 返回数量
memory_types: 过滤的记忆类型
min_importance: 最低重要性阈值
Returns:
记忆条目列表（已按评分排序）
Returns:
记忆条目列表
"""
import time
if min_importance is None:
min_importance = self.MIN_IMPORTANCE
# 生成查询向量
query_embedding = None
if self.embedding_model:
query_embedding = self.embedding_model.embed(query)
# 向量相似度搜索
results = self.vector_store.search(
query_embedding=query_embedding,
query_text=query,
top_k=top_k * 2, # 多取一些，后面过滤
filter_metadata={"memory_type": memory_types} if memory_types else None
)
memories = []
for result in results:
memory_id = result["id"]
# 从索引中获取完整条目
entry = self._memory_index.get(memory_id)
if not entry:
continue
# 过滤重要性
if entry.importance < min_importance:
continue
# 计算综合评分
# 评分 = 相似度 * 重要性 * 时间衰减
time_elapsed = time.time() - entry.last_accessed
time_decay = np.exp(-time_elapsed / (self.MEMORY_EXPIRY / 2))
# 访问次数加权
access_boost = 1 + np.log1p(entry.access_count) * 0.1
result_score = result["similarity"] * entry.importance * time_decay * access_boost
# 更新访问记录
entry.last_accessed = time.time()
entry.access_count += 1
memories.append((result_score, entry))
# 按评分排序
memories.sort(key=lambda x: x[0], reverse=True)
return [m[1] for m in memories[:top_k]]
def update_importance(self, memory_id: str, delta: float):
"""
更新记忆的重要性
当记忆被成功使用时提高重要性，
当记忆被遗忘时降低重要性
Args:
memory_id: 记忆 ID
delta: 变化量（正数增加，负数减少）
"""
entry = self._memory_index.get(memory_id)
if entry:
entry.importance = max(0.0, min(1.0, entry.importance + delta))
def forget(self, memory_id: str):
"""
遗忘记忆
Args:
memory_id: 记忆 ID
"""
# 从向量数据库删除
self.vector_store.delete(memory_id)
# 从索引删除
if memory_id in self._memory_index:
del self._memory_index[memory_id]
def garbage_collect(self) -> int:
"""
垃圾回收：遗忘低价值和过期的记忆
Returns:
被遗忘的记忆数量
"""
import time
forgotten_count = 0
current_time = time.time()
to_forget = []
for memory_id, entry in self._memory_index.items():
# 检查是否过期
age = current_time - entry.created_at
if age > self.MEMORY_EXPIRY:
to_forget.append(memory_id)
continue
# 检查是否低于最低重要性
if entry.importance < self.MIN_IMPORTANCE:
to_forget.append(memory_id)
# 执行遗忘
for memory_id in to_forget:
self.forget(memory_id)
forgotten_count += 1
return forgotten_count
def get_user_profile(self, user_id: str) -> dict:
"""
获取用户画像
从长期记忆中提取用户特征
Args:
user_id: 用户 ID
Returns:
用户画像字典
"""
if not self.llm:
return {}
# 获取用户相关的记忆
memories = self.retrieve(
query=f"user {user_id} preferences characteristics",
top_k=20,
memory_types=["preference", "experience"]
)
if not memories:
return {}
# 构建画像提示
memory_texts = [m.content for m in memories]
profile_prompt = f"""
基于以下用户记忆，提取用户画像：
{chr(10).join(memory_texts)}
请提取：
1. 基本特征（职业、兴趣等）
2. 沟通偏好（语言风格、正式程度）
3. 特殊需求或限制
4. 历史交互总结
以结构化的 JSON 格式输出。
"""
try:
response = self.llm.invoke(profile_prompt)
# 实际应用中应该解析 JSON
return {"raw_profile": response.content}
except:
return {}
class InMemoryVectorStore:
"""
简单的内存向量存储（仅用于演示）
生产环境应使用 ChromaDB、Pinecone、Milvus 等专业向量数据库
"""
def __init__(self):
self._documents: dict[str, dict] = {}
self._counter = 0
def add(self, id: str, embedding, document: str, metadata: dict):
"""添加向量"""
self._documents[id] = {
"embedding": embedding,
"document": document,
"metadata": metadata
}
self._counter += 1
def search(
self,
query_embedding,
query_text: str,
top_k: int = 5,
filter_metadata: dict = None
) -> list[dict]:
"""搜索"""
results = []
for doc_id, doc in self._documents.items():
# 简单的余弦相似度计算
if doc["embedding"] is not None and query_embedding is not None:
similarity = self._cosine_similarity(doc["embedding"], query_embedding)
else:
# 如果没有嵌入向量，使用关键词匹配
similarity = self._keyword_match(doc["document"], query_text)
# 应用元数据过滤
if filter_metadata:
match = True
for key, value in filter_metadata.items():
if doc["metadata"].get(key) != value:
match = False
break
if not match:
continue
results.append({
"id": doc_id,
"similarity": similarity,
"document": doc["document"],
"metadata": doc["metadata"]
})
# 排序
results.sort(key=lambda x: x["similarity"], reverse=True)
return results[:top_k]
def delete(self, id: str):
"""删除"""
if id in self._documents:
del self._documents[id]
def _cosine_similarity(self, a, b) -> float:
"""计算余弦相似度"""
if a is None or b is None:
return 0.0
dot_product = np.dot(a, b)
norm_a = np.linalg.norm(a)
norm_b = np.linalg.norm(b)
if norm_a == 0 or norm_b == 0:
return 0.0
return dot_product / (norm_a * norm_b)
def _keyword_match(self, document: str, query: str) -> float:
"""简单的关键词匹配"""
doc_lower = document.lower()
query_words = query.lower().split()
matches = sum(1 for word in query_words if word in doc_lower)
return matches / len(query_words) if query_words else 0.0六、记忆检索策略：相似度搜索 + 时间衰减 + 重要性评分
6.1 综合评分算法
记忆检索不能只靠相似度，还需要综合考虑多个因素：
┌─────────────────────────────────────────────────────────────────────┐
│ 记忆检索评分算法 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 综合评分 = 相似度 × 重要性 × 时间衰减 × 访问增益 │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 1. 语义相似度 (Semantic Similarity) │ │
│ │ - 使用 Embedding 向量计算 │ │
│ │ - 余弦相似度：cosine(A, B) │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ × │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 2. 重要性评分 (Importance Score) │ │
│ │ - 用户显式标注 │ │
│ │ - 基于交互结果的学习 │ │
│ │ - 0.0 - 1.0 范围 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ × │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 3. 时间衰减 (Time Decay) │ │
│ │ - 指数衰减：exp(-λt) │ │
│ │ - 半衰期：30 天 │ │
│ │ - 近期记忆权重更高 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ × │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 4. 访问增益 (Access Boost) │ │
│ │ - 频繁访问的记忆权重提升 │ │
│ │ - log(1 + access_count) × 0.1 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘6.2 检索策略实现
from typing import Optional, Callable
class MemoryRetrievalStrategy:
"""
记忆检索策略
支持多种检索策略的组合：
1. 语义检索：基于向量相似度
2. 关键词检索：BM25 或关键词匹配
3. 过滤检索：基于元数据过滤
4. 混合检索：组合多种策略
"""
def __init__(
self,
long_term_memory: LongTermMemory,
llm: Optional[ChatOpenAI] = None
):
self.memory = long_term_memory
self.llm = llm
def retrieve_for_task(
self,
task: str,
context: dict,
strategy: str = "hybrid",
top_k: int = 5
) -> list[MemoryEntry]:
"""
为任务检索相关记忆
Args:
task: 当前任务描述
context: 当前上下文（用户信息、会话状态等）
strategy: 检索策略
top_k: 返回数量
Returns:
相关记忆列表
"""
if strategy == "semantic":
return self._semantic_retrieval(task, top_k)
elif strategy == "keyword":
return self._keyword_retrieval(task, top_k)
elif strategy == "contextual":
return self._contextual_retrieval(task, context, top_k)
elif strategy == "hybrid":
return self._hybrid_retrieval(task, context, top_k)
else:
raise ValueError(f"Unknown strategy: {strategy}")
def _semantic_retrieval(self, query: str, top_k: int) -> list[MemoryEntry]:
"""纯语义检索"""
return self.memory.retrieve(query, top_k=top_k)
def _keyword_retrieval(self, query: str, top_k: int) -> list[MemoryEntry]:
"""关键词检索"""
# 提取关键词
keywords = self._extract_keywords(query)
# 简单实现：在所有记忆中匹配关键词
results = []
for memory_id, entry in self.memory._memory_index.items():
score = sum(1 for kw in keywords if kw in entry.content.lower())
if score > 0:
results.append((score, entry))
results.sort(key=lambda x: x[0], reverse=True)
return [r[1] for r in results[:top_k]]
def _contextual_retrieval(
self,
task: str,
context: dict,
top_k: int
) -> list[MemoryEntry]:
"""上下文感知检索"""
user_id = context.get("user_id", "")
# 构建上下文增强的查询
enhanced_query = f"""
Task: {task}
User: {user_id}
Session: {context.get('session_id', '')}
"""
# 获取用户相关记忆
user_memories = self.memory.retrieve(
query=enhanced_query,
top_k=top_k * 2
)
# 根据上下文过滤
filtered = []
for memory in user_memories:
# 优先选择与当前用户相关的记忆
if memory.metadata.get("user_id") == user_id:
filtered.append(memory)
elif memory.memory_type in ["fact", "skill"]:
# 通用知识也保留
filtered.append(memory)
return filtered[:top_k]
def _hybrid_retrieval(
self,
task: str,
context: dict,
top_k: int
) -> list[MemoryEntry]:
"""混合检索：综合语义、关键词、上下文"""
# 并行执行多种检索
semantic_results = self._semantic_retrieval(task, top_k * 2)
keyword_results = self._keyword_retrieval(task, top_k * 2)
contextual_results = self._contextual_retrieval(task, context, top_k * 2)
# 去重并分配权重
memory_scores: dict[str, float] = {}
# 语义检索权重
for i, memory in enumerate(semantic_results):
score = 1.0 / (i + 1) # 位置衰减
memory_scores[memory.id] = memory_scores.get(memory.id, 0) + score * 0.5
# 关键词检索权重
for i, memory in enumerate(keyword_results):
score = 1.0 / (i + 1)
memory_scores[memory.id] = memory_scores.get(memory.id, 0) + score * 0.3
# 上下文检索权重
for i, memory in enumerate(contextual_results):
score = 1.0 / (i + 1)
memory_scores[memory.id] = memory_scores.get(memory.id, 0) + score * 0.2
# 按分数排序
sorted_ids = sorted(memory_scores.keys(), key=lambda x: memory_scores[x], reverse=True)
# 构建最终结果
all_memories = {m.id: m for m in semantic_results + keyword_results + contextual_results}
results = [all_memories[mid] for mid in sorted_ids[:top_k] if mid in all_memories]
return results
def _extract_keywords(self, text: str) -> list[str]:
"""提取关键词"""
if not self.llm:
# 简单的关键词提取
words = text.lower().split()
# 去除停用词
stopwords = {"的", "了", "是", "在", "我", "有", "和", "就", "不", "人", "都", "一", "一个", "上", "也", "很", "到", "说", "要", "去", "你", "会", "着", "没有", "看", "好", "自己", "这"}
return [w for w in words if w not in stopwords and len(w) > 1]
# 使用 LLM 提取关键词
prompt = f"""
从以下文本中提取 5-10 个关键词：
{text}
关键词要求：
1. 名词或动词
2. 有意义的词汇
3. 用逗号分隔
"""
try:
response = self.llm.invoke(prompt)
keywords = [k.strip() for k in response.content.split(',')]
return keywords
except:
return self._extract_keywords(text) # 回退到简单方法七、记忆压缩与遗忘机制
7.1 记忆压缩策略
┌─────────────────────────────────────────────────────────────────────┐
│ 记忆压缩与遗忘机制 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 记忆生命周期 │ │
│ │ │ │
│ │ 创建 ──▶ 活跃 ──▶ 半活跃 ──▶ 休眠 ──▶ 遗忘 │ │
│ │ │ │
│ │ • 活跃：最近访问，权重高 │ │
│ │ • 半活跃：偶尔访问，权重中等 │ │
│ │ • 休眠：长时间未访问，等待回收 │ │
│ │ • 遗忘：被垃圾回收清除 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ 压缩策略： │
│ 1. 摘要压缩：将多条相关记忆合并为摘要 │
│ 2. 层级压缩：从详细记录到概览摘要 │
│ 3. 选择性压缩：保留关键细节，模糊次要信息 │
│ │
└─────────────────────────────────────────────────────────────────────┘7.2 遗忘机制实现
class MemoryForgettingMechanism:
"""
记忆遗忘机制
模拟人类记忆的遗忘曲线：
1. 时间衰减：越久远的记忆越容易被遗忘
2. 重要性衰减：低重要性的记忆更容易遗忘
3. 干扰衰减：相似记忆相互干扰
"""
# 遗忘参数
BASE_DECAY_RATE = 0.1 # 基础衰减率
HALF_LIFE_DAYS = 30 # 半衰期（天）
IMPORTANCE_PROTECTION = 0.2 # 重要性对遗忘的保护系数
def __init__(self, long_term_memory: LongTermMemory):
self.memory = long_term_memory
def calculate_decay(self, entry: MemoryEntry) -> float:
"""
计算记忆的衰减程度
Args:
entry: 记忆条目
Returns:
衰减程度（0-1），1 表示完全衰减
"""
import time
# 时间衰减
days_elapsed = (time.time() - entry.last_accessed) / (24 * 60 * 60)
time_decay = 1 - np.exp(-np.log(2) * days_elapsed / self.HALF_LIFE_DAYS)
# 重要性保护
importance_protection = entry.importance * self.IMPORTANCE_PROTECTION
# 访问频率保护
access_boost = np.log1p(entry.access_count) * 0.05
# 综合衰减
decay = time_decay * (1 - importance_protection - access_boost)
return max(0.0, min(1.0, decay))
def should_forget(self, entry: MemoryEntry) -> tuple[bool, str]:
"""
判断记忆是否应该被遗忘
Args:
entry: 记忆条目
Returns:
(是否遗忘, 原因)
"""
# 检查衰减程度
decay = self.calculate_decay(entry)
if decay > 0.8:
return True, f"衰减程度过高 ({decay:.2f})"
# 检查创建时间
import time
age_days = (time.time() - entry.created_at) / (24 * 60 * 60)
if age_days > 90 and entry.access_count == 0:
return True, "长期未访问且从未被使用"
if age_days > 180 and entry.importance < 0.4:
return True, "超过半年且重要性低"
# 检查最低重要性
if entry.importance < 0.2:
return True, "重要性极低"
return False, ""
def consolidate_memories(self, theme: str) -> Optional[str]:
"""
整合关于某个主题的记忆
将多条相关记忆压缩为一条摘要
Args:
theme: 主题关键词
Returns:
摘要内容
"""
if not self.memory.llm:
return None
# 获取相关记忆
related = self.memory.retrieve(
query=theme,
top_k=10,
min_importance=0.3
)
if len(related) < 2:
return None
# 构建整合提示
memories_text = "\n".join([
f"- {m.content} (重要性: {m.importance:.2f})"
for m in related
])
consolidate_prompt = f"""
请将以下关于「{theme}」的记忆整合为一条简洁的摘要：
{memories_text}
要求：
1. 保留核心信息和关键细节
2. 消除冗余和矛盾
3. 控制在 200 字以内
4. 标注信息来源数量
"""
try:
response = self.memory.llm.invoke(consolidate_prompt)
return f"{response.content}\n\n(整合自 {len(related)} 条相关记忆)"
except:
return None
def run_garbage_collection(self) -> dict:
"""
执行垃圾回收
Returns:
回收报告
"""
to_forget = []
details = []
for memory_id, entry in self.memory._memory_index.items():
should, reason = self.should_forget(entry)
if should:
to_forget.append(memory_id)
details.append({
"id": memory_id,
"content_preview": entry.content[:50],
"reason": reason
})
# 执行遗忘
for memory_id in to_forget:
self.memory.forget(memory_id)
return {
"forgotten_count": len(to_forget),
"details": details
}八、LangGraph 中的记忆集成
8.1 完整的记忆集成工作流
┌─────────────────────────────────────────────────────────────────────┐
│ 记忆集成的 LangGraph 工作流 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 用户输入 │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 记忆检索节点 │ │
│ │ • 短期记忆：获取会话历史 │ │
│ │ • 长期记忆：检索相关记忆 │ │
│ │ • 组合上下文 │ │
│ └──────────────────────────┬──────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ LLM 处理节点 │ │
│ │ • 基于记忆生成回复 │ │
│ │ • 更新任务状态 │ │
│ └──────────────────────────┬──────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 记忆存储节点 │ │
│ │ • 短期记忆：追加消息 │ │
│ │ • 长期记忆：评估是否存储新记忆 │ │
│ │ • 重要性更新 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘8.2 完整实现代码
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
"""
带完整记忆系统的 Agent 状态
"""
# 消息历史（工作记忆）
messages: Annotated[list, add_messages]
# 上下文信息
user_id: str
session_id: str
# 记忆相关
retrieved_memories: list = [] # 检索到的长期记忆
memory_context: str = "" # 组合的记忆上下文
# 任务状态
current_task: str = ""
task_status: str = "pending"
class MemoryIntegratedAgent:
"""
集成记忆系统的 Agent
整合：
1. 工作记忆：messages 列表
2. 短期记忆：Redis 会话存储
3. 长期记忆：向量数据库
"""
def __init__(
self,
short_term_memory: ShortTermMemory,
long_term_memory: LongTermMemory,
llm: ChatOpenAI
):
self.short_term = short_term_memory
self.long_term = long_term_memory
self.llm = llm
# 构建工作流
self.app = self._build_workflow()
def retrieve_memories(self, session_id: str, query: str, user_id: str) -> str:
"""
从各层记忆检索相关信息
Returns:
组合的记忆上下文字符串
"""
context_parts = []
# 1. 短期记忆：获取最近对话
short_messages = self.short_term.get_messages(session_id, limit=10)
if short_messages:
short_context = "\n".join([
f"{m.get('role', 'unknown')}: {m.get('content', '')[:100]}"
for m in short_messages[-5:]
])
context_parts.append(f"[近期对话]\n{short_context}")
# 2. 长期记忆：检索相关记忆
retrieval_strategy = MemoryRetrievalStrategy(
self.long_term,
self.llm
)
relevant_memories = retrieval_strategy.retrieve_for_task(
task=query,
context={"user_id": user_id, "session_id": session_id},
top_k=5
)
if relevant_memories:
memory_context = "\n".join([
f"- {m.content}"
for m in relevant_memories
])
context_parts.append(f"[相关记忆]\n{memory_context}")
return "\n\n".join(context_parts)
def store_interaction(
self,
session_id: str,
user_id: str,
user_message: str,
assistant_response: str,
importance: float = 0.5
):
"""
存储一次交互到各层记忆
"""
# 1. 短期记忆：追加消息
self.short_term.add_message(session_id, "user", user_message)
self.short_term.add_message(session_id, "assistant", assistant_response)
# 2. 长期记忆：根据重要性决定是否存储
if importance >= 0.5:
self.long_term.store_memory(
content=f"用户提问：{user_message}\n助手回答：{assistant_response}",
memory_type="experience",
importance=importance,
metadata={"user_id": user_id, "session_id": session_id}
)
def _build_workflow(self):
"""构建 LangGraph 工作流"""
def memory_retrieval_node(state: AgentState) -> dict:
"""记忆检索节点"""
# 从各层检索记忆
memory_context = self.retrieve_memories(
session_id=state["session_id"],
query=state["messages"][-1].content if state["messages"] else "",
user_id=state["user_id"]
)
return {
"memory_context": memory_context,
"retrieved_memories": [] # 可以存储 MemoryEntry 对象
}
def llm_process_node(state: AgentState) -> dict:
"""LLM 处理节点"""
# 构建包含记忆的提示
system_prompt = f"""
你是一个有帮助的 AI 助手。
以下是相关的上下文信息：
{state.get('memory_context', '无')}
"""
# 准备消息
full_messages = [
{"role": "system", "content": system_prompt}
]
for msg in state["messages"]:
role = msg.type if hasattr(msg, 'type') else 'user'
content = msg.content if hasattr(msg, 'content') else str(msg)
full_messages.append({"role": role, "content": content})
# 调用 LLM
response = self.llm.invoke(full_messages)
return {
"messages": [response],
"task_status": "completed"
}
def memory_storage_node(state: AgentState) -> dict:
"""记忆存储节点"""
if len(state["messages"]) >= 2:
# 获取最后一轮对话
user_msg = state["messages"][-2]
assistant_msg = state["messages"][-1]
user_content = user_msg.content if hasattr(user_msg, 'content') else str(user_msg)
assistant_content = assistant_msg.content if hasattr(assistant_msg, 'content') else str(assistant_msg)
# 评估重要性（简化版本）
importance = 0.5
if len(user_content) > 100:
importance += 0.2
if "重要" in user_content or "记住" in user_content:
importance += 0.3
# 存储
self.store_interaction(
session_id=state["session_id"],
user_id=state["user_id"],
user_message=user_content,
assistant_response=assistant_content,
importance=importance
)
return {}
# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("memory_retrieval", memory_retrieval_node)
workflow.add_node("llm_process", llm_process_node)
workflow.add_node("memory_storage", memory_storage_node)
workflow.add_edge(START, "memory_retrieval")
workflow.add_edge("memory_retrieval", "llm_process")
workflow.add_edge("llm_process", "memory_storage")
workflow.add_edge("memory_storage", END)
return workflow.compile()
# === 使用示例 ===
def main():
"""主函数"""
from langchain_openai import ChatOpenAI
import os
# 初始化组件
llm = ChatOpenAI(model="gpt-4", api_key=os.getenv("OPENAI_API_KEY"))
# 短期记忆（需要 Redis）
# short_term = ShortTermMemory(redis_url="redis://localhost:6379/0", llm=llm)
# 长期记忆
long_term = LongTermMemory(llm=llm)
# 存储一些初始记忆
long_term.store_memory(
content="用户张三是一名 Python 开发者，偏好简洁的技术解释",
memory_type="preference",
importance=0.8,
metadata={"user_id": "user_123"}
)
# 创建 Agent
# agent = MemoryIntegratedAgent(short_term, long_term, llm)
print("✅ 记忆系统初始化完成")
print("\n示例记忆：")
# 检索测试
memories = long_term.retrieve("用户偏好和特点", top_k=5)
for m in memories:
print(f" - {m.content[:60]}... (重要性: {m.importance:.2f})")
if __name__ == "__main__":
main()九、总结
9.1 三层记忆架构回顾
┌─────────────────────────────────────────────────────────────────────┐
│ Agent 记忆系统完整架构总结 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 工作记忆 (Working Memory) │ │
│ │ │ │
│ │ 实现：TypedDict + add_messages │ │
│ │ 生命周期：单轮对话 │ │
│ │ 管理：消息追加 + 长度控制 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ ▲ │
│ │ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 短期记忆 (Short-Term Memory) │ │
│ │ │ │
│ │ 实现：Redis List + Hash │ │
│ │ 生命周期：会话级（可配置 TTL） │ │
│ │ 管理：消息存储 + 对话摘要 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ ▲ │
│ │ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 长期记忆 (Long-Term Memory) │ │
│ │ │ │
│ │ 实现：Vector DB + Knowledge Graph │ │
│ │ 生命周期：永久 │ │
│ │ 管理：语义检索 + 重要性评分 + 时间衰减 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ 核心机制： │
│ • 检索策略：语义 + 关键词 + 上下文 + 混合 │
│ • 遗忘机制：时间衰减 + 重要性保护 + 干扰衰减 │
│ • 压缩机制：摘要压缩 + 层级压缩 │
│ │
└─────────────────────────────────────────────────────────────────────┘9.2 关键设计要点
9.3 生产环境注意事项
存储选型：
开发环境：内存存储
生产环境：Redis + ChromaDB/Pinecone
性能优化：
异步写入：记忆存储不阻塞主流程
批量操作：减少网络往返
数据安全：
敏感信息脱敏
加密存储
参考资料
本文首发于 2026 年 3 月 15 日，风格对标 PySuper（zhengxingtao.com）
评论区