Merge branch 'refactor-2' into refactor-3

.github/workflows/main.yml

@@ -16,4 +16,4 @@ jobs:
       - name: Push to hub
         env:
           HF_TOKEN: ${{ secrets.HF_TOKEN }}
-        run: git push https://HF_USERNAME:[email protected]/spaces/KingZack/ctp-slack-bot main
+        run: git push https://KingZack:[email protected]/spaces/KingZack/ctp-slack-bot main

Dockerfile

@@ -25,8 +25,8 @@ RUN pip install --no-cache-dir .
 RUN useradd -m appuser
 USER appuser
 
-# Expose a volume mount for logs
-VOLUME ./logs
+# Expose a volume mount for logs ― Hugging Face Spaces requires specifically /data.
+VOLUME /data
 
 # Run the application.
 CMD ["python", "-m", "ctp_slack_bot.app"]

pyproject.toml

@@ -34,6 +34,7 @@ dependencies = [
     "webvtt-py>=0.5.1",
     "slack-sdk>=3.35.0",
     "slack_bolt>=1.23.0",
+    "pymongo>=4.11.3 ",
     "motor>=3.7.0",
     "openai>=1.70.0"
     "google-api-python-client>=2.167.0",

src/ctp_slack_bot/core/logging.py

@@ -75,16 +75,16 @@ def setup_logging() -> None:
     )
 
     # Add file handler for non-DEBUG environments.
-    if log_level != "DEBUG":
-        logger.add(
-            "logs/app.log",
-            rotation="10 MB",
-            retention="1 week",
-            compression="zip",
-            format=format_string,
-            level=log_level,
-            serialize=(log_format == "json"),
-        )
+    # if log_level != "DEBUG":
+    #     logger.add(
+    #         "/data/app.log",
+    #         rotation="10 MB",
+    #         retention="1 week",
+    #         compression="zip",
+    #         format=format_string,
+    #         level=log_level,
+    #         serialize=(log_format == "json"),
+    #     )
 
     # Intercept standard logging messages.
     basicConfig(handlers=[InterceptHandler()], level=0, force=True)

src/ctp_slack_bot/db/mongo_db.py

@@ -1,6 +1,7 @@
 from dependency_injector.resources import Resource
 from motor.motor_asyncio import AsyncIOMotorClient
 from pymongo.errors import ConnectionFailure, ServerSelectionTimeoutError
+from pymongo.operations import SearchIndexModel
 from loguru import logger
 from pydantic import BaseModel, PrivateAttr
 from typing import Any, Dict, Optional, Self
@@ -27,6 +28,10 @@ class MongoDB(BaseModel):
         """Initialize MongoDB client with settings."""
         try:
             connection_string = self.settings.MONGODB_URI.get_secret_value()
