update ranking

chatbot_model.py

@@ -36,6 +36,7 @@ class ChatbotConfig:
     max_context_turns: int = 5
     warmup_steps: int = 200
     pretrained_model: str = 'distilbert-base-uncased'
+    cross_encoder_model: str = 'cross-encoder/ms-marco-MiniLM-L-12-v2'
     dtype: str = 'float32'
     freeze_embeddings: bool = False
     embedding_batch_size: int = 64
@@ -190,7 +191,7 @@ class RetrievalChatbot(DeviceAwareModel):
     def _initialize_reranker(self) -> CrossEncoderReranker:
         """Initialize the CrossEncoderReranker."""
         logger.info("Initializing default CrossEncoderReranker...")
-        return CrossEncoderReranker(model_name="cross-encoder/ms-marco-MiniLM-L-12-v2")
+        return CrossEncoderReranker(model_name=self.config.cross_encoder_model)
 
     def _initialize_summarizer(self) -> Summarizer:
         """Initialize the Summarizer."""
@@ -392,7 +393,7 @@ class RetrievalChatbot(DeviceAwareModel):
         
         # Re-rank these boosted candidates
         if not reranker:
-            reranker = CrossEncoderReranker(model_name="cross-encoder/ms-marco-MiniLM-L-12-v2")
+            reranker = CrossEncoderReranker(model_name=self.config.cross_encoder_model)
         
         ce_scores = reranker.rerank(query, texts, max_length=256)
 
@@ -564,8 +565,8 @@ class RetrievalChatbot(DeviceAwareModel):
         boosted.sort(key=lambda x: x[1], reverse=True)
         
         # Print top 10
-        for resp, score in boosted[:150]:
-            logger.debug(f"Candidate: '{resp}' with score {score}")
+        # for resp, score in boosted[:150]:
+        #     logger.debug(f"Candidate: '{resp}' with score {score}")
 
         # 8) Return top_k
         return boosted[:top_k]

chatbot_validator.py

@@ -17,8 +17,8 @@ class ChatbotValidator:
         """
         Initialize the validator.
         Args:
-            chatbot: RetrievalChatbot instance for inference
-            quality_checker: ResponseQualityChecker instance
+            chatbot: RetrievalChatbot for inference
+            quality_checker: ResponseQualityChecker
         """
         self.chatbot = chatbot
         self.quality_checker = quality_checker
@@ -86,7 +86,7 @@ class ChatbotValidator:
         domain_metrics = {}
         
         # Init the cross-encoder reranker to pass to the chatbot
-        reranker = CrossEncoderReranker(model_name="cross-encoder/ms-marco-MiniLM-L-12-v2")
+        reranker = CrossEncoderReranker(model_name=self.chatbot.config.cross_encoder_model)
 
         # Prepare random selection if needed
         rng = random.Random(seed)

response_quality_checker.py

@@ -1,352 +1,194 @@
 import numpy as np
-from typing import List, Tuple, Dict, Any, TYPE_CHECKING
-from sklearn.metrics.pairwise import cosine_similarity
-
+from typing import List, Tuple, Dict, Any
+from tf_data_pipeline import TFDataPipeline
 from logger_config import config_logger
-logger = config_logger(__name__)
 
-if TYPE_CHECKING:
-    from tf_data_pipeline import TFDataPipeline
+logger = config_logger(__name__)
 
 class ResponseQualityChecker:
     """
-    Enhanced quality checking that calculates:
-      - Relevance between query & responses
+    The Response Quality Checker measures:
+      - Relevance (embedding or cross-encoder)
       - Diversity among top responses
-      - Response length scoring
-      - Confidence determination based on multiple thresholds
+      - Length
+      - Score gap
+      - Confidence
     """
-
+    
     def __init__(
         self,
-        data_pipeline: 'TFDataPipeline',
-        confidence_threshold: float = 0.45,
+        data_pipeline: "TFDataPipeline",
+        confidence_threshold: float = 0.40,
         diversity_threshold: float = 0.15,
         min_response_length: int = 5,
-        similarity_cap: float = 0.85,
+        similarity_cap: float = 0.85
     ):
         """
         Args:
-            data_pipeline: Reference to TFDataPipeline for encoding
-            confidence_threshold: Minimum top_score for a 'confident' result
-            diversity_threshold: Minimum required diversity among top responses
-            min_response_length: Minimum words for a decent response
-            similarity_cap: Cap on pairwise similarity for diversity calc
+            data_pipeline: TFDataPipeline for encoding
+            confidence_threshold: Min top_score for 'confident'
+            diversity_threshold: Min average diversity for top responses
+            min_response_length: Min word count - 'valid length'
+            similarity_cap: Cap pairwise similarity to reduce outliers
         """
