added vectorDB and context retrieval services, vectorquery model and MongoDB initialization

src/ctp_slack_bot/db/MongoDB.py

@@ -0,0 +1,122 @@
+from motor.motor_asyncio import AsyncIOMotorClient
+from pymongo import IndexModel, ASCENDING
+import logging
+from typing import Optional
+
+from ctp_slack_bot.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+class MongoDB:
+    """
+    MongoDB connection and initialization class.
+    Handles connection to MongoDB, database selection, and index creation.
+    """
+    def __init__(self):
+        self.client: Optional[AsyncIOMotorClient] = None
+        self.db = None
+        self.vector_collection = None
+        self.initialized = False
+
+    async def connect(self):
+        """
+        Connect to MongoDB using connection string from settings.
+        """
+        if self.client is not None:
+            return
+
+        if not settings.MONGODB_URI:
+            raise ValueError("MONGODB_URI is not set in environment variables")
+
+        try:
+            # Create MongoDB connection
+            self.client = AsyncIOMotorClient(settings.MONGODB_URI.get_secret_value())
+            self.db = self.client[settings.MONGODB_DB_NAME]
+            self.vector_collection = self.db["vector_store"]
+            logger.info(f"Connected to MongoDB: {settings.MONGODB_DB_NAME}")
+        except Exception as e:
+            logger.error(f"Error connecting to MongoDB: {str(e)}")
+            raise
+
+    async def initialize(self):
+        """
+        Initialize MongoDB with required collections and indexes.
+        """
+        if self.initialized:
+            return
+        
+        if not self.client:
+            await self.connect()
+        
+        try:
+            # Create vector index for similarity search
+            await self.create_vector_index()
+            self.initialized = True
+            logger.info("MongoDB initialized successfully")
+        except Exception as e:
+            logger.error(f"Error initializing MongoDB: {str(e)}")
+            raise
+
+    async def create_vector_index(self):
+        """
+        Create vector index for similarity search using MongoDB Atlas Vector Search.
+        """
+        try:
+            # Check if index already exists
+            existing_indexes = await self.vector_collection.list_indexes().to_list(length=None)
+            index_names = [index.get('name') for index in existing_indexes]
+            
+            if "vector_index" not in index_names:
+                # Create vector search index
+                index_definition = {
+                    "mappings": {
+                        "dynamic": True,
+                        "fields": {
+                            "embedding": {
+                                "dimensions": settings.VECTOR_DIMENSION,
+                                "similarity": "cosine",
+                                "type": "knnVector"
+                            }
+                        }
+                    }
+                }
+                
+                # Create the index
+                await self.db.command({
+                    "createIndexes": self.vector_collection.name,
+                    "indexes": [
+                        {
+                            "name": "vector_index",
+                            "key": {"embedding": "vector"},
+                            "weights": {"embedding": 1},
+                            "vectorSearchOptions": index_definition
+                        }
+                    ]
+                })
+                
+                # Create additional metadata indexes for filtering
+                await self.vector_collection.create_index([("metadata.source", ASCENDING)])
+                await self.vector_collection.create_index([("metadata.timestamp", ASCENDING)])
+                
+                logger.info("Vector search index created")
+            else:
+                logger.info("Vector search index already exists")
+                
+        except Exception as e:
+            logger.error(f"Error creating vector index: {str(e)}")
+            raise
+
+    async def close(self):
+        """
+        Close MongoDB connection.
+        """
+        if self.client:
+            self.client.close()
+            self.client = None
+            self.db = None
+            self.vector_collection = None
+            self.initialized = False
+            logger.info("MongoDB connection closed")
+
+# Create a singleton instance
+mongodb = MongoDB() 

src/ctp_slack_bot/models/VectorQuery.py

@@ -0,0 +1,17 @@
+from pydantic import BaseModel, Field, validator
+from typing import Optional, List, Dict, Any
+from ctp_slack_bot.core.config import settings
+
+class VectorQuery(BaseModel):
+    """Model for vector database similarity search queries.
+    
+    Attributes:
+        query_text: The text to be vectorized and used for similarity search
+        k: Number of similar documents to retrieve
+        score_threshold: Minimum similarity score threshold for inclusion in results
+        filter_metadata: Optional filters for metadata fields
+    """
+    query_text: str
+    k: int = Field(default=settings.TOP_K_MATCHES)
+    score_threshold: float = Field(default=0.7)
+    filter_metadata: Optional[Dict[str, Any]] = None

src/ctp_slack_bot/services/ContextRetrievalService.py