+            logger.debug("Connecting to MongoDB using URI: {}", connection_string.replace(
+                connection_string.split('@')[-1].split('/')[0] if '@' in connection_string else '',
+                '[REDACTED]'
+            ))
             
             # Create client with appropriate settings
             self._client = AsyncIOMotorClient(
@@ -54,86 +59,108 @@ class MongoDB(BaseModel):
     @property
     def client(self: Self) -> AsyncIOMotorClient:
         """Get the MongoDB client instance."""
-        if self._client is None:
+        if not hasattr(self, '_client') or self._client is None:
             logger.warning("MongoDB client not initialized. Attempting to initialize.")
             self.connect()
-            if self._client is None:
+            if not hasattr(self, '_client') or self._client is None:
                 raise ConnectionError("Failed to initialize MongoDB client")
         return self._client
     
     @property
     def db(self: Self) -> Any:
         """Get the MongoDB database instance."""
-        if self._db is None:
+        if not hasattr(self, '_db') or self._db is None:
             logger.warning("MongoDB database not initialized. Attempting to initialize client.")
             self.connect()
-            if self._db is None:
+            if not hasattr(self, '_db') or self._db is None:
                 raise ConnectionError("Failed to initialize MongoDB database")
         return self._db
     
     async def ping(self: Self) -> bool:
         """Check if MongoDB connection is alive."""
         try:
-            await self.client.admin.command('ping')
+            # Get client to ensure we're connected
+            client = self.client
+            
+            # Try a simple ping command
+            await client.admin.command('ping')
+            logger.debug("MongoDB connection is active")
             return True
         except (ConnectionFailure, ServerSelectionTimeoutError) as e:
             logger.error("MongoDB connection failed: {}", e)
             return False
+        except Exception as e:
+            logger.error("Unexpected error during MongoDB ping: {}", e)
+            return False
     
     async def get_collection(self: Self, name: str) -> Any:
         """
         Get a collection by name with validation.
         Creates the collection if it doesn't exist.
         """
+        # First ensure we can connect at all
         if not await self.ping():
+            logger.error("Cannot get collection '{}' - MongoDB connection is not available", name)
             raise ConnectionError("MongoDB connection is not available")
         
-        # Get all collection names to check if this one exists
-        collection_names = await self.db.list_collection_names()
-        if name not in collection_names:
-            logger.info(f"Collection {name} does not exist. Creating it.")
-            # Create the collection
-            await self.db.create_collection(name)
+        try:
+            # Get all collection names to check if this one exists
+            logger.debug("Checking if collection '{}' exists", name)
+            collection_names = await self.db.list_collection_names()
             
-        return self.db[name]
+            if name not in collection_names:
+                logger.info("Collection '{}' does not exist. Creating it.", name)
+                # Create the collection
+                await self.db.create_collection(name)
+                logger.debug("Successfully created collection '{}'", name)
+            else:
+                logger.debug("Collection '{}' already exists", name)
+                
+            # Get and return the collection
+            collection = self.db[name]
+            return collection
+        except Exception as e:
+            logger.error("Error accessing collection '{}': {}", name, e)
+            raise
     
-    async def create_indexes(self: Self, collection_name: str, indexes: list = None) -> None:
+    async def create_indexes(self: Self, collection_name: str) -> None:
         """
-        Create indexes on a collection.
-        If no indexes provided and collection needs vector search capability,
-        creates a vector search index using config settings.
+        Create a vector search index on a collection.
+        
+        Args:
+            collection_name: Name of the collection
         """
         collection = await self.get_collection(collection_name)
         
-        if indexes:
-            await collection.create_indexes(indexes)
-            logger.info("Created custom indexes for collection {}: {}", collection_name, indexes)
-        
-        else: # Create vector search index using settings from config
-            try:
-                # Create the vector search index with the proper MongoDB format
-                vector_search_index = {
-                    "mappings": {
-                        "dynamic": True,
-                        "fields": {
-                            "embedding": {
-                                "type": "knnVector",
-                                "dimensions": self.settings.VECTOR_DIMENSION,
-                                "similarity": "cosine"
-                            }
+        try:
+            # Create search index model using MongoDB's recommended approach
+            search_index_model = SearchIndexModel(
+                definition={
+                    "fields": [
+                        {
+                            "type": "vector",
+                            "path": "embedding",
+                            "numDimensions": self.settings.VECTOR_DIMENSION,
+                            "similarity": "cosine",
+                            "quantization": "scalar"
                         }
-                    }
-                }
-                
-                # Using createSearchIndex command which is the proper way to create vector search indexes
-                await self.db.command({
-                    "createSearchIndex": collection_name,
-                    "name": f"{collection_name}_vector_index",
-                    "definition": vector_search_index
-                })
-                
-                logger.info("Created vector search index for collection {}", collection_name)
-            except Exception as e:
+                    ]
+                },
+                name=f"{collection_name}_vector_index",
+                type="vectorSearch"
+            )
+            
+            # Create the search index using the motor collection
+            result = await collection.create_search_index(search_index_model)
+            logger.info("Vector search index '{}' created for collection {}", result, collection_name)
+            
+        except Exception as e:
+            if "command not found" in str(e).lower():
+                logger.warning("Vector search not supported by this MongoDB instance. Some functionality may be limited.")
+                # Create a fallback standard index on embedding field
+                await collection.create_index("embedding")
+                logger.info("Created standard index on 'embedding' field as fallback")
+            else:
                 logger.error("Failed to create vector index: {}", e)
                 raise
     
@@ -147,6 +174,30 @@ class MongoDB(BaseModel):
 
 class MongoDBResource(Resource):
     def init(self: Self, settings: Settings) -> MongoDB:
+        logger.info("Initializing MongoDB connection for database: {}", settings.MONGODB_NAME)
         mongo_db = MongoDB(settings=settings)
         mongo_db.connect()
+        
+        # Test the connection asynchronously - this will run after init returns
+        asyncio.create_task(self._test_connection(mongo_db))
+        
         return mongo_db
+    
+    async def _test_connection(self, mongo_db: MongoDB) -> None:
+        """Test MongoDB connection and log the result."""
+        try:
+            is_connected = await mongo_db.ping()
+            if is_connected:
+                logger.info("MongoDB connection test successful!")
+            else:
+                logger.error("MongoDB connection test failed!")
+        except Exception as e:
+            logger.error("Error testing MongoDB connection: {}", e)
+    
+    async def shutdown(self: Self, mongo_db: MongoDB) -> None:
+        """Close MongoDB connection on shutdown."""
+        try:
+            logger.info("Closing MongoDB connection...")
+            await mongo_db.close()
+        except Exception as e:
+            logger.error("Error closing MongoDB connection: {}", e)

src/ctp_slack_bot/services/content_ingestion_service.py

@@ -34,16 +34,16 @@ class ContentIngestionService(BaseModel):
         #    logger.debug("Ignored content with ID {} because it already exists in the database.", content.id)
         #    return
         chunks = content.get_chunks()
-        self.__vectorize_and_store_chunks_in_database(chunks)
+        await self.__vectorize_and_store_chunks_in_database(chunks)
         logger.debug("Stored {} vectorized chunk(s) in the database.", len(chunks))
 
     async def process_incoming_slack_message(self: Self, slack_message: SlackMessage) -> None:
         logger.debug("Content ingestion service received a Slack message: {}", slack_message.text)
         chunks = slack_message.get_chunks()
-        self.__vectorize_and_store_chunks_in_database(chunks)
+        await self.__vectorize_and_store_chunks_in_database(chunks)
         logger.debug("Stored {} vectorized chunk(s) in the database.", len(chunks))
 
-    def __vectorize_and_store_chunks_in_database(self: Self, chunks: Sequence[Chunk]) -> None:
-        # vectorized_chunks = self.vectorization_service.vectorize(chunks) # TODO
-        # self.vector_database_service.store(vectorized_chunks) # TODO
-        pass
+    async def __vectorize_and_store_chunks_in_database(self: Self, chunks: Sequence[Chunk]) -> None:
+        vectorized_chunks = self.vectorization_service.vectorize(chunks) # TODO
+        await self.vector_database_service.store(vectorized_chunks) # TODO
+

src/ctp_slack_bot/services/language_model_service.py

@@ -34,22 +34,22 @@ class LanguageModelService(BaseModel):
             str: Generated answer
         """
         logger.debug("Generating response for question “{}” using {} context chunks…", question, len(context))
-        # messages = [
-        #     {"role": "system", "content": self.settings.SYSTEM_PROMPT},
-        #     {"role": "user", "content": 
-        #         f"""Student Question: {question}
-
-        #         Context from class materials and transcripts:
-        #         {'\n'.join(chunk.text for chunk in context)}
-
-        #         Please answer the Student Question based on the Context from class materials and transcripts. If the context doesn’t contain relevant information, acknowledge that and suggest asking the professor."""}
-        # ]
-        # response: ChatCompletion = self._open_ai_client.chat.completions.create(
-        #     model=self.settings.CHAT_MODEL,
-        #     messages=messages,
-        #     max_tokens=self.settings.MAX_TOKENS,
-        #     temperature=self.settings.TEMPERATURE
-        # )
+        messages = [
+            {"role": "system", "content": self.settings.SYSTEM_PROMPT},
+            {"role": "user", "content": 
+                f"""Student Question: {question}
+
+                Context from class materials and transcripts:
+                {'\n'.join(chunk.text for chunk in context)}
+
+                Please answer the Student Question based on the Context from class materials and transcripts. If the context doesn’t contain relevant information, acknowledge that and suggest asking the professor."""}
+        ]
+        response: ChatCompletion = self._open_ai_client.chat.completions.create(
+            model=self.settings.CHAT_MODEL,
+            messages=messages,
+            max_tokens=self.settings.MAX_TOKENS,
+            temperature=self.settings.TEMPERATURE
+        )
         
-        # return response.choices[0].message.content
-        return f"Mock response to “{question}”"
+        return response.choices[0].message.content
+        # return f"Mock response to “{question}”"

src/ctp_slack_bot/services/vector_database_service.py

@@ -35,9 +35,11 @@ class VectorDatabaseService(BaseModel): # TODO: this should not rely specificall
             
         try:
             # Get the vector collection - this will create it if it doesn't exist
+            logger.debug("Getting vectors collection for storing {} chunks", len(chunks))
             vector_collection = await self.mongo_db.get_collection("vectors")
             
             # Ensure vector search index exists
+            logger.debug("Creating vector search index for vectors collection")
             await self.mongo_db.create_indexes("vectors")
             
             # Create documents to store, ensuring compatibility with BSON
@@ -57,11 +59,15 @@ class VectorDatabaseService(BaseModel): # TODO: this should not rely specificall
                 documents.append(document)
             
             # Insert into collection as a batch
+            logger.debug("Inserting {} documents into vectors collection", len(documents))
             result = await vector_collection.insert_many(documents)
-            logger.info(f"Stored {len(result.inserted_ids)} vector chunks in database")
+            logger.info("Stored {} vector chunks in database", len(result.inserted_ids))
             
         except Exception as e:
-            logger.error(f"Error storing vector embeddings: {str(e)}")
+            logger.error("Error storing vector embeddings: {}", str(e))
+            # Include more diagnostic information
+            logger.debug("MongoDB connection info: URI defined: {}, DB name: {}", 
+                         bool(self.settings.MONGODB_URI), self.settings.MONGODB_NAME)
             raise
     
     async def content_exists(self: Self, key: str)-> bool: # TODO: implement this.
@@ -85,18 +91,19 @@ class VectorDatabaseService(BaseModel): # TODO: this should not rely specificall
         """
         try:
             # Get the vector collection
+            logger.debug("Getting vectors collection for similarity search")
             vector_collection = await self.mongo_db.get_collection("vectors")
             
-            # Build aggregation pipeline for vector search
+            # Build aggregation pipeline for vector search using official MongoDB format
+            logger.debug("Building vector search pipeline with query embedding dimension: {}", len(query.query_embeddings))
             pipeline = [
                 {
-                    "$search": {
+                    "$vectorSearch": {
                         "index": "vectors_vector_index",
-                        "knnBeta": {
-                            "vector": list(query.query_embeddings),
-                            "path": "embedding",
-                            "k": query.k
-                        }
+                        "path": "embedding",
+                        "queryVector": query.query_embeddings, #list(query.query_embeddings),
+                        "numCandidates": query.k * 10,
+                        "limit": query.k
                     }
                 },
                 {
@@ -105,7 +112,7 @@ class VectorDatabaseService(BaseModel): # TODO: this should not rely specificall
                         "metadata": 1,
                         "parent_id": 1,
                         "chunk_id": 1,
-                        "score": { "$meta": "searchScore" }
+                        "score": { "$meta": "vectorSearchScore" }
                     }
                 }
             ]
@@ -114,17 +121,32 @@ class VectorDatabaseService(BaseModel): # TODO: this should not rely specificall
             if query.filter_metadata:
                 metadata_filter = {f"metadata.{k}": v for k, v in query.filter_metadata.items()}
                 pipeline.insert(1, {"$match": metadata_filter})
+                logger.debug("Added metadata filters to search: {}", query.filter_metadata)
             
-            # Add score threshold filter
+            # Add score threshold filter if needed
             if query.score_threshold > 0:
                 pipeline.append({
                     "$match": {
                         "score": { "$gte": query.score_threshold }
                     }
                 })
+                logger.debug("Added score threshold filter: {}", query.score_threshold)
             
-            # Execute the pipeline
-            results = await vector_collection.aggregate(pipeline).to_list(length=query.k)
+            try:
+                # Execute the vector search pipeline
+                logger.debug("Executing vector search pipeline")
+                results = await vector_collection.aggregate(pipeline).to_list(length=query.k)
+                logger.debug("Vector search returned {} results", len(results))
+            except Exception as e:
+                logger.warning("Vector search failed: {}. Falling back to basic text search.", str(e))
+                # Fall back to basic filtering with limit
+                query_filter = {}
+                if query.filter_metadata:
+                    query_filter.update({f"metadata.{k}": v for k, v in query.filter_metadata.items()})
+                
+                logger.debug("Executing fallback basic search with filter: {}", query_filter)
+                results = await vector_collection.find(query_filter).limit(query.k).to_list(length=query.k)
+                logger.debug("Fallback search returned {} results", len(results))
             
             # Convert results to Chunk objects
             chunks = []
@@ -140,9 +162,14 @@ class VectorDatabaseService(BaseModel): # TODO: this should not rely specificall
                 )
                 chunks.append(chunk)
             
-            logger.info(f"Found {len(chunks)} similar chunks with similarity search")
+            logger.info("Found {} similar chunks with similarity search", len(chunks))
             return chunks
             
         except Exception as e:
-            logger.error(f"Error in similarity search: {str(e)}")
+            logger.error("Error in similarity search: {}", str(e))
+            # Include additional diagnostic information
+            logger.debug("MongoDB connection info: URI defined: {}, DB name: {}", 
+                         bool(self.settings.MONGODB_URI), self.settings.MONGODB_NAME)
+            logger.debug("Query details: k={}, dimension={}", 
+                         query.k, len(query.query_embeddings) if query.query_embeddings else "None")
             raise