1259 字
3 分钟
Agent 记忆系统：短期、长期与向量数据库
「你叫什么名字？」「我叫小智。」
「我叫什么名字？」「抱歉，我不知道您的名字。」
这是很多 AI 系统的真实写照——它们没有记忆。
没有记忆的 Agent 就像《记忆碎片》的主角，每次对话都是全新的开始，无法积累经验，无法建立用户画像，无法提供个性化服务。
记忆系统是 Agent 从「一次性工具」进化为「长期伙伴」的关键基础设施。
本文要点
- Agent 记忆的三层架构
- 短期记忆：上下文窗口管理
- 长期记忆：向量数据库选型与实践
- 情景记忆：任务执行状态追踪
- 记忆检索策略
- 记忆压缩与遗忘机制
一、为什么 Agent 需要记忆？
1.1 没有记忆的问题
用户：「帮我分析一下苹果公司的股票」
AI：「好的，苹果公司股票分析如下...」
（三天后）
用户：「上次那个分析，能再详细说说吗？」
AI：「抱歉，我不知道您指的哪个分析...」
问题：1. 无法记住历史对话2. 无法理解用户指代3. 无法积累用户偏好4. 每次都要从头开始1.2 记忆的价值
flowchart TD
A[记忆系统] --> B[用户画像]
A --> C[任务上下文]
A --> D[知识积累]
A --> E[经验学习]
B --> B1[个性化服务]
C --> C1[多轮任务支持]
D --> D1[知识库构建]
E --> E1[能力持续提升]
记忆带来的核心价值：
| 价值 | 说明 | 示例 |
|---|---|---|
| 连续性 | 支持跨会话对话 | 「上次讨论的项目进展如何？」 |
| 个性化 | 记住用户偏好 | 「按您习惯的格式生成报告」 |
| 学习性 | 从经验中改进 | 避免重复犯同样的错误 |
| 效率 | 减少重复沟通 | 无需每次解释背景信息 |
二、Agent 记忆的三层架构
2.1 架构总览
flowchart TB
subgraph 感知层
A[用户输入]
end
subgraph 记忆层
B[短期记忆<br/>Working Memory]
C[长期记忆<br/>Long-term Memory]
D[情景记忆<br/>Episodic Memory]
end
subgraph 存储层
E[内存/缓存]
F[向量数据库]
G[关系数据库]
end
A --> B
B <--> C
B <--> D
C --> F
D --> G
B --> E
2.2 三层记忆对比
| 类型 | 存储位置 | 生命周期 | 容量 | 用途 |
|---|---|---|---|---|
| 短期记忆 | 内存/缓存 | 会话级 | 有限 | 当前对话上下文 |
| 长期记忆 | 向量数据库 | 持久化 | 大规模 | 用户画像、知识库 |
| 情景记忆 | 关系数据库 | 任务级 | 中等 | 任务执行状态 |
三、短期记忆（Working Memory）
3.1 什么是短期记忆？
短期记忆是 Agent 处理当前任务时的「工作台」，存储正在使用的临时信息。
┌─────────────────────────────────────────────────────────────┐│ 短期记忆结构 │├─────────────────────────────────────────────────────────────┤│ ││ 系统提示词（System Prompt） ││ ├── 角色定义 ││ ├── 能力说明 ││ └── 输出规范 ││ ││ 对话历史（Conversation History） ││ ├── 用户消息 1 ││ ├── 助手回复 1 ││ ├── 用户消息 2 ││ ├── 助手回复 2 ││ └── ... ││ ││ 工具调用记录（Tool Calls） ││ ├── 调用：search("天气") ││ ├── 结果：北京晴 25°C ││ └── ... ││ ││ 当前任务状态（Task State） ││ ├── 已完成步骤 ││ ├── 待执行步骤 ││ └── 中间结果 ││ │└─────────────────────────────────────────────────────────────┘3.2 上下文窗口管理
大模型的上下文窗口有限（如 128K tokens），需要合理管理。
flowchart LR
A[新消息] --> B{窗口是否已满?}
B -->|否| C[直接添加]
B -->|是| D[压缩/摘要]
D --> E[替换旧内容]
E --> C
C --> F[更新窗口]
管理策略
策略 1：滑动窗口（Sliding Window）
class SlidingWindowMemory: """滑动窗口记忆管理"""
def __init__(self, max_messages: int = 20): self.max_messages = max_messages self.messages = []
def add_message(self, role: str, content: str): self.messages.append({"role": role, "content": content}) # 保持窗口大小 if len(self.messages) > self.max_messages: self.messages = self.messages[-self.max_messages:]
def get_context(self) -> list: return self.messages问题：简单粗暴，可能丢失重要信息。
策略 2：摘要压缩（Summarization）
class SummaryMemory: """摘要压缩记忆管理"""
def __init__(self, llm, max_tokens: int = 4000): self.llm = llm self.max_tokens = max_tokens self.summary = "" self.recent_messages = []
def add_message(self, role: str, content: str): self.recent_messages.append({"role": role, "content": content})
# 超过阈值时压缩 if self._estimate_tokens() > self.max_tokens: self._compress()
def _compress(self): """用 LLM 生成摘要""" prompt = f"""请总结以下对话的关键信息：
{self._format_messages(self.recent_messages)}
当前摘要：{self.summary}
请生成新的综合摘要，保留重要信息。"""
self.summary = self.llm.generate(prompt) # 保留最近几条消息 self.recent_messages = self.recent_messages[-5:]
def get_context(self) -> list: context = [] if self.summary: context.append({ "role": "system", "content": f"历史对话摘要：{self.summary}" }) context.extend(self.recent_messages) return context策略 3：混合策略（Hybrid）
class HybridMemory: """混合记忆策略：重要消息保留 + 普通消息压缩"""
def __init__(self, llm, max_tokens: int = 8000): self.llm = llm self.max_tokens = max_tokens self.important_messages = [] # 重要消息永久保留 self.regular_messages = [] # 普通消息可压缩 self.summary = ""
def add_message(self, role: str, content: str, important: bool = False): message = {"role": role, "content": content, "important": important}
if important: self.important_messages.append(message) else: self.regular_messages.append(message)
if self._estimate_tokens() > self.max_tokens: self._compress_regular_messages()
def _compress_regular_messages(self): """压缩普通消息""" if len(self.regular_messages) > 3: to_compress = self.regular_messages[:-3] self.regular_messages = self.regular_messages[-3:]
prompt = f"总结以下对话内容：\n{self._format_messages(to_compress)}" new_summary = self.llm.generate(prompt)
self.summary = f"{self.summary}\n{new_summary}" if self.summary else new_summary四、长期记忆（Long-term Memory）
4.1 什么是长期记忆？
长期记忆用于持久化存储用户信息、知识库、历史交互等，支持跨会话检索。
flowchart TD
A[长期记忆] --> B[用户画像]
A --> C[知识库]
A --> D[历史交互]
B --> B1[偏好设置]
B --> B2[个人信息]
B --> B3[行为习惯]
C --> C1[文档知识]
C --> C2[领域知识]
C --> C3[操作手册]
D --> D1[历史对话]
D --> D2[任务记录]
D --> D3[反馈结果]
4.2 向量数据库选型
长期记忆通常使用向量数据库存储，支持语义检索。
┌─────────────────────────────────────────────────────────────┐│ 主流向量数据库对比 │├─────────────────────────────────────────────────────────────┤│ ││ ChromaDB ││ ├── 特点：轻量级、易上手、开源 ││ ├── 适用：中小规模、快速原型 ││ └── 部署：本地/Docker ││ ││ FAISS（Meta） ││ ├── 特点：高性能、纯向量搜索 ││ ├── 适用：大规模向量检索 ││ └── 部署：本地库 ││ ││ Milvus ││ ├── 特点：分布式、企业级、功能丰富 ││ ├── 适用：大规模生产环境 ││ └── 部署：云原生/Kubernetes ││ ││ Pinecone ││ ├── 特点：托管服务、零运维 ││ ├── 适用：快速上线、不想管基础设施 ││ └── 部署：云服务 ││ ││ Weaviate ││ ├── 特点：GraphQL API、混合搜索 ││ ├── 适用：需要结构化+向量混合查询 ││ └── 部署：Docker/云服务 ││ │└─────────────────────────────────────────────────────────────┘4.3 实现示例：ChromaDB
from chromadb import Clientfrom chromadb.config import Settingsfrom openai import OpenAIfrom typing import List, Dictimport uuid
class LongTermMemory: """基于 ChromaDB 的长期记忆系统"""
def __init__(self, collection_name: str = "agent_memory"): self.client = Client(Settings( chroma_db_impl="duckdb+parquet", persist_directory="./chroma_db" )) self.collection = self.client.get_or_create_collection( name=collection_name, metadata={"hnsw:space": "cosine"} ) self.embedder = OpenAI() # 用于生成 embedding
def _get_embedding(self, text: str) -> List[float]: """生成文本的向量表示""" response = self.embedder.embeddings.create( model="text-embedding-3-small", input=text ) return response.data[0].embedding
def store(self, content: str, metadata: Dict = None) -> str: """存储记忆""" memory_id = str(uuid.uuid4()) embedding = self._get_embedding(content)
self.collection.add( ids=[memory_id], embeddings=[embedding], documents=[content], metadatas=[metadata or {}] )
return memory_id
def recall(self, query: str, n_results: int = 5) -> List[Dict]: """检索相关记忆""" query_embedding = self._get_embedding(query)
results = self.collection.query( query_embeddings=[query_embedding], n_results=n_results, include=["documents", "metadatas", "distances"] )
memories = [] for i in range(len(results["ids"][0])): memories.append({ "id": results["ids"][0][i], "content": results["documents"][0][i], "metadata": results["metadatas"][0][i], "distance": results["distances"][0][i] })
return memories
def forget(self, memory_id: str): """删除记忆""" self.collection.delete(ids=[memory_id])
def update(self, memory_id: str, new_content: str, new_metadata: Dict = None): """更新记忆""" embedding = self._get_embedding(new_content)
self.collection.update( ids=[memory_id], embeddings=[embedding], documents=[new_content], metadatas=[new_metadata or {}] )4.4 用户画像存储
class UserProfileMemory: """用户画像记忆系统"""
def __init__(self, long_term_memory: LongTermMemory): self.memory = long_term_memory
def learn_preference(self, user_id: str, preference: str, value: str): """学习用户偏好""" content = f"用户偏好：{preference} = {value}" metadata = { "type": "preference", "user_id": user_id, "preference_key": preference } self.memory.store(content, metadata)
def get_user_preferences(self, user_id: str) -> Dict[str, str]: """获取用户所有偏好""" # 这里可以扩展为更复杂的查询 results = self.memory.recall(f"用户 {user_id} 偏好", n_results=20)
preferences = {} for item in results: if item["metadata"].get("type") == "preference": key = item["metadata"].get("preference_key") # 解析偏好内容 if "=" in item["content"]: value = item["content"].split("=")[1].strip() preferences[key] = value
return preferences五、情景记忆（Episodic Memory）
5.1 什么是情景记忆？
情景记忆记录特定任务执行的完整过程，包括步骤、决策、结果。
flowchart TD
A[情景记忆] --> B[任务目标]
A --> C[执行步骤]
A --> D[决策过程]
A --> E[执行结果]
A --> F[经验总结]
C --> C1[Step 1: 搜索]
C --> C2[Step 2: 分析]
C --> C3[Step 3: 生成]
D --> D1[为什么选择这个工具]
D --> D2[为什么做出这个决策]
5.2 情景记忆的结构
from dataclasses import dataclass, fieldfrom datetime import datetimefrom typing import List, Dict, Anyimport json
@dataclassclass ExecutionStep: """执行步骤""" step_id: int action: str # 执行的动作 tool: str # 使用的工具 input: Dict # 输入参数 output: Any # 输出结果 success: bool # 是否成功 timestamp: datetime = field(default_factory=datetime.now)
@dataclassclass EpisodicMemory: """情景记忆""" episode_id: str task: str # 任务描述 goal: str # 任务目标 steps: List[ExecutionStep] # 执行步骤 decisions: List[Dict] # 决策过程 outcome: str # 最终结果 success: bool # 是否成功 lessons_learned: str # 经验总结 created_at: datetime = field(default_factory=datetime.now)
def to_dict(self) -> Dict: return { "episode_id": self.episode_id, "task": self.task, "goal": self.goal, "steps": [ { "step_id": s.step_id, "action": s.action, "tool": s.tool, "input": s.input, "output": str(s.output), "success": s.success, "timestamp": s.timestamp.isoformat() } for s in self.steps ], "decisions": self.decisions, "outcome": self.outcome, "success": self.success, "lessons_learned": self.lessons_learned, "created_at": self.created_at.isoformat() }5.3 情景记忆管理器
import sqlite3from typing import List, Optionalfrom datetime import datetime
class EpisodicMemoryManager: """情景记忆管理器 - 使用 SQLite 存储"""
def __init__(self, db_path: str = "./episodes.db"): self.conn = sqlite3.connect(db_path) self._init_db()
def _init_db(self): """初始化数据库""" self.conn.execute(""" CREATE TABLE IF NOT EXISTS episodes ( episode_id TEXT PRIMARY KEY, task TEXT, goal TEXT, steps_json TEXT, decisions_json TEXT, outcome TEXT, success BOOLEAN, lessons_learned TEXT, created_at TIMESTAMP ) """) self.conn.commit()
def save_episode(self, episode: EpisodicMemory): """保存情景记忆""" self.conn.execute(""" INSERT OR REPLACE INTO episodes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( episode.episode_id, episode.task, episode.goal, json.dumps([s.__dict__ for s in episode.steps], default=str), json.dumps(episode.decisions), episode.outcome, episode.success, episode.lessons_learned, episode.created_at )) self.conn.commit()
def get_episode(self, episode_id: str) -> Optional[EpisodicMemory]: """获取特定情景""" cursor = self.conn.execute( "SELECT * FROM episodes WHERE episode_id = ?", (episode_id,) ) row = cursor.fetchone() if row: return self._row_to_episode(row) return None
def search_episodes(self, query: str, limit: int = 10) -> List[EpisodicMemory]: """搜索相关情景""" cursor = self.conn.execute(""" SELECT * FROM episodes WHERE task LIKE ? OR outcome LIKE ? OR lessons_learned LIKE ? ORDER BY created_at DESC LIMIT ? """, (f"%{query}%", f"%{query}%", f"%{query}%", limit))
return [self._row_to_episode(row) for row in cursor.fetchall()]
def get_successful_episodes(self, task_type: str = None) -> List[EpisodicMemory]: """获取成功的情景（用于学习）""" if task_type: cursor = self.conn.execute( "SELECT * FROM episodes WHERE success = 1 AND task LIKE ?", (f"%{task_type}%",) ) else: cursor = self.conn.execute( "SELECT * FROM episodes WHERE success = 1" ) return [self._row_to_episode(row) for row in cursor.fetchall()]六、记忆检索策略
6.1 检索时机
flowchart TD
A[用户请求] --> B{需要记忆?}
B -->|新任务| C[检索长期记忆]
B -->|继续对话| D[使用短期记忆]
B -->|类似任务| E[检索情景记忆]
C --> F[语义检索]
E --> G[关键词+语义]
F --> H[整合上下文]
G --> H
D --> H
H --> I[生成响应]
6.2 混合检索策略
class HybridRetrieval: """混合检索：关键词 + 语义 + 时间"""
def __init__(self, long_term_memory: LongTermMemory): self.memory = long_term_memory
def retrieve(self, query: str, user_id: str = None, top_k: int = 5, recency_weight: float = 0.3) -> List[Dict]: """混合检索"""
# 1. 语义检索 semantic_results = self.memory.recall(query, n_results=top_k * 2)
# 2. 过滤和重排序 scored_results = [] for item in semantic_results: score = 1 - item["distance"] # 语义相似度
# 时间衰减 if "timestamp" in item["metadata"]: age_days = (datetime.now() - datetime.fromisoformat(item["metadata"]["timestamp"])).days recency_score = 1 / (1 + age_days / 30) # 30天半衰期 score = (1 - recency_weight) * score + recency_weight * recency_score
# 用户相关性 if user_id and item["metadata"].get("user_id") == user_id: score *= 1.2 # 用户相关记忆加权
scored_results.append({**item, "score": score})
# 3. 按综合分数排序 scored_results.sort(key=lambda x: x["score"], reverse=True)
return scored_results[:top_k]6.3 记忆重要性评估
class MemoryImportance: """评估记忆的重要性"""
@staticmethod def calculate_importance(memory_item: Dict) -> float: """计算记忆重要性分数"""
score = 0.5 # 基础分
# 1. 访问频率 access_count = memory_item.get("access_count", 0) score += min(access_count * 0.05, 0.2)
# 2. 类型权重 memory_type = memory_item.get("type", "") type_weights = { "preference": 0.3, # 用户偏好很重要 "fact": 0.2, # 事实信息 "experience": 0.25, # 经验教训 "temporary": -0.1 # 临时信息降权 } score += type_weights.get(memory_type, 0)
# 3. 反馈信号 if memory_item.get("positive_feedback"): score += 0.15 if memory_item.get("negative_feedback"): score -= 0.1
# 4. 确定性 confidence = memory_item.get("confidence", 0.5) score += (confidence - 0.5) * 0.2
return max(0, min(1, score)) # 归一化到 [0, 1]七、记忆压缩与遗忘
7.1 为什么需要压缩和遗忘？
问题：1. 存储成本：记忆无限增长会消耗大量存储2. 检索效率：记忆越多，检索越慢3. 噪声干扰：过多低价值记忆影响决策4. 过时信息：旧记忆可能已不相关
解决：1. 压缩：将多条记忆合并为摘要2. 遗忘：删除低价值或过时记忆3. 归档：将冷数据移到低成本存储7.2 记忆压缩策略
flowchart TD
A[原始记忆] --> B{压缩判断}
B -->|相似度高| C[合并]
B -->|时间跨度大| D[摘要]
B -->|同类型多| E[聚类压缩]
C --> F[新记忆条目]
D --> F
E --> F
F --> G[删除原始记忆]
压缩实现示例：
class MemoryCompressor: """记忆压缩器"""
def __init__(self, llm): self.llm = llm self.similarity_threshold = 0.85
def compress_similar_memories(self, memories: List[Dict]) -> Dict: """压缩相似记忆"""
if len(memories) < 2: return memories[0] if memories else None
# 合并内容 combined_content = "\n".join([ f"- {m['content']}" for m in memories ])
# 使用 LLM 生成压缩摘要 prompt = f"""请将以下多条相似记忆压缩为一条，保留关键信息：
{combined_content}
要求：1. 保留所有独特信息2. 去除重复内容3. 保持简洁4. 一句话概括"""
compressed = self.llm.generate(prompt)
# 合并元数据 merged_metadata = self._merge_metadata(memories)
return { "content": compressed, "metadata": merged_metadata, "compressed_from": len(memories) }
def _merge_metadata(self, memories: List[Dict]) -> Dict: """合并元数据""" merged = {} for m in memories: metadata = m.get("metadata", {}) for key, value in metadata.items(): if key not in merged: merged[key] = value elif isinstance(merged[key], list): if value not in merged[key]: merged[key].append(value) elif merged[key] != value: merged[key] = [merged[key], value] return merged7.3 遗忘机制
from datetime import datetime, timedeltafrom typing import List
class ForgettingMechanism: """遗忘机制"""
def __init__(self, memory_manager): self.memory_manager = memory_manager
# 遗忘策略参数 self.max_age_days = 365 # 最大保留天数 self.min_importance = 0.1 # 最低重要性阈值 self.max_access_interval = 90 # 最大未访问天数
def should_forget(self, memory_item: Dict) -> bool: """判断是否应该遗忘"""
# 1. 时间因素 created_at = datetime.fromisoformat( memory_item.get("created_at", datetime.now().isoformat()) ) age_days = (datetime.now() - created_at).days
if age_days > self.max_age_days: return True
# 2. 重要性因素 importance = MemoryImportance.calculate_importance(memory_item) if importance < self.min_importance: return True
# 3. 访问频率因素 last_accessed = memory_item.get("last_accessed") if last_accessed: last_access = datetime.fromisoformat(last_accessed) days_since_access = (datetime.now() - last_access).days
# 长期未访问且重要性低 if days_since_access > self.max_access_interval and importance < 0.3: return True
# 4. 特殊标记 if memory_item.get("marked_for_deletion"): return True
return False
def apply_forgetting(self, batch_size: int = 100) -> int: """批量应用遗忘策略""" forgotten_count = 0
# 获取所有记忆 all_memories = self.memory_manager.get_all_memories(limit=10000)
for memory in all_memories: if self.should_forget(memory): self.memory_manager.delete(memory["id"]) forgotten_count += 1
if forgotten_count >= batch_size: break
return forgotten_count
def archive_old_memories(self, days: int = 180) -> int: """归档旧记忆""" threshold = datetime.now() - timedelta(days=days)
archived = self.memory_manager.move_to_archive( condition={"created_before": threshold.isoformat()} )
return archived7.4 Ebbinghaus 遗忘曲线
xychart-beta
title "Ebbinghaus 遗忘曲线与复习策略"
x-axis ["1天", "2天", "6天", "14天", "30天", "60天"]
y-axis "记忆保留率 %" 0 --> 100
line [33, 25, 20, 15, 10, 5]
line [100, 100, 100, 100, 100, 100]
class SpacedRepetition: """间隔重复：对抗遗忘"""
def __init__(self): # 复习间隔（天） self.intervals = [1, 2, 6, 14, 30, 60]
def get_next_review_date(self, memory_item: Dict) -> datetime: """计算下次复习日期""" review_count = memory_item.get("review_count", 0)
if review_count >= len(self.intervals): # 已完成所有复习，转为长期记忆 return None
interval = self.intervals[review_count] return datetime.now() + timedelta(days=interval)
def review_memory(self, memory_id: str, success: bool): """复习记忆""" memory = self.memory_manager.get(memory_id)
if success: # 复习成功，增加复习次数 memory["review_count"] = memory.get("review_count", 0) + 1 memory["next_review"] = self.get_next_review_date(memory) memory["importance"] = min(1.0, memory.get("importance", 0.5) + 0.1) else: # 复习失败，重置 memory["review_count"] = 0 memory["next_review"] = datetime.now() + timedelta(days=1)
self.memory_manager.update(memory)八、完整记忆系统架构
class AgentMemorySystem: """Agent 完整记忆系统"""
def __init__(self, llm, config: Dict = None): self.config = config or {}
# 三层记忆 self.working_memory = HybridMemory(llm) self.long_term_memory = LongTermMemory() self.episodic_memory = EpisodicMemoryManager()
# 辅助组件 self.retriever = HybridRetrieval(self.long_term_memory) self.compressor = MemoryCompressor(llm) self.forgetting = ForgettingMechanism(self.long_term_memory)
def remember(self, content: str, memory_type: str = "short", metadata: Dict = None): """存储记忆""" if memory_type == "short": self.working_memory.add_message( "system", content, important=metadata.get("important", False) ) elif memory_type == "long": metadata = metadata or {} metadata["created_at"] = datetime.now().isoformat() self.long_term_memory.store(content, metadata)
def recall(self, query: str, include_types: List[str] = None) -> Dict: """检索记忆""" include_types = include_types or ["short", "long", "episodic"]
results = {}
if "short" in include_types: results["working"] = self.working_memory.get_context()
if "long" in include_types: results["long_term"] = self.retriever.retrieve(query)
if "episodic" in include_types: results["episodes"] = self.episodic_memory.search_episodes(query)
return results
def save_episode(self, episode: EpisodicMemory): """保存情景记忆""" self.episodic_memory.save_episode(episode)
# 提取经验教训存入长期记忆 if episode.lessons_learned: self.remember( f"经验：{episode.lessons_learned}", memory_type="long", metadata={"type": "experience", "task": episode.task} )
def maintain(self): """记忆维护：压缩和遗忘""" # 定期执行 forgotten = self.forgetting.apply_forgetting() archived = self.forgetting.archive_old_memories()
return {"forgotten": forgotten, "archived": archived}常见问题 FAQ
Q1：短期记忆和长期记忆如何协同工作？
A：
- 短期记忆处理当前对话，实时更新
- 重要信息从短期记忆提取，存入长期记忆
- 长期记忆通过检索补充短期记忆
Q2：向量数据库检索会不会很慢？
A：
- 现代 向量数据库 支持 ANN（近似最近邻）算法，检索速度快
- 通常在毫秒级别完成 TOP-K 检索
- 可以通过索引优化、分区策略进一步提升
Q3：如何避免记忆中存储敏感信息？
A：
- 在存储前进行敏感信息检测和脱敏
- 设置访问控制，限制记忆的访问范围
- 支持用户主动删除特定记忆
Q4：记忆系统会增加多少成本？
A：
- 向量存储成本：约 $0.1/GB/月
- Embedding 成本：约 $0.0001/1K tokens
- 对于个人用户，月成本通常在 $1-10 范围
Q5：如何评估记忆系统的效果？
A：
- 检索准确率：检索结果与实际需求的匹配度
- 记忆命中率：Agent 使用记忆的频率
- 用户满意度：个性化服务的反馈
小结
记忆系统是 Agent 的「大脑存储」，决定了 Agent 能否持续进化。
核心要点回顾：
┌─────────────────────────────────────────────────────────────┐│ 记忆系统核心总结 │├─────────────────────────────────────────────────────────────┤│ ││ 三层架构：短期 + 长期 + 情景 ││ ││ 短期记忆：上下文窗口管理（滑动窗口/摘要/混合） ││ ││ 长期记忆：向量数据库（ChromaDB/Milvus/FAISS） ││ ││ 情景记忆：任务执行过程的完整记录 ││ ││ 检索策略：语义 + 关键词 + 时间 + 重要性 ││ ││ 压缩遗忘：控制存储成本，保持记忆质量 ││ │└─────────────────────────────────────────────────────────────┘下一步学习：
下篇预告
《Agent 工具调用：Function Calling 与 Tool Use 实战》
深入解析：
- Function Calling 的原理与实现
- OpenAI vs Anthropic 的工具调用对比
- Tool Schema 设计最佳实践
- 多工具协同策略
参考资料
支持与分享
如果这篇文章对你有帮助，欢迎支持作者或分享给更多人
Agent 记忆系统：短期、长期与向量数据库
https://blog.souloss.com/posts/machine-learning/agent-guide/memory-system/ 部分信息可能已经过时
相关文章 智能推荐
1
向量数据库深度解析
AI 深入解析主流向量数据库的架构原理、性能对比与选型指南，涵盖 Milvus、Qdrant、Pinecone、Chroma、Weaviate。
2
Embedding 与向量搜索原理
AI 深入解析文本 Embedding 模型与向量检索——模型原理、索引结构、相似度计算与工程实践。
3
让AI拥有知识：RAG检索增强生成详解
AI 让AI拥有知识——RAG检索增强生成详解
4
AI 工程化实践
AI AI 工程化全景导览——从提示词工程到多模态系统，梳理大模型落地的核心工程能力与知识体系
5
RAG 优化策略深度解析
AI 深入解析 RAG 检索增强生成的优化策略——Query 改写、混合检索、重排序、Context 压缩与引用追踪。