@@ -0,0 +1,76 @@
+import logging
+from typing import List, Dict, Any, Optional
+
+from ctp_slack_bot.models.slack import SlackMessage
+from ctp_slack_bot.models.content import RetreivedContext
+from ctp_slack_bot.models.VectorQuery import VectorQuery
+from ctp_slack_bot.services.VectorizationService import VectorizationService
+from ctp_slack_bot.services.VectorDatabaseService import VectorDatabaseService
+from ctp_slack_bot.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+class ContextRetrievalService:
+    """
+    Service for retrieving relevant context from the vector database based on user questions.
+    """
+    
+    def __init__(self):
+        self.vectorization_service = VectorizationService()
+        self.vector_db_service = VectorDatabaseService()
+    
+    async def initialize(self):
+        """
+        Initialize the required services.
+        """
+        await self.vector_db_service.initialize()
+    
+    async def get_context(self, message: SlackMessage) -> List[RetreivedContext]:
+        """
+        Retrieve relevant context for a given Slack message.
+        
+        This function:
+        1. Extracts the question text from the message
+        2. Vectorizes the question using VectorizationService
+        3. Queries VectorDatabaseService for similar context
+        4. Returns the relevant context as a list of RetreivedContext objects
+        
+        Args:
+            message: The SlackMessage containing the user's question
+            
+        Returns:
+            List[RetreivedContext]: List of retrieved context items with similarity scores
+        """
+        if not message.is_question:
+            logger.debug(f"Message {message.key} is not a question, skipping context retrieval")
+            return []
+        
+        try:
+            # Vectorize the message text
+            embeddings = self.vectorization_service.get_embeddings([message.text])
+            if embeddings is None or len(embeddings) == 0:
+                logger.error(f"Failed to generate embedding for message: {message.key}")
+                return []
+            
+            query_embedding = embeddings[0].tolist()
+            
+            # Create vector query
+            vector_query = VectorQuery(
+                query_text=message.text,
+                k=settings.TOP_K_MATCHES,
+                score_threshold=0.7  # Minimum similarity threshold
+            )
+            
+            # Search for similar content in vector database
+            context_results = await self.vector_db_service.search_by_similarity(
+                query=vector_query,
+                query_embedding=query_embedding
+            )
+            
+            logger.info(f"Retrieved {len(context_results)} context items for message: {message.key}")
+            return context_results
+            
+        except Exception as e:
+            logger.error(f"Error retrieving context for message {message.key}: {str(e)}")
+            return []
+   

src/ctp_slack_bot/services/VectorDatabaseService.py

@@ -0,0 +1,124 @@
+import logging
+from typing import List, Dict, Any, Optional
+# import numpy as np
+
+from ctp_slack_bot.db.MongoDB import mongodb
+from ctp_slack_bot.models.VectorQuery import VectorQuery
+from ctp_slack_bot.models.content import RetreivedContext
+
+logger = logging.getLogger(__name__)
+
+class VectorDatabaseService:
+    """
+    Service for storing and retrieving vector embeddings from MongoDB.
+    """
+    
+    async def initialize(self):
+        """
+        Initialize the database connection.
+        """
+        await mongodb.initialize()
+    
+    async def store(self, text: str, embedding: List[float], metadata: Dict[str, Any]) -> str:
+        """
+        Store text and its embedding vector in the database.
+        
+        Args:
+            text: The text content to store
+            embedding: The vector embedding of the text
+            metadata: Additional metadata about the text (source, timestamp, etc.)
+            
+        Returns:
+            str: The ID of the stored document
+        """
+        if not mongodb.initialized:
+            await mongodb.initialize()
+            
+        try:
+            # Create document to store
+            document = {
+                "text": text,
+                "embedding": embedding,
+                "metadata": metadata
+            }
+            
+            # Insert into collection
+            result = await mongodb.vector_collection.insert_one(document)
+            logger.debug(f"Stored document with ID: {result.inserted_id}")
+            
+            return str(result.inserted_id)
+        except Exception as e:
+            logger.error(f"Error storing embedding: {str(e)}")
+            raise
+    
+    async def search_by_similarity(self, query: VectorQuery, query_embedding: List[float]) -> List[RetreivedContext]:
+        """
+        Query the vector database for similar documents.
+        
+        Args:
+            query: VectorQuery object with search parameters
+            query_embedding: The vector embedding of the query text
+            
+        Returns:
+            List[RetreivedContext]: List of similar documents with similarity scores
+        """
+        if not mongodb.initialized:
+            await mongodb.initialize()
+            
+        try:
+            # Build aggregation pipeline for vector search
+            pipeline = [
+                {
+                    "$search": {
+                        "index": "vector_index",
+                        "knnBeta": {
+                            "vector": query_embedding,
+                            "path": "embedding",
+                            "k": query.k
+                        }
+                    }
+                },
+                {
+                    "$project": {
+                        "_id": 0,
+                        "text": 1,
+                        "metadata": 1,
+                        "score": {"$meta": "searchScore"}
+                    }
+                }
+            ]
+            
+            # Add metadata filters if provided
+            if query.filter_metadata:
+                metadata_filter = {f"metadata.{k}": v for k, v in query.filter_metadata.items()}
+                pipeline.insert(1, {"$match": metadata_filter})
+            
+            # Execute the pipeline
+            results = await mongodb.vector_collection.aggregate(pipeline).to_list(length=query.k)
+            
+            # Convert to RetreivedContext objects directly
+            context_results = []
+            for result in results:
+                # Normalize score to [0,1] range
+                normalized_score = result.get("score", 0)
+                
+                # Skip if below threshold
+                if normalized_score < query.score_threshold:
+                    continue
+                    
+                context_results.append(
+                    RetreivedContext(
+                        contextual_text=result["text"],
+                        metadata_source=result["metadata"].get("source", "unknown"),
+                        similarity_score=normalized_score,
+                        said_by=result["metadata"].get("speaker", None),
+                        in_reation_to_question=result["metadata"].get("related_question", None)
+                    )
+                )
+            
+            logger.debug(f"Found {len(context_results)} similar documents")
+            return context_results
+            
+        except Exception as e:
+            logger.error(f"Error in similarity search: {str(e)}")
+            raise