Adding general response certainty based on median log probability to …

…output

Signed-off-by: Jiri Podivin <[email protected]>

.github/workflows/python-diff-lint.yml

@@ -16,6 +16,7 @@ jobs:
       - name: VCS Diff Lint
         uses: fedora-copr/vcs-diff-lint-action@v1
         id: VCS_Diff_Lint
+        install_rpm_packages: ["python3-numpy"]
 
       - name: Upload artifact with detected defects in SARIF format
         uses: actions/upload-artifact@v4

logdetective/logdetective.py

@@ -3,7 +3,8 @@
 import sys
 
 from logdetective.constants import DEFAULT_ADVISOR
-from logdetective.utils import process_log, initialize_model, retrieve_log_content, format_snippets
+from logdetective.utils import (
+    process_log, initialize_model, retrieve_log_content, format_snippets, compute_certainty)
 from logdetective.extractors import LLMExtractor, DrainExtractor
 
 LOG = logging.getLogger("logdetective")
@@ -92,15 +93,21 @@ def main():
     if args.no_stream:
         stream = False
     response = process_log(log_summary, model, stream)
+    probs = []
     print("Explanation:")
     if args.no_stream:
         print(response["choices"][0]["text"])
+        probs = response["choices"][0]["logprobs"]["top_logprobs"]
     else:
         # Stream the output
         for chunk in response:
+            if isinstance(chunk["choices"][0]["logprobs"], dict):
+                probs.extend(chunk["choices"][0]["logprobs"]["top_logprobs"])
             delta = chunk['choices'][0]['text']
             print(delta, end='', flush=True)
-        print()
+    certainty = compute_certainty(probs)
+
+    print(f"\nResponse certainty: {certainty:.2f}%\n")
 
 
 if __name__ == "__main__":

logdetective/utils.py

@@ -1,10 +1,11 @@
 import logging
 import os
+from typing import Iterator, List, Dict
 from urllib.parse import urlparse
-
+import numpy as np
 import requests
 
-from llama_cpp import Llama
+from llama_cpp import Llama, CreateCompletionResponse, CreateCompletionStreamResponse
 from logdetective.constants import PROMPT_TEMPLATE
 
 
@@ -60,20 +61,36 @@ def initialize_model(model_pth: str, filename_suffix: str = ".gguf", verbose: bo
         model = Llama(
             model_path=model_pth,
             n_ctx=0,  # Maximum context for the model
-            verbose=verbose)
+            verbose=verbose,
+            logits_all=True)
     else:
         model = Llama.from_pretrained(
             model_pth,
             f"*{filename_suffix}",
             n_ctx=0,  # Maximum context for the model
-            verbose=verbose)
+            verbose=verbose,
+            logits_all=True)
 
     return model
 
 
-def process_log(log: str, model: Llama, stream: bool) -> str:
+def compute_certainty(probs: List[Dict[str, float] | None]) -> float:
+    """Compute certainty of repsponse based on average logit probability.
+    Log probability is log(p), isn't really readable for most people, especially in compound.
+    In this case it's just a matter of applying inverse operation exp.
+    Of course that leaves you with a value in range <0, 1> so it needs to be multiplied by 100.
+    Simply put, this is the most straightforward way to get the numbers out.
     """
-    Processes a given log using the provided language model and returns its summary.
+
+    top_logprobs = [
+        np.exp(x) * 100 for e in probs if isinstance(e, dict) for x in e.values()]
+
+    return np.median(top_logprobs, axis=0)
+
+
+def process_log(log: str, model: Llama, stream: bool) -> (
+        CreateCompletionResponse | Iterator[CreateCompletionStreamResponse]):
+    """Processes a given log using the provided language model and returns its summary.
 
     Args:
         log (str): The input log to be processed.
@@ -82,10 +99,13 @@ def process_log(log: str, model: Llama, stream: bool) -> str:
     Returns:
         str: The summary of the given log generated by the language model.
     """
-    return model(
+    response = model(
         prompt=PROMPT_TEMPLATE.format(log),
         stream=stream,
-        max_tokens=0)
+        max_tokens=0,
+        logprobs=1)
+
+    return response
 
 
 def retrieve_log_content(log_path: str) -> str: