use common custom llm call

Author: asamal4

src/lightspeed_evaluation/core/llm/ragas.py

@@ -2,19 +2,20 @@
 
 from typing import Any, Optional
 
-import litellm
 from ragas.llms.base import BaseRagasLLM, Generation, LLMResult
 from ragas.metrics import answer_relevancy, faithfulness
 
+from lightspeed_evaluation.core.llm.custom import BaseCustomLLM
+from lightspeed_evaluation.core.system.exceptions import LLMError
 
-class RagasCustomLLM(BaseRagasLLM):
+
+class RagasCustomLLM(BaseRagasLLM, BaseCustomLLM):
     """Custom LLM for Ragas using LiteLLM parameters."""
 
     def __init__(self, model_name: str, litellm_params: dict[str, Any]):
         """Initialize Ragas custom LLM with model name and LiteLLM parameters."""
-        super().__init__()
-        self.model_name = model_name
-        self.litellm_params = litellm_params
+        BaseRagasLLM.__init__(self)
+        BaseCustomLLM.__init__(self, model_name, litellm_params)
         print(f"✅ Ragas Custom LLM: {self.model_name}")
 
     def generate_text(  # pylint: disable=too-many-arguments,too-many-positional-arguments
@@ -36,31 +37,27 @@ def generate_text(  # pylint: disable=too-many-arguments,too-many-positional-arg
         )
 
         try:
-            response = litellm.completion(
-                model=self.model_name,
-                messages=[{"role": "user", "content": prompt_text}],
-                n=n,
-                temperature=temp,
-                max_tokens=self.litellm_params.get("max_tokens"),
-                timeout=self.litellm_params.get("timeout"),
-                num_retries=self.litellm_params.get("num_retries"),
+            # Use inherited BaseCustomLLM functionality
+            call_kwargs = {}
+            if stop is not None:
+                call_kwargs["stop"] = stop
+
+            responses = self.call(
+                prompt_text, n=n, temperature=temp, return_single=False, **call_kwargs
             )
 
             # Convert to Ragas format
             generations = []
-            for choice in response.choices:  # type: ignore
-                content = choice.message.content  # type: ignore
-                if content is None:
-                    content = ""
-                gen = Generation(text=content.strip())
+            for response_text in responses:
+                gen = Generation(text=response_text)
                 generations.append(gen)
 
             result = LLMResult(generations=[generations])
             return result
 
         except Exception as e:
             print(f"❌ Ragas LLM failed: {e}")
-            raise RuntimeError(f"Ragas LLM evaluation failed: {str(e)}") from e
+            raise LLMError(f"Ragas LLM evaluation failed: {str(e)}") from e
 
     async def agenerate_text(  # pylint: disable=too-many-arguments,too-many-positional-arguments
         self,

src/lightspeed_evaluation/core/metrics/custom.py

@@ -3,12 +3,13 @@
 import re
 from typing import Any, Optional
 
-import litellm
 from pydantic import BaseModel, Field
 
+from lightspeed_evaluation.core.llm.custom import BaseCustomLLM
 from lightspeed_evaluation.core.llm.manager import LLMManager
 from lightspeed_evaluation.core.metrics.tool_eval import evaluate_tool_calls
 from lightspeed_evaluation.core.models import EvaluationScope, TurnData
+from lightspeed_evaluation.core.system.exceptions import LLMError
 
 
 class EvaluationPromptParams(BaseModel):
@@ -35,15 +36,16 @@ def __init__(self, llm_manager: LLMManager):
         Args:
             llm_manager: Pre-configured LLMManager with validated parameters
         """
-        self.model_name = llm_manager.get_model_name()
-        self.litellm_params = llm_manager.get_litellm_params()
+        self.llm = BaseCustomLLM(
+            llm_manager.get_model_name(), llm_manager.get_litellm_params()
+        )
 
         self.supported_metrics = {
             "answer_correctness": self._evaluate_answer_correctness,
             "tool_eval": self._evaluate_tool_calls,
         }
 
-        print(f"✅ Custom Metrics initialized: {self.model_name}")
+        print(f"✅ Custom Metrics initialized: {self.llm.model_name}")
 
     def evaluate(
         self,
@@ -62,31 +64,12 @@ def evaluate(
         except (ValueError, AttributeError, KeyError) as e:
             return None, f"Custom {metric_name} evaluation failed: {str(e)}"
 
-    def _call_llm(self, prompt: str, system_prompt: Optional[str] = None) -> str:
-        """Make a LiteLLM call with the configured parameters."""
-        # Prepare messages
-        messages = []
-        if system_prompt:
-            messages.append({"role": "system", "content": system_prompt})
-        messages.append({"role": "user", "content": prompt})
-
-        try:
-            response = litellm.completion(
-                model=self.model_name,
-                messages=messages,
-                temperature=self.litellm_params.get("temperature", 0.0),
-                max_tokens=self.litellm_params.get("max_tokens"),
-                timeout=self.litellm_params.get("timeout"),
-                num_retries=self.litellm_params.get("num_retries", 3),
-            )
-
-            content = response.choices[0].message.content  # type: ignore
-            if content is None:
-                raise RuntimeError("LLM returned empty response")
-            return content.strip()
-
-        except Exception as e:
-            raise RuntimeError(f"LiteLLM call failed: {str(e)}") from e
+    def _call_llm(self, prompt: str) -> str:
+        """Make an LLM call with the configured parameters."""
+        result = self.llm.call(prompt, return_single=True)
+        if isinstance(result, list):
+            return result[0] if result else ""
+        return result
 
     def _parse_score_response(self, response: str) -> tuple[Optional[float], str]:
         r"""Parse LLM response to extract score and reason.
@@ -232,16 +215,19 @@ def _evaluate_answer_correctness(
         prompt += "- Absence of contradictory information"
 
         # Make LLM call and parse response
-        llm_response = self._call_llm(prompt)
-        score, reason = self._parse_score_response(llm_response)
-
-        if score is None:
-            return (
-                None,
-                f"Could not parse score from LLM response: {llm_response[:100]}...",
-            )
-
-        return score, f"Custom answer correctness: {score:.2f} - {reason}"
+        try:
+            llm_response = self._call_llm(prompt)
+            score, reason = self._parse_score_response(llm_response)
+
+            if score is None:
+                return (
+                    None,
+                    f"Could not parse score from LLM response: {llm_response[:100]}...",
+                )
+
+            return score, f"Custom answer correctness: {score:.2f} - {reason}"
+        except LLMError as e:
+            return None, f"Answer correctness evaluation failed: {str(e)}"
 
     def _evaluate_tool_calls(
         self,