basic_usage.py•6.71 kB
#!/usr/bin/env python3
"""
MemOS Memory Ops SDK 基础使用示例
演示如何使用轻量级SDK进行记忆管理操作
"""
import sys
from pathlib import Path
# 添加父目录到Python路径
sys.path.insert(0, str(Path(__file__).parent.parent))
from memory_ops_sdk import MemoryOps, create_memory_ops, get_global_memory_ops
def basic_operations_example():
"""基础操作示例"""
print("🧪 基础操作示例")
print("=" * 40)
# 方式1: 直接创建实例
mem = MemoryOps(data_dir="./example_memories", verbose=True)
try:
# 添加记忆
print("\n📝 添加记忆...")
mem.add("今天学习了Python编程基础", tags=["学习", "Python"])
mem.add("完成了机器学习项目的数据预处理", tags=["项目", "机器学习"])
mem.add("阅读了关于深度学习的论文", tags=["学习", "深度学习"])
# 查询记忆
print("\n🔍 查询记忆...")
results = mem.query("Python编程")
print(f"找到 {len(results)} 条相关记忆:")
for i, result in enumerate(results, 1):
content = result.get('content', '')
score = result.get('score', 0)
print(f" {i}. {content[:50]}... (分数: {score:.3f})")
# 提供反馈
if results:
print("\n👍 提供反馈...")
memory_id = results[0].get('id')
if memory_id:
mem.feedback(memory_id, "👍")
# 获取统计信息
print("\n📊 获取统计信息...")
stats = mem.stats()
if stats:
print(f"系统状态: {stats.get('status', {}).get('status', '未知')}")
print(f"反馈统计: {stats.get('feedback', {})}")
print("\n✅ 基础操作示例完成")
finally:
mem.close()
def context_manager_example():
"""上下文管理器示例"""
print("\n🔧 上下文管理器示例")
print("=" * 40)
# 使用上下文管理器自动管理资源
with MemoryOps(data_dir="./example_memories", verbose=True) as mem:
# 添加记忆
mem.add("使用上下文管理器添加的记忆", tags=["示例"])
# 查询记忆
results = mem.query("上下文管理器")
print(f"找到 {len(results)} 条相关记忆")
# 健康检查
is_healthy = mem.health()
print(f"系统健康状态: {'✅ 正常' if is_healthy else '❌ 异常'}")
print("✅ 上下文管理器示例完成（资源已自动清理）")
def convenience_functions_example():
"""便捷函数示例"""
print("\n⚡ 便捷函数示例")
print("=" * 40)
# 方式2: 使用便捷函数创建
mem = create_memory_ops("./example_memories", verbose=True)
try:
# 一行代码操作
mem.add("便捷函数创建的记忆", tags=["便捷"])
results = mem.query("便捷函数", limit=3)
print(f"便捷函数查询结果: {len(results)} 条")
# 方式3: 使用全局单例
global_mem = get_global_memory_ops(data_dir="./example_memories", verbose=True)
global_mem.add("全局单例添加的记忆", tags=["全局"])
print("✅ 便捷函数示例完成")
finally:
mem.close()
def advanced_features_example():
"""高级功能示例"""
print("\n🚀 高级功能示例")
print("=" * 40)
with MemoryOps(data_dir="./example_memories", verbose=True) as mem:
# 添加带元数据的记忆
metadata = {
"source": "示例脚本",
"priority": "high",
"category": "技术"
}
mem.add("带元数据的高级记忆", tags=["高级", "元数据"], metadata=metadata)
# 高级查询（禁用某些功能）
results = mem.query(
"高级记忆",
limit=5,
use_reranker=True,
use_time_decay=True,
use_topic_drift=False
)
print(f"高级查询结果: {len(results)} 条")
# 导出记忆
print("\n📁 导出记忆...")
memories = mem.dump("example_backup.json")
print(f"已导出 {len(memories)} 条记忆")
# 系统优化
print("\n⚡ 系统优化...")
success = mem.optimize()
print(f"优化结果: {'✅ 成功' if success else '❌ 失败'}")
print("✅ 高级功能示例完成")
def error_handling_example():
"""错误处理示例"""
print("\n🛡️ 错误处理示例")
print("=" * 40)
# 测试未初始化的情况
mem = MemoryOps(auto_init=False, verbose=True)
try:
# 这应该会抛出异常
mem.add("这会失败")
except RuntimeError as e:
print(f"✅ 正确捕获异常: {e}")
# 手动初始化
if mem.init():
print("✅ 手动初始化成功")
mem.add("手动初始化后的记忆", tags=["测试"])
mem.close()
else:
print("❌ 手动初始化失败")
print("✅ 错误处理示例完成")
def performance_test():
"""性能测试示例"""
print("\n⚡ 性能测试示例")
print("=" * 40)
import time
with MemoryOps(data_dir="./example_memories", verbose=False) as mem:
# 批量添加测试
print("测试批量添加性能...")
start_time = time.time()
for i in range(10):
mem.add(f"性能测试记忆 {i}", tags=["性能测试"])
add_time = time.time() - start_time
print(f"添加10条记忆耗时: {add_time:.2f}秒")
# 批量查询测试
print("测试批量查询性能...")
start_time = time.time()
for i in range(5):
results = mem.query(f"性能测试 {i}")
query_time = time.time() - start_time
print(f"查询5次耗时: {query_time:.2f}秒")
print("✅ 性能测试完成")
def main():
"""主函数"""
print("🧠 MemOS Memory Ops SDK 使用示例")
print("=" * 50)
try:
# 运行所有示例
basic_operations_example()
context_manager_example()
convenience_functions_example()
advanced_features_example()
error_handling_example()
performance_test()
print("\n🎉 所有示例运行完成!")
except Exception as e:
print(f"\n❌ 示例运行失败: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()