MemOS-MCP

#!/usr/bin/env python3 """ ProductMemory - 商品记忆管理模块 基于TreeTextMemory实现商品数据的存储和搜索功能 使用嵌套JSON格式存储商品属性、卖点和关系数据 """ import json import sys from pathlib import Path from typing import Dict, List, Any, Optional from datetime import datetime # 添加项目根目录到Python路径 project_root = Path(__file__).parent sys.path.insert(0, str(project_root)) sys.path.insert(0, str(project_root / "src")) try: from simple_memory_ops_sdk import SimpleMemoryOps except ImportError: print("Error: Cannot import SimpleMemoryOps", file=sys.stderr) sys.exit(1) class ProductMemory: """ 商品记忆管理类 基于SimpleMemoryOps实现商品数据的存储、搜索和管理 支持商品属性、卖点、关系的结构化存储 """ def __init__(self, memory_ops: Optional[SimpleMemoryOps] = None, verbose: bool = False): """ 初始化ProductMemory Args: memory_ops: SimpleMemoryOps实例，如果为None则自动创建 verbose: 是否显示详细日志 """ self.verbose = verbose self.memory_ops = memory_ops or SimpleMemoryOps(verbose=verbose) if self.verbose: print("🛍️ ProductMemory初始化完成") def add_product(self, sku: str, attributes: Dict[str, Any], selling_points: List[str], relationships: Optional[List[Dict]] = None) -> bool: """ 添加商品到记忆系统 Args: sku: 商品SKU编号 attributes: 商品属性字典 (如: {"name": "商品名", "price": 100, "category": "分类"}) selling_points: 商品卖点列表 relationships: 商品关系列表 (可选) Returns: bool: 是否添加成功 Example: product_mem.add_product( sku="SKU-001", attributes={"name": "智能手机", "price": 2999, "brand": "华为"}, selling_points=["5G网络", "长续航", "拍照清晰"], relationships=[{"type": "similar", "target": "SKU-002"}] ) """ try: # 构造商品数据的嵌套JSON结构 product_data = { "sku": sku, "attributes": attributes, "selling_points": selling_points, "relationships": relationships or [], "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat() } # 构造记忆内容 - 包含结构化信息和可搜索文本 memory_content = f"""商品信息: {sku} 商品名称: {attributes.get('name', '未知')} 商品属性: {json.dumps(attributes, ensure_ascii=False, indent=2)} 商品卖点: {', '.join(selling_points)} 关系信息: {json.dumps(relationships or [], ensure_ascii=False)} 结构化数据: {json.dumps(product_data, ensure_ascii=False, indent=2)}""" # 构造元数据 metadata = { "type": "product", "source": "product_memory", "sku": sku, "product_name": attributes.get('name', ''), "category": attributes.get('category', ''), "brand": attributes.get('brand', ''), "price": attributes.get('price'), "product_data": product_data, # 完整的商品数据 "timestamp": datetime.now().isoformat() } # 构造标签 tags = ["product", "商品", sku] if attributes.get('category'): tags.append(attributes['category']) if attributes.get('brand'): tags.append(attributes['brand']) tags.extend(selling_points) # 添加到product_mem success = self.memory_ops.add( text=memory_content, memory_type='product_mem', tags=tags, metadata=metadata ) if success and self.verbose: print(f"✅ 商品 {sku} 添加成功") return success except Exception as e: if self.verbose: print(f"❌ 添加商品 {sku} 失败: {e}") return False def search_product(self, query: str, limit: int = 5) -> List[Dict[str, Any]]: """ 搜索商品 Args: query: 搜索查询 limit: 返回结果数量限制 Returns: List[Dict]: 搜索结果列表，每个结果包含商品信息 Example: results = product_mem.search_product("智能手机") results = product_mem.search_product("SKU-001") """ try: # 在product_mem中搜索 results = self.memory_ops.query(query, memory_type='product_mem', limit=limit) # 格式化搜索结果 formatted_results = [] for result in results: product_data = result.get('metadata', {}).get('product_data', {}) formatted_result = { "id": result.get("id", ""), "sku": product_data.get("sku", ""), "attributes": product_data.get("attributes", {}), "selling_points": product_data.get("selling_points", []), "relationships": product_data.get("relationships", []), "created_at": product_data.get("created_at", ""), "updated_at": product_data.get("updated_at", ""), "memory_content": result.get("memory", ""), "tags": result.get("metadata", {}).get("tags", []) } formatted_results.append(formatted_result) if self.verbose: print(f"🔍 搜索'{query}' 找到 {len(formatted_results)} 个商品") return formatted_results except Exception as e: if self.verbose: print(f"❌ 搜索商品失败: {e}") return [] def get_product_by_sku(self, sku: str) -> Optional[Dict[str, Any]]: """ 根据SKU获取商品信息 Args: sku: 商品SKU编号 Returns: Dict: 商品信息，如果未找到返回None """ results = self.search_product(sku, limit=1) # 查找精确匹配的SKU for result in results: if result.get("sku") == sku: return result return None def update_product(self, sku: str, attributes: Optional[Dict[str, Any]] = None, selling_points: Optional[List[str]] = None, relationships: Optional[List[Dict]] = None) -> bool: """ 更新商品信息 Args: sku: 商品SKU编号 attributes: 新的商品属性（可选） selling_points: 新的商品卖点（可选） relationships: 新的商品关系（可选） Returns: bool: 是否更新成功 """ try: # 获取现有商品信息 existing_product = self.get_product_by_sku(sku) if not existing_product: if self.verbose: print(f"❌ 商品 {sku} 不存在，无法更新") return False # 合并更新信息 new_attributes = existing_product["attributes"].copy() if attributes: new_attributes.update(attributes) new_selling_points = selling_points if selling_points is not None else existing_product["selling_points"] new_relationships = relationships if relationships is not None else existing_product["relationships"] # 删除旧记录并添加新记录 # 注意：这是简化实现，实际应该支持原地更新 return self.add_product(sku, new_attributes, new_selling_points, new_relationships) except Exception as e: if self.verbose: print(f"❌ 更新商品 {sku} 失败: {e}") return False def list_all_products(self) -> List[Dict[str, Any]]: """ 列出所有商品 Returns: List[Dict]: 所有商品列表 """ try: # 获取所有product_mem类型的记忆 all_memories = self.memory_ops.dump(memory_type='product_mem') products = [] for memory in all_memories: product_data = memory.get('metadata', {}).get('product_data', {}) if product_data: products.append({ "sku": product_data.get("sku", ""), "attributes": product_data.get("attributes", {}), "selling_points": product_data.get("selling_points", []), "relationships": product_data.get("relationships", []), "created_at": product_data.get("created_at", ""), "updated_at": product_data.get("updated_at", "") }) if self.verbose: print(f"📋 共找到 {len(products)} 个商品") return products except Exception as e: if self.verbose: print(f"❌ 列出商品失败: {e}") return [] def get_stats(self) -> Dict[str, Any]: """ 获取商品记忆统计信息 Returns: Dict: 统计信息 """ try: stats = self.memory_ops.stats() product_stats = stats.get('product_mem', {}) all_products = self.list_all_products() categories = set() brands = set() for product in all_products: if product['attributes'].get('category'): categories.add(product['attributes']['category']) if product['attributes'].get('brand'): brands.add(product['attributes']['brand']) return { "total_products": len(all_products), "memory_status": product_stats.get("status", "unknown"), "memory_count": product_stats.get("count", 0), "categories": list(categories), "brands": list(brands), "timestamp": datetime.now().isoformat() } except Exception as e: if self.verbose: print(f"❌ 获取统计信息失败: {e}") return {} def close(self): """关闭ProductMemory""" if self.memory_ops: self.memory_ops.close() if self.verbose: print("🔒 ProductMemory已关闭") # 便捷函数 def create_product_memory(verbose: bool = False) -> ProductMemory: """ 创建ProductMemory实例的便捷函数 Args: verbose: 是否显示详细日志 Returns: ProductMemory: 实例 """ return ProductMemory(verbose=verbose) if __name__ == "__main__": # 简单测试 print("🧪 ProductMemory测试") pm = create_product_memory(verbose=True) # 添加测试商品 success = pm.add_product( sku="SKU-001", attributes={ "name": "华为Mate60 Pro", "price": 6999, "category": "智能手机", "brand": "华为" }, selling_points=["麒麟9000S芯片", "卫星通话", "昆仑玻璃", "5000mAh大电池"], relationships=[{"type": "series", "target": "Mate系列"}] ) print(f"添加结果: {success}") # 搜索测试 results = pm.search_product("华为") print(f"搜索结果数量: {len(results)}") # 统计信息 stats = pm.get_stats() print(f"统计信息: {stats}") pm.close()