Buckets:

tahamajs
/

Sysmem2_in_AI

Files

xet

tahamajs/Sysmem2_in_AI / ComputerAssignments /CA5_geometric_reasoning /integrations /experimental_system.py

tahamajs

5 months ago

download

raw

27.6 kB

#!/usr/bin/env python3
"""
Advanced Experimental Geometric Reasoning with LangChain, RAG, and Gemini API
تجربه پیشرفته استدلال هندسی با LangChain، RAG و Gemini API
This is a comprehensive experimental script that demonstrates:
- LangChain agents with custom tools
- RAG system with vector databases
- Gemini API integration
- Multiple solving strategies
- Advanced evaluation and visualization
Author: AI Systems Course Team
Version: 2.0.0
Date: January 2025
"""

import os
import sys
import asyncio
import logging
from typing import Dict, List, Any, Optional
from datetime import datetime
from pathlib import Path

# Add src to path
sys.path.append(str(Path(__file__).parent / "src"))

# Core dependencies
import numpy as np
import pandas as pd
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn

# LangChain imports
from langchain.llms import GoogleGenerativeAI
from langchain.chains import RetrievalQA, ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
from langchain.agents import AgentExecutor, create_conversational_agent
from langchain.tools import Tool
from langchain.prompts import PromptTemplate
from langchain.schema import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import SentenceTransformerEmbeddings
from langchain.vectorstores import Chroma, FAISS
from langchain.document_loaders import TextLoader
from langchain.callbacks import StreamingStdOutCallbackHandler

# LangGraph imports
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver

# Google Gemini API
import google.generativeai as genai

# Custom modules
from src.rag.knowledge_base import KnowledgeBaseManager
from src.agents.geometry_agent import GeometrySolverAgent
from src.tools.geometry_tools import GeometryTools
from src.workflows.solving_workflow import SolvingWorkflow
from src.utils.visualization import GeometryVisualizer
from src.utils.evaluation import GeometryEvaluator

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler("logs/experimental_geometry.log"),
        logging.StreamHandler(),
    ],
)
logger = logging.getLogger(__name__)

# Rich console
console = Console()


