1
import nemo_evaluator_sdk
2
3
4
print(nemo_evaluator_sdk.version)

1
from nemo_evaluator_sdk import Metric, MetricInput, MetricResult
2
3
4
class ArithmeticExpressionCorrectnessMetric(Metric):
5
    type = "arithmetic-expression-correctness"
6
7
    def output_spec(self): ...
8
9
    async def compute_scores(self, metric_input: MetricInput) -> MetricResult: ...

1
from nemo_evaluator_sdk import MetricOutputSpec
2
3
4
def output_spec(self) -> list[MetricOutputSpec]:
5
    return [
6
        MetricOutputSpec.boolean("valid_expression"),
7
        MetricOutputSpec.boolean("correct_value"),
8
    ]

1
import ast
2
import operator
3
from collections.abc import Callable
4
5
6
MAX_EXPRESSION_CHARS = 256
7
MAX_AST_NODES = 64
8
9
_BINARY_OPERATORS: dict[type[ast.operator], Callable[[float, float], float]] = {
10
    ast.Add: operator.add,
11
    ast.Sub: operator.sub,
12
    ast.Mult: operator.mul,
13
    ast.Div: operator.truediv,
14
    ast.FloorDiv: operator.floordiv,
15
    ast.Mod: operator.mod,
16
}
17
18
_UNARY_OPERATORS: dict[type[ast.unaryop], Callable[[float], float]] = {
19
    ast.UAdd: operator.pos,
20
    ast.USub: operator.neg,
21
}
22
23
24
def _validate_expression_size(expression: str, parsed: ast.AST) -> None:
25
    if len(expression) > MAX_EXPRESSION_CHARS:
26
        raise ValueError("expression is too long")
27
28
    if sum(1 for _ in ast.walk(parsed)) > MAX_AST_NODES:
29
        raise ValueError("expression is too complex")
30
31
32
def _evaluate_ast(node: ast.AST) -> float:
33
    if isinstance(node, ast.Expression):
34
        return _evaluate_ast(node.body)
35
36
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)) and not isinstance(node.value, bool):
37
        return float(node.value)
38
39
    if isinstance(node, ast.BinOp):
40
        operator_fn = _BINARY_OPERATORS.get(type(node.op))
41
        if operator_fn is None:
42
            raise ValueError(f"unsupported operator: {type(node.op).__name__}")
43
        return operator_fn(_evaluate_ast(node.left), _evaluate_ast(node.right))
44
45
    if isinstance(node, ast.UnaryOp):
46
        operator_fn = _UNARY_OPERATORS.get(type(node.op))
47
        if operator_fn is None:
48
            raise ValueError(f"unsupported unary operator: {type(node.op).__name__}")
49
        return operator_fn(_evaluate_ast(node.operand))
50
51
    raise ValueError(f"unsupported expression: {type(node).__name__}")
52
53
54
def safe_eval_math_expression(expression: str) -> float:
55
    expression = expression.strip()
56
    parsed = ast.parse(expression, mode="eval")
57
    _validate_expression_size(expression, parsed)
58
    return _evaluate_ast(parsed)

1
import math
2
3
from nemo_evaluator_sdk import (
4
    Metric,
5
    MetricInput,
6
    MetricOutput,
7
    MetricOutputSpec,
8
    MetricResult,
9
)
10
11
12
class ArithmeticExpressionCorrectnessMetric(Metric):
13
    type = "arithmetic-expression-correctness"
14
15
    def output_spec(self) -> list[MetricOutputSpec]:
16
        return [
17
            MetricOutputSpec.boolean("valid_expression"),
18
            MetricOutputSpec.boolean("correct_value"),
19
        ]
20
21
    async def compute_scores(self, metric_input: MetricInput) -> MetricResult:
22
        expression = metric_input.candidate.output_text
23
        if not expression:
24
            return MetricResult(
25
                outputs=[
26
                    MetricOutput(name="valid_expression", value=False),
27
                    MetricOutput(name="correct_value", value=False),
28
                ]
29
            )
30
31
        try:
32
            actual = safe_eval_math_expression(expression)
33
        except (SyntaxError, ValueError, TypeError, ZeroDivisionError, OverflowError, RecursionError):
34
            return MetricResult(
35
                outputs=[
36
                    MetricOutput(name="valid_expression", value=False),
37
                    MetricOutput(name="correct_value", value=False),
38
                ]
39
            )
40
41
        expected = float(metric_input.row.data["expected"])
42
        tolerance = float(metric_input.row.data.get("tolerance", 1e-6))
43
        correct_value = math.isclose(actual, expected, rel_tol=tolerance, abs_tol=tolerance)
44
45
        return MetricResult(
46
            outputs=[
47
                MetricOutput(name="valid_expression", value=True),
48
                MetricOutput(name="correct_value", value=correct_value),
49
            ]
50
        )

