support prompt template loading

src/text_generation/adapters/prompt_template_repository.py

@@ -0,0 +1,19 @@
+import os
+from src.text_generation.ports.abstract_prompt_template_repository import AbstractPromptTemplateRepository
+from langchain_core.prompts import load_prompt, PromptTemplate
+
+
+class PromptTemplateRepository(AbstractPromptTemplateRepository):
+    def __init__(self):
+        super().__init__()
+        self.templates_dir = os.environ.get('PROMPT_TEMPLATES_DIR')
+    
+    def _create_path_from_id(self, id: str) -> str:
+        template_filename = f'{id}.json'
+        return os.path.join(self.templates_dir, template_filename)
+
+    def get(self, id: str) -> PromptTemplate:
+        return load_prompt(self._create_path_from_id(id))
+    
+    def add(self, id: str, prompt_template: PromptTemplate) -> None:
+        prompt_template.save(self._create_path_from_id(id))

src/text_generation/common/constants.py

@@ -2,4 +2,8 @@ class Constants:
     ASSISTANT_TOKEN = "<|assistant|>"
     END_TOKEN = "<|end|>"
     SYSTEM_TOKEN = "<|system|>"
-    USER_TOKEN = "<|user|>"
+    USER_TOKEN = "<|user|>"
+
+    # prompt template IDs
+    class PromptTemplateIds:
+        PHI_3_MINI_4K_INSTRUCT_BASIC = "phi-3-mini-4k-instruct-basic"

src/text_generation/domain/abstract_semantic_similarity_result.py

@@ -0,0 +1,5 @@
+import abc
+
+
+class AbstractSemanticSimilarityResult(abc.ABC):
+    pass

src/text_generation/domain/average.py

@@ -1,3 +0,0 @@
-class Average:
-    def from_list_of_floats(self, floats: list[float]) -> float:
-        return sum(floats) / len(floats)

src/text_generation/domain/semantic_similarity_result.py

@@ -0,0 +1,9 @@
+from typing import List
+from src.text_generation.domain.abstract_semantic_similarity_result import AbstractSemanticSimilarityResult
+
+
+class SemanticSimilarityResult(AbstractSemanticSimilarityResult):
+    def __init__(self, scores: List[float], mean: float):
+        super().__init__()
+        self.scores: List[float] = scores
+        self.mean: float = mean

src/text_generation/ports/abstract_prompt_template_repository.py

@@ -0,0 +1,11 @@
+import abc
+
+
+class AbstractPromptTemplateRepository(abc.ABC):
+    @abc.abstractmethod
+    def get(self, id: str) -> abc.ABC:
+        raise NotImplementedError
+    
+    @abc.abstractmethod
+    def add(self, id: str, prompt_template: abc.ABC) -> None:
+        raise NotImplementedError

src/text_generation/services/guidelines/abstract_rag_enhanced_semantic_similarity_guidelines_service.py

@@ -1,7 +1,7 @@
 import abc
 
 
-class AbstractSemanticSimilarityGuidelinesService(abc.ABC):
+class AbstractRagEnhancedSemanticSimilarityGuidelinesService(abc.ABC):
     @abc.abstractmethod
     def analyze(self, prompt_input_text: str) -> float:
         raise NotImplementedError

src/text_generation/services/guidelines/rag_enhanced_semantic_similarity_guidelines_service.py

@@ -0,0 +1,11 @@
+from src.text_generation.services.guidelines.abstract_rag_enhanced_semantic_similarity_guidelines_service import AbstractRagEnhancedSemanticSimilarityGuidelinesService
+
+
+class RagEnhancedSemanticSimilarityGuidelinesService(AbstractRagEnhancedSemanticSimilarityGuidelinesService):
+    def analyze(self, prompt_input_text: str) -> float:
+
+        
+        # TODO - check semantic similarity score
+        # TODO - retry with summarized prompt? task decomposition - result could contain original score and improved score
+
+        raise NotImplementedError

src/text_generation/services/guidelines/semantic_similarity_guidelines_service.py

@@ -1,7 +0,0 @@
-import abc
-
-
-class SemanticSimilarityGuidelinesService(abc.ABC):
-    @abc.abstractmethod
-    def analyze(self, prompt_input_text: str) -> float:
-        raise NotImplementedError

