process default & turn-level override

Author: asamal4

src/lightspeed_evaluation/core/metrics/manager.py

@@ -0,0 +1,136 @@
+"""Metrics mapping for evaluation."""
+
+from enum import Enum
+from typing import Any, Optional
+
+from ..models.data import EvaluationData, TurnData
+from ..models.system import SystemConfig
+
+
+class MetricLevel(Enum):
+    """Metric level enumeration."""
+
+    TURN = "turn"
+    CONVERSATION = "conversation"
+
+
+class MetricManager:
+    """Manager for both turn and conversation metrics."""
+
+    def __init__(self, system_config: SystemConfig):
+        """Initialize with system configuration."""
+        self.system_config = system_config
+
+    def resolve_metrics(
+        self, metrics: Optional[list[str]], level: MetricLevel
+    ) -> list[str]:
+        """Resolve metrics mapping.
+
+        Options:
+        - None: use system defaults (metrics with default=true)
+        - []: skip evaluation completely
+        - [metrics...]: use specified metrics from turn data
+
+        Args:
+            metrics: The metrics configuration (None, [], or list of metrics)
+            level: Whether this is TURN or CONVERSATION level
+
+        Returns:
+            List of metrics to evaluate
+        """
+        if metrics is None:
+            # None = use system defaults
+            return self._extract_default_metrics(level)
+        if metrics == []:
+            # [] = explicitly skip evaluation
+            return []
+        # Use specified metrics as-is
+        return metrics
+
+    def get_effective_threshold(
+        self,
+        metric_identifier: str,
+        level: MetricLevel,
+        conv_data: Optional[EvaluationData] = None,
+        turn_data: Optional[TurnData] = None,
+    ) -> Optional[float]:
+        """Get effective threshold with priority hierarchy.
+
+        Priority:
+        1. Level-specific metadata (turn-specific for turns, conversation-specific for convs)
+        2. System defaults
+
+        Args:
+            metric_identifier: The metric to get threshold for
+            level: Whether this is TURN or CONVERSATION level
+            conv_data: Conversation data for conversation-level metadata
+            turn_data: Turn data for turn-specific metadata
+
+        Returns:
+            Effective threshold or None if not found
+        """
+        # Check level-specific metadata first
+        level_metadata = self._get_level_metadata(level, conv_data, turn_data)
+        threshold = level_metadata.get(metric_identifier, {}).get("threshold")
+        if threshold is not None:
+            return threshold
+
+        # Fall back to system defaults
+        system_metadata = self._get_system_metadata(level)
+        return system_metadata.get(metric_identifier, {}).get("threshold")
+
+    def _get_level_metadata(
+        self,
+        level: MetricLevel,
+        conv_data: Optional[EvaluationData],
+        turn_data: Optional[TurnData],
+    ) -> dict[str, dict[str, Any]]:
+        """Get level-specific metadata (turn or conversation level)."""
+        if level == MetricLevel.TURN and turn_data and turn_data.turn_metrics_metadata:
+            return turn_data.turn_metrics_metadata
+        if (
+            level == MetricLevel.CONVERSATION
+            and conv_data
+            and conv_data.conversation_metrics_metadata
+        ):
+            return conv_data.conversation_metrics_metadata
+        return {}
+
+    def _get_system_metadata(self, level: MetricLevel) -> dict[str, dict[str, Any]]:
+        """Get system-level metadata for the given level."""
+        if level == MetricLevel.TURN:
+            return self.system_config.default_turn_metrics_metadata
+        return self.system_config.default_conversation_metrics_metadata
+
+    def _extract_default_metrics(self, level: MetricLevel) -> list[str]:
+        """Extract metrics that have default=true from metadata."""
+        metrics_metadata = self._get_system_metadata(level)
+
+        default_metrics = []
+        for metric_name, metadata in metrics_metadata.items():
+            if metadata.get("default", False):  # default=false if not specified
+                default_metrics.append(metric_name)
+        return default_metrics
+
+    def count_metrics_for_conversation(
+        self, conv_data: EvaluationData
+    ) -> dict[str, int]:
+        """Count total metrics that would be evaluated for a conversation."""
+        # Count turn metrics
+        total_turn_metrics = 0
+        for turn_data in conv_data.turns:
+            turn_metrics = self.resolve_metrics(
+                turn_data.turn_metrics, MetricLevel.TURN
+            )
+            total_turn_metrics += len(turn_metrics)
+
+        # Count conversation metrics
+        conversation_metrics = self.resolve_metrics(
+            conv_data.conversation_metrics, MetricLevel.CONVERSATION
+        )
+
+        return {
+            "turn_metrics": total_turn_metrics,
+            "conversation_metrics": len(conversation_metrics),
+            "total_turns": len(conv_data.turns),
+        }

src/lightspeed_evaluation/pipeline/evaluation/errors.py

