feat(evaluator): bundle metrics for plugin execution

packages/nemo_evaluator_sdk/pyproject.toml

@@ -16,6 +16,7 @@ classifiers = [
   "Programming Language :: Python :: 3.11",
 ]
 dependencies = [
+  "cloudpickle>=3.1.1",
   "pydantic>=2.10.6",
   "jinja2>=3.1.6",
   "jsonschema>=4.23.0",

packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/__init__.py

@@ -12,13 +12,22 @@
     EvaluationError,
     EvaluationPhase,
 )
+from nemo_evaluator_sdk.metrics.base import (
+    MetricBundle,
+    MetricBundlePayload,
+    MetricBundler,
+    MetricBundlingError,
+    MetricMetadata,
+)
 from nemo_evaluator_sdk.metrics.bleu import BLEUMetric
+from nemo_evaluator_sdk.metrics.cloudpickle import CloudpickleMetricBundler, CloudpickleMetricPayload
 from nemo_evaluator_sdk.metrics.exact_match import ExactMatchMetric
 from nemo_evaluator_sdk.metrics.f1 import F1Metric
 from nemo_evaluator_sdk.metrics.llm_judge import LLMJudgeMetric
 from nemo_evaluator_sdk.metrics.number_check import NumberCheckMetric
 from nemo_evaluator_sdk.metrics.protocol import (
     Metric,
+    MetricTypeName,
     validate_metric_result,
 )
 from nemo_evaluator_sdk.metrics.remote import NemoAgentToolkitRemoteMetric, RemoteMetric
