为什么 Agent 需要记忆 LLM 本身是无状态的——每次调用都是独立的。对话历史塞进 Context Window 是最简单的"记忆"，但有致命缺陷：token 成本线性增长、上下文长度有上限、无法跨会话持久化。一个真正有用的 Agent 需要像人一样拥有多层次记忆系统。
四种记忆类型 记忆类型 类比人类 生命周期 存储介质 示例 工作记忆 工作台上的文件 当前任务 Redis/内存 当前对话上下文、临时变量 情景记忆 个人经历 天~月 向量数据库 “用户上周问了XX问题” 语义记忆 知识事实 长期 向量数据库 + 图数据库 “用户是Java开发者” 程序记忆 肌肉记忆 长期 规则库/技能库 “调用API的固定流程” 架构设计 ┌──────────────────────────────────────────────────┐ │ Agent 记忆系统 │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ 工作记忆 │ │ 情景记忆 │ │ 语义记忆 │ │ │ │ (Redis) │ │(Qdrant) │ │(Qdrant) │ │ │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ │ │ │ │ └──────────────┼──────────────┘ │ │ ▼ │ │ ┌──────────────┐ │ │ │ 记忆管理器 │ │ │ │ (Memory Mgr) │ │ │ └──────┬───────┘ │ │ ▼ │ │ ┌──────────────┐ │ │ │ 程序记忆 │ │ │ │ (规则库) │ │ │ └──────────────┘ │ └──────────────────────────────────────────────────┘ 核心实现 1. 记忆数据结构 from dataclasses import dataclass, field from datetime import datetime from enum import Enum import uuid class MemoryType(Enum): WORKING = "working" EPISODIC = "episodic" SEMANTIC = "semantic" PROCEDURAL = "procedural" @dataclass class Memory: id: str = field(default_factory=lambda: str(uuid.uuid4())) type: MemoryType = MemoryType.EPISODIC content: str = "" embedding: list[float] = field(default_factory=list) metadata: dict = field(default_factory=dict) importance: float = 0.5 # 0-1, 影响遗忘策略 created_at: datetime = field(default_factory=datetime.now) last_accessed: datetime = field(default_factory=datetime.now) access_count: int = 0 decay_factor: float = 1.0 # 随时间衰减 2. 记忆写入 import redis import json from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct class MemoryStore: def __init__(self): self.redis = redis.Redis(host='localhost', port=6379, db=0) self.qdrant = QdrantClient(host='localhost', port=6333) self._init_collections() def _init_collections(self): """初始化向量集合""" for collection in ["episodic", "semantic"]: if not self.qdrant.collection_exists(collection): self.qdrant.create_collection( collection_name=collection, vectors_config=VectorParams(size=1024, distance=Distance.COSINE) ) async def write(self, memory: Memory): """写入记忆""" if memory.type == MemoryType.WORKING: # 工作记忆写入 Redis，设置过期时间 key = f"working:{memory.id}" self.redis.setex( key, 3600, # 1小时过期 json.dumps({ "content": memory.content, "metadata": memory.metadata }) ) elif memory.type in (MemoryType.EPISODIC, MemoryType.SEMANTIC): # 情景/语义记忆写入向量库 self.qdrant.upsert( collection_name=memory.type.value, points=[PointStruct( id=memory.id, vector=memory.embedding, payload={ "content": memory.content, "metadata": memory.metadata, "importance": memory.importance, "created_at": memory.created_at.isoformat(), "last_accessed": memory.last_accessed.isoformat(), "access_count": memory.access_count } )] ) async def search(self, query_vector, memory_type: MemoryType, top_k=5): """检索记忆""" if memory_type == MemoryType.WORKING: # 工作记忆：返回所有未过期的 keys = self.redis.keys("working:*") results = [] for key in keys: data = json.loads(self.redis.get(key)) results.append(data) return results # 向量检索 hits = self.qdrant.search( collection_name=memory_type.value, query_vector=query_vector, limit=top_k, score_threshold=0.7 ) return [hit.payload for hit in hits] 3. 记忆检索与排序 class MemoryRetriever: """记忆检索器：综合相关性、重要性、时效性""" def __init__(self, store: MemoryStore): self.store = store async def retrieve(self, query: str, query_vector: list, context: dict = None) -> list[dict]: # 并行检索三种记忆 import asyncio tasks = [ self.store.search(query_vector, MemoryType.WORKING), self.store.search(query_vector, MemoryType.EPISODIC, top_k=5), self.store.search(query_vector, MemoryType.SEMANTIC, top_k=5), ] working, episodic, semantic = await asyncio.gather(*tasks) # 融合排序 all_memories = [] for mem in episodic + semantic: score = self._compute_score(mem, query_vector) all_memories.append((mem, score)) all_memories.sort(key=lambda x: x[1], reverse=True) # 返回 top-K + 工作记忆 return { "working": working, "long_term": [m[0] for m in all_memories[:5]] } def _compute_score(self, mem, query_vector): """综合评分 = 相关性 × 时效性 × 重要性 × 访问频率""" import numpy as np from datetime import datetime, timedelta # 相关性（向量相似度，假设payload中存了） relevance = mem.get("score", 0.5) # 时效性衰减 created = datetime.fromisoformat(mem["created_at"]) days_old = (datetime.now() - created).days recency = np.exp(-days_old / 30) # 30天半衰期 # 重要性 importance = mem.get("importance", 0.5) # 访问频率（越常访问越重要） access_factor = np.log1p(mem.get("access_count", 0)) return relevance * (0.3 + 0.4 * recency) * importance * (1 + 0.1 * access_factor) 4. 遗忘机制 class ForgettingManager: """遗忘机制：模拟人类遗忘曲线""" async def decay_memories(self): """定期执行，降低旧记忆的重要性""" from datetime import datetime, timedelta cutoff = datetime.now() - timedelta(days=7) # 在向量库中批量更新 self.store.qdrant.set_payload( collection_name="episodic", payload={"decay_factor": 0.8}, # 每次衰减20% filters={ "must": [ {"field": "last_accessed", "range": {"lt": cutoff.isoformat()}} ] } ) async def cleanup_memories(self): """删除重要性低于阈值的记忆""" self.store.qdrant.delete( collection_name="episodic", points_selector={ "filter": { "must": [ {"field": "importance", "range": {"lt": 0.1}}, {"field": "decay_factor", "range": {"lt": 0.3}} ] } } ) 实战建议 重要性评分策略：不要让 Agent 自己打分（LLM 评分不稳定）。用规则：用户明确要求记住的 → 1.0，任务关键信息 → 0.8，普通对话 → 0.3。
...