+        self.data_pipeline = data_pipeline
         self.confidence_threshold = confidence_threshold
         self.diversity_threshold = diversity_threshold
         self.min_response_length = min_response_length
         self.similarity_cap = similarity_cap
-        self.data_pipeline = data_pipeline
-
-        # Additional thresholds for more refined checks
+        
+        # Additional thresholds
         self.thresholds = {
-            'relevance': 0.30,     # Slightly relaxed
-            'length_score': 0.80,  # Stricter length requirement
-            'score_gap': 0.05      # Gap between top scores
+            'relevance': 0.30,
+            'length_score': 0.80,
+            'score_gap': 0.05
         }
-
+        
     def check_response_quality(
         self,
         query: str,
         responses: List[Tuple[str, float]]
     ) -> Dict[str, Any]:
         """
-        Evaluate the quality of a set of ranked responses for a given query.
-
-        Args:
-            query: The user's original query
-            responses: List of (response_text, score) sorted by descending score
-
-        Returns:
-            Dictionary of metrics, including 'is_confident' and others
+        Evaluate the quality of top-k responses:
+         - response_diversity
+         - query_response_relevance
+         - response_length_score
+         - top_score
+         - top_3_score_gap
+         - is_confident
         """
         if not responses:
             return {
                 'response_diversity': 0.0,
                 'query_response_relevance': 0.0,
-                'is_confident': False,
-                'top_score': 0.0,
                 'response_length_score': 0.0,
-                'top_3_score_gap': 0.0
+                'top_score': 0.0,
+                'top_3_score_gap': 0.0,
+                'is_confident': False
             }
-
-        # 1) Calculate relevant metrics
+            
         metrics = {}
-        metrics['response_diversity'] = self.calculate_diversity(responses)
-        metrics['query_response_relevance'] = self.calculate_relevance(query, responses)
-        metrics['response_length_score'] = self._average_length_score(responses)
+        metrics['response_diversity'] = self._calc_diversity(responses)
+        metrics['query_response_relevance'] = self._calc_relevance(query, responses)
+        metrics['response_length_score'] = self._calc_length_score(responses)
         metrics['top_score'] = responses[0][1]
-        metrics['top_3_score_gap'] = self._calculate_score_gap([s for _, s in responses], top_n=3)
-
-        # 2) Determine confidence
+        metrics['top_3_score_gap'] = self._calc_score_gap([score for _, score in responses])
+        
         metrics['is_confident'] = self._determine_confidence(metrics)
+        
         logger.info(f"Quality metrics: {metrics}")
         return metrics
-
-    def calculate_relevance(self, query: str, responses: List[Tuple[str, float]]) -> float:
+    
+    def _calc_diversity(self, responses: List[Tuple[str, float]]) -> float:
         """
-        Compute an overall 'relevance' metric between the query and the top responses.
-        Uses an exponential transform on the similarity to penalize weaker matches.
-        """
-        if not responses:
-            return 0.0
-
-        # Encode query and responses
-        query_emb = self.data_pipeline.encode_query(query)
-        resp_texts = [r for r, _ in responses]
-        resp_embs = self.data_pipeline.encode_responses(resp_texts)
-
-        # Normalize embeddings
-        query_emb = query_emb / (np.linalg.norm(query_emb) + 1e-12)
-        resp_norms = np.linalg.norm(resp_embs, axis=1, keepdims=True) + 1e-12
-        resp_embs = resp_embs / resp_norms
-
-        # Cosine similarity
-        sims = cosine_similarity([query_emb], resp_embs)[0]
-
-        # Exponential transform: higher sims remain close to 1, lower sims drop quickly
-        sims = np.exp(sims - 1.0)
-
-        # Weighted average: give heavier weighting to higher-ranked items
-        weights = np.exp(-np.arange(len(sims)) / 2.0)
-        weighted_avg = np.average(sims, weights=weights)
-        return float(weighted_avg)
-
-    def calculate_diversity(self, responses: List[Tuple[str, float]]) -> float:
-        """
-        Calculate how 'different' the top responses are from each other.
-        Diversity = 1 - avg_cosine_similarity (capped).
+        Average similarity among top response embeddings, capped by self.similarity_cap.
         """
         if len(responses) < 2:
