Agent 的记忆系统设计：从短期到长期
“Agent 的智能不在于模型有多大，而在于它记得多少、记得多准、记得多久。”
TL;DR
- 记忆的三个层次：Working Memory（秒-分钟级）、Short-term Memory（小时-天级）、Long-term Memory（永久）
- 技术选型：Vector DB 用于语义检索、Graph DB 用于关系推理、传统数据库用于结构化数据
- 关键挑战：遗忘与保留的平衡、多模态记忆融合、隐私与安全的边界
- 实战方案：从简单的上下文窗口到企业级记忆系统的演进路径
📋 本文结构
- 为什么记忆是 Agent 的核心能力
- 记忆的三个层次
- Working Memory：当下的上下文
- Short-term Memory：近期的经验
- Long-term Memory：持久的知识
- Knowledge Graph：关系的网络
- 记忆系统的工程实现
- 写在最后
为什么记忆是 Agent 的核心能力
没有记忆的 Agent 是”傻子”
想象一下和一个”金鱼型”Agent 对话：
User: 我叫 @postcodeeng，是做 CRM SaaS 的。
Agent: 很高兴认识你，@postcodeeng！有什么我可以帮你的？
User: 我想了解一下 Agent OS 的架构。
Agent: 当然可以！不过请问您叫什么名字，在哪个行业？
User: ...我刚说过我叫 @postcodeeng，做 CRM SaaS。
Agent: 哦，抱歉！那么 @postcodeeng，你对 Agent OS 的哪方面感兴趣？
User: 我之前问你推荐过记忆系统的设计方案。
Agent: 抱歉，我不记得我们之前聊过这个...
这样的 Agent 能用吗？显然不能。
记忆决定 Agent 的上限
| 能力维度 | 无记忆 Agent | 有记忆 Agent |
|---|---|---|
| 个性化 | 每次对话从零开始 | 记住用户偏好、历史交互 |
| 连贯性 | 无法理解上下文 | 多轮对话保持连贯 |
| 学习效率 | 每次犯同样的错误 | 从错误中学习改进 |
| 知识积累 | 静态知识库 | 持续学习、动态更新 |
| 业务深度 | 通用回答 | 理解特定业务场景 |
💡 Key Insight
大模型提供了”智商”，记忆系统提供了”情商”和”经验”。
记忆的三个层次
人类记忆的启示
人类记忆分为：
- 感觉记忆：毫秒级，瞬时感知
- 短期记忆：秒-分钟级，当前关注
- 长期记忆：永久存储，可提取的知识
Agent 的记忆系统设计可以借鉴这个分层：
┌─────────────────────────────────────────────┐
│ Working Memory (工作记忆) │
│ • 当前对话上下文 │
│ • 活跃任务状态 │
│ • 临时变量和中间结果 │
│ 容量：有限（类似 7±2 原则） │
│ 持续时间：秒到分钟 │
├─────────────────────────────────────────────┤
│ Short-term Memory (短期记忆) │
│ • 近期对话历史 │
│ • 最近执行的任务 │
│ • 临时学习的内容 │
│ 容量：较大 │
│ 持续时间：小时到天 │
├─────────────────────────────────────────────┤
│ Long-term Memory (长期记忆) │
│ • 用户画像和偏好 │
│ • 业务规则和知识 │
│ • 成功案例和模式 │
│ 容量：几乎无限 │
│ 持续时间：永久 │
└─────────────────────────────────────────────┘
Working Memory：当下的上下文
什么是 Working Memory？
Working Memory 是 Agent 的”脑海”，存储当前正在处理的信息：
- 当前对话的最近几轮
- 正在执行的任务的状态
- 临时计算结果
- 当前关注的实体（客户、订单等）
实现方案
1. 上下文窗口（Context Window）
最简单的方式：维护最近的 N 轮对话
class ContextWindow:
def __init__(self, max_size: int = 10):
self.messages = []
self.max_size = max_size
def add(self, message: Message):
self.messages.append(message)
# 保持窗口大小
if len(self.messages) > self.max_size:
self.messages.pop(0)
def get_context(self) -> List[Message]:
return self.messages
def clear(self):
self.messages = []
问题：当对话很长时，会丢失早期信息。
2. 滑动窗口 + 摘要
对较早的对话进行摘要，保留关键信息：
class SummarizingContext:
def __init__(self, llm: LLM, max_raw_messages: int = 6):
self.llm = llm
self.recent_messages = []
self.summary = ""
self.max_raw = max_raw_messages
def add(self, message: Message):
self.recent_messages.append(message)
# 当原始消息过多时，生成摘要
if len(self.recent_messages) > self.max_raw:
old_messages = self.recent_messages[:-self.max_raw//2]
self.recent_messages = self.recent_messages[-self.max_raw//2:]
# 生成摘要
self.summary = self._summarize(old_messages)
def _summarize(self, messages: List[Message]) -> str:
prompt = f"""
Summarize the following conversation, keeping key facts and decisions:
{messages}
Summary:
"""
return self.llm.generate(prompt)
def get_context(self) -> str:
return f"Summary of earlier conversation: {self.summary}\n\nRecent messages: {self.recent_messages}"
3. 实体追踪（Entity Tracking）
在对话中追踪关键实体的状态：
class EntityTracker:
def __init__(self):
self.active_entities = {} # 当前关注的实体
def update(self, message: Message):
# 使用 NER 提取实体
entities = self.extract_entities(message.content)
for entity in entities:
if entity.id not in self.active_entities:
self.active_entities[entity.id] = entity
else:
# 更新实体信息
self.active_entities[entity.id].update(entity)
def get_entity_context(self) -> str:
"""生成当前活跃实体的描述"""
context = "Currently discussing:\n"
for entity in self.active_entities.values():
context += f"- {entity.type}: {entity.name} ({entity.key_attributes})\n"
return context
Working Memory 的关键设计
| 设计决策 | 建议 | 理由 |
|---|---|---|
| 容量 | 5-10 轮对话 | 平衡上下文理解和成本 |
| 保留策略 | FIFO + 重要性加权 | 简单且有效 |
| 摘要触发 | 每 5-10 轮 | 避免频繁调用 LLM |
| 实体追踪 | 必须 | 保持对话焦点 |
Short-term Memory：近期的经验
什么是 Short-term Memory？
Short-term Memory 存储最近几小时到几天的经验：
- 今天和用户的对话历史
- 最近执行的任务及其结果
- 最近学习的新知识
- 临时的业务状态
技术实现：Vector Database
为什么用 Vector DB？
因为需要语义检索——不是精确匹配关键词，而是理解含义后检索相关内容。
示例：
用户问：”上周那个大客户怎么样了？”
- 关键词检索：找不到包含”大客户”的记录
- 向量检索：找到关于”Acme Corp”、”合同谈判”的记忆，即使没出现”大客户”这个词
实现方案：
from typing import List, Optional
import numpy as np
class ShortTermMemory:
"""
基于向量数据库的短期记忆系统
"""
def __init__(self, embedding_model: EmbeddingModel, vector_store: VectorStore):
self.embedding_model = embedding_model
self.vector_store = vector_store
self.session_id: Optional[str] = None
def start_session(self, session_id: str):
"""开始新会话"""
self.session_id = session_id
def add(self, content: str, memory_type: str = "conversation",
metadata: dict = None, importance: float = 1.0):
"""
添加记忆
Args:
content: 记忆内容
memory_type: 记忆类型 (conversation, action, observation, reflection)
metadata: 附加元数据
importance: 重要性评分 (0-1)，影响保留优先级
"""
# 生成嵌入向量
embedding = self.embedding_model.embed(content)
# 存储到向量数据库
self.vector_store.add(
id=self._generate_id(),
embedding=embedding,
content=content,
metadata={
"session_id": self.session_id,
"type": memory_type,
"timestamp": datetime.now().isoformat(),
"importance": importance,
**(metadata or {})
}
)
def retrieve(self, query: str, k: int = 5,
memory_type: Optional[str] = None,
time_range: Optional[tuple] = None) -> List[Memory]:
"""
检索相关记忆
Args:
query: 查询内容
k: 返回结果数量
memory_type: 筛选特定类型的记忆
time_range: 时间范围过滤 (start, end)
"""
# 生成查询向量
query_embedding = self.embedding_model.embed(query)
# 构建过滤条件
filters = {"session_id": self.session_id}
if memory_type:
filters["type"] = memory_type
if time_range:
filters["timestamp"] = {"$gte": time_range[0], "$lte": time_range[1]}
# 向量检索
results = self.vector_store.search(
query_embedding=query_embedding,
k=k,
filters=filters
)
return [Memory.from_record(r) for r in results]
def get_recent(self, hours: int = 24, k: int = 10) -> List[Memory]:
"""获取最近几小时的记忆"""
since = datetime.now() - timedelta(hours=hours)
return self.vector_store.query(
filters={
"session_id": self.session_id,
"timestamp": {"$gte": since.isoformat()}
},
order_by="timestamp",
descending=True,
limit=k
)
记忆的遗忘机制
短期记忆不能无限增长，需要遗忘：
class ForgettingMechanism:
"""
基于重要性和时间的遗忘机制
"""
def __init__(self, vector_store: VectorStore):
self.vector_store = vector_store
def should_forget(self, memory: Memory) -> bool:
"""判断是否应该遗忘这段记忆"""
age_hours = (datetime.now() - memory.timestamp).total_seconds() / 3600
importance = memory.importance
# 遗忘评分：年龄越高、重要性越低，越容易被遗忘
forgetting_score = age_hours / (importance + 0.1)
return forgetting_score > self.forget_threshold
def consolidate_important_memories(self):
"""
将重要的短期记忆提升到长期记忆
"""
old_memories = self.vector_store.query(
filters={"timestamp": {"$lt": (datetime.now() - timedelta(days=7)).isoformat()}},
)
for memory in old_memories:
if memory.importance > 0.8:
# 提升到长期记忆
self.long_term_memory.store(memory)
# 从短期记忆中删除或标记
self.vector_store.delete(memory.id)
技术选型对比
| Vector DB | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| Pinecone | 全托管、易用、性能好 | 贵、vendor lock-in | 快速启动、不想运维 |
| Milvus/Zilliz | 开源、高性能、功能丰富 | 需要运维 | 大规模、自研能力强的团队 |
| PGVector | PostgreSQL 扩展、成本低 | 性能一般 | 已有 PG 基础设施、小规模 |
| Chroma | 轻量、易嵌入 | 不适合生产 | 原型开发、本地测试 |
| Weaviate | 支持 GraphQL、多模态 | 学习曲线 | 需要复杂查询、多模态 |
Long-term Memory：持久的知识
什么是 Long-term Memory？
Long-term Memory 存储 Agent 的”毕生所学”：
- 用户画像：每个用户的偏好、习惯、历史
- 业务知识：行业规则、最佳实践、成功案例
- 世界知识：组织信息、流程规范、人际关系
- 学习成果：从交互中总结的模式和经验
实现方案
1. 用户画像（User Profile）
class UserProfile:
"""
长期存储的用户画像
"""
def __init__(self, user_id: str, db: Database):
self.user_id = user_id
self.db = db
self._profile = None
def load(self):
"""从数据库加载用户画像"""
self._profile = self.db.query_one(
"SELECT * FROM user_profiles WHERE user_id = ?",
self.user_id
)
if not self._profile:
self._profile = self._create_default_profile()
def get_preference(self, key: str, default=None):
"""获取用户偏好"""
return self._profile.get(f"pref_{key}", default)
def set_preference(self, key: str, value: any):
"""设置用户偏好"""
self._profile[f"pref_{key}"] = value
self._persist()
def learn_from_interaction(self, interaction: Interaction):
"""从交互中学习并更新画像"""
# 提取用户偏好
new_preferences = self._extract_preferences(interaction)
for key, value in new_preferences.items():
self.set_preference(key, value)
# 更新行为模式
self._update_behavior_pattern(interaction)
# 更新成功模式
if interaction.successful:
self._add_success_pattern(interaction)
def _extract_preferences(self, interaction: Interaction) -> dict:
"""使用 LLM 从交互中提取用户偏好"""
prompt = f"""
Based on this interaction, what user preferences can we infer?
User message: {interaction.user_message}
Agent response: {interaction.agent_response}
User feedback: {interaction.feedback}
Extract preferences in JSON format:
communication_style
"""
return json.loads(self.llm.generate(prompt))
2. 业务知识库
class BusinessKnowledgeBase:
"""
结构化的业务知识存储
"""
def __init__(self, db: Database):
self.db = db
def add_rule(self, rule: BusinessRule):
"""添加业务规则"""
self.db.execute("""
INSERT INTO business_rules (name, condition, action, priority)
VALUES (?, ?, ?, ?)
""", rule.name, rule.condition, rule.action, rule.priority)
def get_applicable_rules(self, context: dict) -> List[BusinessRule]:
"""获取适用于当前上下文的规则"""
all_rules = self.db.query("SELECT * FROM business_rules ORDER BY priority")
applicable = []
for rule in all_rules:
if self._evaluate_condition(rule.condition, context):
applicable.append(rule)
return applicable
def add_case_study(self, case: CaseStudy):
"""添加成功案例"""
# 存储到向量数据库用于语义检索
embedding = self.embedding_model.embed(case.description)
self.vector_store.add(
content=case.description,
embedding=embedding,
metadata={
"type": "case_study",
"industry": case.industry,
"outcome": case.outcome,
"solution": case.solution
}
)
def find_similar_cases(self, situation: str, k: int = 3) -> List[CaseStudy]:
"""查找相似的成功案例"""
query_embedding = self.embedding_model.embed(situation)
results = self.vector_store.search(query_embedding, k=k, filters={"type": "case_study"})
return [CaseStudy.from_record(r) for r in results]
3. 组织知识
class OrganizationalKnowledge:
"""
组织结构、流程、人员信息
"""
def __init__(self, db: Database):
self.db = db
def get_org_chart(self) -> OrgChart:
"""获取组织结构"""
return OrgChart.from_db(self.db)
def find_expert(self, topic: str) -> Optional[Person]:
"""找到某领域的专家"""
# 使用向量检索找到最相关的人
query_embedding = self.embedding_model.embed(topic)
results = self.vector_store.search(
query_embedding,
k=1,
filters={"type": "person", "is_expert": True}
)
return Person.from_record(results[0]) if results else None
def get_process(self, process_name: str) -> Process:
"""获取流程定义"""
return self.db.query_one(
"SELECT * FROM processes WHERE name = ?",
process_name
)
Knowledge Graph：关系的网络
为什么需要 Knowledge Graph？
Vector DB 擅长语义相似性检索，但不擅长关系推理。
示例问题：
“谁是 @postcodeeng 的经理的经理？”
- Vector DB：无法理解”经理的经理”这种关系链
- Knowledge Graph：可以轻松遍历关系网络
实现方案
class AgentKnowledgeGraph:
"""
Agent 的知识图谱系统
"""
def __init__(self, graph_db: GraphDatabase):
self.graph = graph_db
# ========== 实体管理 ==========
def add_entity(self, entity: Entity):
"""添加实体"""
self.graph.run("""
MERGE (e:Entity {id: $id})
SET e.name = $name,
e.type = $type,
e.properties = $properties
""", id=entity.id, name=entity.name,
type=entity.type, properties=json.dumps(entity.properties))
def add_relation(self, from_id: str, relation_type: str, to_id: str,
properties: dict = None):
"""添加关系"""
self.graph.run("""
MATCH (from:Entity {id: $from_id})
MATCH (to:Entity {id: $to_id})
MERGE (from)-[r:$relation_type]->(to)
SET r.properties = $properties
""", from_id=from_id, to_id=to_id,
relation_type=relation_type,
properties=json.dumps(properties or {}))
# ========== 查询能力 ==========
def find_related(self, entity_id: str, relation_type: str = None,
depth: int = 1) -> List[Entity]:
"""查找相关实体"""
if relation_type:
query = """
MATCH (e:Entity {id: $entity_id})-[:$relation_type*1..$depth]-(related)
RETURN related
"""
else:
query = """
MATCH (e:Entity {id: $entity_id})-[*1..$depth]-(related)
RETURN related
"""
results = self.graph.run(query, entity_id=entity_id,
relation_type=relation_type, depth=depth)
return [Entity.from_record(r) for r in results]
def find_path(self, from_id: str, to_id: str) -> Optional[List[Relation]]:
"""查找两个实体之间的最短路径"""
result = self.graph.run("""
MATCH path = shortestPath(
(a:Entity {id: $from_id})-[*]-(b:Entity {id: $to_id})
)
RETURN path
""", from_id=from_id, to_id=to_id)
return self._parse_path(result[0]) if result else None
def infer_relation(self, entity_a: str, entity_b: str) -> Optional[str]:
"""推断两个实体之间的关系"""
# 查找共同连接的实体
common = self.graph.run("""
MATCH (a:Entity {id: $a})--(common)--(b:Entity {id: $b})
RETURN common, count(*) as strength
ORDER BY strength DESC
LIMIT 1
""", a=entity_a, b=entity_b)
if common:
return f"Both related to {common[0]['name']}"
return None
# ========== 从文本自动构建 ==========
def extract_from_text(self, text: str):
"""从文本中提取实体和关系"""
prompt = f"""
Extract entities and relations from the following text.
Text: {text}
Output in JSON format:
entities}}
],
"relations": [
from
]
}}
"""
extraction = json.loads(self.llm.generate(prompt))
# 添加到图谱
for entity in extraction["entities"]:
self.add_entity(Entity(**entity))
for relation in extraction["relations"]:
self.add_relation(**relation)
Knowledge Graph 的应用场景
| 场景 | KG 查询示例 |
|---|---|
| 权限查询 | “谁可以审批这个预算？” → 查找汇报链 |
| 影响分析 | “如果这个系统宕机，会影响哪些业务？” → 查找依赖关系 |
| 专家定位 | “谁是机器学习方面的专家？” → 查找技能关系 |
| 流程优化 | “这个流程的瓶颈在哪里？” → 分析流程图 |
| 客户分析 | “这个客户和哪些决策者有关系？” → 查找关系网络 |
记忆系统的工程实现
整体架构
┌─────────────────────────────────────────────────────┐
│ Agent Memory System │
├─────────────────────────────────────────────────────┤
│ Working Memory Short-term Memory │
│ (In-Memory) (Vector DB) │
│ • ContextWindow • Semantic retrieval │
│ • EntityTracker • Temporal decay │
│ • Summarizer • Importance scoring │
├─────────────────────────────────────────────────────┤
│ Long-term Memory Knowledge Graph │
│ (Relational DB) (Graph DB) │
│ • User profiles • Entity relationships │
│ • Business rules • Inference queries │
│ • Case studies • Path finding │
└─────────────────────────────────────────────────────┘
关键设计决策
1. 记忆一致性
如何保证不同层次记忆之间的一致性？
class MemoryConsistencyManager:
"""
确保各层记忆的一致性
"""
def sync_working_to_short_term(self, session_end: bool = False):
"""会话结束时同步到短期记忆"""
if session_end:
# 总结整个会话
summary = self.working_memory.summarize()
self.short_term_memory.add(
content=summary,
memory_type="session_summary",
importance=0.7
)
def consolidate_to_long_term(self):
"""将重要的短期记忆提升到长期记忆"""
important_memories = self.short_term_memory.query(
filters={"importance": {"$>": 0.8}},
older_than=timedelta(days=7)
)
for memory in important_memories:
# 提取结构化知识
if memory.type == "user_preference":
self.long_term_memory.update_user_profile(memory)
elif memory.type == "business_rule":
self.long_term_memory.add_business_rule(memory)
# 更新知识图谱
self.knowledge_graph.extract_from_text(memory.content)
2. 多模态记忆
如何处理文本、图像、音频等多种模态？
class MultiModalMemory:
"""
多模态记忆系统
"""
def add(self, content: Union[str, Image, Audio], modality: str):
"""添加多模态记忆"""
if modality == "text":
embedding = self.text_embedder.embed(content)
elif modality == "image":
embedding = self.image_embedder.embed(content)
elif modality == "audio":
# 先转文本
text = self.speech_to_text.transcribe(content)
embedding = self.text_embedder.embed(text)
self.vector_store.add(
embedding=embedding,
content=content,
metadata={"modality": modality}
)
def retrieve(self, query: str, query_modality: str = "text",
target_modalities: List[str] = None):
"""跨模态检索"""
# 查询向量化
if query_modality == "text":
query_embedding = self.text_embedder.embed(query)
# 检索所有模态
results = self.vector_store.search(
query_embedding,
filters={"modality": {"$in": target_modalities or ["text", "image", "audio"]}}
)
return results
3. 隐私和安全
class PrivacyAwareMemory:
"""
隐私感知的记忆系统
"""
def __init__(self):
self.sensitivity_classifier = SensitivityClassifier()
self.encryption = Encryption()
def add(self, content: str, user_id: str):
"""添加记忆时进行隐私处理"""
# 识别敏感信息
sensitivity = self.sensitivity_classifier.classify(content)
if sensitivity == "HIGH":
# 加密存储
encrypted = self.encryption.encrypt(content, key=user_id)
self.store_encrypted(encrypted, sensitivity)
elif sensitivity == "MEDIUM":
# 匿名化处理
anonymized = self.anonymize(content)
self.store(anonymized, sensitivity)
else:
# 正常存储
self.store(content, sensitivity)
def anonymize(self, content: str) -> str:
"""匿名化处理"""
# 使用 NER 识别 PII
pii_entities = self.ner.extract_pii(content)
# 替换为占位符
for entity in pii_entities:
content = content.replace(entity.text, f"[{entity.type}]")
return content
写在最后
记忆系统是 Agent 的”大脑”，它决定了 Agent 能走多远。
从简单的上下文窗口，到企业级的多层次记忆系统，这是一个逐步演进的过程：
- MVP 阶段：Context Window + 简单的向量检索
- 产品阶段：添加 Short-term Memory + User Profiles
- 企业阶段：完整的 Long-term Memory + Knowledge Graph
关键成功因素：
- 数据质量：垃圾进，垃圾出
- 检索精度：召回率和准确率的平衡
- 隐私合规：敏感数据的处理不能妥协
- 成本控制：嵌入和存储的成本需要优化
最后的话：
一个 Agent 可以没有花哨的界面，可以没有复杂的工作流，但不能没有记忆。
记忆让 Agent 从”工具”变成”伙伴”。
📚 延伸阅读
本系列文章
- Agent OS：SaaS 之后的下一个软件形态
- 为什么你的 SaaS 产品需要 Agent 层？
- 从 Human-driven 到 Agent-driven
- Agent OS 的五层架构模型
- Multi-Agent 协作
外部资源
Agent OS 系列 - 第 5 篇 由 @postcodeeng 整理发布
Published on 2026-04-07 阅读时间：约 18 分钟
下一篇预告：《Multi-Agent 协作》
💬 评论
💡 使用 GitHub 账号登录 即可参与讨论