class ExperimentalGeometrySystem:
    """
    Experimental Geometric Reasoning System
    سیستم تجربی استدلال هندسی
    """

    def __init__(self):
        """Initialize the experimental system"""
        self.console = Console()
        self.setup_directories()
        self.initialize_components()

    def setup_directories(self):
        """Create necessary directories"""
        directories = [
            "logs",
            "results",
            "data/vector_db",
            "data/knowledge_base",
            "visualizations",
            "models",
            "cache",
        ]

        for directory in directories:
            os.makedirs(directory, exist_ok=True)

    def initialize_components(self):
        """Initialize all system components"""
        try:
            # Initialize Gemini API
            self.initialize_gemini()

            # Initialize RAG system
            self.initialize_rag()

            # Initialize agents and tools
            self.initialize_agents()

            # Initialize workflows
            self.initialize_workflows()

            # Initialize utilities
            self.initialize_utilities()

            console.print("✅ All components initialized successfully")

        except Exception as e:
            console.print(f"❌ Error initializing components: {e}")
            logger.error(f"Initialization error: {e}")

    def initialize_gemini(self):
        """Initialize Google Gemini API"""
        try:
            api_key = os.getenv("GOOGLE_API_KEY")
            if not api_key:
                raise ValueError("GOOGLE_API_KEY environment variable not set")

            genai.configure(api_key=api_key)

            self.gemini_model = genai.GenerativeModel(
                model_name="gemini-1.5-pro",
                generation_config=genai.types.GenerationConfig(
                    temperature=0.1,
                    max_output_tokens=2048,
                ),
            )

            self.llm = GoogleGenerativeAI(
                model="gemini-1.5-pro",
                google_api_key=api_key,
                temperature=0.1,
                max_output_tokens=2048,
            )

            console.print("✅ Gemini API initialized")

        except Exception as e:
            console.print(f"❌ Gemini initialization failed: {e}")
            raise

    def initialize_rag(self):
        """Initialize RAG system"""
        try:
            # Initialize embeddings
            self.embeddings = SentenceTransformerEmbeddings(
                model_name="sentence-transformers/all-MiniLM-L6-v2"
            )

            # Initialize vector store
            self.vector_store = Chroma(
                persist_directory="./data/vector_db",
                collection_name="geometry_knowledge",
                embedding_function=self.embeddings,
            )

            # Initialize knowledge base manager
            self.knowledge_base = KnowledgeBaseManager(
                config={
                    "sources": [
                        {
                            "name": "geometry_theorems",
                            "path": "./data/knowledge_base/geometry_theorems.json",
                            "type": "json",
                        }
                    ],
                    "document_processing": {
                        "text_splitter": {
                            "chunk_size": 1000,
                            "chunk_overlap": 200,
                            "separators": ["\n\n", "\n", " ", ""],
                        }
                    },
                },
                embeddings=self.embeddings,
                vector_store=self.vector_store,
            )

            # Create sample knowledge base
            self.knowledge_base.create_sample_knowledge_base()

            console.print("✅ RAG system initialized")

        except Exception as e:
            console.print(f"❌ RAG initialization failed: {e}")
            raise

    def initialize_agents(self):
        """Initialize LangChain agents"""
        try:
            # Initialize memory
            self.memory = ConversationBufferMemory(
                memory_key="chat_history", return_messages=True, max_token_limit=4000
            )

            # Initialize geometry tools
            self.geometry_tools = GeometryTools(
                config={
                    "theorem_lookup": {
                        "parameters": {"database_path": "data/theorems/theorems.json"}
                    }
                },
                llm=self.llm,
            )

            # Create LangChain tools
            self.langchain_tools = [
                Tool(
                    name="coordinate_geometry",
                    description="حل مسائل هندسی با استفاده از هندسه تحلیلی",
                    func=self.geometry_tools.coordinate_geometry_solver,
                ),
                Tool(
                    name="synthetic_geometry",
                    description="حل مسائل هندسی با استفاده از هندسه ترکیبی",
                    func=self.geometry_tools.synthetic_geometry_solver,
                ),
                Tool(
                    name="theorem_lookup",
                    description="جستجوی قضایای مرتبط",
                    func=self.geometry_tools.theorem_lookup,
                ),
                Tool(
                    name="proof_verification",
                    description="تأیید صحت اثبات‌های ریاضی",
                    func=self.geometry_tools.proof_verification,
                ),
                Tool(
                    name="diagram_generation",
                    description="ایجاد نمودارهای هندسی",
                    func=self.geometry_tools.diagram_generation,
                ),
                Tool(
                    name="rag_query",
                    description="جستجو در پایگاه دانش",
                    func=self._rag_query_tool,
                ),
            ]

            # Initialize geometry agent
            self.geometry_agent = GeometrySolverAgent(
                llm=self.llm,
                memory=self.memory,
                config={
                    "name": "ExperimentalGeometryAgent",
                    "description": "عامل تجربی برای حل مسائل هندسی",
                    "tools": [tool.name for tool in self.langchain_tools],
                    "settings": {
                        "max_iterations": 15,
                        "temperature": 0.1,
                        "verbose": True,
                        "return_intermediate_steps": True,
                    },
                },
            )

            console.print("✅ Agents initialized")

        except Exception as e:
            console.print(f"❌ Agent initialization failed: {e}")
            raise

    def initialize_workflows(self):
        """Initialize LangGraph workflows"""
        try:
            self.solving_workflow = SolvingWorkflow(
                config={"workflow": {"max_steps": 20, "checkpointing": True}},
                llm=self.llm,
                tools=self.langchain_tools,
                memory=self.memory,
            )

            console.print("✅ Workflows initialized")

        except Exception as e:
            console.print(f"❌ Workflow initialization failed: {e}")
            raise

    def initialize_utilities(self):
        """Initialize utility components"""
        try:
            self.visualizer = GeometryVisualizer()
            self.evaluator = GeometryEvaluator(
                {
                    "metrics": [
                        "accuracy",
                        "proof_quality",
                        "time_efficiency",
                        "explanation_quality",
                        "confidence_calibration",
                    ]
                }
            )

            console.print("✅ Utilities initialized")

        except Exception as e:
            console.print(f"❌ Utility initialization failed: {e}")
            raise

    def _rag_query_tool(self, query: str) -> str:
        """Tool function for RAG queries"""
        try:
            retriever = self.vector_store.as_retriever(
                search_type="similarity", search_kwargs={"k": 5}
            )

            rag_chain = RetrievalQA.from_chain_type(
                llm=self.llm,
                chain_type="stuff",
                retriever=retriever,
                return_source_documents=True,
            )

            result = rag_chain({"query": query})
            return result["result"]

        except Exception as e:
            return f"خطا در جستجو: {str(e)}"

    async def run_comprehensive_experiment(self, problem: str):
        """
        Run a comprehensive experiment with all methods
        Args:
            problem: Geometric problem to solve
        """
        console.print(
            Panel(
                f"🔬 اجرای آزمایش جامع\nمسئله: {problem}",
                title="Experimental Geometry System",
                style="bold blue",
            )
        )

        results = {}

        with Progress(
            SpinnerColumn(),
            TextColumn("[progress.description]{task.description}"),
            console=self.console,
        ) as progress:

            # Method 1: Gemini Direct
            task1 = progress.add_task("Gemini Direct...", total=None)
            try:
                results["gemini_direct"] = await self._test_gemini_direct(problem)
                progress.update(task1, description="✅ Gemini Direct completed")
            except Exception as e:
                results["gemini_direct"] = {"error": str(e)}
                progress.update(task1, description="❌ Gemini Direct failed")

            # Method 2: LangChain Agent
            task2 = progress.add_task("LangChain Agent...", total=None)
            try:
                results["langchain_agent"] = await self._test_langchain_agent(problem)
                progress.update(task2, description="✅ LangChain Agent completed")
            except Exception as e:
                results["langchain_agent"] = {"error": str(e)}
                progress.update(task2, description="❌ LangChain Agent failed")

            # Method 3: RAG Enhanced
            task3 = progress.add_task("RAG Enhanced...", total=None)
            try:
                results["rag_enhanced"] = await self._test_rag_enhanced(problem)
                progress.update(task3, description="✅ RAG Enhanced completed")
            except Exception as e:
                results["rag_enhanced"] = {"error": str(e)}
                progress.update(task3, description="❌ RAG Enhanced failed")

            # Method 4: LangGraph Workflow
            task4 = progress.add_task("LangGraph Workflow...", total=None)
            try:
                results["langgraph_workflow"] = await self._test_langgraph_workflow(
                    problem
                )
                progress.update(task4, description="✅ LangGraph Workflow completed")
            except Exception as e:
                results["langgraph_workflow"] = {"error": str(e)}
                progress.update(task4, description="❌ LangGraph Workflow failed")

        # Evaluate results
        evaluations = self._evaluate_results(results, problem)

        # Generate visualizations
        visualizations = self._generate_visualizations(results, evaluations)

        # Display comprehensive report
        self._display_comprehensive_report(results, evaluations, visualizations)

        # Save results
        self._save_experimental_results(results, evaluations, visualizations)

        return results, evaluations, visualizations

    async def _test_gemini_direct(self, problem: str) -> Dict[str, Any]:
        """Test Gemini API directly"""
        try:
            prompt = f"""
            شما یک متخصص در هندسه المپیادی هستید. لطفاً مسئله زیر را حل کنید:
            
            مسئله: {problem}
            
            لطفاً:
            1. مسئله را تحلیل کنید
            2. راه‌حل گام به گام ارائه دهید
            3. اثبات کنید که راه‌حل درست است
            4. پاسخ نهایی را مشخص کنید
            
            پاسخ خود را به فارسی ارائه دهید.
            """

            response = await self.gemini_model.generate_content_async(prompt)

            return {
                "method": "gemini_direct",
                "solution": response.text,
                "confidence": 0.8,
                "timestamp": datetime.now().isoformat(),
                "model": "gemini-1.5-pro",
            }

        except Exception as e:
            logger.error(f"Gemini direct test failed: {e}")
            raise

    async def _test_langchain_agent(self, problem: str) -> Dict[str, Any]:
        """Test LangChain agent"""
        try:
            # Create agent executor
            agent_executor = AgentExecutor.from_agent_and_tools(
                agent=self.geometry_agent.create_agent(),
                tools=self.langchain_tools,
                memory=self.memory,
                verbose=True,
                max_iterations=10,
                handle_parsing_errors=True,
            )

            result = await agent_executor.ainvoke(
                {"input": f"لطفاً این مسئله هندسی را حل کنید: {problem}"}
            )

            return {
                "method": "langchain_agent",
                "solution": result["output"],
                "intermediate_steps": result.get("intermediate_steps", []),
                "tools_used": self.geometry_agent._extract_tools_used(
                    result.get("intermediate_steps", [])
                ),
                "confidence": 0.7,
                "timestamp": datetime.now().isoformat(),
            }

        except Exception as e:
            logger.error(f"LangChain agent test failed: {e}")
            raise

    async def _test_rag_enhanced(self, problem: str) -> Dict[str, Any]:
        """Test RAG-enhanced solving"""
        try:
            retriever = self.vector_store.as_retriever(
                search_type="similarity", search_kwargs={"k": 5}
            )

            chain = ConversationalRetrievalChain.from_llm(
                llm=self.llm,
                retriever=retriever,
                memory=self.memory,
                return_source_documents=True,
                verbose=True,
            )

            result = await chain.ainvoke(
                {"question": f"لطفاً این مسئله هندسی را حل کنید: {problem}"}
            )

            return {
                "method": "rag_enhanced",
                "solution": result["answer"],
                "source_documents": [
                    doc.page_content for doc in result["source_documents"]
                ],
                "confidence": 0.75,
                "timestamp": datetime.now().isoformat(),
            }

        except Exception as e:
            logger.error(f"RAG enhanced test failed: {e}")
            raise

    async def _test_langgraph_workflow(self, problem: str) -> Dict[str, Any]:
        """Test LangGraph workflow"""
        try:
            result = await self.solving_workflow.execute_workflow(problem)

            return {
                "method": "langgraph_workflow",
                "solution": result["final_answer"],
                "workflow_steps": result["steps"],
                "confidence": 0.8,
                "timestamp": datetime.now().isoformat(),
            }

        except Exception as e:
            logger.error(f"LangGraph workflow test failed: {e}")
            raise

    def _evaluate_results(
        self, results: Dict[str, Any], problem: str
    ) -> Dict[str, Any]:
        """Evaluate all results"""
        try:
            evaluations = {}

            for method, result in results.items():
                if "error" not in result:
                    evaluation = self.evaluator.evaluate_solution(
                        problem=problem,
                        solution=result.get("solution", ""),
                        method=method,
                    )
                    evaluations[method] = evaluation

            # Compare solutions
            comparison = self.evaluator.compare_solutions(evaluations)
            evaluations["comparison"] = comparison

            return evaluations

        except Exception as e:
            logger.error(f"Evaluation failed: {e}")
            return {}

    def _generate_visualizations(
        self, results: Dict[str, Any], evaluations: Dict[str, Any]
    ) -> Dict[str, str]:
        """Generate visualizations"""
        try:
            visualizations = {}

            # Comparison chart
            if len(results) > 1:
                chart_path = self.visualizer.create_comparison_chart(
                    results, "مقایسه روش‌های حل مسئله"
                )
                visualizations["comparison_chart"] = chart_path

            # Workflow diagram for LangGraph results
            if (
                "langgraph_workflow" in results
                and "workflow_steps" in results["langgraph_workflow"]
            ):
                workflow_path = self.visualizer.create_workflow_diagram(
                    results["langgraph_workflow"]["workflow_steps"],
                    "گردش کار LangGraph",
                )
                visualizations["workflow_diagram"] = workflow_path

            # Sample triangle diagram
            triangle_path = self.visualizer.create_triangle_diagram(
                vertices=[(1, 1), (4, 1), (2.5, 3)],
                labels=["A", "B", "C"],
                angles=[60, 60, 60],
                title="مثلث نمونه",
            )
            visualizations["sample_triangle"] = triangle_path

            return visualizations

        except Exception as e:
            logger.error(f"Visualization generation failed: {e}")
            return {}

    def _display_comprehensive_report(
        self,
        results: Dict[str, Any],
        evaluations: Dict[str, Any],
        visualizations: Dict[str, str],
    ):
        """Display comprehensive experimental report"""

        console.print("\n" + "=" * 80)
        console.print(
            Panel(
                "گزارش جامع آزمایش",
                title="Comprehensive Experimental Report",
                style="bold green",
            )
        )

        # Results summary
        console.print("\n📊 خلاصه نتایج:")
        table = Table(title="نتایج روش‌ها")
        table.add_column("روش", style="cyan")
        table.add_column("وضعیت", style="green")
        table.add_column("اعتماد", style="magenta")
        table.add_column("طول راه‌حل", style="blue")

        for method, result in results.items():
            if "error" in result:
                table.add_row(method, "❌ ناموفق", "N/A", "N/A")
            else:
                confidence = result.get("confidence", 0)
                solution_length = len(result.get("solution", ""))
                table.add_row(
                    method, "✅ موفق", f"{confidence:.2f}", f"{solution_length}"
                )

        console.print(table)

        # Evaluation summary
        if evaluations and "comparison" in evaluations:
            comparison = evaluations["comparison"]
            console.print(f"\n🏆 بهترین روش: {comparison.get('best_method', 'N/A')}")

            if "analysis" in comparison:
                analysis = comparison["analysis"]
                console.print(f"📈 محدوده امتیاز: {analysis.get('score_range', 0):.2f}")
                console.print(f"🎯 سازگاری: {analysis.get('consistency', 'N/A')}")

        # Visualizations
        if visualizations:
            console.print("\n🎨 تصاویر تولید شده:")
            for viz_type, path in visualizations.items():
                console.print(f"  📁 {viz_type}: {path}")

        # Detailed solutions
        console.print("\n📝 راه‌حل‌های تفصیلی:")
        for method, result in results.items():
            if "error" not in result:
                console.print(f"\n🔹 {method.upper()}:")
                solution = result.get("solution", "")
                if solution:
                    # Truncate long solutions
                    display_solution = (
                        solution[:500] + "..." if len(solution) > 500 else solution
                    )
                    console.print(Panel(display_solution, title="راه‌حل", style="green"))

    def _save_experimental_results(
        self,
        results: Dict[str, Any],
        evaluations: Dict[str, Any],
        visualizations: Dict[str, str],
    ):
        """Save experimental results"""
        try:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

            # Save results
            results_file = f"results/experimental_results_{timestamp}.json"
            with open(results_file, "w", encoding="utf-8") as f:
                import json

                json.dump(
                    {
                        "results": results,
                        "evaluations": evaluations,
                        "visualizations": visualizations,
                        "timestamp": timestamp,
                    },
                    f,
                    ensure_ascii=False,
                    indent=2,
                )

            # Save evaluation report
            if evaluations:
                report = self.evaluator.generate_evaluation_report(evaluations)
                report_file = f"results/evaluation_report_{timestamp}.txt"
                with open(report_file, "w", encoding="utf-8") as f:
                    f.write(report)

            console.print(f"\n💾 نتایج ذخیره شد:")
            console.print(f"  📄 نتایج: {results_file}")
            if evaluations:
                console.print(f"  📊 گزارش ارزیابی: {report_file}")

        except Exception as e:
            logger.error(f"Error saving results: {e}")
            console.print(f"❌ خطا در ذخیره نتایج: {e}")


