Add improved logging

Author: MathyouMB

src/apps/dev_cli.py

@@ -1,7 +1,10 @@
-# NOTE: This application is intended to be run for development purposes only.
+import logging
+
+logging.getLogger("sqlalchemy.engine.Engine").disabled = True
 
 from completions import CompletionService
 from infrastructure.db.source_repository import SourceRepository
+from config.logger import logger
 
 if __name__ == "__main__":
     with SourceRepository() as repository:
@@ -11,4 +14,5 @@
             q = input("\nAsk a question (or 'exit'): ")
             if q.lower() in {"exit", "quit"}:
                 break
-            service.create(q)
+            answer = service.create(q)
+            logger.info("dev_cli.answer_displayed", answer=answer)

src/completions/services/completion_service.py

@@ -2,18 +2,23 @@
 from infrastructure.ollama import create_completion
 from infrastructure.db.source_repository import SourceRepository
 from completions.services.context_generation_service import ContextGenerationService
+from config import logger
 
 
 class CompletionService:
     def __init__(self, repository: SourceRepository):
         self.repository = repository
 
     def create(self, query, k=3):
-        prompt_context = ContextGenerationService(query, self.repository).process(k)
+        logger.info("completion_service.create", query=query)
 
-        print("\n🧠 Answering with context:\n")
-        print(prompt_context)
-        print("\n💬 LLM Answer:\n")
+        prompt_context = ContextGenerationService(query, self.repository).process(k)
+        logger.info("completion_service.context_generated", query=query)
 
         answer = create_completion(create_prompt(prompt_context, query))
-        print(answer)
+        logger.info(
+            "completion_service.answer_generated",
+            query=query,
+            answer=answer[0:10] + "...",
+        )
+        return answer

src/completions/services/context_generation_service.py

@@ -1,3 +1,4 @@
+from config import logger
 from infrastructure.db.source_repository import SourceRepository
 
 
@@ -7,6 +8,13 @@ def __init__(self, query: str, repository: SourceRepository):
         self.repository = repository
 
     def process(self, k=3):
+        logger.info("context_generation_service.process", query=self.query)
+
         chunks = self.repository.get_top_k_chunks_by_similarity(self.query, k)
+        logger.info(
+            "context_generation_service.found_context_chunks",
+            chunk_ids=[chunk.id for chunk in chunks],
+        )
+
         context = "\n\n".join(chunk.content for chunk in chunks)
         return context

src/config/__init__.py

@@ -2,3 +2,5 @@
 OLLAMA_MODEL = "mistral"
 OLLAMA_URL = "http://localhost:11434"
 EMBEDDING_MODEL = "all-MiniLM-L6-v2"
+
+from .logger import logger

src/config/logger.py

@@ -0,0 +1,53 @@
+import logging
+import json
+from typing import Optional, Dict, Any
+
+# ANSI escape codes
+LOG_COLORS = {
+    "DEBUG": "\033[94m",  # Blue
+    "INFO": "\033[92m",  # Green
+    "WARNING": "\033[93m",  # Yellow
+    "ERROR": "\033[91m",  # Red
+    "RESET": "\033[0m",  # Reset to default
+}
+
+
+class ColorFormatter(logging.Formatter):
+    def format(self, record: logging.LogRecord) -> str:
+        level_color = LOG_COLORS.get(record.levelname, "")
+        reset = LOG_COLORS["RESET"]
+        record.msg = f"{level_color}{record.msg}{reset}"
+        return super().format(record)
+
+
+class SemanticLogger:
+    def __init__(self, name: Optional[str] = None):
+        self.logger = logging.getLogger(name or __name__)
+        handler = logging.StreamHandler()
+
+        formatter = ColorFormatter("%(asctime)s [%(levelname)s] %(message)s")
+        handler.setFormatter(formatter)
+
+        self.logger.setLevel(logging.INFO)
+        if not self.logger.hasHandlers():
+            self.logger.addHandler(handler)
+
+    def info(self, event: str, **data: Any):
+        self.logger.info(self._format(event, data))
+
+    def warning(self, event: str, **data: Any):
+        self.logger.warning(self._format(event, data))
+
+    def error(self, event: str, **data: Any):
+        self.logger.error(self._format(event, data))
+
+    def debug(self, event: str, **data: Any):
+        self.logger.debug(self._format(event, data))
+
+    def _format(self, event: str, data: Dict[str, Any]) -> str:
+        structured = {"event": event, **data}
+        return json.dumps(structured)
+
+
+# Usage
+logger = SemanticLogger()