为什么 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。

记忆压缩：情景记忆数量超过 1000 条时，对同主题记忆做聚类摘要，将 10 条相似记忆合并为 1 条摘要记忆，保留最关键的 2 条原始记忆。

避免记忆污染：设置写入门槛，只在对话中检测到"事实性信息"或"用户偏好"时才写入语义记忆，避免闲聊内容污染知识库。

记忆审计：定期用 LLM 审查语义记忆，检测过时或矛盾的信息（如用户之前说用 Java，后来说转 Go 了，需要更新）。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。