Spaces:
Running
Running
| # api/database.py | |
| import numpy as np | |
| from sentence_transformers import SentenceTransformer | |
| from .config import MODEL_CACHE_DIR, EMBEDDING_MODEL_DEVICE | |
| import logging | |
| logger = logging.getLogger("database-bot") | |
| # Root Cause vs Logic: the runtime used to initialize a MongoDB/FAISS RAG data plane at import time, | |
| # which caused DNS failures when the index cluster was unavailable; the app now keeps only the embedder hook. | |
| class DatabaseManager: | |
| def __init__(self): | |
| self.embedding_model = None | |
| self.index = None | |
| self.symptom_vectors = None | |
| self.symptom_docs = None | |
| def initialize_embedding_model(self): | |
| """Initialize the SentenceTransformer model""" | |
| logger.info("[Embedder] 📥 Loading SentenceTransformer Model...") | |
| try: | |
| self.embedding_model = SentenceTransformer(MODEL_CACHE_DIR, device=EMBEDDING_MODEL_DEVICE) | |
| self.embedding_model = self.embedding_model.half() # Reduce memory | |
| logger.info("✅ Model Loaded Successfully.") | |
| except Exception as e: | |
| logger.error(f"❌ Model Loading Failed: {e}") | |
| raise | |
| def initialize_mongodb(self): | |
| """Compatibility stub for legacy MongoDB-backed retrieval startup.""" | |
| # Root Cause vs Logic: all MongoDB-backed RAG wiring has been removed from the runtime; | |
| # keeping this as a no-op preserves import compatibility without triggering DNS lookups. | |
| logger.info("[DB] MongoDB initialization skipped (legacy RAG disabled).") | |
| def load_faiss_index(self): | |
| """Legacy FAISS loading disabled.""" | |
| return None | |
| def load_symptom_vectors(self): | |
| """Legacy symptom vector loading disabled.""" | |
| self.symptom_docs = [] | |
| self.symptom_vectors = np.empty((0, 0), dtype=np.float32) | |
| def get_embedding_model(self): | |
| """Get the embedding model""" | |
| if self.embedding_model is None: | |
| self.initialize_embedding_model() | |
| return self.embedding_model | |
| def get_qa_collection(self): | |
| """Legacy QA collection access disabled.""" | |
| return None | |
| def get_symptom_collection(self): | |
| """Legacy symptom collection access disabled.""" | |
| return None | |
| # Global database manager instance | |
| db_manager = DatabaseManager() | |