Buckets:
| """ | |
| Knowledge Base Manager for Geometric Reasoning | |
| مدیر پایگاه دانش برای استدلال هندسی | |
| This module handles the creation, management, and querying of the knowledge base | |
| for geometric reasoning problems using RAG (Retrieval-Augmented Generation). | |
| """ | |
| import os | |
| import json | |
| import logging | |
| from typing import List, Dict, Any, Optional | |
| from pathlib import Path | |
| 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.retrievers import VectorStoreRetriever | |
| from sentence_transformers import SentenceTransformer | |
| logger = logging.getLogger(__name__) | |
| class KnowledgeBaseManager: | |
| """ | |
| Manages the knowledge base for geometric reasoning | |
| مدیریت پایگاه دانش برای استدلال هندسی | |
| """ | |
| def __init__(self, config: Dict[str, Any], embeddings, vector_store): | |
| """ | |
| Initialize the knowledge base manager | |
| Args: | |
| config: Configuration dictionary | |
| embeddings: Embedding model | |
| vector_store: Vector store instance | |
| """ | |
| self.config = config | |
| self.embeddings = embeddings | |
| self.vector_store = vector_store | |
| # Initialize text splitter | |
| splitter_config = config.get("document_processing", {}).get("text_splitter", {}) | |
| self.text_splitter = RecursiveCharacterTextSplitter( | |
| chunk_size=splitter_config.get("chunk_size", 1000), | |
| chunk_overlap=splitter_config.get("chunk_overlap", 200), | |
| length_function=len, | |
| separators=splitter_config.get("separators", ["\n\n", "\n", " ", ""]), | |
| ) | |
| # Knowledge base sources | |
| self.sources = config.get("sources", []) | |
| logger.info("Knowledge Base Manager initialized") | |
| def load_documents(self, source_path: str) -> List[Document]: | |
| """ | |
| Load documents from a source path | |
| Args: | |
| source_path: Path to the source directory or file | |
| Returns: | |
| List of loaded documents | |
| """ | |
| try: | |
| documents = [] | |
| if os.path.isfile(source_path): | |
| # Load single file | |
| if source_path.endswith(".pdf"): | |
| loader = PyPDFLoader(source_path) | |
| else: | |
| loader = TextLoader(source_path, encoding="utf-8") | |
| documents = loader.load() | |
| elif os.path.isdir(source_path): | |
| # Load directory | |
| loader = DirectoryLoader( | |
| source_path, | |
| glob="**/*.txt", | |
| loader_cls=TextLoader, | |
| loader_kwargs={"encoding": "utf-8"}, | |
| ) | |
| documents = loader.load() | |
| logger.info(f"Loaded {len(documents)} documents from {source_path}") | |
| return documents | |
| except Exception as e: | |
| logger.error(f"Error loading documents from {source_path}: {e}") | |
| return [] | |
| def process_documents(self, documents: List[Document]) -> List[Document]: | |
| """ | |
| Process documents by splitting them into chunks | |
| Args: | |
| documents: List of documents to process | |
| Returns: | |
| List of processed document chunks | |
| """ | |
| try: | |
| # Split documents into chunks | |
| chunks = self.text_splitter.split_documents(documents) | |
| # Add metadata | |
| for i, chunk in enumerate(chunks): | |
| chunk.metadata.update( | |
| { | |
| "chunk_id": i, | |
| "processed_at": str(pd.Timestamp.now()), | |
| "source_type": "geometric_knowledge", | |
| } | |
| ) | |
| logger.info( | |
| f"Processed {len(documents)} documents into {len(chunks)} chunks" | |
| ) | |
| return chunks | |
| except Exception as e: | |
| logger.error(f"Error processing documents: {e}") | |
| return [] | |
| def add_to_vector_store(self, chunks: List[Document]): | |
| """ | |
| Add document chunks to the vector store | |
| Args: | |
| chunks: List of document chunks to add | |
| """ | |
| try: | |
| if isinstance(self.vector_store, Chroma): | |
| self.vector_store.add_documents(chunks) | |
| elif isinstance(self.vector_store, FAISS): | |
| # FAISS needs to be rebuilt when adding new documents | |
| existing_docs = self.vector_store.docstore._dict | |
| all_docs = list(existing_docs.values()) + chunks | |
| self.vector_store = FAISS.from_documents(all_docs, self.embeddings) | |
| logger.info(f"Added {len(chunks)} chunks to vector store") | |
| except Exception as e: | |
| logger.error(f"Error adding chunks to vector store: {e}") | |
| def build_knowledge_base(self): | |
| """ | |
| Build the complete knowledge base from all sources | |
| """ | |
| try: | |
| all_chunks = [] | |
| for source in self.sources: | |
| source_path = source["path"] | |
| source_type = source.get("type", "json") | |
| if source_type == "json": | |
| chunks = self._load_json_source(source_path) | |
| else: | |
| documents = self.load_documents(source_path) | |
| chunks = self.process_documents(documents) | |
| all_chunks.extend(chunks) | |
| if all_chunks: | |
| self.add_to_vector_store(all_chunks) | |
| logger.info(f"Knowledge base built with {len(all_chunks)} total chunks") | |
| else: | |
| logger.warning("No chunks found to build knowledge base") | |
| except Exception as e: | |
| logger.error(f"Error building knowledge base: {e}") | |
| def _load_json_source(self, source_path: str) -> List[Document]: | |
| """ | |
| Load documents from a JSON source | |
| Args: | |
| source_path: Path to JSON file | |
| Returns: | |
| List of documents | |
| """ | |
| try: | |
| with open(source_path, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| documents = [] | |
| if isinstance(data, list): | |
| for item in data: | |
| if isinstance(item, dict): | |
| content = item.get("content", "") | |
| metadata = item.get("metadata", {}) | |
| documents.append( | |
| Document(page_content=content, metadata=metadata) | |
| ) | |
| elif isinstance(data, dict): | |
| for key, value in data.items(): | |
| if isinstance(value, dict): | |
| content = value.get("content", str(value)) | |
| metadata = value.get("metadata", {"source": key}) | |
| documents.append( | |
| Document(page_content=content, metadata=metadata) | |
| ) | |
| return documents | |
| except Exception as e: | |
| logger.error(f"Error loading JSON source {source_path}: {e}") | |
| return [] | |
| def create_retriever( | |
| self, search_type: str = "similarity", k: int = 5 | |
| ) -> VectorStoreRetriever: | |
| """ | |
| Create a retriever from the vector store | |
| Args: | |
| search_type: Type of search ("similarity", "mmr", "similarity_score_threshold") | |
| k: Number of documents to retrieve | |
| Returns: | |
| Vector store retriever | |
| """ | |
| try: | |
| retriever = self.vector_store.as_retriever( | |
| search_type=search_type, search_kwargs={"k": k} | |
| ) | |
| logger.info(f"Created retriever with search_type={search_type}, k={k}") | |
| return retriever | |
| except Exception as e: | |
| logger.error(f"Error creating retriever: {e}") | |
| raise | |
| def query_knowledge_base(self, query: str, k: int = 5) -> List[Document]: | |
| """ | |
| Query the knowledge base | |
| Args: | |
| query: Query string | |
| k: Number of documents to retrieve | |
| Returns: | |
| List of relevant documents | |
| """ | |
| try: | |
| retriever = self.create_retriever(k=k) | |
| documents = retriever.get_relevant_documents(query) | |
| logger.info( | |
| f"Retrieved {len(documents)} documents for query: {query[:50]}..." | |
| ) | |
| return documents | |
| except Exception as e: | |
| logger.error(f"Error querying knowledge base: {e}") | |
| return [] | |
| def get_similar_documents(self, document: Document, k: int = 5) -> List[Document]: | |
| """ | |
| Get similar documents to a given document | |
| Args: | |
| document: Reference document | |
| k: Number of similar documents to retrieve | |
| Returns: | |
| List of similar documents | |
| """ | |
| try: | |
| similar_docs = self.vector_store.similarity_search( | |
| document.page_content, k=k | |
| ) | |
| logger.info(f"Found {len(similar_docs)} similar documents") | |
| return similar_docs | |
| except Exception as e: | |
| logger.error(f"Error finding similar documents: {e}") | |
| return [] | |
| def update_knowledge_base(self, new_documents: List[Document]): | |
| """ | |
| Update the knowledge base with new documents | |
| Args: | |
| new_documents: List of new documents to add | |
| """ | |
| try: | |
| chunks = self.process_documents(new_documents) | |
| self.add_to_vector_store(chunks) | |
| logger.info(f"Updated knowledge base with {len(chunks)} new chunks") | |
| except Exception as e: | |
| logger.error(f"Error updating knowledge base: {e}") | |
| def get_knowledge_base_stats(self) -> Dict[str, Any]: | |
| """ | |
| Get statistics about the knowledge base | |
| Returns: | |
| Dictionary containing knowledge base statistics | |
| """ | |
| try: | |
| stats = { | |
| "total_documents": 0, | |
| "total_chunks": 0, | |
| "sources": len(self.sources), | |
| "vector_store_type": type(self.vector_store).__name__, | |
| } | |
| if isinstance(self.vector_store, Chroma): | |
| collection = self.vector_store._collection | |
| stats["total_documents"] = collection.count() | |
| stats["total_chunks"] = collection.count() | |
| elif isinstance(self.vector_store, FAISS): | |
| stats["total_documents"] = len(self.vector_store.docstore._dict) | |
| stats["total_chunks"] = len(self.vector_store.docstore._dict) | |
| return stats | |
| except Exception as e: | |
| logger.error(f"Error getting knowledge base stats: {e}") | |
| return {} | |
| def create_sample_knowledge_base(self): | |
| """ | |
| Create a sample knowledge base with geometric theorems and concepts | |
| """ | |
| sample_data = [ | |
| { | |
| "content": "قضیه فیثاغورث: در مثلث قائمالزاویه، مربع وتر برابر است با مجموع مربعات دو ضلع دیگر. یعنی اگر مثلث ABC قائمالزاویه باشد و زاویه C = 90 درجه، آنگاه AB² = AC² + BC².", | |
| "metadata": { | |
| "theorem": "فیثاغورث", | |
| "category": "مثلث", | |
| "difficulty": "پایه", | |
| "language": "persian", | |
| }, | |
| }, | |
| { | |
| "content": "قضیه تالس: اگر خطی موازی با یکی از اضلاع مثلث رسم شود، مثلثهای مشابه ایجاد میکند. این قضیه برای حل مسائل تناسب و مشابهت بسیار مفید است.", | |
| "metadata": { | |
| "theorem": "تالس", | |
| "category": "مثلث", | |
| "difficulty": "متوسط", | |
| "language": "persian", | |
| }, | |
| }, | |
| { | |
| "content": "قضیه دایره: زاویه مرکزی دو برابر زاویه محیطی است که روی همان کمان قرار دارد. این قضیه برای حل مسائل مربوط به دایره و زاویهها استفاده میشود.", | |
| "metadata": { | |
| "theorem": "زاویه مرکزی و محیطی", | |
| "category": "دایره", | |
| "difficulty": "متوسط", | |
| "language": "persian", | |
| }, | |
| }, | |
| { | |
| "content": "قضیه مماس: خط مماس بر دایره عمود بر شعاع در نقطه تماس است. این قضیه برای حل مسائل مربوط به مماس و دایره استفاده میشود.", | |
| "metadata": { | |
| "theorem": "مماس", | |
| "category": "دایره", | |
| "difficulty": "پایه", | |
| "language": "persian", | |
| }, | |
| }, | |
| { | |
| "content": "قضیه وتر: اگر دو وتر در دایره همدیگر را قطع کنند، حاصل ضرب قطعات یکی برابر حاصل ضرب قطعات دیگری است. یعنی اگر وترهای AB و CD در نقطه E همدیگر را قطع کنند، آنگاه AE × EB = CE × ED.", | |
| "metadata": { | |
| "theorem": "وتر", | |
| "category": "دایره", | |
| "difficulty": "متوسط", | |
| "language": "persian", | |
| }, | |
| }, | |
| { | |
| "content": "قضیه سینوس: در هر مثلث، نسبت هر ضلع به سینوس زاویه مقابل آن برابر است با قطر دایره محیطی. یعنی a/sin(A) = b/sin(B) = c/sin(C) = 2R.", | |
| "metadata": { | |
| "theorem": "سینوس", | |
| "category": "مثلث", | |
| "difficulty": "پیشرفته", | |
| "language": "persian", | |
| }, | |
| }, | |
| { | |
| "content": "قضیه کسینوس: در هر مثلث، مربع هر ضلع برابر است با مجموع مربعات دو ضلع دیگر منهای دو برابر حاصل ضرب آنها در کسینوس زاویه بینشان. یعنی a² = b² + c² - 2bc cos(A).", | |
| "metadata": { | |
| "theorem": "کسینوس", | |
| "category": "مثلث", | |
| "difficulty": "پیشرفته", | |
| "language": "persian", | |
| }, | |
| }, | |
| { | |
| "content": "قضیه هرون: مساحت مثلث با اضلاع a، b، c برابر است با √(s(s-a)(s-b)(s-c)) که در آن s = (a+b+c)/2 نیممحیط مثلث است.", | |
| "metadata": { | |
| "theorem": "هرون", | |
| "category": "مثلث", | |
| "difficulty": "متوسط", | |
| "language": "persian", | |
| }, | |
| }, | |
| ] | |
| # Convert to documents | |
| documents = [] | |
| for item in sample_data: | |
| doc = Document(page_content=item["content"], metadata=item["metadata"]) | |
| documents.append(doc) | |
| # Process and add to vector store | |
| chunks = self.process_documents(documents) | |
| self.add_to_vector_store(chunks) | |
| logger.info(f"Created sample knowledge base with {len(chunks)} chunks") | |
| return chunks | |
Xet Storage Details
- Size:
- 16 kB
- Xet hash:
- cce4c67b192ea49d134890cb06b2075e90772ced87283a552c6a9cfee7128c33
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.