@@ -69,6 +78,10 @@
 __all__ = [
     "BLEUMetric",
     "Agent",
+    "MetricBundle",
+    "MetricBundlePayload",
+    "CloudpickleMetricBundler",
+    "CloudpickleMetricPayload",
     "EvaluationError",
     "EvaluationPhase",
     "DatasetLoadError",
@@ -85,11 +98,15 @@
     "InferenceStructuredOutput",
     "JSONScoreParser",
     "Metric",
+    "MetricBundler",
+    "MetricBundlingError",
     "MetricDescriptor",
     "MetricInput",
+    "MetricMetadata",
     "MetricOutput",
     "MetricOutputSpec",
     "MetricResult",
+    "MetricTypeName",
     "LLMJudgeMetric",
     "BooleanValue",
     "CandidateOutput",

packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/metrics/base.py

@@ -0,0 +1,185 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Backend-neutral metric bundle models and protocols."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from collections.abc import Callable, Mapping
+from typing import Any, Literal, Protocol, cast
+
+from nemo_evaluator_sdk.metrics.protocol import Metric, MetricOutputSpec, MetricTypeName, MetricWithSecrets
+from nemo_evaluator_sdk.values.common import SecretRef
+from pydantic import BaseModel, ConfigDict, Field, SerializeAsAny, field_serializer, field_validator, model_validator
+
+
+class MetricBundlingError(ValueError):
+    """Raised when a metric cannot be bundled or hydrated."""
+
+
+class MetricMetadata(BaseModel):
+    """User-facing metadata captured with a bundled metric."""
+
+    model_config = ConfigDict(extra="allow", revalidate_instances="never")
+
+    description: str | None = None
+    labels: dict[str, str] = Field(default_factory=dict)
+
+    @field_validator("labels")
+    @classmethod
+    def _labels_must_be_strings(cls, value: dict[str, str]) -> dict[str, str]:
+        for key, label_value in value.items():
+            if not isinstance(key, str) or not isinstance(label_value, str):
+                raise ValueError("metric labels must be a mapping of string keys to string values")
+        return value
+
+
+class BundledMetricOutputSpec(BaseModel):
+    """JSON-safe projection of a runtime metric output spec."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    name: str
+    description: str | None = None
+    value_json_schema: dict[str, Any]
+
+    @classmethod
+    def from_output_spec(cls, output: MetricOutputSpec) -> "BundledMetricOutputSpec":
+        """Capture the serializable contract for one runtime output."""
+        return cls(
+            name=output.name,
+            description=output.description,
+            value_json_schema=output.value_json_schema(),
+        )
+
+
+class MetricBundlePayload(BaseModel, ABC):
+    """Base class for concrete Pydantic metric bundle payloads."""
+
+    @property
+    @abstractmethod
+    def kind(self) -> str:
+        """Payload discriminator used to select the bundler implementation."""
+        ...
+
+
+_PAYLOAD_TYPES: dict[str, type[MetricBundlePayload]] = {}
+_BUNDLER_FACTORIES: dict[str, Callable[[], MetricBundler]] = {}
+
+
+def _payload_kind(payload: MetricBundlePayload) -> str:
+    kind = payload.kind
+    if not kind:
+        raise MetricBundlingError("metric bundle payload kind must not be empty")
+    return kind
+
+
+def register_metric_bundle_payload(kind: str, payload_type: type[MetricBundlePayload]) -> None:
+    """Register a concrete Pydantic payload model for a bundle kind."""
+    if not kind:
+        raise ValueError("metric bundle payload kind must not be empty")
+    _PAYLOAD_TYPES[kind] = payload_type
+
+
+def register_metric_bundler(kind: str, factory: Callable[[], MetricBundler]) -> None:
+    """Register a metric bundler factory for a payload kind."""
+    if not kind:
+        raise ValueError("metric bundle payload kind must not be empty")
+    _BUNDLER_FACTORIES[kind] = factory
+
+
+class MetricBundle(BaseModel):
+    """Standalone executable metric bundle entity used by backend execution."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    bundle_kind: Literal["metric-bundle"] = "metric-bundle"
+    bundle_format_version: Literal["v1"] = "v1"
+    metric_type: MetricTypeName
+    metadata: MetricMetadata = Field(default_factory=MetricMetadata)
+    outputs: list[BundledMetricOutputSpec] = Field(min_length=1)
+    secrets: dict[str, SecretRef] = Field(default_factory=dict)
+    payload: SerializeAsAny[MetricBundlePayload]
+    digest: str
+
+    @field_serializer("payload")
+    def _serialize_payload(self, payload: MetricBundlePayload) -> dict[str, Any]:
+        value = payload.model_dump(mode="json")
+        value["kind"] = _payload_kind(payload)
+        return value
+
+    @field_validator("payload", mode="before")
+    @classmethod
+    def _payload_must_have_kind(cls, value: object) -> object:
+        if isinstance(value, MetricBundlePayload):
+            return value
+        if not isinstance(value, Mapping):
+            raise ValueError("metric bundle payload must be an object")
+        payload_data = cast(Mapping[str, object], value)
+        kind = payload_data.get("kind")
+        if not isinstance(kind, str) or not kind:
+            raise ValueError("metric bundle payload must include a non-empty kind")
+        payload_type = _PAYLOAD_TYPES.get(kind)
+        if payload_type is None:
+            raise ValueError(f"unsupported metric bundle payload kind: {kind}")
+        return payload_type.model_validate(value)
+
+    @model_validator(mode="after")
+    def _output_names_must_be_unique(self) -> "MetricBundle":
+        names = [output.name for output in self.outputs]
+        duplicates = sorted({name for name in names if names.count(name) > 1})
+        if duplicates:
+            raise ValueError(f"duplicate metric output names: {duplicates}")
+        return self
+
+
+class MetricBundler(Protocol):
+    """Interface for metric bundle implementations."""
+
+    def bundle(self, metric: Metric) -> MetricBundle:
+        """Serialize an executable metric to a bundle entity."""
+        ...
+
+    def unbundle(self, metric: MetricBundle) -> Metric:
+        """Hydrate an executable metric from a bundle entity."""
+        ...
+
+
+def metric_bundler_for_payload(payload: MetricBundlePayload) -> MetricBundler:
+    """Create the bundler registered for a metric bundle payload."""
+    kind = _payload_kind(payload)
+    factory = _BUNDLER_FACTORIES.get(kind)
+    if factory is None:
+        raise MetricBundlingError(f"unsupported metric bundle payload kind: {kind}")
+    return factory()
+
+
+def validate_metric_type(metric: Metric) -> str:
+    """Return the runtime metric type after validating the protocol contract."""
+    value = metric.type
+    if not isinstance(value, str):
+        raise MetricBundlingError("metric type must be a string")
+    if not value:
+        raise MetricBundlingError("metric type must not be empty")
+    return value
+
+
+def metric_metadata(metric: Metric) -> MetricMetadata:
+    """Capture optional runtime metric metadata."""
+    description = getattr(metric, "description", None)
+    if description is not None and not isinstance(description, str):
+        raise MetricBundlingError("metric description must be a string when provided")
+
+    raw_labels = getattr(metric, "labels", None) or {}
+    if not isinstance(raw_labels, Mapping):
+        raise MetricBundlingError("metric labels must be a mapping when provided")
+    labels = dict(raw_labels)
+    return MetricMetadata(description=description, labels=labels)
+
+
+def metric_secrets(metric: Metric) -> dict[str, SecretRef]:
+    """Capture secret environment mappings needed to execute one metric."""
+    if not isinstance(metric, MetricWithSecrets):
+        return {}
+    return metric.secrets()

packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/metrics/cloudpickle.py

@@ -0,0 +1,118 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Cloudpickle-backed metric bundle implementation."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import pickle
+import platform
+from typing import Literal
+
+import cloudpickle
+from nemo_evaluator_sdk.metrics.base import (
+    BundledMetricOutputSpec,
+    MetricBundle,
+    MetricBundlePayload,
+    MetricBundler,
+    MetricBundlingError,
+    metric_metadata,
+    metric_secrets,
+    register_metric_bundle_payload,
+    register_metric_bundler,
+    validate_metric_type,
+)
+from nemo_evaluator_sdk.metrics.protocol import Metric
+from pydantic import ConfigDict, field_validator
+
+
+class CloudpickleMetricPayload(MetricBundlePayload):
+    """Cloudpickle payload for an executable metric object."""
+
+    model_config = ConfigDict(extra="ignore", ser_json_bytes="base64", val_json_bytes="base64")
+
+    python_version: str
+    cloudpickle_version: str
+    pickle_protocol: int
+    blob: bytes
+
+    @property
+    def kind(self) -> Literal["cloudpickle"]:
+        """Payload discriminator used by the metric bundle registry."""
+        return "cloudpickle"
+
+    @classmethod
+    def from_blob(cls, blob: bytes) -> CloudpickleMetricPayload:
+        """Create a JSON-safe cloudpickle payload from raw bytes."""
+        return cls(
+            python_version=platform.python_version(),
+            cloudpickle_version=cloudpickle.__version__,
+            pickle_protocol=pickle.HIGHEST_PROTOCOL,
+            blob=blob,
+        )
+
+    @field_validator("blob")
+    @classmethod
+    def _blob_must_not_be_empty(cls, value: bytes) -> bytes:
+        if not value:
+            raise ValueError("blob must not be empty")
+        try:
+            encoded = base64.b64encode(value).decode("ascii")
+            base64.b64decode(encoded.encode("ascii"), validate=True)
+        except Exception as e:
+            raise ValueError("blob must be JSON-serializable as base64") from e
+        return value
+
+    def blob_bytes(self) -> bytes:
+        """Return the decoded cloudpickle payload bytes."""
+        return self.blob
+
+
+class CloudpickleMetricBundler(MetricBundler):
+    """Cloudpickle-backed metric bundler.
+
+    Cloudpickle bundles execute arbitrary Python code when hydrated. This
+    implementation is intended for explicit opt-in development/MVP use.
+    """
+
+    def bundle(self, metric: Metric) -> MetricBundle:
+        """Serialize a runtime metric object to a cloudpickle bundle."""
+        if not isinstance(metric, Metric):
+            raise MetricBundlingError("object does not satisfy the Metric protocol")
+
+        blob = cloudpickle.dumps(metric, protocol=pickle.HIGHEST_PROTOCOL)
+        digest = hashlib.sha256(blob).hexdigest()
+        return MetricBundle(
+            metric_type=validate_metric_type(metric),
+            metadata=metric_metadata(metric),
+            outputs=[BundledMetricOutputSpec.from_output_spec(output) for output in metric.output_spec()],
+            secrets=metric_secrets(metric),
+            payload=CloudpickleMetricPayload.from_blob(blob),
+            digest=digest,
+        )
+
+    def unbundle(self, metric: MetricBundle) -> Metric:
+        """Hydrate a metric from a cloudpickle bundle."""
+        payload = CloudpickleMetricPayload.model_validate(metric.payload.model_dump(mode="python"))
+        blob = payload.blob_bytes()
+        digest = hashlib.sha256(blob).hexdigest()
+        if digest != metric.digest:
+            raise MetricBundlingError("metric bundle digest does not match payload")
+
+        hydrated_metric = cloudpickle.loads(blob)
+        if not isinstance(hydrated_metric, Metric):
+            raise MetricBundlingError("unbundled object does not satisfy the Metric protocol")
+
+        output_names = [output.name for output in hydrated_metric.output_spec()]
+        bundled_output_names = [output.name for output in metric.outputs]
+        if output_names != bundled_output_names:
+            raise MetricBundlingError("unbundled metric output spec does not match bundle metadata")
+        if validate_metric_type(hydrated_metric) != metric.metric_type:
+            raise MetricBundlingError("unbundled metric type does not match bundle metadata")
+        return hydrated_metric
+
+
+register_metric_bundle_payload("cloudpickle", CloudpickleMetricPayload)
+register_metric_bundler("cloudpickle", CloudpickleMetricBundler)