File size: 7,344 Bytes
d06fa57 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | # Copyright (c) 2026-present, Yumeow. Licensed under the MIT License.
"""Evaluate Python-defined models inside a data-only OS sandbox."""
from __future__ import annotations
from typing import Any, Dict
import numpy as np
import sr_harness_engine as engine
from ..runtime.sandbox import _run_sandbox_protocol
from .base_tool import BaseTool, ToolMetadata
from .code_executor import CodeExecutorTool
@BaseTool.register("evaluate_code")
class EvaluateCodeTool(BaseTool):
"""Implementation of the evaluate code tool."""
metadata = ToolMetadata(name="evaluate_code")
DEFAULT_TIMEOUT_SECONDS = CodeExecutorTool.DEFAULT_TIMEOUT_SECONDS
DEFAULT_MEMORY_LIMIT_MB = CodeExecutorTool.DEFAULT_MEMORY_LIMIT_MB
DEFAULT_OUTPUT_LIMIT_BYTES = CodeExecutorTool.DEFAULT_OUTPUT_LIMIT_BYTES
MAX_TIMEOUT_SECONDS = CodeExecutorTool.MAX_TIMEOUT_SECONDS
MAX_MEMORY_LIMIT_MB = CodeExecutorTool.MAX_MEMORY_LIMIT_MB
MAX_OUTPUT_LIMIT_BYTES = CodeExecutorTool.MAX_OUTPUT_LIMIT_BYTES
def execute(
self,
model_code: str,
predict_code: str,
y: str = None,
timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS,
memory_limit_mb: int = DEFAULT_MEMORY_LIMIT_MB,
output_limit_bytes: int = DEFAULT_OUTPUT_LIMIT_BYTES,
show_diagnostics: bool = True,
) -> Dict[str, Any]:
"""Evaluate a Python-defined candidate model on the current dataset.
If possible, first use ``read_skill`` to review ``sr-harness-engine-syntax`` and any
relevant companion documentation, such as ``sr-harness-engine-graph-syntax`` for graph
or hypergraph models. Prefer a structured SRHarness Engine formula when it can express
the candidate, and use this tool when it cannot do so conveniently. The code runs in a
restricted sandbox, then the tool computes metrics against the target and returns the
formatted model under the ``formula`` key.
This tool can return candidate formulas for submission when `y` is the target
variable and `predict_code` does not depend on the target variable.
Args:
model_code: Code containing exactly one function with signature
`def func(data)` plus optional top-level imports. `data` is a dictionary
mapping variable names to numeric arrays, including the target variable.
This function should return a Python dict as the fitted `model`, which
will be passed to `predict_code` and `format_code`. The returned model
should contain a `description` field that identifies the model with a
concise mathematical formula (e.g., y = aₖx² + bₖx + cₖ,
yᵢ = MLP1(xᵢ) + Σ Aᵢⱼ MLP2(xᵢ, xⱼ)).
predict_code: Code containing exactly one function with signature
`def func(data, model)` plus optional top-level imports.
The function should return the predicted value for the target `y`,
which must be array-like and compatible with the target shape.
y: Target variable name. Use target variable by default.
Expressions are also supported, e.g., "log(y)", "y - x1"
timeout_seconds: Wall-clock timeout in seconds. The effective value is capped.
memory_limit_mb: Address-space memory limit in MB. The effective value is capped.
output_limit_bytes: Limit on the amount of output (in bytes) that can be produced.
show_diagnostics: Whether metrics should include compact residual diagnostics.
"""
train_data = self.context.train_split.data
validation_data = self.context.validation_split.data
y = (y or self.context.target).strip().strip('"').strip("'")
eq_y = self.parse_formula(y)
timeout_seconds = CodeExecutorTool.bounded_int(
timeout_seconds, self.DEFAULT_TIMEOUT_SECONDS, 1, self.MAX_TIMEOUT_SECONDS
)
memory_limit_mb = CodeExecutorTool.bounded_int(
memory_limit_mb, self.DEFAULT_MEMORY_LIMIT_MB, 64, self.MAX_MEMORY_LIMIT_MB
)
output_limit_bytes = CodeExecutorTool.bounded_int(
output_limit_bytes, self.DEFAULT_OUTPUT_LIMIT_BYTES, 1024, self.MAX_OUTPUT_LIMIT_BYTES
)
groups = {"train": train_data}
if validation_data:
groups["validation"] = validation_data
result = _run_sandbox_protocol(
operation="evaluate-code",
request={
"model_code": CodeExecutorTool.extract_code(model_code),
"predict_code": CodeExecutorTool.extract_code(predict_code),
"target": self.context.target,
},
arrays=groups,
timeout_seconds=timeout_seconds,
memory_limit_mb=memory_limit_mb,
output_limit_bytes=output_limit_bytes,
interruption_event=getattr(self, "cancel_event", None),
)
predictions = result.arrays.get("predictions", {})
opaque_f = engine.parse("__code_model_prediction__")
data_split_results: dict[str, dict[str, Any]] = {}
for split_name, split_values in groups.items():
if split_name not in predictions:
raise ValueError(f"Sandbox did not return {split_name} predictions")
y_pred = np.asarray(predictions[split_name])
y_true = np.asarray(eq_y.eval(split_values))
y_pred, y_true = np.broadcast_arrays(y_pred, y_true)
metrics = self.calculate_metrics(opaque_f, y_true, y_pred)
metrics["complexity"] = len(model_code) + len(predict_code)
metrics.pop("aic", None)
metrics.pop("bic", None)
split_result: dict[str, Any] = {"metrics": metrics}
if show_diagnostics:
split_result["diagnostics"] = self.residual_diagnostics(
y_true=y_true,
y_pred=y_pred,
data=split_values,
target_expression=eq_y.to_str(),
)
data_split_results[split_name] = split_result
return {
"formula": str(result.payload["model_str"]),
"is_candidate": bool(result.payload["is_candidate"])
and eq_y.to_str() == self.context.target,
"data_split_results": data_split_results,
}
@classmethod
def format_result_dict(cls, result: Dict[str, Any]) -> str:
"""Format a tool result for the language model.
Args:
result: Result mapping to format or update.
Returns:
str: The operation result.
"""
text = cls.format_evaluation_result(result, title="Evaluated code-defined model")
marker = "Formula Complexity="
if text.count(marker) > 1:
raise ValueError(
f"Expected at most one '{marker}' field in formatted evaluation output, "
f"but found {text.count(marker)}."
)
text = text.replace(marker, "Code Complexity=")
return (
text + "\n"
"(Note: The Code Complexity is measured as source-code character count, so it is not "
"directly comparable to Formula Complexity defined as the symbolic formula node count.)"
)
|