-            return 1.0  # Single response is trivially 'unique'
-
-        resp_texts = [r for r, _ in responses]
-        embs = self.data_pipeline.encode_responses(resp_texts)
-
-        # Pairwise similarity
-        sim_matrix = cosine_similarity(embs, embs)
+            return 1.0  # Single response
+        
+        texts = [r for r, _ in responses]
+        embs = self.data_pipeline.encode_responses(texts)
+        sim_matrix = self._cosine_similarity(embs, embs)
+        
+        # Zero out diagonal
         np.fill_diagonal(sim_matrix, 0.0)
-
-        # Cap similarity to avoid outliers
+        
+        # Cap similarity
         sim_matrix = np.minimum(sim_matrix, self.similarity_cap)
-
-        # Mean off-diagonal similarity
+        
         sum_sims = np.sum(sim_matrix)
-        num_pairs = len(resp_texts) * (len(resp_texts) - 1)
-        avg_sim = sum_sims / num_pairs if num_pairs > 0 else 0.0
-
-        # Invert to get diversity
+        count = len(responses) * (len(responses) - 1)
+        avg_sim = sum_sims / count if count > 0 else 0.0
+        
         return 1.0 - avg_sim
-
-    def _determine_confidence(self, metrics: Dict[str, float]) -> bool:
-        """
-        Decide if we're 'confident' based on multiple metric thresholds.
-        """
-        primary_conditions = [
-            metrics['top_score'] >= self.confidence_threshold,
-            metrics['response_diversity'] >= self.diversity_threshold,
-            metrics['response_length_score'] >= self.thresholds['length_score']
-        ]
-
-        secondary_conditions = [
-            metrics['query_response_relevance'] >= self.thresholds['relevance'],
-            metrics['top_3_score_gap'] >= self.thresholds['score_gap'],
-            metrics['top_score'] >= (self.confidence_threshold + 0.05)  # Extra buffer
-        ]
-
-        # Must pass all primary checks, and at least 2 of the 3 secondary
-        return all(primary_conditions) and (sum(secondary_conditions) >= 2)
-
-    def _average_length_score(self, responses: List[Tuple[str, float]]) -> float:
+    
+    def _calc_relevance(self, query: str, responses: List[Tuple[str, float]]) -> float:
         """
-        Compute an average length score across all responses.
+        Weighted average of exponential-transformed similarities for top-k.
+        Encourages a high similarity with the top responses.
         """
