Buckets:
tahamajs/Sysmem2_in_AI / ComputerAssignments /CA6_systematic_generalization /integrations /main_experimental.py
| """ | |
| فایل اجرایی اصلی برای آزمایش Systematic Generalization | |
| Main execution file for Systematic Generalization experiments | |
| """ | |
| import os | |
| import sys | |
| import logging | |
| import json | |
| import argparse | |
| from pathlib import Path | |
| from datetime import datetime | |
| from typing import Dict, Any, Optional | |
| # تنظیم مسیر پروژه | |
| PROJECT_ROOT = Path(__file__).parent | |
| sys.path.append(str(PROJECT_ROOT)) | |
| # Import local modules | |
| try: | |
| from gemini_config import ( | |
| GeminiConfig, | |
| GeminiSystematicGeneralization, | |
| create_gemini_config_from_env, | |
| ) | |
| from advanced_rag_system import ( | |
| AdvancedRAGSystem, | |
| RAGConfig, | |
| SystematicGeneralizationRAG, | |
| ) | |
| from systematic_generalization_gemini import ( | |
| SystematicGeneralizationEvaluator, | |
| SystematicGeneralizationBenchmark, | |
| ) | |
| from neural_symbolic_integration import ( | |
| NeuralSymbolicConfig, | |
| SystematicGeneralizationNeuralSymbolic, | |
| ) | |
| except ImportError as e: | |
| print(f"❌ خطا در import ماژول‌ها: {e}") | |
| print("لطفاً مطمئن شوید که تمام فایل‌های مورد نیاز موجود هستند") | |
| sys.exit(1) | |
| # تنظیم لاگینگ | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", | |
| handlers=[ | |
| logging.FileHandler("systematic_generalization.log"), | |
| logging.StreamHandler(), | |
| ], | |
| ) | |
| logger = logging.getLogger(__name__) | |
| class SystematicGeneralizationExperimentRunner: | |
| """اجراکننده آزمایش‌های Systematic Generalization""" | |
| def __init__(self, config_path: Optional[str] = None): | |
| self.config_path = config_path | |
| self.results_dir = PROJECT_ROOT / "experimental_results" | |
| self.results_dir.mkdir(exist_ok=True) | |
| # بارگذاری پیکربندی | |
| self.config = self._load_config() | |
| # ایجاد سیستم‌ها | |
| self._setup_systems() | |
| logger.info("✅ آزمایش‌گر Systematic Generalization آماده شد") | |
| def _load_config(self) -> Dict[str, Any]: | |
| """بارگذاری پیکربندی""" | |
| if self.config_path and Path(self.config_path).exists(): | |
| with open(self.config_path, "r", encoding="utf-8") as f: | |
| config = json.load(f) | |
| else: | |
| # پیکربندی پیش‌فرض | |
| config = { | |
| "gemini": { | |
| "model_name": "gemini-1.5-pro", | |
| "temperature": 0.7, | |
| "max_tokens": 2048, | |
| }, | |
| "rag": { | |
| "embedding_model": "sentence-transformers/all-MiniLM-L6-v2", | |
| "chunk_size": 1000, | |
| "vector_store_type": "faiss", | |
| "top_k": 5, | |
| }, | |
| "neural_symbolic": { | |
| "embedding_dim": 128, | |
| "hidden_dim": 256, | |
| "integration_mode": "hybrid", | |
| "symbolic_weight": 0.6, | |
| "neural_weight": 0.4, | |
| }, | |
| "experiment": { | |
| "save_results": True, | |
| "create_visualizations": True, | |
| "run_comprehensive_evaluation": True, | |
| }, | |
| } | |
| logger.info("✅ پیکربندی بارگذاری شد") | |
| return config | |
| def _setup_systems(self): | |
| """تنظیم سیستم‌ها""" | |
| try: | |
| # تنظیم Gemini | |
| self.gemini_config = create_gemini_config_from_env() | |
| logger.info("✅ پیکربندی Gemini تنظیم شد") | |
| # تنظیم RAG | |
| rag_config = RAGConfig( | |
| embedding_model=self.config["rag"]["embedding_model"], | |
| chunk_size=self.config["rag"]["chunk_size"], | |
| vector_store_type=self.config["rag"]["vector_store_type"], | |
| top_k=self.config["rag"]["top_k"], | |
| ) | |
| self.rag_system = AdvancedRAGSystem(rag_config, self.gemini_config) | |
| logger.info("✅ سیستم RAG تنظیم شد") | |
| # تنظیم Neural-Symbolic | |
| neural_symbolic_config = NeuralSymbolicConfig( | |
| embedding_dim=self.config["neural_symbolic"]["embedding_dim"], | |
| hidden_dim=self.config["neural_symbolic"]["hidden_dim"], | |
| symbolic_weight=self.config["neural_symbolic"]["symbolic_weight"], | |
| neural_weight=self.config["neural_symbolic"]["neural_weight"], | |
| ) | |
| self.neural_symbolic_system = SystematicGeneralizationNeuralSymbolic( | |
| neural_symbolic_config, self.gemini_config | |
| ) | |
| logger.info("✅ سیستم Neural-Symbolic تنظیم شد") | |
| # تنظیم ارزیاب | |
| self.evaluator = SystematicGeneralizationEvaluator(self.gemini_config) | |
| logger.info("✅ ارزیاب تنظیم شد") | |
| except Exception as e: | |
| logger.error(f"❌ خطا در تنظیم سیستم‌ها: {e}") | |
| raise | |
| def run_basic_tests(self) -> Dict[str, Any]: | |
| """اجرای تست‌های پایه""" | |
| logger.info("🧪 شروع تست‌های پایه") | |
| results = { | |
| "timestamp": datetime.now().isoformat(), | |
| "basic_tests": {}, | |
| "summary": {}, | |
| } | |
| # تست Gemini API | |
| try: | |
| gemini_system = GeminiSystematicGeneralization(self.gemini_config) | |
| response = gemini_system.generate_systematic_response( | |
| "چگونه می‌توانم مفهوم ترکیب‌پذیری را در هوش مصنوعی پیاده‌سازی کنم؟" | |
| ) | |
| results["basic_tests"]["gemini_api"] = { | |
| "success": response["success"], | |
| "response_length": len(response["response"]), | |
| "test_passed": True, | |
| } | |
| logger.info("✅ تست Gemini API موفق بود") | |
| except Exception as e: | |
| results["basic_tests"]["gemini_api"] = { | |
| "success": False, | |
| "error": str(e), | |
| "test_passed": False, | |
| } | |
| logger.error(f"❌ تست Gemini API ناموفق: {e}") | |
| # تست RAG System | |
| try: | |
| # افزودن اسناد نمونه | |
| sample_docs = [ | |
| "Systematic Generalization به معنای توانایی ترکیب و تعمیم سیستماتیک اجزای یادگرفته شده است.", | |
| "ترکیب‌پذیری یکی از اصول اصلی Systematic Generalization است.", | |
| "Neural-Symbolic Integration ترکیبی از رویکردهای عصبی و نمادین است.", | |
| ] | |
| # شبیه‌سازی بارگذاری اسناد | |
| from langchain.schema import Document | |
| documents = [Document(page_content=doc) for doc in sample_docs] | |
| split_docs = self.rag_system.split_documents(documents) | |
| self.rag_system.create_vector_store(split_docs) | |
| # تست پرسش | |
| test_query = "Systematic Generalization چیست؟" | |
| rag_response = self.rag_system.query_systematic(test_query) | |
| results["basic_tests"]["rag_system"] = { | |
| "success": rag_response["success"], | |
| "query": test_query, | |
| "response_length": len(rag_response["answer"]), | |
| "test_passed": True, | |
| } | |
| logger.info("✅ تست RAG System موفق بود") | |
| except Exception as e: | |
| results["basic_tests"]["rag_system"] = { | |
| "success": False, | |
| "error": str(e), | |
| "test_passed": False, | |
| } | |
| logger.error(f"❌ تست RAG System ناموفق: {e}") | |
| # تست Neural-Symbolic | |
| try: | |
| test_expression = "2 + 3" | |
| neural_symbolic_result = ( | |
| self.neural_symbolic_system.integrator.process_expression( | |
| test_expression | |
| ) | |
| ) | |
| results["basic_tests"]["neural_symbolic"] = { | |
| "success": True, | |
| "expression": test_expression, | |
| "confidence": neural_symbolic_result["confidence"], | |
| "test_passed": True, | |
| } | |
| logger.info("✅ تست Neural-Symbolic موفق بود") | |
| except Exception as e: | |
| results["basic_tests"]["neural_symbolic"] = { | |
| "success": False, | |
| "error": str(e), | |
| "test_passed": False, | |
| } | |
| logger.error(f"❌ تست Neural-Symbolic ناموفق: {e}") | |
| # خلاصه نتایج | |
| passed_tests = sum( | |
| 1 | |
| for test in results["basic_tests"].values() | |
| if test.get("test_passed", False) | |
| ) | |
| total_tests = len(results["basic_tests"]) | |
| results["summary"] = { | |
| "total_tests": total_tests, | |
| "passed_tests": passed_tests, | |
| "success_rate": passed_tests / total_tests if total_tests > 0 else 0, | |
| } | |
| logger.info( | |
| f"✅ تست‌های پایه تکمیل شد - نرخ موفقیت: {results['summary']['success_rate']:.1%}" | |
| ) | |
| return results | |
| def run_comprehensive_evaluation(self) -> Dict[str, Any]: | |
| """اجرای ارزیابی جامع""" | |
| logger.info("🚀 شروع ارزیابی جامع") | |
| try: | |
| # ارزیابی Neural-Symbolic | |
| neural_symbolic_results = ( | |
| self.neural_symbolic_system.run_comprehensive_evaluation() | |
| ) | |
| # ارزیابی استاندارد | |
| standard_results = self.evaluator.run_comprehensive_evaluation() | |
| comprehensive_results = { | |
| "timestamp": datetime.now().isoformat(), | |
| "neural_symbolic_evaluation": neural_symbolic_results, | |
| "standard_evaluation": standard_results, | |
| "comparison": self._compare_results( | |
| standard_results, neural_symbolic_results | |
| ), | |
| } | |
| logger.info("✅ ارزیابی جامع تکمیل شد") | |
| return comprehensive_results | |
| except Exception as e: | |
| logger.error(f"❌ خطا در ارزیابی جامع: {e}") | |
| return {"error": str(e), "timestamp": datetime.now().isoformat()} | |
| def _compare_results( | |
| self, standard_results: Dict[str, Any], neural_symbolic_results: Dict[str, Any] | |
| ) -> Dict[str, Any]: | |
| """مقایسه نتایج""" | |
| try: | |
| standard_accuracy = standard_results.get("overall_scores", {}).get( | |
| "average_accuracy", 0 | |
| ) | |
| neural_symbolic_accuracy = neural_symbolic_results.get( | |
| "neural_symbolic_evaluation", {} | |
| ).get("overall_accuracy", 0) | |
| improvement = neural_symbolic_accuracy - standard_accuracy | |
| return { | |
| "standard_accuracy": standard_accuracy, | |
| "neural_symbolic_accuracy": neural_symbolic_accuracy, | |
| "improvement": improvement, | |
| "improvement_percentage": ( | |
| (improvement / standard_accuracy * 100) | |
| if standard_accuracy > 0 | |
| else 0 | |
| ), | |
| "better_approach": "neural_symbolic" if improvement > 0 else "standard", | |
| } | |
| except Exception as e: | |
| logger.error(f"❌ خطا در مقایسه نتایج: {e}") | |
| return {"error": str(e)} | |
| def save_results(self, results: Dict[str, Any], filename: str): | |
| """ذخیره نتایج""" | |
| try: | |
| results_file = self.results_dir / filename | |
| with open(results_file, "w", encoding="utf-8") as f: | |
| json.dump(results, f, ensure_ascii=False, indent=2) | |
| logger.info(f"✅ نتایج ذخیره شد: {results_file}") | |
| except Exception as e: | |
| logger.error(f"❌ خطا در ذخیره نتایج: {e}") | |
| def run_full_experiment(self) -> Dict[str, Any]: | |
| """اجرای آزمایش کامل""" | |
| logger.info("🎯 شروع آزمایش کامل Systematic Generalization") | |
| full_results = { | |
| "experiment_timestamp": datetime.now().isoformat(), | |
| "config": self.config, | |
| "basic_tests": {}, | |
| "comprehensive_evaluation": {}, | |
| "final_summary": {}, | |
| } | |
| # اجرای تست‌های پایه | |
| basic_results = self.run_basic_tests() | |
| full_results["basic_tests"] = basic_results | |
| # اجرای ارزیابی جامع (اگر تست‌های پایه موفق بودند) | |
| if basic_results["summary"]["success_rate"] > 0.5: | |
| comprehensive_results = self.run_comprehensive_evaluation() | |
| full_results["comprehensive_evaluation"] = comprehensive_results | |
| else: | |
| logger.warning("⚠️ تست‌های پایه ناموفق بودند - ارزیابی جامع اجرا نمی‌شود") | |
| full_results["comprehensive_evaluation"] = {"skipped": "basic_tests_failed"} | |
| # خلاصه نهایی | |
| full_results["final_summary"] = { | |
| "basic_tests_success_rate": basic_results["summary"]["success_rate"], | |
| "comprehensive_evaluation_completed": "comprehensive_evaluation" | |
| in full_results | |
| and "skipped" not in full_results["comprehensive_evaluation"], | |
| "overall_experiment_success": basic_results["summary"]["success_rate"] | |
| > 0.5, | |
| } | |
| # ذخیره نتایج | |
| if self.config["experiment"]["save_results"]: | |
| self.save_results( | |
| full_results, | |
| f"full_experiment_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", | |
| ) | |
| logger.info("🎉 آزمایش کامل تکمیل شد") | |
| return full_results | |
| def main(): | |
| """تابع اصلی""" | |
| parser = argparse.ArgumentParser( | |
| description="اجرای آزمایش Systematic Generalization" | |
| ) | |
| parser.add_argument("--config", type=str, help="مسیر فایل پیکربندی") | |
| parser.add_argument( | |
| "--mode", | |
| type=str, | |
| choices=["basic", "comprehensive", "full"], | |
| default="full", | |
| help="حالت اجرا", | |
| ) | |
| parser.add_argument("--output", type=str, help="مسیر ذخیره نتایج") | |
| args = parser.parse_args() | |
| try: | |
| # ایجاد اجراکننده آزمایش | |
| runner = SystematicGeneralizationExperimentRunner(args.config) | |
| # اجرای آزمایش بر اساس حالت | |
| if args.mode == "basic": | |
| results = runner.run_basic_tests() | |
| elif args.mode == "comprehensive": | |
| results = runner.run_comprehensive_evaluation() | |
| else: # full | |
| results = runner.run_full_experiment() | |
| # ذخیره نتایج در مسیر مشخص شده | |
| if args.output: | |
| runner.save_results(results, args.output) | |
| # نمایش خلاصه نتایج | |
| print("\n" + "=" * 60) | |
| print("📊 خلاصه نتایج آزمایش") | |
| print("=" * 60) | |
| if "basic_tests" in results: | |
| basic_summary = results["basic_tests"].get("summary", {}) | |
| print( | |
| f"تست‌های پایه: {basic_summary.get('passed_tests', 0)}/{basic_summary.get('total_tests', 0)} موفق" | |
| ) | |
| print(f"نرخ موفقیت: {basic_summary.get('success_rate', 0):.1%}") | |
| if ( | |
| "comprehensive_evaluation" in results | |
| and "skipped" not in results["comprehensive_evaluation"] | |
| ): | |
| print("ارزیابی جامع: ✅ تکمیل شد") | |
| if "final_summary" in results: | |
| final_summary = results["final_summary"] | |
| print( | |
| f"آزمایش کلی: {'✅ موفق' if final_summary.get('overall_experiment_success') else '❌ ناموفق'}" | |
| ) | |
| print("=" * 60) | |
| except Exception as e: | |
| logger.error(f"❌ خطا در اجرای آزمایش: {e}") | |
| sys.exit(1) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 16.5 kB
- Xet hash:
- f498215dc67fcd04309a29a330b8e5a643a0b496c0f30d8cbdc8c1aded573622
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.