Update core/vectorstore/vectorstore_manager.py

core/vectorstore/vectorstore_manager.py

@@ -1,136 +1,136 @@
-# core/vectorstore/vectorstore_manager.py
-import os
-import faiss
-from langchain.text_splitter import RecursiveCharacterTextSplitter
-from langchain_community.docstore.in_memory import InMemoryDocstore
-from langchain_community.vectorstores import FAISS as FAISS_STORE
-from vectorstore.document_processor import DocumentProcessor
-from vectorstore.embeddings import EmbeddingManager
-from vectorstore.distance_strategy import DistanceStrategyManager
-from loguru import logger
-
-
-class VectorStoreManager:
-    """
-    Gestión minimalista de FAISS para EDULLM:
-    - Indexa documentos
-    - Carga/guarda el índice
-    - Expone retriever para RAG
-    """
-
-    def __init__(self, path: str, name: str):
-        self.path = path
-        self.store_path = os.path.join("database", name)
-        self.embeddings = EmbeddingManager.get_embeddings()
-        self.strategy = DistanceStrategyManager().strategy
-        self.vectorstore = None
-        logger.info(f"🔹 Inicializando VectorStoreManager en ruta: {self.store_path}")
-        self._initialize()
-
-    def _initialize(self):
-        if self.exist_vectorstore():
-            logger.info("✅ Índice FAISS encontrado. Cargando desde disco...")
-            self.vectorstore = self.load_vectorstore()
-        else:
-            logger.warning("⚠️ No existe índice previo. Creando índice vacío...")
-            dummy = self.embeddings.embed_query("init")
-            index = faiss.IndexFlatL2(len(dummy))
-            self.vectorstore = FAISS_STORE(
-                embedding_function=self.embeddings,
-                index=index,
-                docstore=InMemoryDocstore(),
-                index_to_docstore_id={},
-                distance_strategy=self.strategy,
-            )
-
-    def create_vectorstore(self) -> None:
-        logger.info(f"🚀 Procesando documentos en '{self.path}' para indexar...")
-        docs = DocumentProcessor(self.path).files_to_texts()
-        splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=400)
-        chunks = splitter.split_documents(docs)
-        self.vectorstore.add_documents(chunks)
-        self.save_vectorstore()
-        logger.success("🎯 Vectorstore creado y guardado correctamente.")
-
-    def save_vectorstore(self) -> None:
-        try:
-            os.makedirs(self.store_path, exist_ok=True)
-            self.vectorstore.save_local(self.store_path)
-            logger.info(f"💾 Índice guardado en '{self.store_path}'.")
-        except Exception as e:
-            logger.error(f"❌ Error al guardar el vectorstore: {e}")
-
-    def load_vectorstore(self):
-        try:
-            logger.info(f"📂 Cargando vectorstore desde '{self.store_path}'.")
-            return FAISS_STORE.load_local(
-                folder_path=self.store_path,
-                embeddings=self.embeddings,
-                allow_dangerous_deserialization=True,
-                distance_strategy=self.strategy,
-            )
-        except Exception as e:
-            logger.error(f"❌ Error al cargar el vectorstore: {e}")
-            raise
-
-    def exist_vectorstore(self) -> bool:
-        """Verifica si el vectorstore existe, creando la carpeta base si es necesario."""
-        base_dir = "database"
-
-        if not os.path.isdir(base_dir):
-            logger.warning(f"📂 Directorio base '{base_dir}' no encontrado. Creando...")
-            os.makedirs(base_dir, exist_ok=True)
-            return False
-
-        if os.path.isdir(self.store_path):
-            logger.info(f"✅ Vectorstore encontrado en '{self.store_path}'.")
-            return True
-        else:
-            logger.info(f"ℹ️ Vectorstore no existe aún en '{self.store_path}'.")
-            return False
-
-    def as_retriever(
-        self,
-        search_type: str = "similarity_score_threshold",
-        k: int = 4,
-        score_threshold: float = 0.75,
-        fallback_to_similarity: bool = True,
-        **kwargs,
-    ):
-        if not self.vectorstore:
-            self.vectorstore = self.load_vectorstore()
-
-        logger.debug(
-            f"🔍 Configurando retriever: type={search_type}, k={k}, threshold={score_threshold}"
-        )
-        search_kwargs = {"k": k, "score_threshold": score_threshold}
-        retriever = self.vectorstore.as_retriever(
-            search_type=search_type, search_kwargs=search_kwargs
-        )
-
-        if fallback_to_similarity:
-            logger.info(
-                "🛡️ Fallback activado: Si no hay resultados, se usará búsqueda por similarity."
-            )
-
-            class SafeRetriever:
-                def __init__(self, primary, fallback):
-                    self.primary = primary
-                    self.fallback = fallback
-
-                def invoke(self, query):
-                    docs = self.primary.invoke(query)
-                    if not docs:
-                        logger.warning(
-                            "⚠️ Sin resultados en threshold. Aplicando fallback a similarity."
-                        )
-                        return self.fallback.invoke(query)
-                    return docs
-
-            fallback_retriever = self.vectorstore.as_retriever(
-                search_type="similarity", search_kwargs={"k": k}
-            )
-
-            return SafeRetriever(retriever, fallback_retriever)
-
-        return retriever
+# core/vectorstore/vectorstore_manager.py
+import os
+import faiss
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain_community.docstore.in_memory import InMemoryDocstore
+from langchain_community.vectorstores import FAISS as FAISS_STORE
+from core.vectorstore.document_processor import DocumentProcessor
+from core.vectorstore.embeddings import EmbeddingManager
+from core.vectorstore.distance_strategy import DistanceStrategyManager
+from loguru import logger
+
+
+class VectorStoreManager:
+    """
+    Gestión minimalista de FAISS para EDULLM:
+    - Indexa documentos
+    - Carga/guarda el índice
+    - Expone retriever para RAG
+    """
+
+    def __init__(self, path: str, name: str):
+        self.path = path
+        self.store_path = os.path.join("database", name)
+        self.embeddings = EmbeddingManager.get_embeddings()
+        self.strategy = DistanceStrategyManager().strategy
+        self.vectorstore = None
+        logger.info(f"🔹 Inicializando VectorStoreManager en ruta: {self.store_path}")
+        self._initialize()
+
+    def _initialize(self):
+        if self.exist_vectorstore():
+            logger.info("✅ Índice FAISS encontrado. Cargando desde disco...")
+            self.vectorstore = self.load_vectorstore()
+        else:
+            logger.warning("⚠️ No existe índice previo. Creando índice vacío...")
+            dummy = self.embeddings.embed_query("init")
+            index = faiss.IndexFlatL2(len(dummy))
+            self.vectorstore = FAISS_STORE(
+                embedding_function=self.embeddings,
+                index=index,
+                docstore=InMemoryDocstore(),
+                index_to_docstore_id={},
+                distance_strategy=self.strategy,
+            )
+
+    def create_vectorstore(self) -> None:
+        logger.info(f"🚀 Procesando documentos en '{self.path}' para indexar...")
+        docs = DocumentProcessor(self.path).files_to_texts()
+        splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=400)
+        chunks = splitter.split_documents(docs)
+        self.vectorstore.add_documents(chunks)
+        self.save_vectorstore()
+        logger.success("🎯 Vectorstore creado y guardado correctamente.")
+
+    def save_vectorstore(self) -> None:
+        try:
+            os.makedirs(self.store_path, exist_ok=True)
+            self.vectorstore.save_local(self.store_path)
+            logger.info(f"💾 Índice guardado en '{self.store_path}'.")
+        except Exception as e:
+            logger.error(f"❌ Error al guardar el vectorstore: {e}")
+
+    def load_vectorstore(self):
+        try:
+            logger.info(f"📂 Cargando vectorstore desde '{self.store_path}'.")
+            return FAISS_STORE.load_local(
+                folder_path=self.store_path,
+                embeddings=self.embeddings,
+                allow_dangerous_deserialization=True,
+                distance_strategy=self.strategy,
+            )
+        except Exception as e:
+            logger.error(f"❌ Error al cargar el vectorstore: {e}")
+            raise
+
+    def exist_vectorstore(self) -> bool:
+        """Verifica si el vectorstore existe, creando la carpeta base si es necesario."""
+        base_dir = "database"
+
+        if not os.path.isdir(base_dir):
+            logger.warning(f"📂 Directorio base '{base_dir}' no encontrado. Creando...")
+            os.makedirs(base_dir, exist_ok=True)
+            return False
+
+        if os.path.isdir(self.store_path):
+            logger.info(f"✅ Vectorstore encontrado en '{self.store_path}'.")
+            return True
+        else:
+            logger.info(f"ℹ️ Vectorstore no existe aún en '{self.store_path}'.")
+            return False
+
+    def as_retriever(
+        self,
+        search_type: str = "similarity_score_threshold",
+        k: int = 4,
+        score_threshold: float = 0.75,
+        fallback_to_similarity: bool = True,
+        **kwargs,
+    ):
+        if not self.vectorstore:
+            self.vectorstore = self.load_vectorstore()
+
+        logger.debug(
+            f"🔍 Configurando retriever: type={search_type}, k={k}, threshold={score_threshold}"
+        )
+        search_kwargs = {"k": k, "score_threshold": score_threshold}
+        retriever = self.vectorstore.as_retriever(
+            search_type=search_type, search_kwargs=search_kwargs
+        )
+
+        if fallback_to_similarity:
+            logger.info(
+                "🛡️ Fallback activado: Si no hay resultados, se usará búsqueda por similarity."
+            )
+
+            class SafeRetriever:
+                def __init__(self, primary, fallback):
+                    self.primary = primary
+                    self.fallback = fallback
+
+                def invoke(self, query):
+                    docs = self.primary.invoke(query)
+                    if not docs:
+                        logger.warning(
+                            "⚠️ Sin resultados en threshold. Aplicando fallback a similarity."
+                        )
+                        return self.fallback.invoke(query)
+                    return docs
+
+            fallback_retriever = self.vectorstore.as_retriever(
+                search_type="similarity", search_kwargs={"k": k}
+            )
+
+            return SafeRetriever(retriever, fallback_retriever)
+
+        return retriever