-        length_scores = []
-        for response, _ in responses:
-            length_scores.append(self._length_score(response))
-        return float(np.mean(length_scores)) if length_scores else 0.0
-
-    def _length_score(self, text: str) -> float:
-        """
-        Calculate how well the text meets our length requirement.
-        Scores 1.0 if text is >= min_response_length and not too long,
-        else it scales down.
-        """
-        words = len(text.split())
-        if words < self.min_response_length:
-            return words / float(self.min_response_length)
-        elif words > 60:  
-            return max(0.5, 60.0 / words)  # Slight penalty for very long
-        return 1.0
-
-    def _calculate_score_gap(self, scores: List[float], top_n: int = 3) -> float:
-        """
-        Calculate the average gap between consecutive scores in the top N.
+        if not responses:
+            return 0.0
+        
+        query_emb = self.data_pipeline.encode_query(query)
+        texts = [r for r, _ in responses]
+        resp_embs = self.data_pipeline.encode_responses(texts)
+        
+        query_emb = query_emb / (np.linalg.norm(query_emb) + 1e-8)
+        norms = (np.linalg.norm(resp_embs, axis=1, keepdims=True) + 1e-8)
+        resp_embs = resp_embs / norms
+        
+        # Cosine similarity, then exponential transform
+        sims = np.sum(query_emb[np.newaxis, :] * resp_embs, axis=1)  # shape [k]
+        sims = np.exp(sims - 1.0)
+        
+        # Weighted average to boost top responses
+        weights = np.exp(-np.arange(len(responses)) / 2.0)
+        weighted_avg = np.average(sims, weights=weights)
+        return float(weighted_avg)
+    
+    def _calc_length_score(self, responses: List[Tuple[str, float]]) -> float:
+        """
+        Average length-based score across top responses.
+        """
+        scores = []
+        for text, _ in responses:
+            words = len(text.strip().split())
+            if words < self.min_response_length:
+                # Penalty for too short
+                s = words / float(self.min_response_length)
+            elif words > 50:
+                # Penalty for excessive length
+                s = max(0.5, 50.0 / words)
+            else:
+                s = 1.0
+            scores.append(s)
+            
+        return float(np.mean(scores)) if scores else 0.0
+    
+    def _calc_score_gap(self, scores: List[float], top_n: int = 3) -> float:
+        """
+        Average difference between consecutive ranks for top_n.
         """
         if len(scores) < 2:
             return 0.0
-        top_n = min(len(scores), top_n)
+        top_n = min(top_n, len(scores))
         gaps = []
         for i in range(top_n - 1):
             gaps.append(scores[i] - scores[i + 1])
         return float(np.mean(gaps)) if gaps else 0.0
-
-# import numpy as np
-# from typing import List, Tuple, Dict, Any, TYPE_CHECKING
-# from sklearn.metrics.pairwise import cosine_similarity
-
-# from logger_config import config_logger
-# logger = config_logger(__name__)
-
-# if TYPE_CHECKING:
-#     from tf_data_pipeline import TFDataPipeline
-
-# class ResponseQualityChecker:
-#     """Enhanced quality checking with dynamic thresholds."""
-
-#     def __init__(
-#         self,
-#         data_pipeline: 'TFDataPipeline',
-#         confidence_threshold: float = 0.4,
-#         diversity_threshold: float = 0.15,
-#         min_response_length: int = 5,
-#         similarity_cap: float = 0.85  # Renamed from max_similarity_ratio and used in diversity calc
-#     ):
-#         self.confidence_threshold = confidence_threshold
-#         self.diversity_threshold = diversity_threshold
-#         self.min_response_length = min_response_length
-#         self.similarity_cap = similarity_cap
-#         self.data_pipeline = data_pipeline  # Reference to TFDataPipeline
-
-#         # Dynamic thresholds based on response patterns
-#         self.thresholds = {
-#             'relevance': 0.35,    
-#             'length_score': 0.85,  
-#             'score_gap': 0.04     
-#         }
-
-#     def check_response_quality(
-#         self,
-#         query: str,
-#         responses: List[Tuple[str, float]]
-#     ) -> Dict[str, Any]:
-#         """
-#         Evaluate the quality of responses based on various metrics.
-
-#         Args:
-#             query: The user's query
-#             responses: List of (response_text, score) tuples
-
-#         Returns:
-#             Dict containing quality metrics and confidence assessment
-#         """
-#         if not responses:
-#             return {
-#                 'response_diversity': 0.0,
-#                 'query_response_relevance': 0.0,
-#                 'is_confident': False,
-#                 'top_score': 0.0,
-#                 'response_length_score': 0.0,
-#                 'top_3_score_gap': 0.0
-#             }
-
-#         # Calculate core metrics
-#         metrics = {
-#             'response_diversity': self.calculate_diversity(responses),
-#             'query_response_relevance': self.calculate_relevance(query, responses),
-#             'response_length_score': np.mean([
-#                 self._calculate_length_score(response) for response, _ in responses
-#             ]),
-#             'top_score': responses[0][1],
-#             'top_3_score_gap': self._calculate_score_gap([score for _, score in responses], top_n=3)
-#         }
-
-#         # Determine confidence using thresholds
-#         metrics['is_confident'] = self._determine_confidence(metrics)
-
-#         logger.info(f"Quality metrics: {metrics}")
-#         return metrics
-
-#     def calculate_relevance(self, query: str, responses: List[Tuple[str, float]]) -> float:
-#         """Calculate relevance with stricter scoring."""
-#         if not responses:
-#             return 0.0
-
-#         query_embedding = self.data_pipeline.encode_query(query)
-#         response_texts = [resp for resp, _ in responses]
-#         response_embeddings = self.data_pipeline.encode_responses(response_texts)
-
-#         # Normalize embeddings
-#         query_embedding = query_embedding / np.linalg.norm(query_embedding)
-#         response_embeddings = response_embeddings / np.linalg.norm(response_embeddings, axis=1)[:, np.newaxis]
-
-#         # Compute similarities with exponential decay for far matches
-#         similarities = cosine_similarity([query_embedding], response_embeddings)[0]
-#         similarities = np.exp(similarities - 1)  # Penalize lower similarities more strongly
-
-#         # Apply stronger position weighting
-#         weights = np.exp(-np.arange(len(similarities)) / 2)
+    
+    def _determine_confidence(self, m: Dict[str, float]) -> bool:
+        """
+        Require:
+         - top_score >= self.confidence_threshold
+         - response_diversity >= self.diversity_threshold
+         - response_length_score >= self.thresholds['length_score']
+         
+        Secondary conditions (2 of 3 required):
+         - query_response_relevance >= self.thresholds['relevance']
+         - top_3_score_gap >= self.thresholds['score_gap']
+         - top_score >= (confidence_threshold + 0.05)
+        """
+        primary = [
+            m['top_score'] >= self.confidence_threshold,
+            m['response_diversity'] >= self.diversity_threshold,
+            m['response_length_score'] >= self.thresholds['length_score']
+        ]
+        secondary = [
+            m['query_response_relevance'] >= self.thresholds['relevance'],
+            m['top_3_score_gap'] >= self.thresholds['score_gap'],
+            m['top_score'] >= (self.confidence_threshold + 0.05)
+        ]
         