@@ -30,9 +30,9 @@ def mark_all_metrics_as_error(
         error_results = []
 
         # Mark all turn-level metrics as ERROR
-        if conv_data.turn_metrics:
-            for turn_data in conv_data.turns:
-                for metric_identifier in conv_data.turn_metrics:
+        for turn_data in conv_data.turns:
+            if turn_data.turn_metrics:
+                for metric_identifier in turn_data.turn_metrics:
                     error_result = EvaluationResult(
                         conversation_group_id=conv_data.conversation_group_id,
                         turn_id=turn_data.turn_id,

src/lightspeed_evaluation/pipeline/evaluation/evaluator.py

@@ -7,13 +7,9 @@
 from ...core.llm.manager import LLMManager
 from ...core.metrics.custom import CustomMetrics
 from ...core.metrics.deepeval import DeepEvalMetrics
+from ...core.metrics.manager import MetricLevel, MetricManager
 from ...core.metrics.ragas import RagasMetrics
-from ...core.models import (
-    EvaluationData,
-    EvaluationRequest,
-    EvaluationResult,
-    EvaluationScope,
-)
+from ...core.models import EvaluationRequest, EvaluationResult, EvaluationScope
 from ...core.system import ConfigLoader
 
 logger = logging.getLogger(__name__)
@@ -22,8 +18,13 @@
 class MetricsEvaluator:
     """Handles individual metric evaluation with proper scoring and status determination."""
 
-    def __init__(self, llm_manager: LLMManager, config_loader: ConfigLoader) -> None:
-        """Initialize with LLM manager and config."""
+    def __init__(
+        self,
+        llm_manager: LLMManager,
+        config_loader: ConfigLoader,
+        metric_manager: MetricManager,
+    ) -> None:
+        """Initialize with LLM manager, config, and metric manager."""
         self.config_loader = config_loader
         self.config = config_loader.system_config
 
@@ -39,6 +40,8 @@ def __init__(self, llm_manager: LLMManager, config_loader: ConfigLoader) -> None
             "custom": self.custom_metrics,
         }
 
+        self.metric_manager = metric_manager
+
     def evaluate_metric(self, request: EvaluationRequest) -> Optional[EvaluationResult]:
         """Evaluate a single metric and return result."""
         start_time = time.time()
@@ -81,9 +84,14 @@ def evaluate_metric(self, request: EvaluationRequest) -> Optional[EvaluationResu
             if score is None:
                 return self._create_error_result(request, reason, execution_time)
 
-            # Get threshold and determine status
-            threshold = self._get_effective_threshold(
-                request.conv_data, request.metric_identifier, request.is_conversation
+            # Get threshold
+            level = (
+                MetricLevel.CONVERSATION
+                if request.is_conversation
+                else MetricLevel.TURN
+            )
+            threshold = self.metric_manager.get_effective_threshold(
+                request.metric_identifier, level, request.conv_data, request.turn_data
             )
             status = self._determine_status(score, threshold)
 
@@ -124,37 +132,6 @@ def _create_error_result(
             execution_time=execution_time,
         )
 
-    def _get_effective_threshold(
-        self, conv_data: EvaluationData, metric_identifier: str, is_conversation: bool
-    ) -> Optional[float]:
-        """Get effective threshold for metric (conversation-specific or system default)."""
-        # Check conversation-specific metadata first
-        if is_conversation:
-            metadata = (conv_data.conversation_metrics_metadata or {}).get(
-                metric_identifier, {}
-            )
-        else:
-            metadata = (conv_data.turn_metrics_metadata or {}).get(
-                metric_identifier, {}
-            )
-
-        if "threshold" in metadata:
-            return metadata["threshold"]
-
-        # Fall back to system defaults
-        if self.config is None:
-            raise ValueError("SystemConfig must be loaded")
-        if is_conversation:
-            default_metadata = (
-                self.config.default_conversation_metrics_metadata or {}
-            ).get(metric_identifier, {})
-        else:
-            default_metadata = (self.config.default_turn_metrics_metadata or {}).get(
-                metric_identifier, {}
-            )
-
-        return default_metadata.get("threshold")
-
     def _determine_status(self, score: float, threshold: Optional[float]) -> str:
         """Determine evaluation status based on score and threshold."""
         if threshold is None:

src/lightspeed_evaluation/pipeline/evaluation/pipeline.py

@@ -5,13 +5,14 @@
 
 from ...core.api import APIClient
 from ...core.llm.manager import LLMManager
+from ...core.metrics.manager import MetricManager
 from ...core.models import EvaluationData, EvaluationResult
 from ...core.output.data_persistence import save_evaluation_data
 from ...core.system import ConfigLoader, DataValidator
 from .amender import APIDataAmender
 from .errors import EvaluationErrorHandler
 from .evaluator import MetricsEvaluator
-from .processor import ConversationProcessor
+from .processor import ConversationProcessor, ProcessorComponents
 
 logger = logging.getLogger(__name__)
 
@@ -52,25 +53,29 @@ def _initialize_components(self) -> None:
         # LLM Manager
         llm_manager = LLMManager.from_llm_config(self.config.llm)
 
+        # Metric manager
+        metric_manager = MetricManager(self.config)
+
         # Create pipeline components
-        api_client = self._create_api_client()
-        api_amender = APIDataAmender(api_client)
+        self.api_client = self._create_api_client()
+        api_amender = APIDataAmender(self.api_client)
         error_handler = EvaluationErrorHandler()
-        metrics_evaluator = MetricsEvaluator(llm_manager, self.config_loader)
-        # Group components for easier access
-        self.components = {
-            "api_client": api_client,
-            "api_amender": api_amender,
-            "error_handler": error_handler,
-            "metrics_evaluator": metrics_evaluator,
-        }
+        metrics_evaluator = MetricsEvaluator(
+            llm_manager, self.config_loader, metric_manager
+        )
+
+        # Create processor components
+        processor_components = ProcessorComponents(
+            metrics_evaluator=metrics_evaluator,
+            api_amender=api_amender,
+            error_handler=error_handler,
+            metric_manager=metric_manager,
+        )
 
         # Conversation processor
         self.conversation_processor = ConversationProcessor(
             self.config_loader,
-            self.components["metrics_evaluator"],
-            self.components["api_amender"],
-            self.components["error_handler"],
+            processor_components,
         )
 
     def _create_api_client(self) -> Optional[APIClient]:
@@ -163,6 +168,5 @@ def _save_updated_data(self, evaluation_data: list[EvaluationData]) -> None:
 
     def close(self) -> None:
         """Clean up resources."""
-        api_client = self.components.get("api_client")
-        if api_client:
-            api_client.close()
+        if self.api_client:
+            self.api_client.close()