#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 代码示例 03: WorkingMemory实现详解 展示工作记忆的混合检索策略和TTL机制 """ import time from datetime import datetime, timedelta from typing import List, Dict, Any from hello_agents.tools import MemoryTool from hello_agents.memory import MemoryItem class WorkingMemoryDemo: """工作记忆演示类""" def __init__(self): self.memory_tool = MemoryTool( user_id="working_memory_demo", memory_types=["working"] # 只启用工作记忆 ) def demonstrate_capacity_management(self): """演示容量管理和TTL机制""" print("🧠 工作记忆容量管理演示") print("=" * 50) print("工作记忆特点:") print("• 容量有限（默认50条）") print("• TTL机制（默认60分钟）") print("• 自动清理过期记忆") print("• 优先级管理（重要性排序）") # 添加多条记忆来演示容量管理 print(f"\n📝 添加测试记忆...") for i in range(10): importance = 0.3 + (i * 0.07) # 递增重要性 self.memory_tool.execute("add", content=f"工作记忆测试项目 {i+1} - 重要性 {importance:.2f}", memory_type="working", importance=importance, test_id=i+1, category="capacity_test" ) # 查看当前状态 stats = self.memory_tool.execute("stats") print(f"当前状态: {stats}") # 演示重要性排序 print(f"\n🔍 按重要性搜索:") result = self.memory_tool.execute("search", query="测试项目", memory_type="working", limit=5 ) print(result) def demonstrate_mixed_retrieval_strategy(self): """演示混合检索策略""" print("\n🔍 混合检索策略演示") print("-" * 40) print("混合检索策略包括:") print("• TF-IDF向量化语义检索") print("• 关键词匹配检索") print("• 时间衰减因子") print("• 重要性权重调整") # 添加不同类型的记忆用于检索测试 test_memories = [ { "content": "Python是一种高级编程语言，语法简洁清晰", "importance": 0.8, "topic": "programming", "language": "python" }, { "content": "机器学习是人工智能的重要分支，包括监督学习和无监督学习", "importance": 0.9, "topic": "ai", "domain": "machine_learning" }, { "content": "数据结构包括数组、链表、栈、队列等基本结构", "importance": 0.7, "topic": "computer_science", "category": "data_structures" }, { "content": "算法复杂度分析使用大O记号来描述时间和空间复杂度", "importance": 0.8, "topic": "algorithms", "analysis": "complexity" } ] print(f"\n📝 添加测试记忆...") for i, memory in enumerate(test_memories): content = memory.pop("content") importance = memory.pop("importance") self.memory_tool.execute("add", content=content, memory_type="working", importance=importance, **memory ) # 测试不同类型的检索 search_tests = [ ("Python编程", "测试语义匹配"), ("学习", "测试关键词匹配"), ("复杂度", "测试部分匹配"), ("人工智能机器学习", "测试多词匹配") ] print(f"\n🔍 混合检索测试:") for query, description in search_tests: print(f"\n查询: '{query}' ({description})") result = self.memory_tool.execute("search", query=query, memory_type="working", limit=2 ) print(f"结果: {result}") def demonstrate_time_decay_mechanism(self): """演示时间衰减机制""" print("\n⏰ 时间衰减机制演示") print("-" * 40) print("时间衰减机制:") print("• 新记忆权重更高") print("• 旧记忆权重衰减") print("• 模拟人类记忆特点") print("• 平衡新旧信息重要性") # 添加不同时间的记忆（模拟） time_test_memories = [ ("最新的重要信息 - 刚刚学习的概念", 0.7, "newest"), ("较新的信息 - 昨天学习的内容", 0.7, "recent"), ("较旧的信息 - 上周学习的内容", 0.7, "older"), ("最旧的信息 - 很久以前的内容", 0.7, "oldest") ] print(f"\n📝 添加不同时期的记忆...") for content, importance, age_category in time_test_memories: self.memory_tool.execute("add", content=content, memory_type="working", importance=importance, age_category=age_category, timestamp_category=age_category ) # 搜索测试时间衰减效果 print(f"\n🔍 时间衰减效果测试:") result = self.memory_tool.execute("search", query="学习的内容", memory_type="working", limit=4 ) print("搜索结果（注意时间因素对排序的影响）:") print(result) def demonstrate_automatic_cleanup(self): """演示自动清理机制""" print("\n🧹 自动清理机制演示") print("-" * 40) print("自动清理机制:") print("• 过期记忆自动清理") print("• 容量超限时清理低优先级记忆") print("• 保持系统性能和响应速度") print("• 模拟工作记忆的有限容量") # 获取清理前的状态 stats_before = self.memory_tool.execute("stats") print(f"\n清理前状态: {stats_before}") # 添加一些低重要性的记忆 print(f"\n📝 添加低重要性记忆...") for i in range(5): self.memory_tool.execute("add", content=f"低重要性临时记忆 {i+1}", memory_type="working", importance=0.1 + i * 0.05, temporary=True, cleanup_test=True ) # 触发基于重要性的清理 print(f"\n🧹 执行基于重要性的清理...") cleanup_result = self.memory_tool.execute("forget", strategy="importance_based", threshold=0.3 ) print(f"清理结果: {cleanup_result}") # 获取清理后的状态 stats_after = self.memory_tool.execute("stats") print(f"\n清理后状态: {stats_after}") def demonstrate_performance_characteristics(self): """演示性能特征""" print("\n⚡ 性能特征演示") print("-" * 40) print("工作记忆性能特点:") print("• 纯内存存储，访问速度极快") print("• 无需磁盘I/O，响应时间短") print("• 适合频繁访问的临时数据") print("• 系统重启后数据丢失（符合设计）") # 性能测试 print(f"\n⏱️ 性能测试:") # 批量添加测试 start_time = time.time() for i in range(20): self.memory_tool.execute("add", content=f"性能测试记忆 {i+1}", memory_type="working", importance=0.5, performance_test=True ) add_time = time.time() - start_time print(f"批量添加20条记忆耗时: {add_time:.3f}秒") # 批量搜索测试 start_time = time.time() for i in range(10): self.memory_tool.execute("search", query=f"性能测试", memory_type="working", limit=3 ) search_time = time.time() - start_time print(f"批量搜索10次耗时: {search_time:.3f}秒") # 获取最终统计 final_stats = self.memory_tool.execute("stats") print(f"\n📊 最终统计: {final_stats}") def main(): """主函数""" print("🧠 WorkingMemory实现详解") print("展示工作记忆的核心特性和实现机制") print("=" * 60) try: demo = WorkingMemoryDemo() # 1. 容量管理演示 demo.demonstrate_capacity_management() # 2. 混合检索策略演示 demo.demonstrate_mixed_retrieval_strategy() # 3. 时间衰减机制演示 demo.demonstrate_time_decay_mechanism() # 4. 自动清理机制演示 demo.demonstrate_automatic_cleanup() # 5. 性能特征演示 demo.demonstrate_performance_characteristics() print("\n" + "=" * 60) print("🎉 WorkingMemory实现演示完成！") print("=" * 60) print("\n✨ 工作记忆核心特性:") print("1. 🧠 有限容量 - 模拟人类工作记忆限制") print("2. ⚡ 高速访问 - 纯内存存储，响应迅速") print("3. 🔍 混合检索 - 语义+关键词+时间+重要性") print("4. ⏰ 时间衰减 - 新信息优先，旧信息衰减") print("5. 🧹 自动清理 - TTL机制+优先级管理") print("\n🎯 设计理念:") print("• 临时性 - 存储当前会话的临时信息") print("• 高效性 - 快速访问和处理能力") print("• 智能性 - 自动管理和优化策略") print("• 仿生性 - 模拟人类工作记忆特点") except Exception as e: print(f"\n❌ 演示过程中发生错误: {e}") import traceback traceback.print_exc() if __name__ == "__main__": main()