MindCarver Blog

MindCarver

MindCarver

技术服务于产品

记忆系统最佳实践

16分钟阅读

记忆系统最佳实践

1. 记忆类型选择策略

1.1 决策树

是否需要长期保存？
├─ 否 → 工作记忆
└─ 是
    ├─ 是否需要时间序列？
    │   ├─ 是 → 情景记忆
    │   └─ 否
    │       ├─ 是否涉及抽象知识和概念？
    │       │   ├─ 是 → 语义记忆
    │       │   └─ 否
    │           └─ 是否是多模态数据？
    │               ├─ 是 → 感知记忆
    │               └─ 否 → 语义记忆

1.2 场景映射表

使用场景推荐记忆类型理由
当前对话上下文working临时、需要快速访问
临时状态跟踪working会话结束后可丢弃
历史对话记录episodic需要时间序列和完整性
重要事件里程碑episodic需要长期保存和回顾
用户偏好semantic抽象概念、需要关系推理
领域知识semantic知识体系、概念关联
技能学习进度episodic学习经历追踪
图像/视频内容perceptual多模态支持
语音信息perceptual音频编码和检索

2. 重要性评分指南

2.1 评分标准

重要性范围含义适用内容
0.0-0.3低优先级临时信息、测试数据、重复内容
0.4-0.6中等优先级一般对话、普通事件、日常信息
0.7-0.8高优先级重要事件、学习记录、有意义的对话
0.9-1.0核心优先级关键偏好、核心知识、重要里程碑

2.2 自动化评分策略

def calculate_importance(
    content: str,
    event_type: str = None,
    user_feedback: float = None
) -> float:
    """计算记忆重要性"""
 
    base_score = 0.5
 
    # 基于事件类型
    event_weights = {
        "milestone": 0.9,
        "question": 0.6,
        "preference": 0.85,
        "learning": 0.8,
        "casual": 0.5
    }
 
    if event_type:
        base_score = event_weights.get(event_type, 0.5)
 
    # 基于内容长度（长内容可能更重要）
    if len(content) > 200:
        base_score += 0.1
 
    # 基于用户反馈
    if user_feedback:
        base_score = max(base_score, user_feedback)
 
    return min(1.0, max(0.0, base_score))
 
# 使用示例
importance = calculate_importance(
    content="用户完成了第一个Python项目",
    event_type="milestone"
)
# importance = 0.9

3. 性能优化技巧

3.1 批量操作

# ❌ 低效：逐条添加
for item in items:
    memory_tool.execute("add", **item)
 
# ✅ 高效：批量添加（如果支持）
memory_tool.execute("batch_add", items=items)

3.2 合理设置限制

# 搜索时设置合理的limit
result = memory_tool.execute("search",
    query="机器学习",
    limit=10  # 根据实际需求调整
)
 
# 使用memory_type过滤减少搜索范围
result = memory_tool.execute("search",
    query="Python",
    memory_type="semantic",  # 只搜索语义记忆
    limit=5
)

3.3 缓存常用查询

from functools import lru_cache
 
class CachedMemoryTool:
    def __init__(self, memory_tool):
        self.tool = memory_tool
 
    @lru_cache(maxsize=100)
    def search(self, query: str, memory_type: str, limit: int = 5):
        return self.tool.execute("search",
            query=query,
            memory_type=memory_type,
            limit=limit
        )

3.4 异步操作

import asyncio
 
async def async_add_memory(memory_tool, content, **kwargs):
    """异步添加记忆"""
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        None,
        lambda: memory_tool.execute("add", content=content, **kwargs)
    )
    return result
 
# 使用示例
async def main():
    tasks = [
        async_add_memory(tool, f"记忆{i}", **kwargs)
        for i, kwargs in enumerate(data)
    ]
    await asyncio.gather(*tasks)

4. 记忆维护策略

4.1 定期整合

def scheduled_consolidation():
    """定期整合记忆"""
 
    # 每天执行一次
    schedule.every().day.at("02:00").do(
        memory_tool.execute,
        "consolidate",
        from_type="working",
        to_type="episodic",
        importance_threshold=0.7
    )
 
    # 每周执行一次更深层次的整合
    schedule.every().week.do(
        memory_tool.execute,
        "consolidate",
        from_type="episodic",
        to_type="semantic",
        importance_threshold=0.85
    )

