Buckets:

tahamajs
/

Sysmem2_in_AI

Files
xet
tahamajs/Sysmem2_in_AI / ComputerAssignments /CA5_geometric_reasoning /scripts /main.py
tahamajs
5 months ago
download
raw
30.1 kB
#!/usr/bin/env python3
"""
Advanced Geometric Reasoning System with LangChain, RAG, and Gemini API
پروژه پیشرفته استدلال هندسی با LangChain، RAG و Gemini API
This module implements a comprehensive geometric reasoning system that combines:
- LangChain for agent orchestration and tool management
- RAG (Retrieval-Augmented Generation) for knowledge retrieval
- Gemini API for advanced mathematical reasoning
- Multiple solving strategies and verification methods
Author: AI Systems Course Team
Version: 2.0.0
Date: January 2025
"""

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

# 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, ConversationSummaryMemory
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, PyPDFLoader, DirectoryLoader
from langchain.callbacks import StreamingStdOutCallbackHandler

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

# RAG and Vector Database imports
import chromadb
from chromadb.config import Settings
import faiss
from sentence_transformers import SentenceTransformer

# 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/geometry_solver.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# Rich console for beautiful output
console = Console()

class AdvancedGeometrySolver:
    """
    Advanced Geometric Reasoning System with LangChain, RAG, and Gemini API
    سیستم پیشرفته استدلال هندسی با LangChain، RAG و Gemini API
    """
    
    def __init__(self, config_path: str = "config.yaml"):
        """
        Initialize the Advanced Geometry Solver
        
        Args:
            config_path: Path to configuration file
        """
        self.config_path = config_path
        self.config = self._load_config()
        self.console = Console()
        
        # Initialize components
        self._initialize_gemini()
        self._initialize_rag_system()
        self._initialize_agents()
        self._initialize_tools()
        self._initialize_workflows()
        self._initialize_evaluator()
        
        logger.info("Advanced Geometry Solver initialized successfully")
    
    def _load_config(self) -> Dict[str, Any]:
        """Load configuration from YAML file"""
        try:
            with open(self.config_path, 'r', encoding='utf-8') as file:
                config = yaml.safe_load(file)
            
            # Replace environment variables
            for key, value in config.get('api_keys', {}).items():
                if isinstance(value, str) and value.startswith('${') and value.endswith('}'):
                    env_var = value[2:-1]
                    config['api_keys'][key] = os.getenv(env_var, '')
            
            return config
        except Exception as e:
            logger.error(f"Error loading config: {e}")
            raise
    
    def _initialize_gemini(self):
        """Initialize Google Gemini API"""
        try:
            api_key = self.config['api_keys']['google_api_key']
            if not api_key:
                raise ValueError("Google API key not found in environment variables")
            
            genai.configure(api_key=api_key)
            
            # Initialize Gemini models
            self.gemini_model = genai.GenerativeModel(
                model_name=self.config['ai_models']['gemini']['model_name'],
                generation_config=genai.types.GenerationConfig(
                    temperature=self.config['ai_models']['gemini']['temperature'],
                    max_output_tokens=2048,
                ),
                safety_settings=[
                    genai.types.SafetySetting(
                        category=genai.types.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
                        threshold=genai.types.HarmBlockThreshold.BLOCK_NONE,
                    ),
                    genai.types.SafetySetting(
                        category=genai.types.HarmCategory.HARM_CATEGORY_HARASSMENT,
                        threshold=genai.types.HarmBlockThreshold.BLOCK_NONE,
                    ),
                ]
            )
            
            # Initialize LangChain Gemini LLM
            self.llm = GoogleGenerativeAI(
                model=self.config['ai_models']['gemini']['model_name'],
                google_api_key=api_key,
                temperature=self.config['ai_models']['gemini']['temperature'],
                max_output_tokens=2048,
            )
            
            logger.info("Gemini API initialized successfully")
            
        except Exception as e:
            logger.error(f"Error initializing Gemini: {e}")
            raise
    
    def _initialize_rag_system(self):
        """Initialize RAG (Retrieval-Augmented Generation) system"""
        try:
            # Initialize embeddings
            self.embeddings = SentenceTransformerEmbeddings(
                model_name=self.config['rag']['embeddings']['model_name'],
                model_kwargs=self.config['rag']['embeddings']['model_kwargs'],
                encode_kwargs=self.config['rag']['embeddings']['encode_kwargs']
            )
            
            # Initialize vector store
            vector_store_config = self.config['rag']['vector_store']
            
            if vector_store_config['type'] == 'chromadb':
                self.vector_store = Chroma(
                    persist_directory=vector_store_config['persist_directory'],
                    collection_name=vector_store_config['collection_name'],
                    embedding_function=self.embeddings
                )
            elif vector_store_config['type'] == 'faiss':
                # FAISS will be initialized when documents are added
                self.vector_store = None
            else:
                raise ValueError(f"Unsupported vector store type: {vector_store_config['type']}")
            
            # Initialize knowledge base manager
            self.knowledge_base = KnowledgeBaseManager(
                config=self.config['rag']['knowledge_base'],
                embeddings=self.embeddings,
                vector_store=self.vector_store
            )
            
            # Initialize text splitter
            splitter_config = self.config['rag']['document_processing']['text_splitter']
            self.text_splitter = RecursiveCharacterTextSplitter(
                chunk_size=splitter_config['chunk_size'],
                chunk_overlap=splitter_config['chunk_overlap'],
                length_function=len,
                separators=splitter_config['separators']
            )
            
            logger.info("RAG system initialized successfully")
            
        except Exception as e:
            logger.error(f"Error initializing RAG system: {e}")
            raise
    
    def _initialize_agents(self):
        """Initialize LangChain agents"""
        try:
            # Initialize memory
            memory_config = self.config['ai_models']['langchain']['memory']
            self.memory = ConversationBufferMemory(
                memory_key=memory_config.get('memory_key', 'chat_history'),
                return_messages=memory_config.get('return_messages', True),
                max_token_limit=memory_config.get('max_tokens', 4000)
            )
            
            # Initialize geometry solver agent
            self.geometry_agent = GeometrySolverAgent(
                llm=self.llm,
                memory=self.memory,
                config=self.config['agents']['geometry_solver']
            )
            
            logger.info("LangChain agents initialized successfully")
            
        except Exception as e:
            logger.error(f"Error initializing agents: {e}")
            raise
    
    def _initialize_tools(self):
        """Initialize geometry tools"""
        try:
            self.geometry_tools = GeometryTools(
                config=self.config['tools'],
                llm=self.llm
            )
            
            # Create LangChain tools
            self.langchain_tools = [
                Tool(
                    name="coordinate_geometry",
                    description="Solve geometric problems using coordinate geometry",
                    func=self.geometry_tools.coordinate_geometry_solver
                ),
                Tool(
                    name="synthetic_geometry",
                    description="Solve geometric problems using synthetic geometry",
                    func=self.geometry_tools.synthetic_geometry_solver
                ),
                Tool(
                    name="theorem_lookup",
                    description="Look up relevant geometric theorems",
                    func=self.geometry_tools.theorem_lookup
                ),
                Tool(
                    name="proof_verification",
                    description="Verify mathematical proofs",
                    func=self.geometry_tools.proof_verification
                ),
                Tool(
                    name="diagram_generation",
                    description="Generate geometric diagrams",
                    func=self.geometry_tools.diagram_generation
                ),
                Tool(
                    name="rag_query",
                    description="Query the knowledge base for relevant information",
                    func=self._rag_query_tool
                )
            ]
            
            logger.info("Geometry tools initialized successfully")
            
        except Exception as e:
            logger.error(f"Error initializing tools: {e}")
            raise
    
    def _initialize_workflows(self):
        """Initialize LangGraph workflows"""
        try:
            self.solving_workflow = SolvingWorkflow(
                config=self.config['ai_models']['langgraph'],
                llm=self.llm,
                tools=self.langchain_tools,
                memory=self.memory
            )
            
            logger.info("LangGraph workflows initialized successfully")
            
        except Exception as e:
            logger.error(f"Error initializing workflows: {e}")
            raise
    
    def _initialize_evaluator(self):
        """Initialize evaluation system"""
        try:
            self.evaluator = GeometryEvaluator(
                config=self.config['evaluation']
            )
            
            logger.info("Evaluation system initialized successfully")
            
        except Exception as e:
            logger.error(f"Error initializing evaluator: {e}")
            raise
    
    def _rag_query_tool(self, query: str) -> str:
        """Tool function for RAG queries"""
        try:
            # Create retriever
            retriever = self.vector_store.as_retriever(
                search_type="similarity",
                search_kwargs={"k": 5}
            )
            
            # Create RAG chain
            rag_chain = RetrievalQA.from_chain_type(
                llm=self.llm,
                chain_type="stuff",
                retriever=retriever,
                return_source_documents=True
            )
            
            # Execute query
            result = rag_chain({"query": query})
            
            return result['result']
            
            except Exception as e:
            logger.error(f"Error in RAG query: {e}")
            return f"Error retrieving information: {str(e)}"
    
    async def solve_problem(self, problem_statement: str, solver_type: str = "all") -> Dict[str, Any]:
        """
        Solve a geometric problem using multiple strategies
        
        Args:
            problem_statement: The geometric problem to solve
            solver_type: Type of solver to use ("all", "gemini", "agent", "rag", "workflow")
        
        Returns:
            Dictionary containing solutions from different methods
        """
        results = {}
        
        with Progress(
            SpinnerColumn(),
            TextColumn("[progress.description]{task.description}"),
            console=self.console
        ) as progress:
            
            # Gemini Direct Reasoning
            if solver_type in ["all", "gemini"]:
                task = progress.add_task("Solving with Gemini...", total=None)
                try:
                    results["gemini_direct"] = await self._solve_with_gemini_direct(problem_statement)
                    progress.update(task, description="✅ Gemini Direct completed")
                except Exception as e:
                    logger.error(f"Gemini direct solving failed: {e}")
                    results["gemini_direct"] = {"error": str(e)}
            
            # LangChain Agent
            if solver_type in ["all", "agent"]:
                task = progress.add_task("Solving with LangChain Agent...", total=None)
                try:
                    results["langchain_agent"] = await self._solve_with_langchain_agent(problem_statement)
                    progress.update(task, description="✅ LangChain Agent completed")
                except Exception as e:
                    logger.error(f"LangChain agent solving failed: {e}")
                    results["langchain_agent"] = {"error": str(e)}
            
            # RAG Enhanced Solving
            if solver_type in ["all", "rag"]:
                task = progress.add_task("Solving with RAG...", total=None)
                try:
                    results["rag_enhanced"] = await self._solve_with_rag(problem_statement)
                    progress.update(task, description="✅ RAG Enhanced completed")
                except Exception as e:
                    logger.error(f"RAG solving failed: {e}")
                    results["rag_enhanced"] = {"error": str(e)}
            
            # LangGraph Workflow
            if solver_type in ["all", "workflow"]:
                task = progress.add_task("Solving with LangGraph Workflow...", total=None)
                try:
                    results["langgraph_workflow"] = await self._solve_with_langgraph_workflow(problem_statement)
                    progress.update(task, description="✅ LangGraph Workflow completed")
                except Exception as e:
                    logger.error(f"LangGraph workflow solving failed: {e}")
                    results["langgraph_workflow"] = {"error": str(e)}
        
        # Compare and evaluate solutions
        comparison = self._compare_solutions(results)
        results["comparison"] = comparison
        
        return results
    
    async def _solve_with_gemini_direct(self, problem_statement: str) -> Dict[str, Any]:
        """Solve problem using Gemini API directly"""
        try:
            prompt = f"""
            شما یک متخصص در هندسه المپیادی هستید. لطفاً مسئله زیر را حل کنید:
            
            مسئله: {problem_statement}
            
            لطفاً:
            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()
            }
            
        except Exception as e:
            logger.error(f"Error in Gemini direct solving: {e}")
            raise
    
    async def _solve_with_langchain_agent(self, problem_statement: str) -> Dict[str, Any]:
        """Solve problem using 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
            )
            
            # Execute agent
            result = await agent_executor.ainvoke({
                "input": f"لطفاً این مسئله هندسی را حل کنید: {problem_statement}"
            })
            
            return {
                "method": "langchain_agent",
                "solution": result["output"],
                "intermediate_steps": result.get("intermediate_steps", []),
                "tools_used": result.get("tools_used", []),
                "confidence": 0.7,
                "timestamp": datetime.now().isoformat()
            }
            
        except Exception as e:
            logger.error(f"Error in LangChain agent solving: {e}")
            raise
    
    async def _solve_with_rag(self, problem_statement: str) -> Dict[str, Any]:
        """Solve problem using RAG-enhanced approach"""
        try:
            # Create conversational retrieval chain
            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
            )
            
            # Execute chain
            result = await chain.ainvoke({
                "question": f"لطفاً این مسئله هندسی را حل کنید: {problem_statement}"
            })
            
            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"Error in RAG solving: {e}")
            raise
    
    async def _solve_with_langgraph_workflow(self, problem_statement: str) -> Dict[str, Any]:
        """Solve problem using LangGraph workflow"""
        try:
            # Execute workflow
            result = await self.solving_workflow.execute_workflow(problem_statement)
            
            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"Error in LangGraph workflow solving: {e}")
            raise
    
    def _compare_solutions(self, results: Dict[str, Any]) -> Dict[str, Any]:
        """Compare solutions from different methods"""
        try:
            comparison = {
                "methods_used": list(results.keys()),
                "successful_methods": [],
                "failed_methods": [],
                "confidence_scores": {},
                "solution_lengths": {},
                "recommendation": ""
            }
            
            for method, result in results.items():
                if "error" in result:
                    comparison["failed_methods"].append(method)
                else:
                    comparison["successful_methods"].append(method)
                    comparison["confidence_scores"][method] = result.get("confidence", 0)
                    comparison["solution_lengths"][method] = len(result.get("solution", ""))
            
            # Determine recommendation
            if comparison["successful_methods"]:
                best_method = max(
                    comparison["successful_methods"],
                    key=lambda m: comparison["confidence_scores"].get(m, 0)
                )
                comparison["recommendation"] = f"Best solution from {best_method}"
            else:
                comparison["recommendation"] = "All methods failed"
            
            return comparison
            
        except Exception as e:
            logger.error(f"Error comparing solutions: {e}")
            return {"error": str(e)}
    
    def setup_knowledge_base(self, data_directory: str = "data/knowledge_base"):
        """Setup the knowledge base with geometric data"""
        try:
            console.print(Panel("Setting up knowledge base...", title="Knowledge Base Setup"))
            
            # Load documents
            loader = DirectoryLoader(
                data_directory,
                glob="**/*.txt",
                loader_cls=TextLoader,
                loader_kwargs={"encoding": "utf-8"}
            )
            documents = loader.load()
            
            if not documents:
                console.print("No documents found. Creating sample knowledge base...")
                self._create_sample_knowledge_base()
                return
            
            # Split documents
            texts = self.text_splitter.split_documents(documents)
            
            # Add to vector store
            if self.config['rag']['vector_store']['type'] == 'chromadb':
                self.vector_store.add_documents(texts)
            elif self.config['rag']['vector_store']['type'] == 'faiss':
                self.vector_store = FAISS.from_documents(texts, self.embeddings)
            
            console.print(f"✅ Knowledge base setup complete with {len(texts)} chunks")
            
        except Exception as e:
            logger.error(f"Error setting up knowledge base: {e}")
            console.print(f"❌ Error setting up knowledge base: {e}")
    
    def _create_sample_knowledge_base(self):
        """Create a sample knowledge base with geometric theorems"""
        sample_data = [
            "قضیه فیثاغورث: در مثلث قائم‌الزاویه، مربع وتر برابر است با مجموع مربعات دو ضلع دیگر.",
            "قضیه تالس: اگر خطی موازی با یکی از اضلاع مثلث رسم شود، مثلث‌های مشابه ایجاد می‌کند.",
            "قضیه دایره: زاویه مرکزی دو برابر زاویه محیطی است که روی همان کمان قرار دارد.",
            "قضیه مماس: خط مماس بر دایره عمود بر شعاع در نقطه تماس است.",
            "قضیه وتر: اگر دو وتر در دایره همدیگر را قطع کنند، حاصل ضرب قطعات یکی برابر حاصل ضرب قطعات دیگری است."
        ]
        
        documents = [Document(page_content=text, metadata={"source": "sample"}) for text in sample_data]
        texts = self.text_splitter.split_documents(documents)
        
        if self.config['rag']['vector_store']['type'] == 'chromadb':
            self.vector_store.add_documents(texts)
        elif self.config['rag']['vector_store']['type'] == 'faiss':
            self.vector_store = FAISS.from_documents(texts, self.embeddings)
        
        console.print("✅ Sample knowledge base created")
    
    def display_results(self, results: Dict[str, Any]):
        """Display results in a beautiful format"""
        console.print("\n" + "="*80)
        console.print(Panel("نتایج حل مسئله هندسی", title="Geometric Problem Solutions", style="bold blue"))
        
        for method, result in results.items():
            if method == "comparison":
                continue
                
            if "error" in result:
                console.print(f"\n❌ {method.upper()}: {result['error']}")
            else:
                console.print(f"\n✅ {method.upper()}:")
                console.print(f"   روش: {result.get('method', 'N/A')}")
                console.print(f"   اعتماد: {result.get('confidence', 0):.2f}")
                console.print(f"   زمان: {result.get('timestamp', 'N/A')}")
                
                solution = result.get('solution', '')
                if solution:
                    console.print(Panel(solution, title="راه‌حل", style="green"))
        
        # Display comparison
        if "comparison" in results:
            comparison = results["comparison"]
            console.print("\n" + "="*50)
            console.print(Panel("مقایسه نتایج", title="Results Comparison", style="bold yellow"))
            
            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 in comparison["methods_used"]:
                if method in comparison["successful_methods"]:
                    status = "✅ موفق"
                    confidence = f"{comparison['confidence_scores'].get(method, 0):.2f}"
                    length = f"{comparison['solution_lengths'].get(method, 0)}"
                else:
                    status = "❌ ناموفق"
                    confidence = "N/A"
                    length = "N/A"
                
                table.add_row(method, status, confidence, length)
            
            console.print(table)
            console.print(f"\n💡 توصیه: {comparison['recommendation']}")
    
    def save_results(self, results: Dict[str, Any], filename: str = None):
        """Save results to file"""
        if filename is None:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"results/geometry_solution_{timestamp}.json"
        
        os.makedirs(os.path.dirname(filename), exist_ok=True)
        
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(results, f, ensure_ascii=False, indent=2)
        
        console.print(f"✅ Results saved to {filename}")

