Merge branch 'awarno/pydantic-config-validation' of https://github.com/NVIDIA-NeMo/Evaluator into awarno/pydantic-config-validation

Author: AWarno

packages/nemo-evaluator-launcher/src/nemo_evaluator_launcher/package_info.py

@@ -16,7 +16,7 @@
 # Below is the _next_ version that will be published, not the currently published one.
 MAJOR = 0
 MINOR = 1
-PATCH = 31
+PATCH = 32
 PRE_RELEASE = ""
 
 # Use the following formatting: (major, minor, patch, pre-release)

packages/nemo-evaluator/pyproject.toml

@@ -65,7 +65,7 @@ repository = "https://github.com/NVIDIA-NeMo/Evaluator/packages/nemo-evaluator"
 # END(if-changed)
 
 [dependency-groups]
-test = ["pytest", "pytest-cov", "pytest-subtests", "pytest-httpserver", "nvidia-simple-evals"]
+test = ["pytest", "pytest-asyncio", "pytest-cov", "pytest-subtests", "pytest-httpserver", "nvidia-simple-evals"]
 
 docs = [
     "sphinx",

packages/nemo-evaluator/src/nemo_evaluator/adapters/adapter_config.py

@@ -207,36 +207,19 @@ class AdapterConfig(BaseModel):
         description="Type of the endpoint to run the adapter for",
         default="chat",
     )
-    caching_dir: str | None = Field(
-        description="Directory for caching responses (legacy field)",
-        default=None,
-    )
-    generate_html_report: bool = Field(
-        description="Whether to generate HTML report (legacy field)",
-        default=True,
-    )
     log_failed_requests: bool = Field(
-        description="Whether to log failed requests (legacy field)",
+        description="Whether to log failed requests",
         default=False,
     )
-    tracking_requests_stats: bool = Field(
-        description="Whether to enable request statistics tracking. When enabled, response statistics including token usage, status codes, finish reasons, tool calls, and latency metrics will be collected and added to eval_factory_metrics.json for comprehensive evaluation analysis.",
-        default=True,
-    )
-    html_report_size: int | None = Field(
-        description="Number of request-response pairs to track in HTML report. If this is larger than max_saved_responses or max_saved_requests, it will override those values.",
-        default=5,
-    )
 
     @classmethod
     def get_legacy_defaults(cls) -> dict[str, Any]:
         """Get default values for legacy configuration parameters."""
         return {
-            "generate_html_report": cls.model_fields["generate_html_report"].default,
-            "html_report_size": cls.model_fields["html_report_size"].default,
-            "tracking_requests_stats": cls.model_fields[
-                "tracking_requests_stats"
-            ].default,
+            "generate_html_report": True,
+            "html_report_size": 5,
+            "tracking_requests_stats": True,
+            "caching_dir": None,
             "log_failed_requests": cls.model_fields["log_failed_requests"].default,
             "endpoint_type": cls.model_fields["endpoint_type"].default,
             # Boolean defaults for optional features
@@ -254,7 +237,6 @@ def get_legacy_defaults(cls) -> dict[str, Any]:
             "use_raise_client_errors": False,
             "include_json": True,
             "custom_system_prompt": None,
-            "caching_dir": None,
             "output_dir": None,
             "params_to_add": None,
             "params_to_remove": None,
@@ -303,6 +285,24 @@ def merge_discovery(
             run_config.get("target", {}).get("api_endpoint", {}).get("adapter_config")
         )
 
+        # Validate that legacy parameters are not mixed with interceptors
+        legacy_defaults = cls.get_legacy_defaults()
+        model_fields = set(cls.model_fields.keys())
+        legacy_only_params = set(legacy_defaults.keys()) - model_fields
+
+        for config_name, config in [
+            ("global_adapter_config", global_cfg),
+            ("target.api_endpoint.adapter_config", local_cfg),
+        ]:
+            if config and config.get("interceptors"):
+                found_legacy = [p for p in legacy_only_params if p in config]
+                if found_legacy:
+                    raise ValueError(
+                        f"Cannot use legacy configuration parameters when interceptors are explicitly defined in {config_name}. "
+                        f"Found: {', '.join(sorted(found_legacy))}. "
+                        f"Please remove these and configure using interceptors instead."
+                    )
+
         if not global_cfg and not local_cfg:
             # Create default adapter config with caching enabled by default
             return cls.from_legacy_config({}, run_config)
@@ -746,11 +746,7 @@ def from_legacy_config(
             interceptors=interceptors,
             post_eval_hooks=post_eval_hooks,
             endpoint_type=legacy_config["endpoint_type"],
-            caching_dir=legacy_config["caching_dir"],
-            generate_html_report=legacy_config["generate_html_report"],
             log_failed_requests=legacy_config["log_failed_requests"],
-            tracking_requests_stats=legacy_config["tracking_requests_stats"],
-            html_report_size=legacy_config["html_report_size"],
         )
 
     def get_interceptor_configs(self) -> dict[str, dict[str, Any]]:

packages/nemo-evaluator/src/nemo_evaluator/adapters/interceptors/progress_tracking_interceptor.py

@@ -18,7 +18,8 @@
 import os
 import pathlib
 import threading
-from typing import Optional, final
+import time
+from typing import Annotated, Optional, final
 
 import requests
 from pydantic import Field
@@ -48,10 +49,16 @@ class Params(BaseLoggingParams):
             default="http://localhost:8000",
             description="URL to post the number of processed samples to. Supports expansion of shell variables if present.",
         )