4.2 智能遗忘策略

def smart_forget():
    """智能遗忘策略"""
 
    # 1. 清理低重要性且过期的记忆
    memory_tool.execute("forget",
        strategy="time_based",
        max_age_days=90
    )
 
    # 2. 清理极低重要性的记忆（不管时间）
    memory_tool.execute("forget",
        strategy="importance_based",
        threshold=0.2
    )
 
    # 3. 检查容量，必要时清理
    stats = memory_tool.execute("stats")
    if stats["total_count"] > 10000:
        memory_tool.execute("forget",
            strategy="capacity_based",
            threshold=0.3
        )

4.3 记忆质量评估

def evaluate_memory_quality():
    """评估记忆质量"""
 
    stats = memory_tool.execute("stats")
 
    # 检查重要性分布
    importance_avg = stats["average_importance"]
    if importance_avg < 0.5:
        print("⚠️ 警告：记忆平均重要性较低，考虑清理低质量记忆")
 
    # 检查时间分布
    oldest = stats["oldest_memory"]
    if (datetime.now() - oldest).days > 365:
        print("⚠️ 警告：存在超过1年的记忆，考虑清理")
 
    # 检查类型分布
    if stats["working_memory_count"] > 100:
        print("⚠️ 警告：工作记忆过多，考虑整合到长期记忆")

5. 错误处理和容错

5.1 完整的错误处理

def safe_add_memory(memory_tool, content, **kwargs):
    """安全地添加记忆"""
 
    try:
        # 验证参数
        if not content or not content.strip():
            return "❌ 记忆内容不能为空"
 
        if len(content) > 10000:
            return "❌ 记忆内容过长（最大10000字符）"
 
        # 添加记忆
        result = memory_tool.execute("add", content=content, **kwargs)
 
        # 验证结果
        if "✅" in result:
            return result
        else:
            raise Exception("添加记忆返回异常结果")
 
    except MemoryError:
        return "❌ 内存不足，无法添加记忆"
 
    except ConnectionError:
        return "❌ 数据库连接失败"
 
    except Exception as e:
        logging.error(f"添加记忆失败: {e}")
        return f"❌ 添加记忆失败: {str(e)}"

5.2 重试机制

from tenacity import retry, stop_after_attempt, wait_exponential
 
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def add_memory_with_retry(memory_tool, content, **kwargs):
    """带重试的记忆添加"""
    return memory_tool.execute("add", content=content, **kwargs)

5.3 数据备份

def backup_memories():
    """备份记忆数据"""
 
    try:
        # 导出所有记忆
        memories = []
        for memory_type in ["working", "episodic", "semantic", "perceptual"]:
            result = memory_tool.execute("search",
                query="",  # 空查询返回所有
                memory_type=memory_type,
                limit=1000
            )
            memories.extend(result)
 
        # 保存到文件
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_file = f"./memory_backups/backup_{timestamp}.json"
 
        with open(backup_file, 'w', encoding='utf-8') as f:
            json.dump(memories, f, ensure_ascii=False, indent=2)
 
        print(f"✅ 记忆已备份到: {backup_file}")
 
    except Exception as e:
        print(f"❌ 备份失败: {e}")

6. 监控和分析

6.1 记忆使用监控

def monitor_memory_usage():
    """监控记忆使用情况"""
 
    stats = memory_tool.execute("stats")
 
    print(f"""
    📊 记忆使用情况
    ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
    总记忆数: {stats['total_count']}
    工作记忆: {stats['working_memory_count']} ({stats['working_memory_count']/stats['total_count']*100:.1f}%)
    情景记忆: {stats['episodic_memory_count']} ({stats['episodic_memory_count']/stats['total_count']*100:.1f}%)
    语义记忆: {stats['semantic_memory_count']} ({stats['semantic_memory_count']/stats['total_count']*100:.1f}%)
    感知记忆: {stats['perceptual_memory_count']} ({stats['perceptual_memory_count']/stats['total_count']*100:.1f}%)
 
    平均重要性: {stats['average_importance']:.2f}
    最近记忆: {stats['latest_memory']}
    最早记忆: {stats['oldest_memory']}
    """)
 
    # 检查告警条件
    if stats['total_count'] > 5000:
        print("⚠️ 警告：记忆总数过多，建议清理")
 
    if stats['average_importance'] < 0.5:
        print("⚠️ 警告：平均重要性较低，建议整合重要记忆")