src/text_generation/services/logging/abstract_test_run_logging_service.py

@@ -0,0 +1,16 @@
+import abc
+from typing import Any, Dict, List
+
+
+class AbstractTestRunLoggingService(abc.ABC):
+    @abc.abstractmethod
+    def log_results(
+            self, 
+            scores: List[float], 
+            mean: float,
+            max: float):
+        raise NotImplementedError
+    
+    @abc.abstractmethod
+    def get_logs(self) -> List[Dict[str, Any]]:
+        raise NotImplementedError

src/text_generation/services/logging/test_run_logging_service.py

@@ -0,0 +1,54 @@
+import calendar
+import json
+import os
+import threading
+import time
+
+from datetime import datetime
+from typing import Any, Dict, List
+
+from src.text_generation.services.logging.abstract_test_run_logging_service import AbstractTestRunLoggingService
+
+
+class TestRunLoggingService(AbstractTestRunLoggingService):
+    def __init__(self, test_id: int):
+        self._lock = threading.Lock()
+        timestamp = calendar.timegm(time.gmtime())
+        self.log_file_path = f"./tests/logs/test_{test_id}_logs_{timestamp}.json"
+        self._ensure_log_file_exists()
+
+    def _ensure_log_file_exists(self):
+        if not os.path.exists(self.log_file_path):
+            with open(self.log_file_path, 'w') as f:
+                json.dump([], f)
+
+    def _read_logs(self) -> List[Dict[str, Any]]:
+        try:
+            with open(self.log_file_path, 'r') as f:
+                return json.load(f)
+        except (json.JSONDecodeError, FileNotFoundError):
+            return []
+
+    def _write_logs(self, logs: List[Dict[str, Any]]):
+        with open(self.log_file_path, 'w') as f:
+            json.dump(logs, f, indent=2, ensure_ascii=False)
+
+    def log_results(
+            self, 
+            scores: List[float],
+            mean: float,
+            max: float):
+        with self._lock:
+            logs = self._read_logs()
+            log_entry = {
+                "scores": scores,
+                "mean": mean,
+                "max": max,
+                "timestamp": datetime.now().isoformat()
+            }
+            logs.append(log_entry)
+            self._write_logs(logs)
+    
+    def get_logs(self) -> List[Dict[str, Any]]:
+        with self._lock:
+            return self._read_logs()

src/text_generation/services/nlp/abstract_prompt_template_service.py

@@ -0,0 +1,7 @@
+import abc
+
+
+class AbstractPromptTemplateService(abc.ABC):
+    @abc.abstractmethod
+    def get(self, id: str) -> abc.ABC:
+        raise NotImplementedError

src/text_generation/services/nlp/prompt_template_service.py

@@ -0,0 +1,18 @@
+from langchain_core.prompts import PromptTemplate
+from src.text_generation.ports.abstract_prompt_template_repository import AbstractPromptTemplateRepository
+from src.text_generation.services.nlp.abstract_prompt_template_service import AbstractPromptTemplateService
+
+
+class PromptTemplateService(AbstractPromptTemplateService):
+    def __init__(
+            self, 
+            prompt_template_repository: AbstractPromptTemplateRepository):
+        super().__init__()
+        self.prompt_template_repository = prompt_template_repository
+
+    def get(self, id: str) -> PromptTemplate:
+        prompt_template: PromptTemplate = self.prompt_template_repository.get(id)
+        return prompt_template
+    
+    def add(self, id: str, prompt_template: PromptTemplate) -> None:
+        self.prompt_template_repository.add(id, prompt_template)

src/text_generation/services/nlp/semantic_similarity_service.py

@@ -1,7 +1,9 @@
-from numpy import float64, array
+from numpy import array
 from sklearn.metrics.pairwise import cosine_similarity
 
 from src.text_generation.common.constants import Constants
+from src.text_generation.domain.abstract_semantic_similarity_result import AbstractSemanticSimilarityResult
+from src.text_generation.domain.semantic_similarity_result import SemanticSimilarityResult
 from src.text_generation.services.nlp.abstract_semantic_similarity_service import AbstractSemanticSimilarityService
 from src.text_generation.ports.abstract_embedding_model import AbstractEmbeddingModel
 