async def main():
    """Main experimental function"""
    console.print(
        Panel(
            "سیستم تجربی استدلال هندسی\nExperimental Geometric Reasoning System",
            title="CA5: Advanced Experimental System",
            style="bold blue",
        )
    )

    try:
        # Initialize system
        system = ExperimentalGeometrySystem()

        # Example problems
        example_problems = [
            "در مثلث ABC، زاویه A = 60 درجه، AB = 5، AC = 7. طول BC را پیدا کنید.",
            "دایره‌ای با شعاع 5 رسم شده است. اگر وتری به طول 8 رسم شود، فاصله آن از مرکز دایره چقدر است؟",
            "در مثلث متساوی‌الاضلاع ABC، نقطه D روی ضلع BC قرار دارد. اگر AD = 6 و BD = 2، طول DC را پیدا کنید.",
            "در دایره‌ای، دو وتر AB و CD همدیگر را در نقطه E قطع می‌کنند. اگر AE = 3، EB = 4، CE = 2، طول ED را پیدا کنید.",
        ]

        console.print("\n📚 مثال‌های مسئله:")
        for i, problem in enumerate(example_problems, 1):
            console.print(f"{i}. {problem}")

        # Get user input
        console.print("\n" + "=" * 50)
        user_problem = console.input(
            "[bold green]مسئله هندسی خود را وارد کنید (یا Enter برای مثال اول): [/bold green]"
        )

        if not user_problem.strip():
            user_problem = example_problems[0]

        # Run comprehensive experiment
        results, evaluations, visualizations = (
            await system.run_comprehensive_experiment(user_problem)
        )

        console.print("\n✅ آزمایش جامع تکمیل شد!")

        # Ask for another experiment
        another = console.input(
            "\n[bold yellow]آیا می‌خواهید آزمایش دیگری انجام دهید؟ (y/n): [/bold yellow]"
        )
        if another.lower() in ["y", "yes", "بله"]:
            await main()

    except KeyboardInterrupt:
        console.print("\n⏹️ آزمایش توسط کاربر متوقف شد.")
    except Exception as e:
        console.print(f"\n❌ خطا در آزمایش: {e}")
        logger.error(f"Main experimental error: {e}")


