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) # from fastembed import TextEmbedding # Original # FASTEMBED_AVAILABLE = False # Original from core.multimodal_embedder import MultimodalEmbedder # NEW # 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.multimodal_embedder = None # New attribute for multimodal embedder 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" # Initialize multimodal embedder try: gemini_api_key = os.environ.get("GEMINI_API_KEY", "YOUR_GEMINI_API_KEY") self.multimodal_embedder = MultimodalEmbedder(api_key=gemini_api_key) self.vector_size = self.multimodal_embedder.get_vector_size() self.logger.info(f"Multimodal Embedder initialized (Vector size: {self.vector_size})") except Exception as e: raise RuntimeError(f"!! FATAL: Multimodal Embedder init failed: {e}") if not skip_connection: self._connect_qdrant() 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 async def _generate_multimodal_embedding( self, text_input: Optional[str] = None, image_input: Optional[bytes] = None ) -> List[float]: """Generate multimodal embedding using the MultimodalEmbedder.""" if not self.multimodal_embedder: raise ConnectionError("!! FATAL: Multimodal Embedder not initialized") try: return await self.multimodal_embedder.generate_embedding( text_input=text_input, image_input=image_input ) except Exception as e: raise RuntimeError(f"!! FATAL: Multimodal 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 async def add(self, content: Optional[str] = None, metadata: Dict = None, doc_id: Optional[str] = None, image_data: Optional[bytes] = None) -> Dict[str, Any]: """ Add content (text and/or image) to Qdrant with auto-generated multimodal embedding. FAIL LOUD - raises exception on failure. """ if not self.qdrant: raise ConnectionError("!! FATAL: Qdrant not connected") if not content and not image_data: raise ValueError("At least one of content or image_data must be provided.") # If no doc_id is provided, generate a deterministic one from content + image hash if doc_id is None: combined_content = (content or "") + (hashlib.sha256(image_data).hexdigest() if image_data else "") doc_id = self._generate_id(combined_content) embedding = await self._generate_multimodal_embedding(text_input=content, image_input=image_data) payload = {"content": content} if image_data: payload["has_image"] = True # Indicate presence of image 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}") async def search(self, query: Optional[str] = None, image_query: Optional[bytes] = None, top_k: int = 5) -> List[VectorDocument]: """ Search Qdrant for similar content using multimodal query. FAIL LOUD - raises exception on failure. """ if not self.qdrant: raise ConnectionError("!! FATAL: Qdrant not connected") if not query and not image_query: raise ValueError("At least one of query or image_query must be provided.") query_embedding = await self._generate_multimodal_embedding(text_input=query, image_input=image_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}") async def unified_search(self, query: Optional[str] = None, image_query: Optional[bytes] = None, top_k: int = 5) -> List[VectorDocument]: """Alias for search() for backwards compatibility.""" return await self.search(query=query, image_query=image_query, top_k=top_k)