6.2 记忆增长趋势

def track_memory_growth():
    """追踪记忆增长趋势"""
 
    import matplotlib.pyplot as plt
 
    # 获取历史数据（需要提前记录）
    history = get_memory_history()
 
    dates = [h['date'] for h in history]
    counts = [h['count'] for h in history]
 
    plt.figure(figsize=(12, 6))
    plt.plot(dates, counts, marker='o')
    plt.title('记忆数量增长趋势')
    plt.xlabel('日期')
    plt.ylabel('记忆数量')
    plt.grid(True)
    plt.xticks(rotation=45)
    plt.tight_layout()
    plt.savefig('memory_growth.png')
    plt.show()

6.3 主题分析

def analyze_memory_topics():
    """分析记忆主题分布"""
 
    # 使用关键词或主题模型
    from sklearn.feature_extraction.text import TfidfVectorizer
    from sklearn.cluster import KMeans
 
    # 获取所有记忆内容
    contents = []
    for memory_type in ["episodic", "semantic"]:
        result = memory_tool.execute("search",
            query="",
            memory_type=memory_type,
            limit=100
        )
        contents.extend(result)
 
    # TF-IDF特征提取
    vectorizer = TfidfVectorizer(max_features=100)
    X = vectorizer.fit_transform(contents)
 
    # K-means聚类
    kmeans = KMeans(n_clusters=5, random_state=42)
    kmeans.fit(X)
 
    # 输出主题
    feature_names = vectorizer.get_feature_names_out()
    for i, center in enumerate(kmeans.cluster_centers_):
        top_indices = center.argsort()[-10:][::-1]
        top_words = [feature_names[idx] for idx in top_indices]
        print(f"主题 {i+1}: {', '.join(top_words)}")

7. 高级应用场景

7.1 个性化推荐

def personalized_recommendation(user_id: str):
    """基于记忆的个性化推荐"""
 
    # 1. 检索用户偏好
    preferences = memory_tool.execute("search",
        query="用户偏好",
        memory_type="semantic",
        user_id=user_id,
        limit=10
    )
 
    # 2. 检索历史行为
    behaviors = memory_tool.execute("search",
        query="用户行为",
        memory_type="episodic",
        user_id=user_id,
        limit=20
    )
 
    # 3. 分析偏好和行为模式
    # ... 分析逻辑 ...
 
    # 4. 生成推荐
    recommendations = generate_recommendations(preferences, behaviors)
 
    return recommendations

7.2 学习路径规划

def plan_learning_path(user_id: str, topic: str):
    """基于学习记录规划学习路径"""
 
    # 1. 检索学习进度
    progress = memory_tool.execute("search",
        query=f"{topic} 学习进度",
        memory_type="episodic",
        user_id=user_id,
        limit=10
    )
 
    # 2. 分析已掌握的知识
    mastered_knowledge = []
    for item in progress:
        if "掌握" in item or "完成" in item:
            mastered_knowledge.append(item)
 
    # 3. 检索相关知识图谱
    knowledge_graph = memory_tool.execute("search",
        query=topic,
        memory_type="semantic",
        limit=20
    )
 
    # 4. 规划下一步学习内容
    next_steps = plan_next_steps(mastered_knowledge, knowledge_graph)
 
    return next_steps

7.3 情感记忆

def add_emotional_memory(content: str, emotion: str, intensity: float):
    """添加带有情感标记的记忆"""
 
    # 情感权重
    emotion_weights = {
        "happy": 1.1,
        "sad": 0.9,
        "angry": 1.0,
        "excited": 1.15,
        "calm": 1.0
    }
 
    # 计算重要性
    base_importance = 0.6
    emotion_weight = emotion_weights.get(emotion, 1.0)
    final_importance = min(1.0, base_importance * emotion_weight * (1 + intensity * 0.2))
 
    # 添加记忆
    memory_tool.execute("add",
        content=content,
        memory_type="episodic",
        importance=final_importance,
        emotion=emotion,
        emotion_intensity=intensity
    )
 