@@ -17,18 +19,17 @@ class SemanticSimilarityService(AbstractSemanticSimilarityService):
     def use_comparison_texts(self, comparison_texts: list[str]):
         self.comparison_texts = comparison_texts
 
-    def analyze(self, text: str) -> float:
-        query_embedding = self.embeddings.embed_query(text)
-        doc_embeddings = self.embeddings.embed_documents(self.comparison_texts)
+    def analyze(self, text: str) -> AbstractSemanticSimilarityResult:
+        """
+            Perfect alignment (similarity) results in a score of 1; opposite is 0
+        """
+        query_embedding = array(self.embeddings.embed_query(text)).reshape(1, -1)
+        doc_embeddings = array(self.embeddings.embed_documents(self.comparison_texts))
 
-        query_embedding = array(query_embedding).reshape(1, -1)
-        doc_embeddings = array(doc_embeddings)
-        similarity_scores: list[float64] = cosine_similarity(query_embedding, doc_embeddings)[0]
-        scores = list()
+        similarity_scores = cosine_similarity(query_embedding, doc_embeddings)
 
-        # perfect alignment (similarity) results in a score of 1;
-        # opposite is -1
+        all_scores = list()
         for _, score in enumerate(similarity_scores):
-            scores.append(score)
-
-        return max(scores)
+            all_scores.append(score)
+    
+        return SemanticSimilarityResult(scores=all_scores, mean=similarity_scores.mean())

src/text_generation/services/nlp/text_generation_completion_service.py

@@ -3,15 +3,20 @@ from langchain_core.output_parsers import StrOutputParser
 from langchain_core.runnables import RunnablePassthrough
 
 from src.text_generation.common.constants import Constants
+from src.text_generation.services.nlp.abstract_prompt_template_service import AbstractPromptTemplateService
 from src.text_generation.services.nlp.abstract_text_generation_completion_service import AbstractTextGenerationCompletionService
 from src.text_generation.ports.abstract_foundation_model import AbstractFoundationModel
 
 
 class TextGenerationCompletionService(AbstractTextGenerationCompletionService):
     
-    def __init__(self, foundation_model: AbstractFoundationModel):
+    def __init__(
+            self, 
+            foundation_model: AbstractFoundationModel,
+            prompt_template_service: AbstractPromptTemplateService):
         super().__init__()
         self.language_model_pipeline = foundation_model.create_pipeline()
+        self.prompt_template_service = prompt_template_service
         self.constants = Constants()
 
     def _extract_assistant_response(self, text):
@@ -19,21 +24,12 @@ class TextGenerationCompletionService(AbstractTextGenerationCompletionService):
             return text.split(self.constants.ASSISTANT_TOKEN)[-1].strip()
         return text
 
-    # TODO - get from config?
-    def _get_template(self):
-        return """<|system|>
-            Act as a conversational agent to respond to the end user's prompt. 
-            <|user|>\n
-            Question: {question}<|end|>\n
-            <|assistant|>
-            """
-
     def create_chain(self):
-        template = self._get_template()
-        prompt = PromptTemplate.from_template(template)
+        prompt_template_id=self.constants.PromptTemplateIds.PHI_3_MINI_4K_INSTRUCT_BASIC
+        prompt_template = self.prompt_template_service.get(id=prompt_template_id)
         return (
-            {"question": RunnablePassthrough()}
-            | prompt
+            { "question": RunnablePassthrough() }
+            | prompt_template
             | self.language_model_pipeline
             | StrOutputParser()
             | self._extract_assistant_response
@@ -42,9 +38,8 @@ class TextGenerationCompletionService(AbstractTextGenerationCompletionService):
     def invoke(self, user_prompt: str) -> str:
         if not user_prompt:
             raise ValueError(f"Parameter 'user_prompt' cannot be empty or None")
-        chain = self.create_chain()
         try:
-            response = chain.invoke(user_prompt)
-            return response
+            chain = self.create_chain()
+            return chain.invoke(user_prompt)
         except Exception as e:
             raise e