fixes

Author: benjibc

eval_protocol/adapters/huggingface.py

@@ -188,12 +188,12 @@ def get_evaluation_rows(
             return
 
         # Create completion parameters
-        completion_params = CompletionParams(
-            model=model_name,
-            temperature=temperature,
-            max_tokens=max_tokens,
+        completion_params: CompletionParams = {
+            "model": model_name,
+            "temperature": temperature,
+            "max_tokens": max_tokens,
             **completion_params_kwargs,
-        )
+        }
 
         # Convert each row
         for i in range(offset, end_idx):

eval_protocol/adapters/langfuse.py

@@ -8,7 +8,7 @@
 from datetime import datetime
 from typing import Any, Dict, Iterator, List, Optional
 
-from eval_protocol.models import CompletionParams, EvaluationRow, InputMetadata, Message
+from eval_protocol.models import EvaluationRow, InputMetadata, Message
 
 logger = logging.getLogger(__name__)
 
@@ -277,20 +277,20 @@ def _create_input_metadata(self, trace: Any, observations: List[Any]) -> InputMe
             InputMetadata object
         """
         # Extract completion parameters from observations
-        completion_params = CompletionParams()
+        completion_params = {}
 
         # Look for model parameters in observations
         for obs in observations:
             if hasattr(obs, "model") and obs.model:
-                completion_params.model = obs.model
+                completion_params["model"] = obs.model
             if hasattr(obs, "model_parameters") and obs.model_parameters:
                 params = obs.model_parameters
                 if "temperature" in params:
-                    completion_params.temperature = params["temperature"]
+                    completion_params["temperature"] = params["temperature"]
                 if "max_tokens" in params:
-                    completion_params.max_tokens = params["max_tokens"]
+                    completion_params["max_tokens"] = params["max_tokens"]
                 if "top_p" in params:
-                    completion_params.top_p = params["top_p"]
+                    completion_params["top_p"] = params["top_p"]
                 break
 
         # Create dataset info from trace metadata

eval_protocol/benchmarks/suites/gpqa.py

@@ -1,3 +1,4 @@
+import asyncio
 import csv
 import io
 import re
@@ -8,9 +9,11 @@
 from eval_protocol.benchmarks.registry import export_benchmark
 from eval_protocol.models import EvaluateResult, EvaluationRow, Message, MetricResult
 from eval_protocol.pytest.default_single_turn_rollout_process import (
-    default_single_turn_rollout_processor,
+    SingleTurnRolloutProcessor,
 )
 from eval_protocol.pytest.evaluation_test import evaluation_test
+from eval_protocol.pytest.rollout_processor import RolloutProcessor
+from eval_protocol.pytest.types import RolloutProcessorConfig
 
 SYSTEM_PROMPT = (
     "You are a helpful assistant. Read the question and options carefully. "
@@ -60,27 +63,40 @@ def _strip_gt_messages(msgs: List[Message]) -> List[Message]:
     return [m for m in msgs if not (m.role == "system" and (m.content or "").startswith("__GT__:"))]
 
 
-async def gpqa_strip_gt_rollout_processor(rows: List[EvaluationRow], config) -> List[EvaluationRow]:
-    """Preprocess rows to set ground_truth and remove __GT__ messages, then delegate to default processor."""
-    processed: List[EvaluationRow] = []
-    for r in rows:
-        gt_tokens = [m.content for m in r.messages if m.role == "system" and (m.content or "").startswith("__GT__:")]
-        if gt_tokens:
-            gt_val = gt_tokens[-1].split(":", 1)[1].strip()
-            r.ground_truth = gt_val
-            r.messages = [
-                m for m in r.messages if not (m.role == "system" and (m.content or "").startswith("__GT__:"))
+class GPQAStripGTRolloutProcessor(RolloutProcessor):
+    """Preprocess rows to set ground_truth and remove __GT__ messages, then delegate to SingleTurnRolloutProcessor."""
+
+    def __init__(self):
+        super().__init__()
+        self.single_turn_processor = SingleTurnRolloutProcessor()
+
+    def __call__(self, rows: List[EvaluationRow], config: RolloutProcessorConfig) -> List[asyncio.Task[EvaluationRow]]:
+        """Preprocess rows and delegate to SingleTurnRolloutProcessor."""
+        processed: List[EvaluationRow] = []
+
+        for r in rows:
+            gt_tokens = [
+                m.content for m in r.messages if m.role == "system" and (m.content or "").startswith("__GT__:")
             ]
-        processed.append(r)
-    return await default_single_turn_rollout_processor(processed, config)
+            if gt_tokens:
+                gt_val = gt_tokens[-1].split(":", 1)[1].strip()
+                r.ground_truth = gt_val
+                r.messages = [
+                    m for m in r.messages if not (m.role == "system" and (m.content or "").startswith("__GT__:"))
+                ]
+            processed.append(r)
+
+        # Delegate to SingleTurnRolloutProcessor
+        return self.single_turn_processor(processed, config)
 
 
 @export_benchmark("gpqa")
 @evaluation_test(
-    model=["fireworks_ai/accounts/fireworks/models/gpt-oss-120b"],
     input_messages=_GPQA_INPUT_MESSAGES,
-    rollout_input_params=[{"extra_body": {"reasoning_effort": "low"}}],
-    rollout_processor=gpqa_strip_gt_rollout_processor,
+    completion_params=[
+        {"extra_body": {"reasoning_effort": "low"}, "model": "fireworks_ai/accounts/fireworks/models/gpt-oss-120b"}
+    ],
+    rollout_processor=GPQAStripGTRolloutProcessor(),
     aggregation_method="mean",
     passed_threshold=None,
     num_runs=8,

eval_protocol/dataset_logger/__init__.py

@@ -1,17 +1,32 @@
 import os
 
+from eval_protocol.dataset_logger.dataset_logger import DatasetLogger
 from eval_protocol.dataset_logger.sqlite_dataset_logger_adapter import SqliteDatasetLoggerAdapter
 
+
 # Allow disabling sqlite logger to avoid environment-specific constraints in simple CLI runs.
-if os.getenv("EP_SQLITE_LOG", "0").strip() == "1":
-    default_logger = SqliteDatasetLoggerAdapter()
-else:
+def _get_default_logger():
+    if os.getenv("DISABLE_EP_SQLITE_LOG", "0").strip() != "1":
+        return SqliteDatasetLoggerAdapter()
+    else:
+
+        class _NoOpLogger(DatasetLogger):
+            def log(self, row):
+                return None
+
+            def read(self, rollout_id=None):
+                return []
+
+        return _NoOpLogger()
+
+
+# Lazy property that creates the logger only when accessed
+class _LazyLogger(DatasetLogger):
+    def log(self, row):
+        return _get_default_logger().log(row)
 
-    class _NoOpLogger:
-        def log(self, row):
-            return None
+    def read(self, rollout_id=None):
+        return _get_default_logger().read(rollout_id)
 
-        def read(self, rollout_id=None):
-            return []
 
-    default_logger = _NoOpLogger()
+default_logger: DatasetLogger = _LazyLogger()

eval_protocol/pytest/default_single_turn_rollout_process.py

@@ -2,129 +2,117 @@
 import logging
 import os
 import time
-from typing import AsyncIterator, List
+from typing import List
 
-import litellm
 from litellm import acompletion
 from openai.types.chat.chat_completion_message import ChatCompletionMessageToolCall
 
 from eval_protocol.dataset_logger import default_logger
 from eval_protocol.models import EvaluationRow, Message
+from eval_protocol.pytest.rollout_processor import RolloutProcessor
 from eval_protocol.pytest.types import RolloutProcessorConfig
 
 logger = logging.getLogger(__name__)
 
 
-async def default_single_turn_rollout_processor(
-    rows: List[EvaluationRow], config: RolloutProcessorConfig
-) -> AsyncIterator[EvaluationRow]:
-    """Generate a single response from any supported model provider using LiteLLM."""
-
-    # Quiet LiteLLM logs in test runs unless user overrode
-    try:
-        if os.environ.get("LITELLM_LOG") is None:
-            os.environ["LITELLM_LOG"] = "ERROR"
-        _llog = logging.getLogger("LiteLLM")
-        _llog.setLevel(logging.CRITICAL)
-        _llog.propagate = False
-        for _h in list(_llog.handlers):
-            _llog.removeHandler(_h)
-    except Exception:
-        pass
-
-    # Do not modify global LiteLLM cache. Disable caching per-request instead.
-
-    async def process_row(row: EvaluationRow) -> EvaluationRow:
-        """Process a single row asynchronously."""
-        if len(row.messages) == 0:
-            raise ValueError("Messages is empty. Please provide a non-empty dataset")
-
-        messages_payload = [{"role": m.role, "content": m.content} for m in row.messages]
-
-        request_params = {"model": config.model, "messages": messages_payload, **config.input_params}
-        # Ensure caching is disabled only for this request (review feedback)
-        request_params["cache"] = {"no-cache": True}
-        # Single-level reasoning effort: expect `reasoning_effort` only
-        effort_val = None
-        if isinstance(config.input_params, dict):
-            if "reasoning_effort" in config.input_params:
-                effort_val = str(config.input_params["reasoning_effort"])  # flat shape
+class SingleTurnRolloutProcessor(RolloutProcessor):
+    """Single turn rollout processor for direct LLM calls."""
+
+    def __call__(self, rows: List[EvaluationRow], config: RolloutProcessorConfig) -> List[asyncio.Task[EvaluationRow]]:
+        """Generate single turn rollout tasks and return them for external handling."""
+
+        # Quiet LiteLLM logs in test runs unless user overrode
+        try:
+            if os.environ.get("LITELLM_LOG") is None:
+                os.environ["LITELLM_LOG"] = "ERROR"
+            _llog = logging.getLogger("LiteLLM")
+            _llog.setLevel(logging.CRITICAL)
+            _llog.propagate = False
+            for _h in list(_llog.handlers):
+                _llog.removeHandler(_h)
+        except Exception:
+            pass
+
+        # Do not modify global LiteLLM cache. Disable caching per-request instead.
+
+        async def process_row(row: EvaluationRow) -> EvaluationRow:
+            """Process a single row asynchronously."""
+            if len(row.messages) == 0:
+                raise ValueError("Messages is empty. Please provide a non-empty dataset")
+
+            messages_payload = [{"role": m.role, "content": m.content} for m in row.messages]
+
+            request_params = {"messages": messages_payload, **config.completion_params}
+            # Ensure caching is disabled only for this request (review feedback)
+            request_params["cache"] = {"no-cache": True}
+            # Single-level reasoning effort: expect `reasoning_effort` only
+            effort_val = None
+
+            if "reasoning_effort" in config.completion_params:
+                effort_val = str(config.completion_params["reasoning_effort"])  # flat shape
             elif (
-                isinstance(config.input_params.get("extra_body"), dict)
-                and "reasoning_effort" in config.input_params["extra_body"]
+                isinstance(config.completion_params.get("extra_body"), dict)
+                and "reasoning_effort" in config.completion_params["extra_body"]
             ):
                 # Accept if user passed it directly inside extra_body
-                effort_val = str(config.input_params["extra_body"]["reasoning_effort"])  # already in extra_body
-
-        if effort_val:
-            # Always under extra_body so LiteLLM forwards to provider-specific param set
-            request_params.setdefault("extra_body", {})
-            request_params["extra_body"]["reasoning_effort"] = effort_val
-            # Ensure unsupported top-level keys are not present
-            if "reasoning_effort" in request_params:
-                request_params.pop("reasoning_effort", None)
-
-        if row.tools is not None:
-            request_params["tools"] = row.tools
-
-        # Dynamic import to avoid static dependency/lint errors if LiteLLM isn't installed yet
-        import importlib
-
-        _litellm = importlib.import_module("litellm")
-        acompletion = getattr(_litellm, "acompletion")
-        response = await acompletion(**request_params)
-
-        assistant_content = response.choices[0].message.content or ""
-        tool_calls = response.choices[0].message.tool_calls if response.choices[0].message.tool_calls else None
-
-        converted_tool_calls = None
-        if tool_calls:
-            converted_tool_calls = [
-                ChatCompletionMessageToolCall(
-                    id=tool_call.id,
-                    type=tool_call.type,
-                    function={
-                        "name": tool_call.function.name,
-                        "arguments": tool_call.function.arguments,
-                    },
+                effort_val = str(config.completion_params["extra_body"]["reasoning_effort"])  # already in extra_body
+
+            if effort_val:
+                # Always under extra_body so LiteLLM forwards to provider-specific param set
+                request_params.setdefault("extra_body", {})
+                request_params["extra_body"]["reasoning_effort"] = effort_val
+                # Ensure unsupported top-level keys are not present
+                if "reasoning_effort" in request_params:
+                    request_params.pop("reasoning_effort", None)
+
+            if row.tools is not None:
+                request_params["tools"] = row.tools
+
+            # Dynamic import to avoid static dependency/lint errors if LiteLLM isn't installed yet
+            import importlib
+
+            _litellm = importlib.import_module("litellm")
+            acompletion = getattr(_litellm, "acompletion")
+            response = await acompletion(**request_params)
+
+            assistant_content = response.choices[0].message.content or ""
+            tool_calls = response.choices[0].message.tool_calls if response.choices[0].message.tool_calls else None
+
+            converted_tool_calls = None
+            if tool_calls:
+                converted_tool_calls = [
+                    ChatCompletionMessageToolCall(
+                        id=tool_call.id,
+                        type=tool_call.type,
+                        function={
+                            "name": tool_call.function.name,
+                            "arguments": tool_call.function.arguments,
+                        },
+                    )
+                    for tool_call in tool_calls
+                ]
+
+            messages = list(row.messages) + [
+                Message(
+                    role="assistant",
+                    content=assistant_content,
+                    tool_calls=converted_tool_calls,
                 )
-                for tool_call in tool_calls
             ]
 
-        messages = list(row.messages) + [
-            Message(
-                role="assistant",
-                content=assistant_content,
-                tool_calls=converted_tool_calls,
-            )
-        ]
-
-        row.messages = messages
-        default_logger.log(row)
-        return row
-
-    # Process rows with bounded concurrency and yield as they complete
-    max_concurrent = getattr(config, "max_concurrent_rollouts", 8) or 8
-    semaphore = asyncio.Semaphore(max_concurrent)
-
-    async def _sem_wrapper(r: EvaluationRow) -> EvaluationRow:
-        async with semaphore:
-            try:
-                return await process_row(r)
-            except Exception:
-                return r
-
-    # Create all tasks
-    tasks = [asyncio.create_task(_sem_wrapper(row)) for row in rows]
-
-    # Yield results as they complete (note that they're not necessarily in original order)
-    try:
-        for task in asyncio.as_completed(tasks):
-            try:
-                yield await task
-            except Exception:
-                logger.exception("Error processing row")
-    finally:
-        for t in tasks:
-            t.cancel()
-        await asyncio.gather(*tasks, return_exceptions=True)
+            row.messages = messages
+            default_logger.log(row)
+            return row
+
+        # Process rows with bounded concurrency
+        max_concurrent = getattr(config, "max_concurrent_rollouts", 8) or 8
+        semaphore = asyncio.Semaphore(max_concurrent)
+
+        async def _sem_wrapper(r: EvaluationRow) -> EvaluationRow:
+            async with semaphore:
+                result = await process_row(r)
+                return result
+
+        # Create and return tasks for external handling
+        tasks = [asyncio.create_task(_sem_wrapper(row)) for row in rows]
+        return tasks