# 使用示例
add_emotional_memory(
    content="用户第一次成功运行机器学习模型，非常兴奋",
    emotion="excited",
    intensity=0.9
)

7.4 跨会话记忆

def maintain_cross_session_memory(user_id: str):
    """维护跨会话记忆"""
 
    # 1. 获取所有会话ID
    all_sessions = get_all_sessions(user_id)
 
    # 2. 分析会话间的关联
    session_relations = analyze_session_relations(all_sessions)
 
    # 3. 整合相关会话到长期记忆
    for relation in session_relations:
        if relation["similarity"] > 0.8:
            consolidate_sessions(relation["session1"], relation["session2"])
 
def consolidate_sessions(session1_id: str, session2_id: str):
    """整合两个相关会话"""
 
    # 获取两个会话的记忆
    memories1 = get_session_memories(session1_id)
    memories2 = get_session_memories(session2_id)
 
    # 合并重要记忆
    all_memories = memories1 + memories2
    important_memories = [
        m for m in all_memories
        if m.importance > 0.7
    ]
 
    # 整合到语义记忆
    for memory in important_memories:
        memory_tool.execute("consolidate",
            from_type="episodic",
            to_type="semantic",
            importance_threshold=0.8
        )

8. 调试和故障排除

8.1 记忆不生效

def debug_memory_not_working():
    """调试记忆不生效的问题"""
 
    # 1. 检查工具是否正确注册
    print("检查工具注册...")
    if not agent.tool_registry.is_registered("memory"):
        print("❌ 记忆工具未注册")
        return
 
    # 2. 检查用户ID是否正确
    print("检查用户ID...")
    print(f"当前用户ID: {memory_tool.user_id}")
 
    # 3. 测试添加记忆
    print("测试添加记忆...")
    result = memory_tool.execute("add",
        content="测试记忆",
        memory_type="working"
    )
    print(f"添加结果: {result}")
 
    # 4. 测试检索记忆
    print("测试检索记忆...")
    result = memory_tool.execute("search", query="测试")
    print(f"检索结果: {result}")
 
    # 5. 检查数据库连接
    print("检查数据库连接...")
    stats = memory_tool.execute("stats")
    print(f"数据库状态: {stats}")

8.2 检索结果不准确

def debug_inaccurate_search():
    """调试检索结果不准确的问题"""
 
    # 1. 检查查询是否清晰
    print("检查查询...")
    query = "机器学习"
    print(f"查询词: {query}")
    print(f"查询长度: {len(query)}")
 
    # 2. 尝试不同的记忆类型
    print("尝试不同记忆类型...")
    for memory_type in ["working", "episodic", "semantic"]:
        result = memory_tool.execute("search",
            query=query,
            memory_type=memory_type,
            limit=3
        )
        print(f"{memory_type}: {len(result)} 条结果")
 
    # 3. 检查重要性阈值
    print("检查重要性阈值...")
    result = memory_tool.execute("search",
        query=query,
        min_importance=0.1  # 降低阈值
    )
    print(f"低阈值结果数: {len(result)}")
 
    # 4. 检查记忆内容
    print("检查记忆内容...")
    all_memories = memory_tool.execute("search", query="")
    print(f"总记忆数: {len(all_memories)}")

8.3 性能问题

def diagnose_performance_issues():
    """诊断性能问题"""
 
    import time
 
    # 1. 测试添加性能
    print("测试添加性能...")
    start = time.time()
    for i in range(10):
        memory_tool.execute("add",
            content=f"测试记忆 {i}",
            memory_type="working"
        )
    end = time.time()
    print(f"添加10条记忆耗时: {end - start:.2f}秒")
 
    # 2. 测试检索性能
    print("测试检索性能...")
    start = time.time()
    for i in range(10):
        memory_tool.execute("search", query=f"测试 {i}")
    end = time.time()
    print(f"检索10次耗时: {end - start:.2f}秒")
 
    # 3. 检查记忆数量
    print("检查记忆数量...")
    stats = memory_tool.execute("stats")
    print(f"总记忆数: {stats['total_count']}")
 
    # 4. 建议
    if stats['total_count'] > 10000:
        print("💡 建议：记忆数量过多，考虑清理或分片存储")