1
from nemo_evaluator_sdk import Evaluator, FieldMapping, RunConfig
2
3
4
dataset = [
5
    {
6
        "question": "A box has 12 rows of pencils with 4 pencils in each row. Then 7 pencils are added. How many pencils are there?",
7
        "expected": 55,
8
        "tolerance": 1e-6,
9
        "model_expression": "(12 * 4) + 7",
10
    },
11
    {
12
        "question": "A server processed 125 requests, then processed 3 more batches of 25 requests. How many requests were processed?",
13
        "expected": 200,
14
        "tolerance": 1e-6,
15
        "model_expression": "125 + 25",
16
    },
17
    {
18
        "question": "A tank starts with 90 liters and loses 18 liters each hour for 3 hours. How many liters remain?",
19
        "expected": 36,
20
        "tolerance": 1e-6,
21
        "model_expression": "90 - (18 * 3)",
22
    },
23
    {
24
        "question": "A package contains 12 items and 4 packages are used. How many items are used?",
25
        "expected": 48,
26
        "tolerance": 1e-6,
27
        "model_expression": "__import__('os').system('echo nope')",
28
    },
29
]
30
31
metric = ArithmeticExpressionCorrectnessMetric()
32
evaluator = Evaluator()
33
result = evaluator.run_sync(
34
    metrics=metric,
35
    dataset=dataset,
36
    config=RunConfig(parallelism=1),
37
    field_mapping=FieldMapping(output="model_expression"),
38
)
39
40
result.print_summary()

1
import os
2
3
from nemo_platform import NeMoPlatform
4
5
6
WORKSPACE = "custom-python-metrics"
7
MODEL_REF = os.environ.get("NMP_EVAL_MODEL_REF", "default/my-model")
8
9
client = NeMoPlatform(
10
    base_url=os.environ.get("NMP_BASE_URL", "http://localhost:8080"),
11
    workspace=WORKSPACE,
12
)

1
from nemo_platform import ConflictError
2
3
4
try:
5
    client.workspaces.create(name=WORKSPACE)
6
    print(f"Workspace '{WORKSPACE}' created")
7
except ConflictError:
8
    print(f"Workspace '{WORKSPACE}' already exists, continuing...")

1
import json
2
from pathlib import Path
3
4
from nemo_evaluator.sdk import FilesetRef
5
6
7
DATASET_NAME = "arithmetic-expression-data"
8
DATASET_FILE = "math-expressions.jsonl"
9
10
remote_rows = [
11
    {
12
        "question": "A box has 12 rows of pencils with 4 pencils in each row. Then 7 pencils are added. How many pencils are there?",
13
        "expected": 55,
14
        "tolerance": 1e-6,
15
    },
16
    {
17
        "question": "A server processed 125 requests, then processed 3 more batches of 25 requests. How many requests were processed?",
18
        "expected": 200,
19
        "tolerance": 1e-6,
20
    },
21
    {
22
        "question": "A tank starts with 90 liters and loses 18 liters each hour for 3 hours. How many liters remain?",
23
        "expected": 36,
24
        "tolerance": 1e-6,
25
    },
26
]
27
28
dataset_path = Path(DATASET_FILE)
29
dataset_path.write_text(
30
    "".join(json.dumps(row) + "\n" for row in remote_rows),
31
    encoding="utf-8",
32
)
33
34
try:
35
    fileset = client.files.filesets.create(
36
        name=DATASET_NAME,
37
        description="Math expression evaluation dataset",
38
        purpose="dataset",
39
        metadata={
40
            "dataset": {
41
                "schema": {
42
                    "type": "object",
43
                    "properties": {
44
                        "question": {"type": "string"},
45
                        "expected": {"type": "number"},
46
                        "tolerance": {"type": "number"},
47
                    },
48
                    "required": ["question", "expected"],
49
                    "additionalProperties": True,
50
                }
51
            }
52
        },
53
    )
54
    print(f"Created fileset: {fileset.workspace}/{fileset.name}")
55
except ConflictError:
56
    fileset = client.files.filesets.retrieve(name=DATASET_NAME)
57
    print(f"Fileset exists: {fileset.workspace}/{fileset.name}")
58
59
client.files.upload(
60
    fileset=fileset.name,
61
    local_path=str(dataset_path),
62
    remote_path=DATASET_FILE,
63
)
64
65
dataset_ref = FilesetRef(root=f"{fileset.workspace}/{fileset.name}").with_fragment(DATASET_FILE)
66
print(f"Using dataset: {dataset_ref.root}")

1
from nemo_evaluator.shared.metric_bundles.cloudpickle import CloudpickleMetricBundlePackager
2
from nemo_evaluator_sdk import InferenceParams, ModelRef, RunConfigOnlineModel
3
4
5
job = client.evaluator.submit(
6
    metric=metric,
7
    dataset=dataset_ref,
8
    config=RunConfigOnlineModel(
9
        parallelism=2,
10
        limit_samples=3,
11
        inference=InferenceParams(
12
            temperature=0.0,
13
            max_tokens=32,
14
        ),
15
    ),
16
    target=ModelRef(root=MODEL_REF),
17
    prompt_template=(
18
        "Return exactly one valid Python arithmetic expression that evaluates to the answer.\n"
19
        "Start immediately with the expression. Do not start with a newline.\n"
20
        "Do not include markdown, code fences, prose, units, the final answer, or any explanation.\n"
21
        "Use only numbers, whitespace, parentheses, and these operators: +, -, *, /, //, %.\n\n"
22
        "Question: {{item.question}}\n"
23
        "Expression:"
24
    ),
25
    metric_bundle_packager=CloudpickleMetricBundlePackager(),
26
)
27
28
print(f"Submitted job: {job.name}")
29
job.wait_until_done()
30
remote_result = job.get_result()
31
remote_result.print_summary()

1
for score in remote_result.aggregate_scores.scores:
2
    print(f"{score.name}: mean={score.mean}, count={score.count}, nan_count={score.nan_count}")

1
for row in remote_result.row_scores:
2
    print("Question:", row.item["question"])
3
    print("Model output:", row.sample.get("output_text"))
4
    print("Metric outputs:", row.metrics)
5
    print()