diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/_executor.py b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/_executor.py index ff192a29ba..6d273323f0 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/_executor.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/_executor.py @@ -13,7 +13,8 @@ from typing import Any, AsyncIterator, Iterator, cast import httpx -from nemo_evaluator.jobs.evaluate import EvaluateJob, EvaluateSpec +from nemo_evaluator.jobs.evaluate import EvaluateJob, EvaluateSpec, TargetSpec +from nemo_evaluator.resolvers import PlatformModelResolver from nemo_evaluator.sdk import http_utils from nemo_evaluator.sdk.fs_utils import EvaluatorLocalRunResult from nemo_evaluator.sdk.job_resources import ( @@ -27,11 +28,13 @@ from nemo_evaluator_sdk import Evaluator as SDKEvaluator from nemo_evaluator_sdk.datasets.loader import prepare_dataset_rows from nemo_evaluator_sdk.execution.config import EvaluationRequest, normalize_params +from nemo_evaluator_sdk.execution.metric_execution import run_sync from nemo_evaluator_sdk.metrics.protocol import Metric from nemo_evaluator_sdk.values import ( Agent, DatasetInput, Model, + ModelRef, RunConfig, RunConfigOnline, RunConfigOnlineModel, @@ -48,6 +51,7 @@ _DEFAULT_PENDING_TIMEOUT_SECONDS = 600.0 _ResolvedDataset = DatasetInput | str | Path +SubmitTargetSpec = TargetSpec | ModelRef class MetricBundlePackagerPolicyError(RuntimeError): @@ -63,6 +67,41 @@ def _require_metric_bundle_packager(metric_bundle_packager: MetricBundlePackager return metric_bundle_packager +def _submit_params( + params: RunConfig | RunConfigOnline | RunConfigOnlineModel | None, + target: SubmitTargetSpec | None, +) -> RunConfig | RunConfigOnline | RunConfigOnlineModel: + if isinstance(target, ModelRef) and not isinstance(params, RunConfigOnlineModel): + raise TypeError("ModelRef target requires RunConfigOnlineModel") + if isinstance(target, Model) and not isinstance(params, RunConfigOnlineModel): + raise TypeError("model target requires RunConfigOnlineModel") + if isinstance(target, Agent) and not isinstance(params, RunConfigOnline): + raise TypeError("agent target requires RunConfigOnline") + if target is None: + return params or RunConfig() + if params is None: + raise TypeError("targeted evaluation requires params") + return params + + +def _resolve_submit_target( + platform: NeMoPlatform, + target: SubmitTargetSpec | None, +) -> Model | Agent | None: + if isinstance(target, ModelRef): + return run_sync(lambda: PlatformModelResolver(platform).resolve_model(target)) + return target + + +async def _resolve_submit_target_async( + platform: AsyncNeMoPlatform, + target: SubmitTargetSpec | None, +) -> Model | Agent | None: + if isinstance(target, ModelRef): + return await PlatformModelResolver(platform).resolve_model(target) + return target + + def _dataset_config(request: EvaluationRequest) -> list[dict[str, Any]] | FilesetRef: """Return the dataset payload to store in an evaluator plugin job spec.""" if isinstance(request.dataset, FilesetRef): @@ -272,7 +311,7 @@ def submit( metric: Metric, dataset: PluginDatasetInput, params: RunConfig | RunConfigOnline | RunConfigOnlineModel | None = None, - target: Model | Agent | None = None, + target: SubmitTargetSpec | None = None, dataset_glob_pattern: str | None = None, prompt_template: str | dict[str, Any] | None = None, metric_bundle_packager: MetricBundlePackager | None = None, @@ -280,8 +319,8 @@ def submit( """Submit a remote evaluator plugin metric job and return the job resource.""" request = EvaluationRequest( dataset=dataset, - params=normalize_params(params, target), - target=target, + params=_submit_params(params, target), + target=_resolve_submit_target(self._platform, target), dataset_glob_pattern=dataset_glob_pattern, prompt_template=prompt_template, ) @@ -391,7 +430,7 @@ async def submit( metric: Metric, dataset: PluginDatasetInput, params: RunConfig | RunConfigOnline | RunConfigOnlineModel | None = None, - target: Model | Agent | None = None, + target: SubmitTargetSpec | None = None, dataset_glob_pattern: str | None = None, prompt_template: str | dict[str, Any] | None = None, metric_bundle_packager: MetricBundlePackager | None = None, @@ -399,8 +438,8 @@ async def submit( """Submit a remote evaluator plugin metric job and return the job resource.""" request = EvaluationRequest( dataset=dataset, - params=normalize_params(params, target), - target=target, + params=_submit_params(params, target), + target=await _resolve_submit_target_async(self._platform, target), dataset_glob_pattern=dataset_glob_pattern, prompt_template=prompt_template, ) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/resources.py b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/resources.py index 2acdb7bad8..f8f6a694e0 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/resources.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/resources.py @@ -10,6 +10,7 @@ from nemo_evaluator.sdk import http_utils from nemo_evaluator.sdk._executor import ( + SubmitTargetSpec, _AsyncEvaluatorPluginExecutor, _SyncEvaluatorPluginExecutor, ) @@ -82,7 +83,7 @@ def submit( metric: Metric, dataset: PluginDatasetInput, config: RunConfig | RunConfigOnline | RunConfigOnlineModel | None = None, - target: Model | Agent | None = None, + target: SubmitTargetSpec | None = None, dataset_glob_pattern: str | None = None, prompt_template: str | dict[str, Any] | None = None, metric_bundle_packager: MetricBundlePackager | None = None, @@ -194,7 +195,7 @@ async def submit( metric: Metric, dataset: PluginDatasetInput, config: RunConfig | RunConfigOnline | RunConfigOnlineModel | None = None, - target: Model | Agent | None = None, + target: SubmitTargetSpec | None = None, dataset_glob_pattern: str | None = None, prompt_template: str | dict[str, Any] | None = None, metric_bundle_packager: MetricBundlePackager | None = None, diff --git a/plugins/nemo-evaluator/tests/test_sdk.py b/plugins/nemo-evaluator/tests/test_sdk.py index 93b6e5dbb5..d230aa6682 100644 --- a/plugins/nemo-evaluator/tests/test_sdk.py +++ b/plugins/nemo-evaluator/tests/test_sdk.py @@ -6,6 +6,7 @@ from __future__ import annotations from pathlib import Path +from types import SimpleNamespace from typing import Any, cast from unittest.mock import AsyncMock, MagicMock @@ -34,7 +35,7 @@ from nemo_evaluator_sdk.execution.config import EvaluationRequest from nemo_evaluator_sdk.metrics.exact_match import ExactMatchMetric from nemo_evaluator_sdk.metrics.protocol import Metric -from nemo_evaluator_sdk.values import Model, RunConfig, RunConfigOnlineModel +from nemo_evaluator_sdk.values import Model, ModelRef, RunConfig, RunConfigOnlineModel from nemo_evaluator_sdk.values.results import AggregatedMetricResult, EvaluationResult from nemo_platform import AsyncNeMoPlatform, NeMoPlatform from nemo_platform_plugin.jobs.schemas import PlatformJobStatus @@ -87,6 +88,24 @@ def load(self, payload: MetricBundlePayload) -> Metric: raise NotImplementedError("test packager only exercises submission-side packaging") +class _FakeModels: + def __init__(self) -> None: + self.retrieved: list[tuple[str, str]] = [] + + def retrieve(self, name: str, *, workspace: str) -> SimpleNamespace: + self.retrieved.append((workspace, name)) + return SimpleNamespace(model_providers=["default/provider"]) + + def get_model_entity_route_openai_url(self, model_entity: object) -> str: + del model_entity + return "https://igw.example.test/v1/chat/completions" + + +class _FakeProviders: + def retrieve(self, name: str, *, workspace: str) -> SimpleNamespace: + return SimpleNamespace(name=name, workspace=workspace, host_url="http://nim.example.test:8000") + + class _SyncPlatform: def __init__(self) -> None: self.base_url = "http://test:8000" @@ -94,6 +113,8 @@ def __init__(self) -> None: self.default_headers = {"Authorization": "Bearer sync-platform-token"} self.timeout = httpx.Timeout(42.0) self._client = MagicMock(spec=httpx.Client) + self.models = _FakeModels() + self.inference = SimpleNamespace(providers=_FakeProviders()) class _AsyncPlatform: @@ -495,6 +516,62 @@ def test_sync_executor_runs_evaluator_job_locally(mocker: MockerFixture) -> None to_thread.assert_not_called() +def test_sync_executor_submit_resolves_model_ref_target_before_building_job_spec() -> None: + """Durable submit resolves platform ModelRef targets before creating the evaluator job.""" + platform = _SyncPlatform() + platform.models = _FakeModels() + platform.inference = SimpleNamespace(providers=_FakeProviders()) + platform._client.post.return_value = httpx.Response( + 201, + request=httpx.Request("POST", "http://test:8000/apis/evaluator/v2/workspaces/platform-ws/evaluate/jobs"), + json={"name": "job-123", "status": "created", "spec": _EXACT_MATCH_SPEC}, + ) + executor = _SyncEvaluatorPluginExecutor(platform=cast(NeMoPlatform, platform)) + + job = executor.submit( + metric=ExactMatchMetric(reference="{{item.expected}}", candidate="{{item.output}}"), + dataset=[{"expected": "a", "output": "a"}], + params=RunConfigOnlineModel(), + target=ModelRef(root="default/model-a"), + prompt_template="Answer: {{item.input}}", + metric_bundle_packager=CloudpickleMetricBundlePackager(), + ) + + assert job.name == "job-123" + assert platform.models.retrieved == [("default", "model-a")] + request_json = platform._client.post.call_args.kwargs["json"] + target = request_json["spec"]["target"] + assert target["name"] == "model-a" + assert target["url"] == "https://igw.example.test/v1/chat/completions" + assert target["host_url"] == "http://nim.example.test:8000" + + +def test_sync_executor_submit_rejects_model_ref_target_without_online_model_params() -> None: + executor = _SyncEvaluatorPluginExecutor(platform=cast(NeMoPlatform, _SyncPlatform())) + + with pytest.raises(TypeError, match="ModelRef target requires RunConfigOnlineModel"): + executor.submit( + metric=ExactMatchMetric(reference="{{item.expected}}", candidate="{{item.output}}"), + dataset=[{"expected": "a", "output": "a"}], + target=ModelRef(root="default/model-a"), + prompt_template="Answer: {{item.input}}", + metric_bundle_packager=CloudpickleMetricBundlePackager(), + ) + + +def test_sync_executor_submit_rejects_model_target_without_online_model_params() -> None: + executor = _SyncEvaluatorPluginExecutor(platform=cast(NeMoPlatform, _SyncPlatform())) + + with pytest.raises(TypeError, match="model target requires RunConfigOnlineModel"): + executor.submit( + metric=ExactMatchMetric(reference="{{item.expected}}", candidate="{{item.output}}"), + dataset=[{"expected": "a", "output": "a"}], + target=Model(url="https://model.test/v1", name="model-a"), + prompt_template="Answer: {{item.input}}", + metric_bundle_packager=CloudpickleMetricBundlePackager(), + ) + + class TestEvaluatorSubmit: """Tests for ``Evaluator.submit`` request construction.""" @@ -532,6 +609,37 @@ def test_builds_request_from_unpacked_fields(self, mocker: MockerFixture) -> Non metric_bundle_packager=packager, ) + def test_accepts_model_ref_target(self, mocker: MockerFixture) -> None: + """Submit should forward platform ModelRef targets to the plugin executor.""" + platform = _SyncPlatform() + resource = Evaluator(cast(NeMoPlatform, platform)) + expected_job = mocker.Mock(spec=EvaluatorJobResource) + submit = mocker.patch.object(resource._executor, "submit", return_value=expected_job) + metric = ExactMatchMetric(reference="{{item.expected}}", candidate="{{item.output}}") + dataset = [{"expected": "a", "output": "a"}] + model_ref = ModelRef(root="default/model-a") + packager = CloudpickleMetricBundlePackager() + + job = resource.submit( + metric=metric, + dataset=dataset, + config=RunConfigOnlineModel(), + target=model_ref, + prompt_template="Answer: {{item.input}}", + metric_bundle_packager=packager, + ) + + assert job is expected_job + submit.assert_called_once_with( + metric=metric, + dataset=dataset, + params=RunConfigOnlineModel(), + target=model_ref, + dataset_glob_pattern=None, + prompt_template="Answer: {{item.input}}", + metric_bundle_packager=packager, + ) + def test_accepts_fileset_ref_dataset(self, mocker: MockerFixture) -> None: """Submit should forward FilesetRef datasets unchanged to the executor.""" platform = _SyncPlatform()