MemOS-MCP

""" MemOS API兼容性验证器 验证API实现是否符合MemOS官方规范 """ import json import time from typing import Dict, Any, List, Optional from datetime import datetime from pathlib import Path from mvp_memory import create_mvp_memory_manager class APICompatibilityValidator: """API兼容性验证器""" def __init__(self): self.validation_results = [] self.mvp_manager = None def log_validation(self, test_name: str, passed: bool, message: str, details: Optional[Dict] = None): """记录验证结果""" result = { "test_name": test_name, "passed": passed, "message": message, "details": details or {}, "timestamp": datetime.now().isoformat() } self.validation_results.append(result) status = "✅" if passed else "❌" print(f"{status} {test_name}: {message}") def validate_response_format(self) -> bool: """验证响应格式标准化""" print("\n📋 验证响应格式标准化...") try: # 初始化MVP管理器 self.mvp_manager = create_mvp_memory_manager(use_official_config=True) # 测试成功响应格式 success = self.mvp_manager.remember("测试响应格式", tags=["验证"]) if success: self.log_validation( "成功响应格式", True, "记忆创建成功，响应格式符合标准" ) else: self.log_validation( "成功响应格式", False, "记忆创建失败" ) return False # 测试搜索响应格式 results = self.mvp_manager.recall("测试", top_k=3) if results and isinstance(results, list): # 验证搜索结果结构 first_result = results[0] required_fields = ["content", "score"] has_required_fields = all(field in first_result for field in required_fields) self.log_validation( "搜索响应格式", has_required_fields, f"搜索结果包含必需字段: {required_fields}" if has_required_fields else "搜索结果缺少必需字段" ) return has_required_fields else: self.log_validation( "搜索响应格式", False, "搜索结果格式不正确" ) return False except Exception as e: self.log_validation( "响应格式验证", False, f"验证过程中发生错误: {str(e)}" ) return False def validate_error_handling(self) -> bool: """验证错误处理机制""" print("\n🚨 验证错误处理机制...") try: # 测试无效输入处理 try: # 尝试创建空内容记忆 success = self.mvp_manager.remember("", tags=[]) # 空内容应该被处理（可能成功也可能失败，但不应该崩溃） self.log_validation( "空内容处理", True, f"空内容记忆处理正常: {'成功' if success else '被拒绝'}" ) except Exception as e: self.log_validation( "空内容处理", False, f"空内容处理异常: {str(e)}" ) return False # 测试无效反馈处理 try: feedback_success = self.mvp_manager.provide_feedback("invalid_id", "invalid_type") self.log_validation( "无效反馈处理", True, f"无效反馈处理正常: {'接受' if feedback_success else '被拒绝'}" ) except Exception as e: self.log_validation( "无效反馈处理", False, f"无效反馈处理异常: {str(e)}" ) return False return True except Exception as e: self.log_validation( "错误处理验证", False, f"验证过程中发生错误: {str(e)}" ) return False def validate_performance_monitoring(self) -> bool: """验证性能监控功能""" print("\n📊 验证性能监控功能...") try: # 测试性能指标获取 metrics = self.mvp_manager.get_performance_metrics() if "error" in metrics: self.log_validation( "性能指标获取", False, f"性能指标获取失败: {metrics['error']}" ) return False # 验证必需的性能指标字段 required_metrics = [ "memory_add_avg_time", "memory_search_avg_time", "memory_total_count", "system_health" ] missing_metrics = [metric for metric in required_metrics if metric not in metrics] if missing_metrics: self.log_validation( "性能指标完整性", False, f"缺少性能指标: {missing_metrics}" ) return False else: self.log_validation( "性能指标完整性", True, "所有必需的性能指标都可用" ) # 测试容量报告 capacity_report = self.mvp_manager.get_capacity_report() if "error" in capacity_report: self.log_validation( "容量报告获取", False, f"容量报告获取失败: {capacity_report['error']}" ) return False else: self.log_validation( "容量报告获取", True, "容量报告获取成功" ) return True except Exception as e: self.log_validation( "性能监控验证", False, f"验证过程中发生错误: {str(e)}" ) return False def validate_capacity_management(self) -> bool: """验证容量管理功能""" print("\n📈 验证容量管理功能...") try: # 测试系统状态获取 status = self.mvp_manager.get_status_info() # 验证容量管理是否启用 capacity_mgmt = status.get("capacity_management", {}) if not capacity_mgmt.get("enabled"): self.log_validation( "容量管理启用", False, "容量管理未启用" ) return False else: self.log_validation( "容量管理启用", True, f"容量管理已启用，监控状态: {capacity_mgmt.get('monitoring_status', 'unknown')}" ) # 测试手动优化 optimization_success = self.mvp_manager.trigger_manual_optimization() self.log_validation( "手动优化功能", optimization_success, "手动优化功能正常" if optimization_success else "手动优化功能异常" ) return optimization_success except Exception as e: self.log_validation( "容量管理验证", False, f"验证过程中发生错误: {str(e)}" ) return False def validate_feedback_mechanism(self) -> bool: """验证反馈机制""" print("\n👍 验证反馈机制...") try: # 创建测试记忆 test_content = f"反馈测试记忆 - {datetime.now().strftime('%H:%M:%S')}" success = self.mvp_manager.remember(test_content, tags=["反馈测试"]) if not success: self.log_validation( "反馈测试记忆创建", False, "无法创建测试记忆" ) return False # 搜索刚创建的记忆 results = self.mvp_manager.recall("反馈测试", top_k=1) if not results: self.log_validation( "反馈测试记忆搜索", False, "无法找到测试记忆" ) return False # 获取记忆ID（模拟） memory_id = str(results[0].get("id", "test_id")) # 测试正面反馈 positive_feedback = self.mvp_manager.provide_feedback(memory_id, "thumbs_up") self.log_validation( "正面反馈功能", positive_feedback, "正面反馈功能正常" if positive_feedback else "正面反馈功能异常" ) # 测试负面反馈 negative_feedback = self.mvp_manager.provide_feedback(memory_id, "thumbs_down") self.log_validation( "负面反馈功能", negative_feedback, "负面反馈功能正常" if negative_feedback else "负面反馈功能异常" ) # 测试反馈统计 feedback_stats = self.mvp_manager.get_feedback_stats() has_stats = feedback_stats and isinstance(feedback_stats, dict) self.log_validation( "反馈统计功能", has_stats, "反馈统计功能正常" if has_stats else "反馈统计功能异常" ) return positive_feedback and negative_feedback and has_stats except Exception as e: self.log_validation( "反馈机制验证", False, f"验证过程中发生错误: {str(e)}" ) return False def validate_official_config_integration(self) -> bool: """验证官方配置集成""" print("\n⚙️ 验证官方配置集成...") try: status = self.mvp_manager.get_status_info() # 验证官方配置是否启用 official_config = status.get("official_config", False) if not official_config: self.log_validation( "官方配置启用", False, "官方配置未启用" ) return False # 验证配置类型 config_type = status.get("config_type", "") expected_type = "official_best_practice" if config_type != expected_type: self.log_validation( "官方配置类型", False, f"配置类型不匹配: 期望 {expected_type}, 实际 {config_type}" ) return False # 验证内存后端 memory_backend = status.get("memory_backend", "") expected_backend = "general_text" if memory_backend != expected_backend: self.log_validation( "内存后端配置", False, f"内存后端不匹配: 期望 {expected_backend}, 实际 {memory_backend}" ) return False # 验证容量规划 capacity_planning = status.get("capacity_planning", {}) expected_capacities = { "working_memory_capacity": 20, "user_memory_capacity": 500, "long_term_memory_capacity": 2000 } capacity_match = all( capacity_planning.get(key) == value for key, value in expected_capacities.items() ) if not capacity_match: self.log_validation( "容量规划配置", False, f"容量规划不匹配: 期望 {expected_capacities}, 实际 {capacity_planning}" ) return False # 验证优化配置 optimizations = status.get("optimizations", {}) expected_optimizations = { "batch_processing": True, "memory_compression": True } optimization_match = all( optimizations.get(key) == value for key, value in expected_optimizations.items() ) if not optimization_match: self.log_validation( "优化配置", False, f"优化配置不匹配: 期望 {expected_optimizations}, 实际 {optimizations}" ) return False self.log_validation( "官方配置集成", True, "所有官方配置项都正确集成" ) return True except Exception as e: self.log_validation( "官方配置验证", False, f"验证过程中发生错误: {str(e)}" ) return False def run_comprehensive_validation(self) -> Dict[str, Any]: """运行综合兼容性验证""" print("🔍 开始MemOS API兼容性验证") print("=" * 60) start_time = time.time() # 运行各项验证 validations = [ ("响应格式标准化", self.validate_response_format), ("错误处理机制", self.validate_error_handling), ("性能监控功能", self.validate_performance_monitoring), ("容量管理功能", self.validate_capacity_management), ("反馈机制", self.validate_feedback_mechanism), ("官方配置集成", self.validate_official_config_integration) ] validation_results = {} for validation_name, validation_func in validations: try: result = validation_func() validation_results[validation_name] = result except Exception as e: print(f"❌ {validation_name}验证失败: {e}") validation_results[validation_name] = False total_duration = time.time() - start_time # 统计结果 passed_validations = sum(1 for result in validation_results.values() if result) total_validations = len(validation_results) compatibility_score = (passed_validations / total_validations) * 100 if total_validations > 0 else 0 summary = { "total_validations": total_validations, "passed_validations": passed_validations, "failed_validations": total_validations - passed_validations, "compatibility_score": compatibility_score, "validation_results": validation_results, "detailed_results": self.validation_results, "total_duration": total_duration } print("\n" + "=" * 60) print("📊 兼容性验证结果汇总:") print(f" 总验证项: {total_validations}") print(f" 通过验证: {passed_validations}") print(f" 失败验证: {total_validations - passed_validations}") print(f" 兼容性评分: {compatibility_score:.1f}%") print(f" 总耗时: {total_duration:.3f}s") if compatibility_score >= 90: print("✅ API完全符合MemOS官方规范！") elif compatibility_score >= 70: print("⚠️ API基本符合规范，但有改进空间") else: print("❌ API存在兼容性问题，需要修复") return summary def save_validation_report(self, summary: Dict[str, Any], filename: str = "api_compatibility_report.json"): """保存验证报告""" with open(filename, 'w', encoding='utf-8') as f: json.dump(summary, f, indent=2, ensure_ascii=False) print(f"\n📄 兼容性验证报告已保存到: {filename}") if __name__ == "__main__": # 运行兼容性验证 validator = APICompatibilityValidator() try: summary = validator.run_comprehensive_validation() validator.save_validation_report(summary) except KeyboardInterrupt: print("\n\n⏹️ 验证被用户中断") except Exception as e: print(f"\n❌ 验证过程中发生错误: {e}") import traceback traceback.print_exc()