import os import logging import hashlib from typing import List, Dict, Any, Optional from dataclasses import dataclass try: from qdrant_client import QdrantClient from qdrant_client.http import models QDRANT_AVAILABLE = True except ImportError: QdrantClient = None QDRANT_AVAILABLE = False # Try fastembed for local embeddings (no API costs) try: from fastembed import TextEmbedding FASTEMBED_AVAILABLE = True except ImportError: TextEmbedding = None FASTEMBED_AVAILABLE = False # Import secrets loader for secure credential management try: from secrets_loader import get_qdrant_config, QdrantConfig SECRETS_AVAILABLE = True except ImportError: SECRETS_AVAILABLE = False get_qdrant_config = None QdrantConfig = None class VectorDocument: def __init__(self, id, content, score, metadata): self.id = id self.content = content self.score = score self.metadata = metadata class VectorManager: """ Manages connections to Vector Databases (Qdrant). Credentials loaded from environment or secrets.env file. See: secrets_loader.py for configuration. """ def __init__(self, qdrant_config: Optional['QdrantConfig'] = None, skip_connection: bool = False): self.logger = logging.getLogger("VectorManager") self.qdrant = None self.embedder = None self.available = False # Load config from secrets if not provided if qdrant_config is None: if SECRETS_AVAILABLE and get_qdrant_config: qdrant_config = get_qdrant_config() else: self.logger.warning("No secrets_loader available for Qdrant config") qdrant_config = None # Store config (may be None if not configured) self._config = qdrant_config self.collection_name = qdrant_config.collection_name if qdrant_config else "genesis_memories" self._init_embedder() if not skip_connection: self._connect_qdrant() def _init_embedder(self): """Initialize embedding model. FAIL LOUD if unavailable.""" if not FASTEMBED_AVAILABLE: raise ImportError("!! FATAL: fastembed not installed. Run: pip install fastembed") try: # BAAI/bge-small-en-v1.5 - 384 dimensions, fast, good quality self.embedder = TextEmbedding(model_name="BAAI/bge-small-en-v1.5") self.vector_size = 384 print("✓ Embedder initialized: BAAI/bge-small-en-v1.5") except Exception as e: raise RuntimeError(f"!! FATAL: Embedder init failed: {e}") def _connect_qdrant(self): """Connect to Qdrant. Gracefully degrades if unavailable.""" if not QDRANT_AVAILABLE: self.logger.warning("qdrant-client not installed. Run: pip install qdrant-client") self.available = False return if not self._config or not self._config.is_configured: self.logger.warning("Qdrant not configured. Set GENESIS_QDRANT_* environment variables.") self.available = False return try: self.qdrant = QdrantClient( url=self._config.url, api_key=self._config.api_key, timeout=10 ) # Ensure collection exists with correct dimensions try: collection_info = self.qdrant.get_collection(self.collection_name) # Check dimensions (BAAI/bge-small is 384, old might be 768) current_size = collection_info.config.params.vectors.size if current_size != self.vector_size: print(f"⚠️ Qdrant dimension mismatch: expected {self.vector_size}, got {current_size}. Recreating...") self.qdrant.delete_collection(self.collection_name) raise ValueError("Recreation needed for dimension alignment") print(f"[OK] Qdrant collection '{self.collection_name}' connected") except Exception as e: if "Not Found" in str(e) or "doesn't exist" in str(e).lower() or "Recreation" in str(e): self.qdrant.create_collection( collection_name=self.collection_name, vectors_config=models.VectorParams( size=self.vector_size, # 384 for bge-small distance=models.Distance.COSINE ) ) print(f"[OK] Created Qdrant collection '{self.collection_name}' (dim: {self.vector_size})") else: raise self.available = True except Exception as e: self.logger.error(f"Qdrant connection failed: {e}") self.available = False self.qdrant = None def _generate_embedding(self, text: str) -> List[float]: """Generate embedding for text. FAIL LOUD.""" try: embeddings = list(self.embedder.embed([text])) return embeddings[0].tolist() except Exception as e: raise RuntimeError(f"!! FATAL: Embedding generation failed: {e}") def _generate_id(self, content: str) -> str: """Generate deterministic UUID from content.""" import uuid # Namespace for Genesis memory GENESIS_NAMESPACE = uuid.UUID('e2a6d7f8-b3c4-4d5e-8f90-a1b2c3d4e5f6') return str(uuid.uuid5(GENESIS_NAMESPACE, content)) def health(self) -> Dict[str, bool]: """Check health of all backends.""" health = { "qdrant": False, "embedder": self.embedder is not None } if self.qdrant: try: self.qdrant.get_collections() health["qdrant"] = True except: pass return health def add(self, content: str, metadata: Dict = None, doc_id: Optional[str] = None) -> Dict[str, Any]: """ Add content to Qdrant with auto-generated embedding. FAIL LOUD - raises exception on failure. """ if not self.qdrant: raise ConnectionError("!! FATAL: Qdrant not connected") if doc_id is None: doc_id = self._generate_id(content) embedding = self._generate_embedding(content) payload = {"content": content} if metadata: payload.update(metadata) try: self.qdrant.upsert( collection_name=self.collection_name, points=[ models.PointStruct( id=doc_id, vector=embedding, payload=payload ) ] ) return {"qdrant": True, "id": doc_id} except Exception as e: raise RuntimeError(f"!! FATAL: Qdrant upsert failed: {e}") def search(self, query: str, top_k: int = 5) -> List[VectorDocument]: """ Search Qdrant for similar content. FAIL LOUD - raises exception on failure. """ if not self.qdrant: raise ConnectionError("!! FATAL: Qdrant not connected") query_embedding = self._generate_embedding(query) try: # Check if traditional search exists, otherwise use query_points if hasattr(self.qdrant, "search"): results = self.qdrant.search( collection_name=self.collection_name, query_vector=query_embedding, limit=top_k ) elif hasattr(self.qdrant, "query_points"): # New API in recently updated qdrant-client response = self.qdrant.query_points( collection_name=self.collection_name, query=query_embedding, limit=top_k ) results = response.points else: raise AttributeError("QdrantClient has no 'search' or 'query_points' method") return [ VectorDocument( id=str(r.id), content=r.payload.get("content", ""), score=r.score if hasattr(r, 'score') else 0.0, metadata={k: v for k, v in r.payload.items() if k != "content"} ) for r in results ] except Exception as e: raise RuntimeError(f"!! FATAL: Qdrant search failed: {e}") def unified_search(self, query: str, top_k: int = 5) -> List[VectorDocument]: """Alias for search() for backwards compatibility.""" return self.search(query, top_k)