if __name__ == "__main__":
    # Run the experimental system
    asyncio.run(main())

Xet Storage Details

Size:: 27.6 kB
Xet hash:: 51241c4cfef9fcb33028e46893d3d5bdb606443bbc0ce2e0482c17ae83c95284

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	#!/usr/bin/env python3
	"""
	Advanced Experimental Geometric Reasoning with LangChain, RAG, and Gemini API
	تجربه پیشرفته استدلال هندسی با LangChain، RAG و Gemini API

	This is a comprehensive experimental script that demonstrates:
	- LangChain agents with custom tools
	- RAG system with vector databases
	- Gemini API integration
	- Multiple solving strategies
	- Advanced evaluation and visualization

	Author: AI Systems Course Team
	Version: 2.0.0
	Date: January 2025
	"""

	import os
	import sys
	import asyncio
	import logging
	from typing import Dict, List, Any, Optional
	from datetime import datetime
	from pathlib import Path

	# Add src to path
	sys.path.append(str(Path(__file__).parent / "src"))

	# Core dependencies
	import numpy as np
	import pandas as pd
	from rich.console import Console
	from rich.table import Table
	from rich.panel import Panel
	from rich.progress import Progress, SpinnerColumn, TextColumn

	# LangChain imports
	from langchain.llms import GoogleGenerativeAI
	from langchain.chains import RetrievalQA, ConversationalRetrievalChain
	from langchain.memory import ConversationBufferMemory
	from langchain.agents import AgentExecutor, create_conversational_agent
	from langchain.tools import Tool
	from langchain.prompts import PromptTemplate
	from langchain.schema import Document
	from langchain.text_splitter import RecursiveCharacterTextSplitter
	from langchain.embeddings import SentenceTransformerEmbeddings
	from langchain.vectorstores import Chroma, FAISS
	from langchain.document_loaders import TextLoader
	from langchain.callbacks import StreamingStdOutCallbackHandler

	# LangGraph imports
	from langgraph.graph import StateGraph, END
	from langgraph.prebuilt import ToolNode
	from langgraph.checkpoint.memory import MemorySaver

	# Google Gemini API
	import google.generativeai as genai

	# Custom modules
	from src.rag.knowledge_base import KnowledgeBaseManager
	from src.agents.geometry_agent import GeometrySolverAgent
	from src.tools.geometry_tools import GeometryTools
	from src.workflows.solving_workflow import SolvingWorkflow
	from src.utils.visualization import GeometryVisualizer
	from src.utils.evaluation import GeometryEvaluator

	# Setup logging
	logging.basicConfig(
	level=logging.INFO,
	format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
	handlers=[
	logging.FileHandler("logs/experimental_geometry.log"),
	logging.StreamHandler(),
	],
	)
	logger = logging.getLogger(__name__)

	# Rich console
	console = Console()


	class ExperimentalGeometrySystem:
	"""
	Experimental Geometric Reasoning System
	سیستم تجربی استدلال هندسی
	"""

	def __init__(self):
	"""Initialize the experimental system"""
	self.console = Console()
	self.setup_directories()
	self.initialize_components()

	def setup_directories(self):
	"""Create necessary directories"""
	directories = [
	"logs",
	"results",
	"data/vector_db",
	"data/knowledge_base",
	"visualizations",
	"models",
	"cache",
	]

	for directory in directories:
	os.makedirs(directory, exist_ok=True)

	def initialize_components(self):
	"""Initialize all system components"""
	try:
	# Initialize Gemini API
	self.initialize_gemini()

	# Initialize RAG system
	self.initialize_rag()

	# Initialize agents and tools
	self.initialize_agents()

	# Initialize workflows
	self.initialize_workflows()

	# Initialize utilities
	self.initialize_utilities()

	console.print("✅ All components initialized successfully")

	except Exception as e:
	console.print(f"❌ Error initializing components: {e}")
	logger.error(f"Initialization error: {e}")

	def initialize_gemini(self):
	"""Initialize Google Gemini API"""
	try:
	api_key = os.getenv("GOOGLE_API_KEY")
	if not api_key:
	raise ValueError("GOOGLE_API_KEY environment variable not set")

	genai.configure(api_key=api_key)

	self.gemini_model = genai.GenerativeModel(
	model_name="gemini-1.5-pro",
	generation_config=genai.types.GenerationConfig(
	temperature=0.1,
	max_output_tokens=2048,
	),
	)

	self.llm = GoogleGenerativeAI(
	model="gemini-1.5-pro",
	google_api_key=api_key,
	temperature=0.1,
	max_output_tokens=2048,
	)

	console.print("✅ Gemini API initialized")

	except Exception as e:
	console.print(f"❌ Gemini initialization failed: {e}")
	raise

	def initialize_rag(self):
	"""Initialize RAG system"""
	try:
	# Initialize embeddings
	self.embeddings = SentenceTransformerEmbeddings(
	model_name="sentence-transformers/all-MiniLM-L6-v2"
	)

	# Initialize vector store
	self.vector_store = Chroma(
	persist_directory="./data/vector_db",
	collection_name="geometry_knowledge",
	embedding_function=self.embeddings,
	)

	# Initialize knowledge base manager
	self.knowledge_base = KnowledgeBaseManager(
	config={
	"sources": [
	{
	"name": "geometry_theorems",
	"path": "./data/knowledge_base/geometry_theorems.json",
	"type": "json",
	}
	],
	"document_processing": {
	"text_splitter": {
	"chunk_size": 1000,
	"chunk_overlap": 200,
	"separators": ["\n\n", "\n", " ", ""],
	}
	},
	},
	embeddings=self.embeddings,
	vector_store=self.vector_store,
	)

	# Create sample knowledge base
	self.knowledge_base.create_sample_knowledge_base()

	console.print("✅ RAG system initialized")

	except Exception as e:
	console.print(f"❌ RAG initialization failed: {e}")
	raise

	def initialize_agents(self):
	"""Initialize LangChain agents"""
	try:
	# Initialize memory
	self.memory = ConversationBufferMemory(
	memory_key="chat_history", return_messages=True, max_token_limit=4000
	)

	# Initialize geometry tools
	self.geometry_tools = GeometryTools(
	config={
	"theorem_lookup": {
	"parameters": {"database_path": "data/theorems/theorems.json"}
	}
	},
	llm=self.llm,
	)

	# Create LangChain tools
	self.langchain_tools = [
	Tool(
	name="coordinate_geometry",
	description="حل مسائل هندسی با استفاده از هندسه تحلیلی",
	func=self.geometry_tools.coordinate_geometry_solver,
	),
	Tool(
	name="synthetic_geometry",
	description="حل مسائل هندسی با استفاده از هندسه ترکیبی",
	func=self.geometry_tools.synthetic_geometry_solver,
	),
	Tool(
	name="theorem_lookup",
	description="جستجوی قضایای مرتبط",
	func=self.geometry_tools.theorem_lookup,
	),
	Tool(
	name="proof_verification",
	description="تأیید صحت اثبات‌های ریاضی",
	func=self.geometry_tools.proof_verification,
	),
	Tool(
	name="diagram_generation",
	description="ایجاد نمودارهای هندسی",
	func=self.geometry_tools.diagram_generation,
	),
	Tool(
	name="rag_query",
	description="جستجو در پایگاه دانش",
	func=self._rag_query_tool,
	),
	]

	# Initialize geometry agent
	self.geometry_agent = GeometrySolverAgent(
	llm=self.llm,
	memory=self.memory,
	config={
	"name": "ExperimentalGeometryAgent",
	"description": "عامل تجربی برای حل مسائل هندسی",
	"tools": [tool.name for tool in self.langchain_tools],
	"settings": {
	"max_iterations": 15,
	"temperature": 0.1,
	"verbose": True,
	"return_intermediate_steps": True,
	},
	},
	)

	console.print("✅ Agents initialized")

	except Exception as e:
	console.print(f"❌ Agent initialization failed: {e}")
	raise

	def initialize_workflows(self):
	"""Initialize LangGraph workflows"""
	try:
	self.solving_workflow = SolvingWorkflow(
	config={"workflow": {"max_steps": 20, "checkpointing": True}},
	llm=self.llm,
	tools=self.langchain_tools,
	memory=self.memory,
	)

	console.print("✅ Workflows initialized")

	except Exception as e:
	console.print(f"❌ Workflow initialization failed: {e}")
	raise

	def initialize_utilities(self):
	"""Initialize utility components"""
	try:
	self.visualizer = GeometryVisualizer()
	self.evaluator = GeometryEvaluator(
	{
	"metrics": [
	"accuracy",
	"proof_quality",
	"time_efficiency",
	"explanation_quality",
	"confidence_calibration",
	]
	}
	)

	console.print("✅ Utilities initialized")

	except Exception as e:
	console.print(f"❌ Utility initialization failed: {e}")
	raise

	def _rag_query_tool(self, query: str) -> str:
	"""Tool function for RAG queries"""
	try:
	retriever = self.vector_store.as_retriever(
	search_type="similarity", search_kwargs={"k": 5}
	)

	rag_chain = RetrievalQA.from_chain_type(
	llm=self.llm,
	chain_type="stuff",
	retriever=retriever,
	return_source_documents=True,
	)

	result = rag_chain({"query": query})
	return result["result"]

	except Exception as e:
	return f"خطا در جستجو: {str(e)}"

	async def run_comprehensive_experiment(self, problem: str):
	"""
	Run a comprehensive experiment with all methods

	Args:
	problem: Geometric problem to solve
	"""
	console.print(
	Panel(
	f"🔬 اجرای آزمایش جامع\nمسئله: {problem}",
	title="Experimental Geometry System",
	style="bold blue",
	)
	)

	results = {}

	with Progress(
	SpinnerColumn(),
	TextColumn("[progress.description]{task.description}"),
	console=self.console,
	) as progress:

	# Method 1: Gemini Direct
	task1 = progress.add_task("Gemini Direct...", total=None)
	try:
	results["gemini_direct"] = await self._test_gemini_direct(problem)
	progress.update(task1, description="✅ Gemini Direct completed")
	except Exception as e:
	results["gemini_direct"] = {"error": str(e)}
	progress.update(task1, description="❌ Gemini Direct failed")

	# Method 2: LangChain Agent
	task2 = progress.add_task("LangChain Agent...", total=None)
	try:
	results["langchain_agent"] = await self._test_langchain_agent(problem)
	progress.update(task2, description="✅ LangChain Agent completed")
	except Exception as e:
	results["langchain_agent"] = {"error": str(e)}
	progress.update(task2, description="❌ LangChain Agent failed")

	# Method 3: RAG Enhanced
	task3 = progress.add_task("RAG Enhanced...", total=None)
	try:
	results["rag_enhanced"] = await self._test_rag_enhanced(problem)
	progress.update(task3, description="✅ RAG Enhanced completed")
	except Exception as e:
	results["rag_enhanced"] = {"error": str(e)}
	progress.update(task3, description="❌ RAG Enhanced failed")

	# Method 4: LangGraph Workflow
	task4 = progress.add_task("LangGraph Workflow...", total=None)
	try:
	results["langgraph_workflow"] = await self._test_langgraph_workflow(
	problem
	)
	progress.update(task4, description="✅ LangGraph Workflow completed")
	except Exception as e:
	results["langgraph_workflow"] = {"error": str(e)}
	progress.update(task4, description="❌ LangGraph Workflow failed")

	# Evaluate results
	evaluations = self._evaluate_results(results, problem)

	# Generate visualizations
	visualizations = self._generate_visualizations(results, evaluations)

	# Display comprehensive report
	self._display_comprehensive_report(results, evaluations, visualizations)

	# Save results
	self._save_experimental_results(results, evaluations, visualizations)

	return results, evaluations, visualizations

	async def _test_gemini_direct(self, problem: str) -> Dict[str, Any]:
	"""Test Gemini API directly"""
	try:
	prompt = f"""
	شما یک متخصص در هندسه المپیادی هستید. لطفاً مسئله زیر را حل کنید:

	مسئله: {problem}

	لطفاً:
	1. مسئله را تحلیل کنید
	2. راه‌حل گام به گام ارائه دهید
	3. اثبات کنید که راه‌حل درست است
	4. پاسخ نهایی را مشخص کنید

	پاسخ خود را به فارسی ارائه دهید.
	"""

	response = await self.gemini_model.generate_content_async(prompt)

	return {
	"method": "gemini_direct",
	"solution": response.text,
	"confidence": 0.8,
	"timestamp": datetime.now().isoformat(),
	"model": "gemini-1.5-pro",
	}

	except Exception as e:
	logger.error(f"Gemini direct test failed: {e}")
	raise

	async def _test_langchain_agent(self, problem: str) -> Dict[str, Any]:
	"""Test LangChain agent"""
	try:
	# Create agent executor
	agent_executor = AgentExecutor.from_agent_and_tools(
	agent=self.geometry_agent.create_agent(),
	tools=self.langchain_tools,
	memory=self.memory,
	verbose=True,
	max_iterations=10,
	handle_parsing_errors=True,
	)

	result = await agent_executor.ainvoke(
	{"input": f"لطفاً این مسئله هندسی را حل کنید: {problem}"}
	)

	return {
	"method": "langchain_agent",
	"solution": result["output"],
	"intermediate_steps": result.get("intermediate_steps", []),
	"tools_used": self.geometry_agent._extract_tools_used(
	result.get("intermediate_steps", [])
	),
	"confidence": 0.7,
	"timestamp": datetime.now().isoformat(),
	}

	except Exception as e:
	logger.error(f"LangChain agent test failed: {e}")
	raise

	async def _test_rag_enhanced(self, problem: str) -> Dict[str, Any]:
	"""Test RAG-enhanced solving"""
	try:
	retriever = self.vector_store.as_retriever(
	search_type="similarity", search_kwargs={"k": 5}
	)

	chain = ConversationalRetrievalChain.from_llm(
	llm=self.llm,
	retriever=retriever,
	memory=self.memory,
	return_source_documents=True,
	verbose=True,
	)

	result = await chain.ainvoke(
	{"question": f"لطفاً این مسئله هندسی را حل کنید: {problem}"}
	)

	return {
	"method": "rag_enhanced",
	"solution": result["answer"],
	"source_documents": [
	doc.page_content for doc in result["source_documents"]
	],
	"confidence": 0.75,
	"timestamp": datetime.now().isoformat(),
	}

	except Exception as e:
	logger.error(f"RAG enhanced test failed: {e}")
	raise

	async def _test_langgraph_workflow(self, problem: str) -> Dict[str, Any]:
	"""Test LangGraph workflow"""
	try:
	result = await self.solving_workflow.execute_workflow(problem)

	return {
	"method": "langgraph_workflow",
	"solution": result["final_answer"],
	"workflow_steps": result["steps"],
	"confidence": 0.8,
	"timestamp": datetime.now().isoformat(),
	}

	except Exception as e:
	logger.error(f"LangGraph workflow test failed: {e}")
	raise

	def _evaluate_results(
	self, results: Dict[str, Any], problem: str
	) -> Dict[str, Any]:
	"""Evaluate all results"""
	try:
	evaluations = {}

	for method, result in results.items():
	if "error" not in result:
	evaluation = self.evaluator.evaluate_solution(
	problem=problem,
	solution=result.get("solution", ""),
	method=method,
	)
	evaluations[method] = evaluation

	# Compare solutions
	comparison = self.evaluator.compare_solutions(evaluations)
	evaluations["comparison"] = comparison

	return evaluations

	except Exception as e:
	logger.error(f"Evaluation failed: {e}")
	return {}

	def _generate_visualizations(
	self, results: Dict[str, Any], evaluations: Dict[str, Any]
	) -> Dict[str, str]:
	"""Generate visualizations"""
	try:
	visualizations = {}

	# Comparison chart
	if len(results) > 1:
	chart_path = self.visualizer.create_comparison_chart(
	results, "مقایسه روش‌های حل مسئله"
	)
	visualizations["comparison_chart"] = chart_path

	# Workflow diagram for LangGraph results
	if (
	"langgraph_workflow" in results
	and "workflow_steps" in results["langgraph_workflow"]
	):
	workflow_path = self.visualizer.create_workflow_diagram(
	results["langgraph_workflow"]["workflow_steps"],
	"گردش کار LangGraph",
	)
	visualizations["workflow_diagram"] = workflow_path

	# Sample triangle diagram
	triangle_path = self.visualizer.create_triangle_diagram(
	vertices=[(1, 1), (4, 1), (2.5, 3)],
	labels=["A", "B", "C"],
	angles=[60, 60, 60],
	title="مثلث نمونه",
	)
	visualizations["sample_triangle"] = triangle_path

	return visualizations

	except Exception as e:
	logger.error(f"Visualization generation failed: {e}")
	return {}

	def _display_comprehensive_report(
	self,
	results: Dict[str, Any],
	evaluations: Dict[str, Any],
	visualizations: Dict[str, str],
	):
	"""Display comprehensive experimental report"""

	console.print("\n" + "=" * 80)
	console.print(
	Panel(
	"گزارش جامع آزمایش",
	title="Comprehensive Experimental Report",
	style="bold green",
	)
	)

	# Results summary
	console.print("\n📊 خلاصه نتایج:")
	table = Table(title="نتایج روش‌ها")
	table.add_column("روش", style="cyan")
	table.add_column("وضعیت", style="green")
	table.add_column("اعتماد", style="magenta")
	table.add_column("طول راه‌حل", style="blue")

	for method, result in results.items():
	if "error" in result:
	table.add_row(method, "❌ ناموفق", "N/A", "N/A")
	else:
	confidence = result.get("confidence", 0)
	solution_length = len(result.get("solution", ""))
	table.add_row(
	method, "✅ موفق", f"{confidence:.2f}", f"{solution_length}"
	)

	console.print(table)

	# Evaluation summary
	if evaluations and "comparison" in evaluations:
	comparison = evaluations["comparison"]
	console.print(f"\n🏆 بهترین روش: {comparison.get('best_method', 'N/A')}")

	if "analysis" in comparison:
	analysis = comparison["analysis"]
	console.print(f"📈 محدوده امتیاز: {analysis.get('score_range', 0):.2f}")
	console.print(f"🎯 سازگاری: {analysis.get('consistency', 'N/A')}")

	# Visualizations
	if visualizations:
	console.print("\n🎨 تصاویر تولید شده:")
	for viz_type, path in visualizations.items():
	console.print(f" 📁 {viz_type}: {path}")

	# Detailed solutions
	console.print("\n📝 راه‌حل‌های تفصیلی:")
	for method, result in results.items():
	if "error" not in result:
	console.print(f"\n🔹 {method.upper()}:")
	solution = result.get("solution", "")
	if solution:
	# Truncate long solutions
	display_solution = (
	solution[:500] + "..." if len(solution) > 500 else solution
	)
	console.print(Panel(display_solution, title="راه‌حل", style="green"))

	def _save_experimental_results(
	self,
	results: Dict[str, Any],
	evaluations: Dict[str, Any],
	visualizations: Dict[str, str],
	):
	"""Save experimental results"""
	try:
	timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

	# Save results
	results_file = f"results/experimental_results_{timestamp}.json"
	with open(results_file, "w", encoding="utf-8") as f:
	import json

	json.dump(
	{
	"results": results,
	"evaluations": evaluations,
	"visualizations": visualizations,
	"timestamp": timestamp,
	},
	f,
	ensure_ascii=False,
	indent=2,
	)

	# Save evaluation report
	if evaluations:
	report = self.evaluator.generate_evaluation_report(evaluations)
	report_file = f"results/evaluation_report_{timestamp}.txt"
	with open(report_file, "w", encoding="utf-8") as f:
	f.write(report)

	console.print(f"\n💾 نتایج ذخیره شد:")
	console.print(f" 📄 نتایج: {results_file}")
	if evaluations:
	console.print(f" 📊 گزارش ارزیابی: {report_file}")

	except Exception as e:
	logger.error(f"Error saving results: {e}")
	console.print(f"❌ خطا در ذخیره نتایج: {e}")


	async def main():
	"""Main experimental function"""
	console.print(
	Panel(
	"سیستم تجربی استدلال هندسی\nExperimental Geometric Reasoning System",
	title="CA5: Advanced Experimental System",
	style="bold blue",
	)
	)

	try:
	# Initialize system
	system = ExperimentalGeometrySystem()

	# Example problems
	example_problems = [
	"در مثلث ABC، زاویه A = 60 درجه، AB = 5، AC = 7. طول BC را پیدا کنید.",
	"دایره‌ای با شعاع 5 رسم شده است. اگر وتری به طول 8 رسم شود، فاصله آن از مرکز دایره چقدر است؟",
	"در مثلث متساوی‌الاضلاع ABC، نقطه D روی ضلع BC قرار دارد. اگر AD = 6 و BD = 2، طول DC را پیدا کنید.",
	"در دایره‌ای، دو وتر AB و CD همدیگر را در نقطه E قطع می‌کنند. اگر AE = 3، EB = 4، CE = 2، طول ED را پیدا کنید.",
	]

	console.print("\n📚 مثال‌های مسئله:")
	for i, problem in enumerate(example_problems, 1):
	console.print(f"{i}. {problem}")

	# Get user input
	console.print("\n" + "=" * 50)
	user_problem = console.input(
	"[bold green]مسئله هندسی خود را وارد کنید (یا Enter برای مثال اول): [/bold green]"
	)

	if not user_problem.strip():
	user_problem = example_problems[0]

	# Run comprehensive experiment
	results, evaluations, visualizations = (
	await system.run_comprehensive_experiment(user_problem)
	)

	console.print("\n✅ آزمایش جامع تکمیل شد!")

	# Ask for another experiment
	another = console.input(
	"\n[bold yellow]آیا می‌خواهید آزمایش دیگری انجام دهید؟ (y/n): [/bold yellow]"
	)
	if another.lower() in ["y", "yes", "بله"]:
	await main()

	except KeyboardInterrupt:
	console.print("\n⏹️ آزمایش توسط کاربر متوقف شد.")
	except Exception as e:
	console.print(f"\n❌ خطا در آزمایش: {e}")
	logger.error(f"Main experimental error: {e}")


	if __name__ == "__main__":
	# Run the experimental system
	asyncio.run(main())