diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py index c072ae07b3..0d35cfce51 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py @@ -13,12 +13,16 @@ from nemo_evaluator.jobs.utils import resolve_run_dataset from nemo_evaluator.resolvers import PlatformModelResolver from nemo_evaluator.sdk.values.filesets import FilesetRef -from nemo_evaluator.shared.metric_bundles.bundles import MetricBundle, unbundle_metric +from nemo_evaluator.shared.metric_bundles.bundles import ( + MetricBundle, + bundle_metric, + metric_bundle_packager_for_payload, + unbundle_metric, +) from nemo_evaluator.shared.metric_bundles.cloudpickle import CloudpickleMetricPayload # noqa: F401 from nemo_evaluator_sdk import Evaluator from nemo_evaluator_sdk.execution._protocols import JobParamsConfigurableMetric from nemo_evaluator_sdk.execution.config import normalize_params -from nemo_evaluator_sdk.execution.metric_execution import run_sync from nemo_evaluator_sdk.metrics.protocol import Metric, MetricWithModels from nemo_evaluator_sdk.values import ( Agent, @@ -61,8 +65,51 @@ class EvaluationResultFiles: artifacts_dir: Path -class EvaluateSpec(BaseModel): - """Inline SDK evaluation input for the first evaluator plugin job.""" +def _hydrate_metrics(metrics: list[MetricBundle]) -> list[Metric]: + return [unbundle_metric(bundle) for bundle in metrics] + + +def _unresolved_model_refs(metrics: list[Metric]) -> list[str]: + refs = [ + model_ref.root + for item in metrics + if isinstance(item, MetricWithModels) + for model_ref in item.model_refs().values() + ] + return sorted(refs) + + +async def _resolve_metric_models( + metrics: list[Metric], + resolver: PlatformModelResolver, +) -> None: + """Resolve ModelRef fields on metric configs before SDK execution.""" + for item in metrics: + if isinstance(item, MetricWithModels): + await item.resolve_models(resolver) + + +def _apply_metric_job_params( + metrics: list[Metric], + params: RunConfig | RunConfigOnline | RunConfigOnlineModel, +) -> bool: + """Apply evaluation job params to metrics that support runtime configuration.""" + applied = False + for item in metrics: + if isinstance(item, JobParamsConfigurableMetric): + item.apply_evaluation_job_params(params) + applied = True + return applied + + +def _bundle_resolved_metric(metric: Metric, source_bundle: MetricBundle) -> MetricBundle: + packager = metric_bundle_packager_for_payload(source_bundle.payload) + resolved_bundle = bundle_metric(metric, packager) + return resolved_bundle.model_copy(update={"metadata": source_bundle.metadata}) + + +class EvaluateInputSpec(BaseModel): + """Submitter-facing SDK evaluation input for the evaluator plugin job.""" model_config = ConfigDict(extra="forbid") @@ -84,38 +131,29 @@ def normalize_params_for_target(self) -> Self: return self +class EvaluateSpec(EvaluateInputSpec): + """Canonical SDK evaluation spec with platform model references resolved.""" + + @model_validator(mode="after") + def reject_unresolved_metric_model_refs(self) -> Self: + unresolved_refs = _unresolved_model_refs(_hydrate_metrics(self.metrics)) + if unresolved_refs: + raise ValueError( + "EvaluateSpec metric models must be resolved before compile/run: " + ", ".join(unresolved_refs) + ) + return self + + class EvaluateJob(NemoJob): """Run one evaluator SDK metric against inline rows.""" name: ClassVar[str] = "evaluate" description: ClassVar[str] = "Run an inline evaluator SDK metric against inline dataset rows." container: ClassVar[str] = "cpu-tasks" + input_spec_schema: ClassVar[type[BaseModel] | None] = EvaluateInputSpec spec_schema: ClassVar[type[BaseModel] | None] = EvaluateSpec job_collection_path: ClassVar[str | None] = "/evaluate/jobs" - @staticmethod - async def _resolve_metric_models( - metrics: list[Metric], - resolver: PlatformModelResolver, - params: RunConfig | RunConfigOnline | RunConfigOnlineModel, - ) -> None: - """Resolve ModelRef fields on metric configs before local SDK execution.""" - for item in metrics: - if isinstance(item, JobParamsConfigurableMetric): - item.apply_evaluation_job_params(params) - if isinstance(item, MetricWithModels): - await item.resolve_models(resolver) - - @staticmethod - def _unresolved_model_refs(metrics: list[Metric]) -> list[str]: - refs = [ - model_ref.root - for item in metrics - if isinstance(item, MetricWithModels) - for model_ref in item.model_refs().values() - ] - return sorted(refs) - @classmethod async def compile( cls, @@ -137,7 +175,7 @@ async def compile( @staticmethod def _hydrate_metrics(metrics: MetricSpec) -> list[Metric]: - return [unbundle_metric(bundle) for bundle in metrics] + return _hydrate_metrics(metrics) @staticmethod def _write_result_files(result: EvaluationArtifactResult, persistent_dir: Path) -> EvaluationResultFiles: @@ -162,21 +200,51 @@ def _write_result_files(result: EvaluationArtifactResult, persistent_dir: Path) artifacts_dir=artifacts_dir, ) + @classmethod + async def to_spec( + cls, + input_spec: BaseModel, + *, + workspace: str, + entity_client: object, + async_sdk: AsyncNeMoPlatform | None, + is_local: bool, + ) -> BaseModel: + """Resolve submitter-facing model references into the canonical evaluation spec.""" + del workspace, entity_client, is_local + submit_spec = ( + input_spec.model_copy(deep=True) + if isinstance(input_spec, EvaluateInputSpec) + else EvaluateInputSpec.model_validate(input_spec.model_dump()) + ) + metrics = _hydrate_metrics(submit_spec.metrics) + applied_params = _apply_metric_job_params( + metrics, + normalize_params(submit_spec.params, submit_spec.target), + ) + unresolved_refs = _unresolved_model_refs(metrics) + if unresolved_refs: + if async_sdk is None: + raise ValueError( + "ModelRef metrics require `async_sdk` for spec resolution: " + ", ".join(unresolved_refs) + ) + await _resolve_metric_models( + metrics, + PlatformModelResolver(async_sdk), + ) + if applied_params or unresolved_refs: + submit_spec.metrics = [ + _bundle_resolved_metric(metric, bundle) + for metric, bundle in zip(metrics, submit_spec.metrics, strict=True) + ] + return EvaluateSpec.model_validate(submit_spec.model_dump(mode="python")) + def run(self, config: dict, *, ctx: JobContext, sdk: object | None = None, async_sdk: object | None = None) -> dict: """Run the evaluator job locally and persist its result artifact.""" spec = EvaluateSpec.model_validate(config) evaluator = Evaluator() - platform_sdk = async_sdk or sdk params = normalize_params(spec.params, spec.target) metrics = self._hydrate_metrics(spec.metrics) - if platform_sdk is None: - unresolved_refs = self._unresolved_model_refs(metrics) - if unresolved_refs: - raise ValueError( - "ModelRef metrics require `sdk` or `async_sdk` for local execution: " + ", ".join(unresolved_refs) - ) - else: - run_sync(lambda: self._resolve_metric_models(metrics, PlatformModelResolver(platform_sdk), params)) dataset = resolve_run_dataset( spec.dataset, ctx=ctx, diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/_executor.py b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/_executor.py index e4042028f4..a0f3855456 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/_executor.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/_executor.py @@ -7,16 +7,12 @@ import asyncio from collections.abc import Sequence -from contextlib import asynccontextmanager, contextmanager -from pathlib import Path -from tempfile import TemporaryDirectory -from typing import Any, AsyncIterator, Iterator, cast +from typing import Any, TypeAlias, cast import httpx -from nemo_evaluator.jobs.evaluate import EvaluateJob, EvaluateSpec -from nemo_evaluator.jobs.utils import download_dataset, download_dataset_sync +from nemo_evaluator.jobs.evaluate import EvaluateInputSpec, EvaluateJob, EvaluateSpec from nemo_evaluator.sdk import http_utils -from nemo_evaluator.sdk.fs_utils import EvaluatorLocalRunResult +from nemo_evaluator.sdk.fs_utils import EvaluatorLocalRunResult, local_result_path from nemo_evaluator.sdk.job_resources import ( AsyncEvaluatorJobResource, EvaluatorJob, @@ -26,13 +22,13 @@ from nemo_evaluator.sdk.utils import filter_benchmark_result, filter_evaluation_result from nemo_evaluator.sdk.values.filesets import FilesetRef from nemo_evaluator.shared.metric_bundles.bundles import MetricBundle, MetricBundlePackager, bundle_metric -from nemo_evaluator_sdk import Evaluator as SDKEvaluator +from nemo_evaluator.shared.metric_bundles.cloudpickle import CloudpickleMetricBundlePackager 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, RunConfig, RunConfigOnline, @@ -46,8 +42,7 @@ _DEFAULT_POLL_INTERVAL_SECONDS = 10.0 _DEFAULT_JOB_TIMEOUT_SECONDS = 3600.0 _DEFAULT_PENDING_TIMEOUT_SECONDS = 600.0 - -_ResolvedDataset = DatasetInput | str | Path +EvaluateRequestSpec: TypeAlias = EvaluateInputSpec | EvaluateSpec class MetricBundlePackagerPolicyError(RuntimeError): @@ -78,58 +73,12 @@ def _dataset_config(request: EvaluationRequest) -> list[dict[str, Any]] | Filese ) -def _fileset_dataset(request: EvaluationRequest) -> FilesetRef: - if not isinstance(request.dataset, FilesetRef): - raise TypeError("request dataset is not a FilesetRef") - if request.dataset_glob_pattern is None: - return request.dataset - if "#" in request.dataset.root: - raise ValueError("dataset_glob_pattern cannot be used when FilesetRef already includes a fragment.") - return request.dataset.with_fragment(request.dataset_glob_pattern) - - -@contextmanager -def _sync_resolved_dataset( - request: EvaluationRequest, platform: NeMoPlatform -) -> Iterator[tuple[_ResolvedDataset, str | None]]: - if not isinstance(request.dataset, FilesetRef): - yield request.dataset, request.dataset_glob_pattern - return - - dataset = _fileset_dataset(request) - with TemporaryDirectory(prefix="nemo-evaluator-fileset-") as temp_dir: - resolved = download_dataset_sync( - sdk=platform, - dataset=dataset, - destination=str(Path(temp_dir) / "dataset"), - ) - yield cast(_ResolvedDataset, resolved), None - - -@asynccontextmanager -async def _async_resolved_dataset( - request: EvaluationRequest, platform: AsyncNeMoPlatform -) -> AsyncIterator[tuple[_ResolvedDataset, str | None]]: - if not isinstance(request.dataset, FilesetRef): - yield request.dataset, request.dataset_glob_pattern - return - - dataset = _fileset_dataset(request) - with TemporaryDirectory(prefix="nemo-evaluator-fileset-") as temp_dir: - resolved = await download_dataset( - sdk=platform, - dataset=dataset, - destination=str(Path(temp_dir) / "dataset"), - ) - yield cast(_ResolvedDataset, resolved), None - - def _build_evaluate_spec( *, metrics: Metric | Sequence[Metric], request: EvaluationRequest, metric_bundle_packager: MetricBundlePackager | None = None, -) -> EvaluateSpec: +) -> EvaluateInputSpec: """Build the evaluator plugin spec shared by local and remote execution.""" effective_packager = _require_metric_bundle_packager(metric_bundle_packager) spec = { @@ -141,7 +90,51 @@ def _build_evaluate_spec( spec["target"] = request.target.model_dump(mode="json") if request.prompt_template is not None: spec["prompt_template"] = request.prompt_template - return EvaluateSpec.model_validate(spec) + return EvaluateInputSpec.model_validate(spec) + + +def _resolve_sync_local_spec( + spec: EvaluateRequestSpec, + *, + platform: NeMoPlatform, + workspace: str, +) -> EvaluateSpec: + """Return a canonical local spec, resolving input-only model references with the sync SDK.""" + if isinstance(spec, EvaluateSpec): + return spec + return cast( + EvaluateSpec, + run_sync( + lambda: EvaluateJob.to_spec( + spec, + workspace=workspace, + entity_client=None, + async_sdk=cast(AsyncNeMoPlatform, platform), + is_local=True, + ) + ), + ) + + +async def _resolve_async_local_spec( + spec: EvaluateRequestSpec, + *, + platform: AsyncNeMoPlatform, + workspace: str, +) -> EvaluateSpec: + """Return a canonical local spec, resolving input-only model references with the async SDK.""" + if isinstance(spec, EvaluateSpec): + return spec + return cast( + EvaluateSpec, + await EvaluateJob.to_spec( + spec, + workspace=workspace, + entity_client=None, + async_sdk=platform, + is_local=True, + ), + ) class _SyncEvaluatorPluginExecutor: @@ -167,7 +160,7 @@ def __init__( def create( self, *, - spec: EvaluateSpec, + spec: EvaluateRequestSpec, workspace: str | None = None, wait_until_done: bool = False, ) -> EvaluatorJobResource: @@ -199,12 +192,18 @@ def create( ) return job_resource - def run_local(self, *, spec: EvaluateSpec, workspace: str | None = None) -> EvaluatorLocalRunResult: + def run_local(self, *, spec: EvaluateRequestSpec, workspace: str | None = None) -> EvaluatorLocalRunResult: """Run an evaluator plugin job locally with a sync platform client.""" + resolved_workspace = http_utils.resolve_workspace(self._platform, workspace) + canonical_spec = _resolve_sync_local_spec( + spec, + platform=self._platform, + workspace=resolved_workspace, + ) payload = NemoJobScheduler().run_local( EvaluateJob, - spec.model_dump(mode="json"), - workspace=http_utils.resolve_workspace(self._platform, workspace), + canonical_spec.model_dump(mode="json"), + workspace=resolved_workspace, sdk=self._platform, ) @@ -255,15 +254,17 @@ def evaluate( prompt_template=prompt_template, aggregate_fields=aggregate_fields, ) - with _sync_resolved_dataset(request, self._platform) as (resolved_dataset, resolved_pattern): - result = SDKEvaluator().run_sync( - metrics=metric, - dataset=resolved_dataset, - config=request.params, - target=request.target, - dataset_glob_pattern=resolved_pattern, - prompt_template=request.prompt_template, - ) + spec = _build_evaluate_spec( + metrics=metric, + request=request, + metric_bundle_packager=CloudpickleMetricBundlePackager(), + ) + payload = self.run_local( + spec=spec, + workspace=http_utils.resolve_workspace(self._platform, self._workspace, strict=True), + ) + result_path = local_result_path(payload) + result = EvaluationResult.model_validate_json(result_path.read_text(encoding="utf-8")) return filter_evaluation_result(result, aggregate_fields) def submit( @@ -304,15 +305,17 @@ def evaluate_benchmark( request: EvaluationRequest, ) -> BenchmarkEvaluationResult: """Evaluate multiple metrics through local in-process plugin execution.""" - with _sync_resolved_dataset(request, self._platform) as (resolved_dataset, resolved_pattern): - result = SDKEvaluator().run_sync( - metrics=metrics, - dataset=resolved_dataset, - config=request.params, - target=request.target, - dataset_glob_pattern=resolved_pattern, - prompt_template=request.prompt_template, - ) + spec = _build_evaluate_spec( + metrics=metrics, + request=request, + metric_bundle_packager=CloudpickleMetricBundlePackager(), + ) + payload = self.run_local( + spec=spec, + workspace=http_utils.resolve_workspace(self._platform, self._workspace, strict=True), + ) + result_path = local_result_path(payload) + result = BenchmarkEvaluationResult.model_validate_json(result_path.read_text(encoding="utf-8")) return filter_benchmark_result(result, request.aggregate_fields) @@ -339,7 +342,7 @@ def __init__( async def create( self, *, - spec: EvaluateSpec, + spec: EvaluateRequestSpec, workspace: str | None = None, wait_until_done: bool = False, ) -> AsyncEvaluatorJobResource: @@ -371,16 +374,22 @@ async def create( ) return job_resource - async def run_local(self, *, spec: EvaluateSpec, workspace: str | None = None) -> EvaluatorLocalRunResult: + async def run_local(self, *, spec: EvaluateRequestSpec, workspace: str | None = None) -> EvaluatorLocalRunResult: """Run an evaluator plugin job locally without blocking the event loop.""" + resolved_workspace = http_utils.resolve_workspace(self._platform, workspace) + canonical_spec = await _resolve_async_local_spec( + spec, + platform=self._platform, + workspace=resolved_workspace, + ) scheduler = NemoJobScheduler() # Leverages programmatic dispatch as described in # packages/nemo_platform_plugin/src/nemo_platform_plugin/docs/ARCHITECTURE.md#job-entry-point-keys payload = await asyncio.to_thread( scheduler.run_local, EvaluateJob, - spec.model_dump(mode="json"), - workspace=http_utils.resolve_workspace(self._platform, workspace), + canonical_spec.model_dump(mode="json"), + workspace=resolved_workspace, async_sdk=self._platform, ) return EvaluatorLocalRunResult.model_validate(payload) @@ -461,15 +470,18 @@ async def evaluate( prompt_template=prompt_template, aggregate_fields=aggregate_fields, ) - async with _async_resolved_dataset(request, self._platform) as (resolved_dataset, resolved_pattern): - result = await SDKEvaluator().run( - metrics=metric, - dataset=resolved_dataset, - config=request.params, - target=request.target, - dataset_glob_pattern=resolved_pattern, - prompt_template=request.prompt_template, - ) + spec = _build_evaluate_spec( + metrics=metric, + request=request, + metric_bundle_packager=CloudpickleMetricBundlePackager(), + ) + payload = await self.run_local( + spec=spec, + workspace=http_utils.resolve_workspace(self._platform, self._workspace, strict=True), + ) + result_path = local_result_path(payload) + result_text = await asyncio.to_thread(result_path.read_text, encoding="utf-8") + result = EvaluationResult.model_validate_json(result_text) return filter_evaluation_result(result, aggregate_fields) async def evaluate_benchmark( @@ -479,15 +491,18 @@ async def evaluate_benchmark( request: EvaluationRequest, ) -> BenchmarkEvaluationResult: """Evaluate multiple metrics through local in-process plugin execution.""" - async with _async_resolved_dataset(request, self._platform) as (resolved_dataset, resolved_pattern): - result = await SDKEvaluator().run( - metrics=metrics, - dataset=resolved_dataset, - config=request.params, - target=request.target, - dataset_glob_pattern=resolved_pattern, - prompt_template=request.prompt_template, - ) + spec = _build_evaluate_spec( + metrics=metrics, + request=request, + metric_bundle_packager=CloudpickleMetricBundlePackager(), + ) + payload = await self.run_local( + spec=spec, + workspace=http_utils.resolve_workspace(self._platform, self._workspace, strict=True), + ) + result_path = local_result_path(payload) + result_text = await asyncio.to_thread(result_path.read_text, encoding="utf-8") + result = BenchmarkEvaluationResult.model_validate_json(result_text) return filter_benchmark_result(result, request.aggregate_fields) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/http_utils.py b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/http_utils.py index bbc1eb31d6..396e7d9fc1 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/http_utils.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/http_utils.py @@ -8,7 +8,7 @@ from typing import Any from urllib.parse import quote, urljoin -from nemo_evaluator.jobs.evaluate import EvaluateSpec +from nemo_evaluator.jobs.evaluate import EvaluateInputSpec, EvaluateSpec from nemo_platform import AsyncNeMoPlatform, NeMoPlatform PlatformClient = NeMoPlatform | AsyncNeMoPlatform @@ -47,7 +47,7 @@ def platform_default_headers(platform: PlatformClient) -> dict[str, str]: return {str(key): value for key, value in platform.default_headers.items() if isinstance(value, str)} -def create_job_payload(spec: EvaluateSpec) -> dict[str, dict[str, Any]]: +def create_job_payload(spec: EvaluateInputSpec | EvaluateSpec) -> dict[str, dict[str, Any]]: """Serialize an evaluator job creation request body.""" return {"spec": spec.model_dump(mode="json")} diff --git a/plugins/nemo-evaluator/tests/test_evaluate_job.py b/plugins/nemo-evaluator/tests/test_evaluate_job.py index e7b9f5ff93..aa913b6b1c 100644 --- a/plugins/nemo-evaluator/tests/test_evaluate_job.py +++ b/plugins/nemo-evaluator/tests/test_evaluate_job.py @@ -18,6 +18,7 @@ DEFAULT_FILE_NAME, DEFAULT_RESULT_NAME, ROW_SCORES_RESULT_NAME, + EvaluateInputSpec, EvaluateJob, EvaluateSpec, ) @@ -48,7 +49,9 @@ RunConfigOnline, RunConfigOnlineModel, SecretRef, + SupportedJobTypes, ) +from nemo_evaluator_sdk.values.llm_judge_defaults import default_judge_prompt_template_chat from nemo_evaluator_sdk.values.models import ModelRef from nemo_evaluator_sdk.values.scores import JSONScoreParser, RangeScore from nemo_platform.types.jobs.platform_job_spec import PlatformJobSpec @@ -319,7 +322,7 @@ def test_metric_bundle_validation_strips_payload_kind_before_payload_validation( assert isinstance(bundle.payload, _StrictMetricPayload) -def test_evaluate_job_resolves_metric_model_refs_before_sdk_run( +async def test_evaluate_job_resolves_metric_model_refs_before_sdk_run( tmp_path: Path, mocker: MockerFixture, ) -> None: @@ -334,13 +337,21 @@ async def compute_scores(metric: LLMJudgeMetric, input) -> MetricResult: mocker.patch.object(LLMJudgeMetric, "compute_scores", compute_scores) ctx = _make_job_context(tmp_path) + spec = await EvaluateJob.to_spec( + EvaluateInputSpec.model_validate( + { + "metrics": [_bundle_payload(_llm_judge_ref_metric())], + "dataset": [{"output_text": "hello"}], + } + ), + workspace="default", + entity_client=object(), + async_sdk=cast(Any, _FakeSDK()), + is_local=True, + ) run_result = EvaluateJob().run( - { - "metrics": [_bundle_payload(_llm_judge_ref_metric())], - "dataset": [{"output_text": "hello"}], - }, + spec.model_dump(mode="json"), ctx=ctx, - sdk=_FakeSDK(), ) payload = _load_artifact_payload(run_result) @@ -369,18 +380,43 @@ async def test_evaluate_job_compile_produces_cpu_task_step() -> None: assert config["dataset"] == _exact_match_spec()["dataset"] -async def test_evaluate_job_compile_preserves_bundled_metric_model_refs_for_runtime_resolution() -> None: - compiled = await EvaluateJob.compile( - workspace="default", - spec=EvaluateSpec.model_validate( +def test_evaluate_spec_rejects_unresolved_bundled_metric_model_refs() -> None: + with pytest.raises(ValueError, match="EvaluateSpec metric models must be resolved"): + EvaluateSpec.model_validate( + { + "metrics": [_bundle_payload(_llm_judge_ref_metric())], + "dataset": [{"output_text": "hello"}], + } + ) + + +async def test_evaluate_job_to_spec_resolves_bundled_metric_model_refs_before_compile() -> None: + canonical = await EvaluateJob.to_spec( + EvaluateInputSpec.model_validate( { "metrics": [_bundle_payload(_llm_judge_ref_metric())], "dataset": [{"output_text": "hello"}], } ), + workspace="default", + entity_client=object(), + async_sdk=cast(Any, _FakeSDK()), + is_local=False, + ) + assert isinstance(canonical, EvaluateSpec) + canonical_metric = unbundle_metric(canonical.metrics[0]) + assert isinstance(canonical_metric, LLMJudgeMetric) + assert isinstance(canonical_metric.model, Model) + assert canonical_metric.model.name == "judge" + assert canonical_metric.model.url == "https://igw.example.test/v1/chat/completions" + assert canonical_metric.model.host_url == "http://nim.example.test:8000" + + compiled = await EvaluateJob.compile( + workspace="default", + spec=canonical, entity_client=object(), job_name=None, - async_sdk=_FakeSDK(), + async_sdk=object(), ) job_spec = PlatformJobSpec.model_validate(compiled) @@ -388,8 +424,43 @@ async def test_evaluate_job_compile_preserves_bundled_metric_model_refs_for_runt metric_bundle = MetricBundle.model_validate(config["metrics"][0]) metric = unbundle_metric(metric_bundle) assert isinstance(metric, LLMJudgeMetric) - assert isinstance(metric.model, ModelRef) - assert metric.model.root == "default/judge" + assert isinstance(metric.model, Model) + assert metric.model.name == "judge" + + +async def test_evaluate_job_to_spec_applies_job_params_without_model_refs() -> None: + canonical = await EvaluateJob.to_spec( + EvaluateInputSpec.model_validate( + { + "metrics": [ + _bundle_payload( + LLMJudgeMetric( + model=Model(url="http://judge.test/v1/chat/completions", name="judge"), + scores=[ + RangeScore( + name="quality", + minimum=0, + maximum=1, + parser=JSONScoreParser(json_path="quality"), + ) + ], + ) + ) + ], + "dataset": [{"output_text": "hello"}], + "params": RunConfig(), + } + ), + workspace="default", + entity_client=object(), + async_sdk=cast(Any, _FakeSDK()), + is_local=False, + ) + + assert isinstance(canonical, EvaluateSpec) + metric = unbundle_metric(canonical.metrics[0]) + assert isinstance(metric, LLMJudgeMetric) + assert metric.prompt_template == default_judge_prompt_template_chat(SupportedJobTypes.OFFLINE) async def test_evaluate_job_compile_produces_online_model_job() -> None: diff --git a/plugins/nemo-evaluator/tests/test_sdk.py b/plugins/nemo-evaluator/tests/test_sdk.py index 4bc19ba921..2741d00684 100644 --- a/plugins/nemo-evaluator/tests/test_sdk.py +++ b/plugins/nemo-evaluator/tests/test_sdk.py @@ -11,7 +11,7 @@ import httpx import pytest -from nemo_evaluator.jobs.evaluate import EvaluateJob, EvaluateSpec +from nemo_evaluator.jobs.evaluate import EvaluateInputSpec, EvaluateJob, EvaluateSpec from nemo_evaluator.sdk import http_utils from nemo_evaluator.sdk._executor import ( MetricBundlePackagerPolicyError, @@ -64,13 +64,24 @@ _EXACT_MATCH_EVALUATE_SPEC_JSON = _EXACT_MATCH_EVALUATE_SPEC.model_dump(mode="json") -def _single_metric(spec: EvaluateSpec) -> MetricBundle: +def _single_metric(spec: EvaluateInputSpec | EvaluateSpec) -> MetricBundle: """Return the single metric from an evaluator job spec.""" if len(spec.metrics) != 1: raise AssertionError("Expected a single metric spec.") return spec.metrics[0] +def _local_run_result(tmp_path: Path, result: EvaluationResult) -> EvaluatorLocalRunResult: + result_path = tmp_path / "evaluation-results.json" + result_path.write_text(result.model_dump_json(), encoding="utf-8") + return EvaluatorLocalRunResult.model_validate( + { + "status": "completed", + "artifact": {"name": "evaluation-results", "artifact_url": f"file://{result_path}"}, + } + ) + + class _RecordingMetricBundlePackager(MetricBundlePackager): """Test packager that records all runtime metrics selected for packaging.""" @@ -644,13 +655,14 @@ def test_run_uses_local_executor_execution(self, mocker: MockerFixture) -> None: remote_evaluate.assert_not_called() -def test_sync_executor_evaluate_calls_sdk_directly_without_packaging(mocker: MockerFixture) -> None: +def test_sync_executor_evaluate_runs_local_job_with_packaged_input( + tmp_path: Path, + mocker: MockerFixture, +) -> None: platform = _SyncPlatform() executor = _SyncEvaluatorPluginExecutor(platform=cast(NeMoPlatform, platform)) expected = EvaluationResult(row_scores=[], aggregate_scores=AggregatedMetricResult(scores=[])) - sdk_evaluator = mocker.Mock() - sdk_evaluator.run_sync.return_value = expected - sdk_evaluator_cls = mocker.patch("nemo_evaluator.sdk._executor.SDKEvaluator", return_value=sdk_evaluator) + run_local = mocker.patch.object(executor, "run_local", return_value=_local_run_result(tmp_path, expected)) metric = ExactMatchMetric(reference="{{item.expected}}", candidate="{{item.output}}") dataset = [{"expected": "a", "output": "a"}] @@ -660,30 +672,24 @@ def test_sync_executor_evaluate_calls_sdk_directly_without_packaging(mocker: Moc params=RunConfig(parallelism=2), ) - assert result is expected - sdk_evaluator_cls.assert_called_once_with() - sdk_evaluator.run_sync.assert_called_once_with( - metrics=metric, - dataset=dataset, - config=RunConfig(parallelism=2), - target=None, - dataset_glob_pattern=None, - prompt_template=None, - ) + assert result == expected + run_local.assert_called_once() + assert run_local.call_args.kwargs["workspace"] == "platform-ws" + spec = run_local.call_args.kwargs["spec"] + assert isinstance(spec, EvaluateInputSpec) + assert _single_metric(spec).metric_type == "exact-match" + assert spec.dataset == dataset + assert spec.params == RunConfig(parallelism=2) -def test_sync_executor_evaluate_resolves_fileset_ref_before_calling_sdk(mocker: MockerFixture) -> None: +def test_sync_executor_evaluate_encodes_fileset_ref_before_local_job( + tmp_path: Path, + mocker: MockerFixture, +) -> None: platform = _SyncPlatform() executor = _SyncEvaluatorPluginExecutor(platform=cast(NeMoPlatform, platform)) expected = EvaluationResult(row_scores=[], aggregate_scores=AggregatedMetricResult(scores=[])) - sdk_evaluator = mocker.Mock() - sdk_evaluator.run_sync.return_value = expected - mocker.patch("nemo_evaluator.sdk._executor.SDKEvaluator", return_value=sdk_evaluator) - downloaded_path = Path("/tmp/downloaded-dataset") - download_dataset_sync = mocker.patch( - "nemo_evaluator.sdk._executor.download_dataset_sync", - return_value=downloaded_path, - ) + run_local = mocker.patch.object(executor, "run_local", return_value=_local_run_result(tmp_path, expected)) metric = ExactMatchMetric(reference="{{item.expected}}", candidate="{{item.output}}") dataset = FilesetRef(root="default/helpsteer2") @@ -693,18 +699,11 @@ def test_sync_executor_evaluate_resolves_fileset_ref_before_calling_sdk(mocker: dataset_glob_pattern="validation/*.jsonl", ) - assert result is expected - download_dataset_sync.assert_called_once() - assert download_dataset_sync.call_args.kwargs["sdk"] is platform - assert download_dataset_sync.call_args.kwargs["dataset"] == FilesetRef(root="default/helpsteer2#validation/*.jsonl") - sdk_evaluator.run_sync.assert_called_once_with( - metrics=metric, - dataset=downloaded_path, - config=RunConfig(), - target=None, - dataset_glob_pattern=None, - prompt_template=None, - ) + assert result == expected + run_local.assert_called_once() + spec = run_local.call_args.kwargs["spec"] + assert isinstance(spec, EvaluateInputSpec) + assert spec.dataset == FilesetRef(root="default/helpsteer2#validation/*.jsonl") def test_sync_executor_evaluate_remote_submits_waits_and_downloads(mocker: MockerFixture) -> None: @@ -1089,13 +1088,18 @@ async def test_async_executor_remote_submit_uses_platform_async_client_headers_a @pytest.mark.asyncio -async def test_async_executor_evaluate_calls_sdk_directly_without_packaging(mocker: MockerFixture) -> None: +async def test_async_executor_evaluate_runs_local_job_with_packaged_input( + tmp_path: Path, + mocker: MockerFixture, +) -> None: platform = _AsyncPlatform() executor = _AsyncEvaluatorPluginExecutor(platform=cast(AsyncNeMoPlatform, platform)) expected = EvaluationResult(row_scores=[], aggregate_scores=AggregatedMetricResult(scores=[])) - sdk_evaluator = mocker.Mock() - sdk_evaluator.run = AsyncMock(return_value=expected) - sdk_evaluator_cls = mocker.patch("nemo_evaluator.sdk._executor.SDKEvaluator", return_value=sdk_evaluator) + run_local = mocker.patch.object( + executor, + "run_local", + new=AsyncMock(return_value=_local_run_result(tmp_path, expected)), + ) metric = ExactMatchMetric(reference="{{item.expected}}", candidate="{{item.output}}") dataset = [{"expected": "a", "output": "a"}] @@ -1105,16 +1109,14 @@ async def test_async_executor_evaluate_calls_sdk_directly_without_packaging(mock params=RunConfig(parallelism=2), ) - assert result is expected - sdk_evaluator_cls.assert_called_once_with() - sdk_evaluator.run.assert_awaited_once_with( - metrics=metric, - dataset=dataset, - config=RunConfig(parallelism=2), - target=None, - dataset_glob_pattern=None, - prompt_template=None, - ) + assert result == expected + run_local.assert_awaited_once() + assert run_local.call_args.kwargs["workspace"] == "platform-ws" + spec = run_local.call_args.kwargs["spec"] + assert isinstance(spec, EvaluateInputSpec) + assert _single_metric(spec).metric_type == "exact-match" + assert spec.dataset == dataset + assert spec.params == RunConfig(parallelism=2) @pytest.mark.asyncio