import hashlib from typing import List from pathlib import Path from langchain_community.document_loaders import DirectoryLoader, TextLoader, PyPDFLoader, BSHTMLLoader from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_core.documents import Document from config import * class DocumentProcessor: """Procesador de documentos para el sistema RAG.""" def __init__(self, docs_path: str = "docs", chroma_path: str = "./chroma_db"): self.docs_path = Path(docs_path) self.chroma_path = Path(chroma_path) self.embeddings = OpenAIEmbeddings(model=EMBEDDINGS_MODEL) self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len, separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""] ) def load_documents(self) -> List[Document]: """Carga documentos markdown, texto y PDF del directorio docs.""" print(f" Cargando documentos desde {self.docs_path}") documents = [] # Cargar archivos .md loader_md = DirectoryLoader( str(self.docs_path), glob="*.md", loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"} ) documents.extend(loader_md.load()) # Cargar archivos .txt loader_txt = DirectoryLoader( str(self.docs_path), glob="*.txt", loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"} ) documents.extend(loader_txt.load()) # Cargar archivos .pdf loader_pdf = DirectoryLoader( str(self.docs_path), glob="*.pdf", loader_cls=PyPDFLoader ) documents.extend(loader_pdf.load()) # Cargar archivos .html loader_html = DirectoryLoader( str(self.docs_path), glob="*.html", loader_cls=BSHTMLLoader, loader_kwargs={ "open_encoding": "utf-8", "bs_kwargs": {"features": "html.parser"}, "get_text_separator": "\n", }, ) documents.extend(loader_html.load()) # Enriquecer metadatos for doc in documents: filename = Path(doc.metadata["source"]).stem doc.metadata.update({ "filename": filename, "doc_type": self._get_doc_type(filename), "doc_id": self._generate_doc_id(doc.page_content) }) print(f"✅ Cargados {len(documents)} documentos") return documents def _chunk_id(self, chunk: Document) -> str: """ID estable: mismo archivo + misma página + mismo texto = mismo ID.""" base = f"{chunk.metadata.get('source')}|{chunk.metadata.get('page', 0)}|{chunk.page_content}" return hashlib.md5(base.encode()).hexdigest() def _get_doc_type(self, filename: str) -> str: """Determina el tipo de documento basado en el nombre.""" if "faq" in filename.lower(): return "faq" elif "manual" in filename.lower(): return "manual" elif "troubleshooting" in filename.lower(): return "troubleshooting" else: return "general" def _generate_doc_id(self, content: str) -> str: """Genera un ID único para el documento.""" return hashlib.md5(content.encode()).hexdigest()[:8] def split_documents(self, documents: List[Document]) -> List[Document]: """Divide documentos en chunks más pequeños.""" print("✂️ Dividiendo documentos en chunks...") chunks = self.text_splitter.split_documents(documents) # Agregar metadatos de chunk for i, chunk in enumerate(chunks): chunk.metadata.update({ "chunk_id": i, "chunk_size": len(chunk.page_content) }) print(f"✅ Creados {len(chunks)} chunks") return chunks def create_vectorstore(self, chunks: List[Document]): """Abre la base existente y agrega solo los chunks nuevos.""" print("🔄 Actualizando vectorstore con ChromaDB...") vectorstore = Chroma( persist_directory=str(self.chroma_path), embedding_function=self.embeddings, collection_name="helpdesk_knowledge", ) # IDs que ya están guardados existentes = set(vectorstore.get()["ids"]) # Quedarse solo con los chunks que aún no están nuevos, nuevos_ids = [], [] vistos = set() for chunk in chunks: cid = self._chunk_id(chunk) if cid not in existentes and cid not in vistos: vistos.add(cid) nuevos.append(chunk) nuevos_ids.append(cid) if nuevos: vectorstore.add_documents(nuevos, ids=nuevos_ids) print(f"➕ Agregados {len(nuevos)} chunks nuevos") else: print("✅ No hay chunks nuevos, base actualizada") return vectorstore def load_existing_vectorstore(self) -> Chroma: """Carga vectorstore existente.""" if not self.chroma_path.exists(): raise FileNotFoundError(f"Vectorstore no encontrado en {self.chroma_path}") return Chroma( persist_directory=str(self.chroma_path), embedding_function=self.embeddings, collection_name="helpdesk_knowledge", ) def setup_rag_system(self, force_rebuild: bool = False): """Configura el sistema RAG completo.""" print("🚀 Configurando sistema RAG...") # Verificar si ya existe y no forzar rebuild if self.chroma_path.exists() and not force_rebuild: print("📦 Vectorstore existente encontrado") return self.load_existing_vectorstore() # Cargar y procesar documentos documents = self.load_documents() if not documents: print("⚠️ No se encontraron documentos para procesar") return None # Dividir documentos chunks = self.split_documents(documents) # Crear vectorstore vectorstore = self.create_vectorstore(chunks) print("✅ Sistema RAG configurado exitosamente") return vectorstore def test_search(self, vectorstore: Chroma, query: str = "resetear contraseña"): """Prueba la funcionalidad de búsqueda.""" print(f"\n🔍 Probando búsqueda: '{query}'") results = vectorstore.similarity_search(query, k=3) for i, doc in enumerate(results, 1): print(f"\n📄 Resultado {i}:") print(f"Tipo: {doc.metadata.get('doc_type', 'unknown')}") print(f"Archivo: {doc.metadata.get('filename', 'unknown')}") print(f"Contenido: {doc.page_content[:200]}...") return results def main(): """Función principal para configurar RAG.""" print("Configuración RAG - Helpdesk Erick Henk") print("=" * 40) # Configurar procesador processor = DocumentProcessor(docs_path=DOCS_PATH, chroma_path=CHROMADB_PATH) # Configurar sistema RAG vectorstore = processor.setup_rag_system(force_rebuild=True) if vectorstore: # Probar búsquedas test_queries = [ "¿por qué elegir CEI?", "¿Qué áreas de formación ofrece la escuela?", "¿Las formaciones son presenciales, online o híbridas?", "¿Qué tipos de formación ofrece?" ] for query in test_queries: processor.test_search(vectorstore, query) print("\n✅ Configuración completada") if __name__ == "__main__": main()