-#         return float(np.average(similarities, weights=weights))
-
-#     def calculate_diversity(self, responses: List[Tuple[str, float]]) -> float:
-#         """Calculate diversity with length normalization and similarity capping."""
-#         if not responses:
-#             return 0.0
-
-#         response_texts = [resp for resp, _ in responses]
-#         embeddings = self.data_pipeline.encode_responses(response_texts)
-#         if len(embeddings) < 2:
-#             return 1.0
-
-#         # Calculate pairwise cosine similarities
-#         similarity_matrix = cosine_similarity(embeddings)
-#         np.fill_diagonal(similarity_matrix, 0)  # Exclude self-similarity
-
-#         # Apply similarity cap
-#         similarity_matrix = np.minimum(similarity_matrix, self.similarity_cap)
-
-#         # Calculate average similarity
-#         sum_similarities = np.sum(similarity_matrix)
-#         num_pairs = len(embeddings) * (len(embeddings) - 1)
-#         avg_similarity = sum_similarities / num_pairs if num_pairs > 0 else 0.0
-
-#         # Diversity is inversely related to average similarity
-#         diversity_score = 1 - avg_similarity
-#         return diversity_score
-
-#     def _determine_confidence(self, metrics: Dict[str, float]) -> bool:
-#         """Determine confidence using primary and secondary conditions."""
-#         # Primary conditions (must all be met)
-#         primary_conditions = [
-#             metrics['top_score'] >= self.confidence_threshold,
-#             metrics['response_diversity'] >= self.diversity_threshold,
-#             metrics['response_length_score'] >= self.thresholds['length_score']
-#         ]
-
-#         # Secondary conditions (majority must be met)
-#         secondary_conditions = [
-#             metrics['query_response_relevance'] >= self.thresholds['relevance'],
-#             metrics['top_3_score_gap'] >= self.thresholds['score_gap'],
-#             metrics['top_score'] >= (self.confidence_threshold * 1.1)  # Extra confidence boost
-#         ]
-
-#         return all(primary_conditions) and sum(secondary_conditions) >= 2
-
-#     def _calculate_length_score(self, response: str) -> float:
-#         """Calculate length score with penalty for very short or long responses."""
-#         words = len(response.split())
-
-#         if words < self.min_response_length:
-#             return words / self.min_response_length
-#         elif words > 50:  # Penalty for very long responses
-#             return min(1.0, 50 / words)
-#         return 1.0
-
-#     def _calculate_score_gap(self, scores: List[float], top_n: int = 3) -> float:
-#         """Calculate average gap between top N scores."""
-#         if len(scores) < top_n + 1:
-#             return 0.0
-#         gaps = [scores[i] - scores[i + 1] for i in range(min(len(scores) - 1, top_n))]
-#         return np.mean(gaps)
+        if all(primary) and sum(secondary) >= 2:
+            return True
+        return False
+    
+    def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> np.ndarray:
+        """Manual cosine sim matrix: a-> shape [N, d], b-> shape [M, d]. Return shape [N, M]."""
+        a_norm = a / (np.linalg.norm(a, axis=1, keepdims=True) + 1e-8)
+        b_norm = b / (np.linalg.norm(b, axis=1, keepdims=True) + 1e-8)
+        return np.dot(a_norm, b_norm.T)
+