simple_memory_ops_sdk.py•12.1 kB
#!/usr/bin/env python3
"""
简化版Memory Ops SDK
专注于MultiMemoryMemCube的适配，避免复杂的依赖问题
基于现有的MultiMemoryOps类实现
"""
import os
import sys
from pathlib import Path
from typing import List, Dict, Any, Optional
from datetime import datetime
# 添加项目根目录到Python路径
sys.path.insert(0, str(Path(__file__).parent))
# 添加src目录到Python路径
sys.path.insert(0, str(Path(__file__).parent / "src"))
try:
# 导入bootstrap来解决依赖问题
import bootstrap
bootstrap.setup_mock_modules()
except ImportError:
pass # bootstrap是可选的
class SimpleMemoryOps:
"""
简化版记忆操作SDK
基于MultiMemoryMemCube实现，支持多Memory类型的统一管理
"""
def __init__(self, data_dir: Optional[str] = None, verbose: bool = False):
"""
初始化Simple Memory Ops SDK
Args:
data_dir: 数据目录路径，默认为当前目录下的memos_data
verbose: 是否显示详细日志
"""
self.data_dir = data_dir or "./memos_data"
self.verbose = verbose
self._cube = None
self._initialized = False
# 配置多Memory模块 - 使用简单的字典配置
self.memory_types = {
'general_mem': '通用记忆',
'code_snippet_mem': '代码片段记忆',
'error_log_mem': '错误日志记忆',
'product_mem': '产品记忆'
}
# 内存存储（用于演示）
self._memory_storage = {mem_type: [] for mem_type in self.memory_types.keys()}
# 持久化存储文件路径
self._storage_file = Path(self.data_dir) / "simple_memory_storage.json"
# 加载已存在的记忆
self._load_from_file()
# 自动初始化
self.init()
def init(self) -> bool:
"""
初始化SDK
Returns:
bool: 初始化是否成功
"""
try:
if self.verbose:
print(f"🚀 初始化Simple Memory Ops SDK...")
print(f" 数据目录: {self.data_dir}")
print(f" Memory类型: {list(self.memory_types.keys())}")
# 简单的初始化，不依赖复杂的外部模块
self._initialized = True
if self.verbose:
print("✅ Simple Memory Ops SDK初始化成功")
return True
except Exception as e:
if self.verbose:
print(f"❌ Simple Memory Ops SDK初始化失败: {e}")
return False
def _ensure_initialized(self):
"""确保SDK已初始化"""
if not self._initialized:
raise RuntimeError("SDK未初始化，请先调用init()方法")
def _load_from_file(self):
"""从文件加载记忆"""
try:
if self._storage_file.exists():
import json
with open(self._storage_file, 'r', encoding='utf-8') as f:
data = json.load(f)
if isinstance(data, dict):
for mem_type, memories in data.items():
if mem_type in self._memory_storage:
self._memory_storage[mem_type] = memories
if self.verbose:
total_memories = sum(len(memories) for memories in self._memory_storage.values())
print(f"📁 已从文件加载 {total_memories} 条记忆")
except Exception as e:
if self.verbose:
print(f"⚠️ 加载记忆文件失败: {e}")
def _save_to_file(self):
"""保存记忆到文件"""
try:
# 确保目录存在
self._storage_file.parent.mkdir(parents=True, exist_ok=True)
import json
with open(self._storage_file, 'w', encoding='utf-8') as f:
json.dump(self._memory_storage, f, ensure_ascii=False, indent=2)
if self.verbose:
total_memories = sum(len(memories) for memories in self._memory_storage.values())
print(f"💾 已保存 {total_memories} 条记忆到文件")
except Exception as e:
if self.verbose:
print(f"❌ 保存记忆文件失败: {e}")
def add(self, text: str, memory_type: str = 'general_mem', tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None) -> bool:
"""
添加记忆
Args:
text: 记忆内容
memory_type: Memory类型
tags: 标签列表
metadata: 元数据字典
Returns:
bool: 是否添加成功
"""
self._ensure_initialized()
try:
# 检查memory_type是否有效
if memory_type not in self.memory_types:
if self.verbose:
print(f"❌ 无效的Memory类型: {memory_type}")
print(f" 可用类型: {list(self.memory_types.keys())}")
return False
# 自动添加时间戳到元数据
if metadata is None:
metadata = {}
if 'timestamp' not in metadata:
metadata['timestamp'] = datetime.now().isoformat()
# 添加标签到元数据
if tags:
metadata['tags'] = tags
# 构造记忆项
memory_item = {
"id": f"{memory_type}_{len(self._memory_storage[memory_type])}",
"memory": text,
"metadata": metadata,
"memory_type": memory_type
}
# 添加到内存存储
self._memory_storage[memory_type].append(memory_item)
# 保存到文件
self._save_to_file()
if self.verbose:
print(f"✅ 成功添加到 {memory_type}: {text[:50]}...")
return True
except Exception as e:
if self.verbose:
print(f"❌ 添加记忆失败: {e}")
return False
def query(self, q: str, memory_type: Optional[str] = None, limit: int = 5) -> List[Dict[str, Any]]:
"""
查询记忆
Args:
q: 查询文本
memory_type: Memory类型，None表示搜索所有类型
limit: 返回结果数量限制
Returns:
List[Dict]: 查询结果列表
"""
self._ensure_initialized()
try:
all_results = []
# 确定要搜索的Memory类型
search_types = [memory_type] if memory_type else list(self.memory_types.keys())
for mem_type in search_types:
if mem_type not in self.memory_types:
continue
# 简单的文本匹配搜索
for item in self._memory_storage[mem_type]:
if q.lower() in item['memory'].lower():
all_results.append(item.copy())
if len(all_results) >= limit:
break
if len(all_results) >= limit:
break
# 限制结果数量
all_results = all_results[:limit]
if self.verbose:
print(f"🔍 查询'{q}' 找到 {len(all_results)} 条结果")
return all_results
except Exception as e:
if self.verbose:
print(f"❌ 查询失败: {e}")
return []
def list_memory_types(self) -> Dict[str, str]:
"""
列出所有可用的Memory类型
Returns:
Dict[str, str]: Memory类型和描述的映射
"""
return self.memory_types.copy()
def stats(self) -> Dict[str, Any]:
"""
获取记忆统计信息
Returns:
Dict: 统计信息字典
"""
self._ensure_initialized()
try:
stats = {}
for mem_type in self.memory_types.keys():
count = len(self._memory_storage[mem_type])
stats[mem_type] = {
"count": count,
"status": "active",
"type": "SimpleMemory"
}
stats["timestamp"] = datetime.now().isoformat()
if self.verbose:
print("📊 获取统计信息成功")
return stats
except Exception as e:
if self.verbose:
print(f"❌ 获取统计信息失败: {e}")
return {}
def dump(self, memory_type: Optional[str] = None, output_file: Optional[str] = None) -> List[Dict[str, Any]]:
"""
导出记忆
Args:
memory_type: Memory类型，None表示导出所有类型
output_file: 输出文件路径（可选）
Returns:
List[Dict]: 记忆列表
"""
self._ensure_initialized()
try:
all_memories = []
# 确定要导出的Memory类型
export_types = [memory_type] if memory_type else list(self.memory_types.keys())
for mem_type in export_types:
if mem_type not in self.memory_types:
continue
all_memories.extend(self._memory_storage[mem_type])
if output_file:
import json
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(all_memories, f, ensure_ascii=False, indent=2)
if self.verbose:
print(f"📁 已导出 {len(all_memories)} 条记忆到 {output_file}")
return all_memories
except Exception as e:
if self.verbose:
print(f"❌ 导出记忆失败: {e}")
return []
def close(self):
"""
关闭SDK，清理资源
"""
if self._initialized:
self._initialized = False
if self.verbose:
print("🔒 Simple Memory Ops SDK已关闭")
def __enter__(self):
"""上下文管理器入口"""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""上下文管理器出口"""
self.close()
def __repr__(self):
"""字符串表示"""
status = "已初始化" if self._initialized else "未初始化"
memory_types = list(self.memory_types.keys())
return f"SimpleMemoryOps(status={status}, memory_types={memory_types}, data_dir='{self.data_dir}')"
# 便捷函数
def create_simple_memory_ops(data_dir: Optional[str] = None, **kwargs) -> SimpleMemoryOps:
"""
创建SimpleMemoryOps实例的便捷函数
Args:
data_dir: 数据目录
**kwargs: 其他参数传递给SimpleMemoryOps构造函数
Returns:
SimpleMemoryOps: SDK实例
"""
return SimpleMemoryOps(data_dir=data_dir, **kwargs)
# 全局实例管理
_global_instance = None
def get_global_simple_memory_ops(**kwargs) -> SimpleMemoryOps:
"""
获取全局SimpleMemoryOps单例实例
Args:
**kwargs: 首次创建时的参数
Returns:
SimpleMemoryOps: 全局SDK实例
"""
global _global_instance
if _global_instance is None:
_global_instance = SimpleMemoryOps(**kwargs)
return _global_instance