Release/v1.0.0 (#6)

* feat: adjustments to connect to open search

* chore: deleting aws vars

* fix: adjust the flow to make queries

* feat: improve the server initaization

* chore: delete logs

* feat: create virtual network

Author: JosemaPereira

.flaskenv

@@ -1,3 +1,3 @@
 FLASK_APP = "main"
-FLASK_RUN_PORT = "8000"
+FLASK_RUN_PORT = "8080"
 FLASK_DEBUG=true

Dockerfile

@@ -43,7 +43,7 @@ USER appuser
 COPY . .
 
 # Expose the port that the application listens on.
-EXPOSE 8000
+EXPOSE 8080
 
 # Run the application.
-CMD python3 -m flask run  --host=0.0.0.0
+CMD python3 -m flask run --host=0.0.0.0

compose.yaml

@@ -13,41 +13,28 @@ class Config:
     Configuration class
     """
 
-    DEBUG = environ.get("DEBUG")
     LOG_LEVEL = environ.get("LOG_LEVEL")
-    S3_BUCKET = environ.get("S3_BUCKET")
     OPENSEARCH_INDEX = environ.get("OPENSEARCH_INDEX")
     OPENSEARCH_HOST = environ.get("OPENSEARCH_HOST")
     OPENSEARCH_PORT = environ.get("OPENSEARCH_PORT")
-    OPENSEARCH_USE_SSL = environ.get("OPENSEARCH_USE_SSL")
-    OPENSEARCH_VERIFY_CERTS = environ.get("OPENSEARCH_VERIFY_CERTS")
-    IS_LOCAL = environ.get("IS_LOCAL")
-    S3_URL = None
-    S3_INDEX_PATH = environ.get("S3_INDEX_PATH")
     OPENSEARCH_USER = environ.get("OPENSEARCH_USER")
     OPENSEARCH_PASS = environ.get("OPENSEARCH_PASS")
+    OPENSEARCH_USE_SSL = environ.get("OPENSEARCH_USE_SSL")
+    OPENSEARCH_VERIFY_CERTS = environ.get("OPENSEARCH_VERIFY_CERTS")
+    S3_URL = environ.get("S3_URL")
 
 
 class DevelopmentConfig(Config):
     """
     Development configuration
     """
 
-    DEBUG = True
     LOG_LEVEL = "DEBUG"
-    OPENSEARCH_CLUSTER_URL = "http://host.docker.internal:9200"
     OPENSEARCH_INDEX = "clone-vector-index"
-    # OPENSEARCH_HOST = "localhost"
     OPENSEARCH_HOST = "host.docker.internal"
     OPENSEARCH_PORT = "9200"
     OPENSEARCH_USER = ""
     OPENSEARCH_PASS = ""
     OPENSEARCH_USE_SSL = False
     OPENSEARCH_VERIFY_CERTS = False
-    S3_BUCKET = "clone-ingestion-messages"
-    IS_LOCAL = True
     S3_URL = "http://host.docker.internal:4566"
-    AWS_ACCESS_KEY_ID = "test"
-    AWS_SECRET_ACCESS_KEY = "test"
-    AWS_DEFAULT_REGION = "us-east-1"
-    S3_INDEX_PATH = "/indexes"

config.py

@@ -29,14 +29,20 @@ class AbstractLlamaIndexService(ABC):
 
     @abstractmethod
     def vector_store_index(
-        self, twin_id: str, source_name: str, file_uuid: str, documents: list
+        self,
+        twin_id: str,
+        source_name: str,
+        channelId: str,
+        file_uuid: str,
+        documents: list,
     ) -> str:
         """
         Abstract method to indexing documents and store vectors in OpenSearch.
 
         Args:
             twin_id (str): Identifier for the twin.
             source_name (str): Name of the data source.
+            channelId (str): Channel identifier.
             file_uuid (str): UUID of the file containing the documents.
             documents (list): List of dictionaries representing documents.
 
@@ -48,28 +54,29 @@ def vector_store_index(
     @abstractmethod
     def vectorize_string(self, text_input: str) -> list:
         """
-            Abstract method to indexing documents and store vectors in OpenSearch.
+        Abstract method to indexing documents and store vectors in OpenSearch.
 
-            Args:
-                text_input (str): A string to vectorize
+        Args:
+            text_input (str): A string to vectorize
 
-            Returns:
-                list: a list of float values representing a vector
-            """
+        Returns:
+            list: a list of float values representing a vector
+        """
         pass
 
 
 class AbstractOpensearchService(ABC):
     """
-       Abstract class for Opensearch services
-   """
+    Abstract class for Opensearch services
+    """
+
     @abstractmethod
     def search(self, query: dict) -> list:
         """
         Abstract method to query an opensearch index
 
         Args:
-            query (dict): Opensearch DSL query string
+            query (dict): Opensearch query string
 
         Returns:
             list: a list of results

core/abstracts/services.py

@@ -2,7 +2,7 @@
 from logging import Logger
 from typing import Any, Dict, Tuple
 
-from flask import jsonify, Response
+from flask import Response, jsonify
 
 from core.abstracts.controller import AbstractVectorController
 from core.abstracts.usescases import AbstractVectorizeUsecase
@@ -42,18 +42,24 @@ def vectoring(self, request: Dict[str, Any]) -> Tuple[Response, int]:
 
         try:
             self.usecase.vectorize_and_index(s3_bucket, s3_object_key)
-            return jsonify({"message": "Object vectorization succeeded!"}), HTTPStatus.OK
+            return (
+                jsonify({"message": "Object vectorization succeeded!"}),
+                HTTPStatus.OK,
+            )
         except Exception as e:
             self.logger.error(f"Failed to vectorize object {s3_object_key}")
             return jsonify({"error": str(e)}), HTTPStatus.INTERNAL_SERVER_ERROR
 
     def search(self, request: Dict[str, Any]) -> Tuple[Response, int]:
         query = request["q"]
         if query is None or query.strip() == "":
-            return jsonify({'error': 'query param "q" is required'}), HTTPStatus.BAD_REQUEST
+            return (
+                jsonify({"error": 'query param "q" is required'}),
+                HTTPStatus.BAD_REQUEST,
+            )
 
         try:
             result = self.usecase.search(query)
-            return jsonify({'results': result}), HTTPStatus.OK
+            return jsonify({"results": result}), HTTPStatus.OK
         except Exception as e:
-            return jsonify({'error': str(e)}), HTTPStatus.INTERNAL_SERVER_ERROR
+            return jsonify({"error": str(e)}), HTTPStatus.INTERNAL_SERVER_ERROR

core/controller/vector.py

@@ -19,9 +19,9 @@ class LlamaIndexService(AbstractLlamaIndexService):
     """
 
     def __init__(
-            self,
-            vector_store: OpensearchVectorStore,
-            logger: Logger,
+        self,
+        vector_store: OpensearchVectorStore,
+        logger: Logger,
     ):
         """
         Initialize the LlamaIndexService.
@@ -37,14 +37,20 @@ def __init__(
         self.embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")
 
     def vector_store_index(
-            self, twin_id: str, source_name: str, file_uuid: str, documents: list
+        self,
+        twin_id: str,
+        source_name: str,
+        channelId: str,
+        file_uuid: str,
+        documents: list,
     ) -> str:
         """
         Index documents and store vectors in OpenSearch.
 
         Args:
             twin_id (str): Identifier for the twin.
             source_name (str): Name of the data source.
+            channelId (str): Identifier for the channel.
             file_uuid (str): UUID of the file containing the documents.
             documents (list): list of dictionary representing documents.
 
@@ -64,28 +70,29 @@ def vector_store_index(
                     metadata={
                         "raw_text": message["text"],
                         "user_name": message["user_name"],
+                        "user_id": message["user_id"],
                         "processed_user": processed_user,
                         "twin_id": twin_id,
                         "source_name": source_name,
                         "file_uuid": file_uuid,
+                        "channelId": channelId,
                     },
                     metadata_seperator=":",
                     embedding=embed_value,
                 )
             )
-        self.logger.info(docs)
         try:
             index = VectorStoreIndex.from_documents(
                 documents=docs,
                 storage_context=self.storage_context,
                 embed_model=self.embed_model,
             )
             self.logger.info(
-                f"Indexing documents for {twin_id}/{source_name}/{file_uuid}: {index.summary}"
+                f"Indexing documents for {twin_id}/{source_name}/{channelId}/{file_uuid}"
             )
             return index.summary
         except Exception as e:
-            message_error = f"Error while indexing documents for {twin_id}/{source_name}/{file_uuid}"
+            message_error = f"Error while indexing documents for {twin_id}/{source_name}/{channelId}/{file_uuid}"
             self.logger.error(e)
             raise ValueError(message_error)
 

core/service/llama_index_service.py

@@ -1,10 +1,9 @@
-import json
 from logging import Logger
 
-from core.abstracts.services import AbstractOpensearchService
-
 from opensearchpy import OpenSearch
 
+from core.abstracts.services import AbstractOpensearchService
+
 
 class OpensearchService(AbstractOpensearchService):
     """
@@ -24,19 +23,19 @@ def __init__(self, opensearch_client: OpenSearch, index: str, logger: Logger):
         self.index = index
         self.logger = logger
 
-    def search(self, query: str) -> list:
+    def search(self, query: dict) -> list:
         """
         Performs a query to the configured index
 
         Args:
-            query (str): Opensearch DSL query string
-
+            query (dict): the query to perform
         Returns:
             list: a list of dictionaries with the opensearch query document results
         """
         try:
             response = self.client.search(index=self.index, body=query)
-            return response['hits']['hits']
+            return response["hits"]["hits"]
         except Exception as e:
             error_message = f"Error while searching in OpenSearch: {str(e)}"
+            self.logger.error(error_message)
             raise Exception(error_message)