| |
| """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.)" |
| ) |
|
|