9. 安全和隐私

9.1 敏感信息过滤

def filter_sensitive_info(content: str) -> str:
    """过滤敏感信息"""
 
    import re
 
    # 手机号
    content = re.sub(r'1[3-9]\d{9}', '[手机号已隐藏]', content)
 
    # 邮箱
    content = re.sub(r'\w+@\w+\.\w+', '[邮箱已隐藏]', content)
 
    # 身份证号
    content = re.sub(r'\d{17}[\dXx]', '[身份证已隐藏]', content)
 
    # 银行卡号
    content = re.sub(r'\d{16,19}', '[银行卡已隐藏]', content)
 
    return content
 
# 使用示例
memory_tool.execute("add",
    content=filter_sensitive_info(user_input),
    memory_type="episodic"
)

9.2 数据加密

from cryptography.fernet import Fernet
 
class EncryptedMemoryTool:
    """加密的记忆工具"""
 
    def __init__(self, memory_tool, encryption_key):
        self.tool = memory_tool
        self.cipher = Fernet(encryption_key)
 
    def execute(self, action, **kwargs):
        if action == "add" and "content" in kwargs:
            # 加密内容
            encrypted = self.cipher.encrypt(
                kwargs["content"].encode()
            )
            kwargs["content"] = encrypted.decode()
 
        result = self.tool.execute(action, **kwargs)
 
        if action == "search" and "🔍" in result:
            # 解密结果
            result = self._decrypt_result(result)
 
        return result

9.3 访问控制

def check_access_permission(user_id: str, target_user_id: str) -> bool:
    """检查访问权限"""
 
    # 管理员可以访问所有数据
    if is_admin(user_id):
        return True
 
    # 用户只能访问自己的数据
    return user_id == target_user_id
 
def secure_memory_search(user_id: str, query: str, **kwargs):
    """安全的记忆检索"""
 
    # 检查权限
    target_user_id = kwargs.get("user_id")
    if not check_access_permission(user_id, target_user_id):
        return "❌ 无权访问"
 
    # 执行检索
    result = memory_tool.execute("search", query=query, **kwargs)
 
    # 脱敏敏感信息
    result = mask_sensitive_data(result)
 
    return result

10. 总结

10.1 核心原则

  1. 合理选择记忆类型：根据场景选择最适合的记忆类型
  2. 设置合适的重要性：影响记忆的保留和检索优先级
  3. 定期维护：整合、清理、优化记忆系统
  4. 性能优化：合理设置限制、使用缓存、异步操作
  5. 错误处理：完善的异常处理和重试机制
  6. 安全隐私：过滤敏感信息、加密存储、访问控制

10.2 最佳实践清单

  • 根据场景选择正确的记忆类型
  • 为记忆设置合适的重要性分数
  • 定期执行记忆整合和清理
  • 监控记忆使用情况和性能
  • 实现完善的错误处理
  • 保护用户隐私和敏感信息
  • 做好数据备份和恢复
  • 定期分析和优化记忆系统

10.3 常见问题

Q: 如何选择记忆类型？ A: 参考场景映射表，根据需求选择：临时→工作记忆，历史事件→情景记忆，抽象知识→语义记忆，多媒体→感知记忆。

Q: 重要性如何设置？ A: 0.0-0.3低优先级，0.4-0.6中等优先级，0.7-0.8高优先级，0.9-1.0核心优先级。

Q: 记忆太多怎么办？ A: 使用forget策略清理，定期整合到长期记忆，设置合理的容量限制。

Q: 如何提高检索准确率？ A: 优化查询词，使用正确的记忆类型，设置合适的重要性阈值，考虑混合检索策略。

Q: 如何保护隐私？ A: 过滤敏感信息，使用加密，实现访问控制，遵循数据保护法规。

关系图谱