-        progress_tracking_interval: int = Field(
+        progress_tracking_interval: Annotated[int, Field(gt=0)] = Field(
             default=1,
             description="How often (every how many samples) to send a progress information.",
         )
+        progress_tracking_interval_seconds: Optional[
+            Annotated[float | None, Field(gt=0)]
+        ] = Field(
+            default=None,
+            description="How often (every N seconds) to send a progress information in addition to progress_tracking_interval.",
+        )
         request_method: str = Field(
             default="PATCH",
             description="Request method to use for updating the evaluation progress.",
@@ -83,15 +90,30 @@ def __init__(self, params: Params):
         else:
             self.progress_filepath = None
         self._samples_processed = self._initialize_samples_processed()
+        self._last_updated_samples_processed = self._samples_processed
         self._lock = threading.Lock()
 
         # Get logger for this interceptor with interceptor context
         self.logger = get_logger(self.__class__.__name__)
 
+        # Optional update on timer
+        self.progress_tracking_interval_seconds = (
+            params.progress_tracking_interval_seconds
+        )
+        if self.progress_tracking_interval_seconds:
+            self._timer_stopped = False
+            self._update_on_timer_thread = threading.Thread(
+                target=self._update_on_timer,
+                kwargs={"interval_seconds": self.progress_tracking_interval_seconds},
+                daemon=True,
+            )
+            self._update_on_timer_thread.start()
+
         self.logger.info(
             "Progress tracking interceptor initialized",
             progress_tracking_url=self.progress_tracking_url,
             progress_tracking_interval=self.progress_tracking_interval,
+            progress_tracking_interval_seconds=self.progress_tracking_interval_seconds,
             output_dir=str(self.progress_filepath) if self.progress_filepath else None,
             initial_samples_processed=self._samples_processed,
         )
@@ -151,6 +173,34 @@ def _send_progress(self, num_samples: int) -> requests.Response:
                 samples_processed=num_samples,
             )
 
+    def _update_on_timer(self, interval_seconds: float):
+        """
+        Sends an update on a timed interval if there has been a change since the last update.
+        This is a blocking function that is expected to be executed in a thread.
+        """
+        assert interval_seconds > 0
+        while True:
+            time.sleep(interval_seconds)
+            with self._lock:
+                if self._timer_stopped:
+                    return
+                if self._last_updated_samples_processed == self._samples_processed:
+                    continue
+                curr_samples = self._samples_processed
+
+            if self.progress_tracking_url is not None:
+                self._send_progress(curr_samples)
+            if self.progress_filepath is not None:
+                self._write_progress(curr_samples)
+
+            self.logger.info(
+                "Progress milestone updated on time interval",
+                samples_processed=curr_samples,
+                interval=self.progress_tracking_interval,
+            )
+            with self._lock:
+                self._last_updated_samples_processed = curr_samples
+
     @final
     def intercept_response(
         self, ar: AdapterResponse, context: AdapterGlobalContext
@@ -177,13 +227,20 @@ def intercept_response(
                 samples_processed=curr_samples,
                 interval=self.progress_tracking_interval,
             )
+            with self._lock:
+                self._last_updated_samples_processed = curr_samples
 
         return ar
 
     def post_eval_hook(self, context: AdapterGlobalContext) -> None:
         self.logger.info(
             "Post-eval hook executed", total_samples_processed=self._samples_processed
         )
+        with self._lock:
+            if self.progress_tracking_interval_seconds:
+                self._timer_stopped = True
+            if self._samples_processed == self._last_updated_samples_processed:
+                return
 
         if self.progress_tracking_url is not None:
             self._send_progress(self._samples_processed)

packages/nemo-evaluator/src/nemo_evaluator/package_info.py

@@ -16,7 +16,7 @@
 # Below is the _next_ version that will be published, not the currently published one.
 MAJOR = 0
 MINOR = 1
-PATCH = 29
+PATCH = 30
 PRE_RELEASE = ""
 
 # Use the following formatting: (major, minor, patch, pre-release)

packages/nemo-evaluator/tests/unit_tests/adapters/interceptors/test_progress_tracking_interceptor.py

@@ -13,14 +13,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 import os
 import threading
-import time
-from typing import List
 from unittest.mock import patch
 
+import pytest
 import requests
-from flask import Flask, request
+from pydantic_core import ValidationError
 
 from nemo_evaluator.adapters.interceptors.progress_tracking_interceptor import (
     ProgressTrackingInterceptor,
@@ -30,49 +30,7 @@
     AdapterRequestContext,
     AdapterResponse,
 )
-
-
-class FakeProgressTrackingServer:
-    """Test server to receive progress tracking webhooks."""
-
-    def __init__(self, port: int = 8000, request_method="PATCH"):
-        self.port = port
-        self.app = Flask(__name__)
-        self.received_updates: List[dict] = []
-        self.lock = threading.Lock()
-
-        @self.app.route("/", methods=[request_method])
-        def progress_webhook():
-            """Receive progress updates."""
-            data = request.get_json()
-            with self.lock:
-                self.received_updates.append(data)
-            return {"status": "ok"}
-
-    def start(self):
-        """Start the server in a background thread."""
-        self.thread = threading.Thread(
-            target=self.app.run, kwargs={"host": "0.0.0.0", "port": self.port}
-        )
-        self.thread.daemon = True
-        self.thread.start()
-        # Give the server time to start
-        time.sleep(0.5)
-
-    def stop(self):
-        """Stop the server."""
-        # Flask doesn't have a clean shutdown, so we'll just let it run as daemon
-        pass
-
-    def get_updates(self) -> List[dict]:
-        """Get all received updates."""
-        with self.lock:
-            return self.received_updates.copy()
-
-    def clear_updates(self):
-        """Clear received updates."""
-        with self.lock:
-            self.received_updates.clear()
+from tests.unit_tests.adapters.testing_utils import FakeProgressTrackingServer
 
 
 class TestProgressTrackingInterceptor:
@@ -255,6 +213,19 @@ def test_network_error_handling(self, mock_request):
         # Verify that the request was attempted
         mock_request.assert_called_once()
 
+    def test_interval_configuration_validation(self):
+        with pytest.raises(ValidationError):
+            ProgressTrackingInterceptor.Params(
+                progress_tracking_url="http://test",
+                progress_tracking_interval=0,
+            )
+
+        with pytest.raises(ValidationError):
+            ProgressTrackingInterceptor.Params(
+                progress_tracking_url="http://test",
+                progress_tracking_interval=-2,
+            )
+
     def test_interval_configuration(self):
         """Test different interval configurations."""
         # Start test server
@@ -367,6 +338,68 @@ def test_configured_method(self):
         finally:
             server.stop()
 
+    def test_interval_timer_validation(self):
+        with pytest.raises(ValidationError):
+            ProgressTrackingInterceptor.Params(
+                progress_tracking_interval_seconds=-1,
+            )
+
+    @pytest.mark.asyncio
+    async def test_interval_timer(self):
+        # Start test server
+        server = FakeProgressTrackingServer(port=8007)
+        server.start()
+
+        try:
+            params = ProgressTrackingInterceptor.Params(
+                progress_tracking_url="http://localhost:8007",
+                progress_tracking_interval=50,
+                progress_tracking_interval_seconds=0.2,
+            )
+            interceptor = ProgressTrackingInterceptor(params)
+            assert interceptor.progress_tracking_url == "http://localhost:8007"
+            assert interceptor.progress_tracking_interval == 50
+            assert interceptor.progress_tracking_interval_seconds == 0.2
+
+            # Create mock response and context
+            mock_response = AdapterResponse(
+                r=requests.Response(),
+                rctx=AdapterRequestContext(),
+            )
+            context = AdapterGlobalContext(output_dir="/tmp", url="http://test")
+
+            # Verify no update until timer interval
+            interceptor.intercept_response(mock_response, context)
+            interceptor.intercept_response(mock_response, context)
+            updates = server.get_updates()
+            assert len(updates) == 0, "no updates until timer interval"
+
+            # Verify first timer interval calls update
+            await asyncio.sleep(0.5)
+            updates = server.get_updates()
+            assert len(updates) == 1, "only expected one update"
+            assert updates[0]["samples_processed"] == 2
+
+            # Verify subsequent timer interval calls update
+            interceptor.intercept_response(mock_response, context)
+            await asyncio.sleep(0.5)
+            updates = server.get_updates()
+            assert len(updates) == 2, "expected second update"
+            assert updates[1]["samples_processed"] == 3
+
+            # No calls to update after timer is stopped
+            interceptor.post_eval_hook(context)
+            interceptor.intercept_response(mock_response, context)
+            assert interceptor._samples_processed == 4
+            await asyncio.sleep(0.5)
+            updates = server.get_updates()
+            assert len(updates) == 2, (
+                "expected post_eval_hook to skip posting update on no change and no updates after post_eval_hook cancels timed updates"
+            )
+
+        finally:
+            server.stop()
+
 
 if __name__ == "__main__":
     # Simple test runner for manual testing

packages/nemo-evaluator/tests/unit_tests/adapters/interceptors/test_reasoning.py

@@ -204,7 +204,7 @@ def test_migration(
     url = f"http://{AdapterServer.DEFAULT_ADAPTER_HOST}:{adapter_server_migration.port}"
 
     # Wait for server to be ready
-    wait_for_server("localhost", 3825)
+    wait_for_server("localhost", adapter_server_migration.port)
 
     # We parametrize the response of the openai fake server.
     response_data = {