async def main():
    """Main function to run the geometric reasoning system"""
    console.print(Panel(
        "سیستم پیشرفته استدلال هندسی\nAdvanced Geometric Reasoning System",
        title="CA5: Geometric Reasoning",
        style="bold blue"
    ))
    
    try:
        # Initialize solver
        solver = AdvancedGeometrySolver()
        
        # Setup knowledge base
        solver.setup_knowledge_base()
        
        # Example problems
        example_problems = [
            "در مثلث ABC، زاویه A = 60 درجه، AB = 5، AC = 7. طول BC را پیدا کنید.",
            "دایره‌ای با شعاع 5 رسم شده است. اگر وتری به طول 8 رسم شود، فاصله آن از مرکز دایره چقدر است؟",
            "در مثلث متساوی‌الاضلاع ABC، نقطه D روی ضلع BC قرار دارد. اگر AD = 6 و BD = 2، طول DC را پیدا کنید."
        ]
        
        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]
        
        # Solve problem
        console.print(f"\n🔍 حل مسئله: {user_problem}")
        results = await solver.solve_problem(user_problem)
        
        # Display results
        solver.display_results(results)
        
        # Save results
        solver.save_results(results)
        
        console.print("\n✅ حل مسئله تکمیل شد!")
        
    except KeyboardInterrupt:
        console.print("\n⏹️ عملیات توسط کاربر متوقف شد.")
    except Exception as e:
        console.print(f"\n❌ خطا: {e}")
        logger.error(f"Main execution error: {e}")

if __name__ == "__main__":
    # Create necessary directories
    os.makedirs("logs", exist_ok=True)
    os.makedirs("results", exist_ok=True)
    os.makedirs("data/vector_db", exist_ok=True)
    os.makedirs("data/knowledge_base", exist_ok=True)
    
    # Run main function
    asyncio.run(main())
Xet Storage Details

Size:: 30.1 kB
Xet hash:: b3093281c6fa8e062c2a4265f985d154e6a0610d80b1e50ab0fd3d61501f925a
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.