# Copyright (c) 2026-present, Yumeow. Licensed under the MIT License. """工具基类定义。 所有工具都应继承自 BaseTool，并提供统一的接口。 """ from __future__ import annotations import re import time import warnings from numbers import Real import numpy as np import sr_harness_engine as engine from pathlib import Path from logging import getLogger from scipy import stats from docstring_parser import DocstringStyle, parse from abc import ABC, abstractmethod from types import NoneType, UnionType from inspect import Parameter, signature from typing import TYPE_CHECKING, Any, Callable, Dict, List, Literal, Union, get_args, get_origin, get_type_hints from ..utils import FactoryMixin, log_exception from ..core import AgentContext from ..core.tool import ToolCallResult, ToolMetadata from ..evaluator import utils as evaluator_utils if TYPE_CHECKING: from ..skills import SkillManager _logger = getLogger(f'sr_harness.{__name__}') _ANSI_ESCAPE_RE = re.compile( r"(?:\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)|(?:\x1b\[|\x9b)[0-?]*[ -/]*[@-~]|\x1b[@-_])" ) class ToolRunAbort(RuntimeError): """Raise from a tool to bypass BaseTool.__call__ error handling.""" class BaseTool(ABC, FactoryMixin): """工具基类。所有工具都应继承此类，并设置 / 实现以下字段和方法： - metadata: ToolMetadata 实例，提供工具的名称、描述和参数 schema（若不提供则尝试自动推断） - execute(): 工具的核心执行方法，接受 LLM 生成的参数并返回结果字典。工具的 execute 方法应该尽量保持参数简单，复杂的上下文信息（如数据）可以通过工具实例的 context 属性传入。 - format_result_dict(): 可选的类方法，用于将 execute 的结果字典格式化为字符串，供 LLM 阅读。默认实现是直接转换为字符串，不同工具可以根据需要重写此方法以提供更友好的结果展示。 """ metadata: ToolMetadata = None MAX_FORMULA_LENGTH = 10000 MAX_RESULT_STR_LENGTH = 64 * 1024 FORMULA_DISPLAY_NUMBER_FORMAT = ".8g" def __init_subclass__(cls, **kwargs: Any) -> None: """在子类定义时自动设置元数据""" super().__init_subclass__(**kwargs) if cls.metadata is None: cls.metadata = ToolMetadata() if cls.metadata.description is None: cls.metadata.description = cls.infer_tool_description() if cls.metadata.parameters is None: cls.metadata.parameters = cls.infer_tool_parameters() def __init__(self, context: AgentContext | None = None) -> None: """Initialize a tool with its authoritative runtime context. Args: context: Context used by the tool. An empty context is created when omitted for tools that do not require runtime data. Raises: TypeError: If ``context`` is not an :class:`AgentContext`. """ if context is not None and not isinstance(context, AgentContext): raise TypeError("context must be an AgentContext or None") self.context = context if context is not None else AgentContext() @classmethod def get_doc(cls) -> dict[str, str] | None: """Return optional documentation to expose as a runtime skill.""" return None @abstractmethod def execute(self) -> Dict[str, Any]: """Execute the tool and return its result. The parameters of this method are generated by the LLM and should therefore remain simple. Pass complex runtime context, such as data, through the tool instance's ``context`` attribute. Implementations must be safe for concurrent process or thread execution. The text before ``Args`` supplies ``metadata.description``; argument descriptions supply the parameter schema descriptions. The signature and type hints supply its JSON schema. Automatic inference supports common scalar, list, and dictionary types. Define ``ToolMetadata.parameters`` explicitly for more complex schemas. Returns: A structured result dictionary.""" pass @classmethod def format_result_dict(cls, result: Dict[str, Any]) -> str: """Format a tool result as text for the language model. Subclasses may override this method to provide a clearer presentation. Args: result: Structured result returned by ``execute``. Returns: Text to append to the model conversation.""" return str(result) def cancel(self) -> None: """Request cancellation when a tool offers no stronger primitive.""" def __call__(self, *args, **kwargs) -> ToolCallResult: """工具调用入口""" start_time = time.time() try: cancellation_event = getattr(self, "cancel_event", None) if cancellation_event is not None and cancellation_event.is_set(): raise InterruptedError("用户强制中止") result = self.execute(*args, **kwargs) if cancellation_event is not None and cancellation_event.is_set(): raise InterruptedError("用户强制中止") if not isinstance(result, dict): _logger.critical(f"Tool {self.metadata.name} execute() should return a dict, but got {type(result)}, please check the implementation.") result_str = self.format_result_dict(result) result_str = self.truncate_result_str(_ANSI_ESCAPE_RE.sub("", str(result_str))) meta_data = { "timestamp": start_time, "execution_time": time.time() - start_time, "tool": self.metadata.name } return ToolCallResult(ok=True, result=result, result_str=result_str, meta_data=meta_data) except ToolRunAbort: raise except Exception as e: _logger.error(f"Error executing {self.metadata.name}: {log_exception(e)}") exception_message = str(e).strip() or repr(e) error_msg = ( f"Error executing {self.metadata.name}:\n" f" {type(e).__name__}: {exception_message}" ) error_msg = self.truncate_result_str(_ANSI_ESCAPE_RE.sub("", error_msg)) meta_data = { "timestamp": start_time, "execution_time": time.time() - start_time, "tool": self.metadata.name } return ToolCallResult(ok=False, result={'error': error_msg}, result_str=error_msg, meta_data=meta_data) @classmethod def normalize_formula(cls, eq: str, *, strip_modules: bool = True) -> str: """Validate and normalize an agent-provided formula before parsing it. Args: eq: Formula text supplied by the agent. strip_modules: Whether to remove supported module prefixes. Returns: The normalized formula text.""" if not isinstance(eq, str): raise TypeError(f"Formula must be a string, got {type(eq).__name__}.") eq = eq.strip() if len(eq) > cls.MAX_FORMULA_LENGTH: raise ValueError( f"Formula is too long: {len(eq)} characters; " f"maximum allowed is {cls.MAX_FORMULA_LENGTH}." ) eq = eq.replace("^", "**") if strip_modules: eq = eq.replace("np.", "").replace("numpy.", "").replace("math.", "") return eq @classmethod def parse_formula(cls, eq: str) -> engine.Expression: """Normalize and parse a formula with constants supported by all tools. Args: eq: Formula text supplied by the agent. Returns: The parsed symbolic expression.""" return engine.parse( cls.normalize_formula(eq), symbols={"pi": np.pi, "e": np.e}, ) @classmethod def truncate_result_str(cls, text: str) -> str: """Bound text sent to the LLM while preserving the full raw result. Args: text: Tool-result text to limit. Returns: The original text or a length-limited prefix with a truncation note.""" text = str(text) limit = cls.MAX_RESULT_STR_LENGTH if len(text) <= limit: return text suffix = "... (characters ignored)" keep = max(0, limit - len(suffix) - 32) ignored = len(text) - keep suffix = f"... ({ignored} characters ignored)" keep = max(0, limit - len(suffix)) ignored = len(text) - keep suffix = f"... ({ignored} characters ignored)" return text[: max(0, limit - len(suffix))] + suffix @classmethod def to_tool_list(cls, tools_used: list[str] | None = None) -> list[dict]: """Load OpenAI-compatible function-tool definitions. Args: tools_used: Tool names to include, or ``None`` for all registered tools. Returns: Function-tool definitions in OpenAI-compatible format.""" tool_list = [] for name, tool_cls in cls.REGISTRY_DICT.items(): if tools_used is None or tool_cls.metadata.name in tools_used: tool_list.append(tool_cls.to_dict()) return tool_list @classmethod def load_tool_list(cls, tools_used: list[str] | None = None) -> list[dict]: """Load tool metadata for text and JSON parsers. Args: tools_used: Tool names to include, or ``None`` for all registered tools. Returns: Names, descriptions, and parameter schemas for the selected tools.""" tool_list = [] for name, tool_cls in cls.REGISTRY_DICT.items(): if tools_used is None or tool_cls.metadata.name in tools_used: tool_list.append({ "name": tool_cls.metadata.name, "description": tool_cls.metadata.description, "parameters": tool_cls.metadata.parameters, }) return tool_list @classmethod def load_tool_classes(cls, tools_used: list[str] | None = None) -> list[type["BaseTool"]]: """Load tool classes for native function calling. Args: tools_used: Tool names to include, or ``None`` for all registered tools. Returns: The selected registered tool classes.""" tool_list = [] for name, tool_cls in cls.REGISTRY_DICT.items(): if tools_used is None or tool_cls.metadata.name in tools_used: tool_list.append(tool_cls) return tool_list @classmethod def load_custom_tool(cls, source: str, *, path: str | Path) -> dict: """Inspect one sandboxed ``BaseTool`` stored in a skill's ``tool.py``. Args: source: Python source defining the custom tool. path: Origin path used for diagnostics and source identity. Returns: Metadata and a sandbox-backed proxy class for the tool.""" path = Path(path).resolve() if path.suffix != ".py": raise ValueError(f"Custom tool source must be a Python file: {path}") from .sandboxed_custom_tool import create_sandboxed_custom_tool for registered_name, registered_cls in list(cls.REGISTRY_DICT.items()): if getattr(registered_cls, "source_path", None) == path: del cls.REGISTRY_DICT[registered_name] registry_before = dict(cls.REGISTRY_DICT) try: if len(source) > 50_000: raise ValueError("Custom tool source exceeds 50000 characters") # Parsing is side-effect free and preserves a useful SyntaxError for authors; # importing and inspecting the module still happens only in the sandbox. compile(source, str(path), "exec", dont_inherit=True) tool_cls = create_sandboxed_custom_tool(source, path) tool_name = tool_cls.metadata.name.strip() if tool_name in registry_before: raise ValueError(f"Tool name '{tool_name}' is already registered.") cls.register(tool_name)(tool_cls) return { "tool_name": tool_name, "path": str(path), "class_name": tool_cls.__name__, "tool_class": tool_cls, } except Exception: cls.REGISTRY_DICT.clear() cls.REGISTRY_DICT.update(registry_before) raise @classmethod def discover_custom_tools(cls, skill_manager: SkillManager) -> list[type["BaseTool"]]: # 将 skill_manager 中发现的自定义工具注册到 BaseTool，并返回 tool class list """Discover custom tools. Args: skill_manager: The skill manager value. Returns: list[type['BaseTool']]: The operation result. """ custom_tool_cls_list = [] for skill, source in skill_manager.load_tool_skills(): tool_path = skill.skill_directory / "tool.py" try: loaded = cls.load_custom_tool(source, path=tool_path) custom_tool_cls_list.append(cls.REGISTRY_DICT[loaded["tool_name"]]) except Exception as e: _logger.warning(f"Skipping custom tool {tool_path}: {log_exception(e)}") return custom_tool_cls_list @classmethod def to_dict(cls) -> dict: """Export an OpenAI-compatible function-tool definition. Returns: The tool name, description, and JSON parameter schema.""" return { "type": "function", "function": { "name": cls.metadata.name, "description": cls.metadata.description, "parameters": cls.metadata.parameters }, } @classmethod def infer_tool_description(cls) -> str: """Extract a tool description from the text before ``Args`` in ``execute``. Returns: The inferred description or a placeholder when none is provided.""" doc = getattr(cls.execute, "__doc__", None) or "" description_lines = [] for line in doc.splitlines(): line = line.strip() if line == "Args:": break if line: description_lines.append(line) return "\n".join(description_lines) if description_lines else "(no description provided)" @classmethod def infer_tool_parameters(cls) -> Dict[str, Any]: """Infer a parameter schema from the ``execute`` signature and docstring. Automatic inference covers common Python and typing annotations. Complex constraints, enumerations, and formats should be declared explicitly in ``ToolMetadata.parameters``. Returns: An object-shaped JSON schema for model-supplied arguments.""" properties = {} required = [] try: sig = signature(cls.execute) type_hints = get_type_hints(cls.execute) descriptions = cls.parse_args_docstring(cls.execute) except Exception as e: _logger.warning(f"Failed to parse signature or type hints for {cls.__name__} since [{type(e).__name__}] {e}") return {} for param_name, param in sig.parameters.items(): if param_name == 'self': continue if param.kind in (Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD): _logger.warning( f"Parameter '{param_name}' in {cls.__name__}.execute is a variable positional or keyword parameter, " f"which is not supported for automatic schema inference and will be ignored. " f"Please specify its schema manually in ToolMetadata.parameters if needed." ) continue schema = {} annotation = type_hints.get(param_name, param.annotation) # e.g., List[int], Optional[str], etc. schema |= cls.parse_args_typehints(annotation) # e.g., {"type": "array", "items": {"type": "integer"}} for List[int] schema |= {'description': descriptions[param_name]} if param.default is not Parameter.empty: schema |= {'default': param.default} else: required.append(param_name) properties[param_name] = schema return { "type": "object", "properties": properties, "required": required, } @staticmethod def parse_args_docstring(func: callable) -> dict[str, str]: """Extract argument descriptions from a Google-style docstring. Args: func: Callable whose docstring describes its parameters. Returns: Descriptions keyed by parameter name.""" docstring = getattr(func, "__doc__", None) parsed = parse(docstring, style=DocstringStyle.GOOGLE) description_dict = {} for param in parsed.params: if param.description is not None: description_dict[param.arg_name] = param.description set1 = set(description_dict.keys()) set2 = set(signature(func).parameters.keys()) - {"self"} if set1 - set2: _logger.warning( f"Descriptions found for parameters {set1 - set2} in {func.__qualname__} docstring, " f"but these parameters are not in the function signature. Please check for typos or remove these descriptions." ) if set2 - set1: _logger.warning( f"Parameters {set2 - set1} in {func.__qualname__} signature do not have descriptions in the docstring. " f"Please add descriptions for better LLM understanding." ) description_dict = description_dict | {param: "(no description provided)" for param in set2 - set1} return description_dict @classmethod def parse_args_typehints(cls, annotation: Any) -> Dict[str, Any]: """Convert a common Python type annotation to JSON Schema. Args: annotation: Type annotation to convert. Returns: A JSON Schema fragment.""" if annotation is Parameter.empty or annotation is Any: return {} origin = get_origin(annotation) # e.g., list, dict, Union, etc. args = get_args(annotation) # e.g., (int,) for List[int], (str, NoneType) for Optional[str], etc. if origin is Literal: values = list(args) schema = {"enum": values} if non_none_values := [v for v in values if v is not None]: schema["type"] = cls.parse_json_type(type(non_none_values[0])) return schema elif origin is list: return {"type": "array", "items": cls.parse_args_typehints(args[0] if args else Any)} elif origin is dict: return {"type": "object"} elif origin is tuple: # Tuple without type args if not args: return {"type": "array", "items": {}} # Variable-length tuple, e.g. Tuple[str, ...] elif args[-1] is Ellipsis: return {"type": "array", "items": cls.parse_args_typehints(args[0] if args else Any)} item_schemas = [cls.parse_args_typehints(arg) for arg in args] # Homogeneous fixed tuple, e.g. Tuple[str, str] if all(schema == item_schemas[0] for schema in item_schemas): return {"type": "array", "items": item_schemas[0], "minItems": len(args), "maxItems": len(args)} # Heterogeneous fixed tuple, e.g. Tuple[str, int] else: # This is less position-strict than prefixItems, but more compatible with OpenAI/Azure/OpenRouter tool schemas. # It means each item may match any of the tuple element schemas, while minItems/maxItems still enforce tuple length. _logger.warning( f"Tuple parameter with heterogeneous types detected in {cls.__name__}.execute. " f"This is not recommended since it may lead to less precise schema and LLM confusion. " f"Please consider using a more specific schema or adding descriptions to clarify the expected format." ) return {"type": "array", "items": {"anyOf": item_schemas}, "minItems": len(args), "maxItems": len(args)} elif origin in (Union, UnionType): schemas = [{"type": "null"} if arg is NoneType else cls.parse_args_typehints(arg) for arg in args] return {"anyOf": schemas} elif json_type := cls.parse_json_type(annotation): return {"type": json_type} else: _logger.warning( f"Unsupported type annotation '{annotation}' in {cls.__name__}.execute. " f"Automatic schema inference will not be able to parse this type, and it will be treated as an untyped parameter. " f"Please consider using basic types (int, float, str, bool) or common generics (List, Dict) for better compatibility, " f"or directly specify the parameter schema in ToolMetadata.parameters if a more complex structure is needed." ) return {} @staticmethod def parse_json_type(value_type: Any) -> str: """Map a Python literal type to a JSON Schema type name. Args: value_type: Python type to inspect. Returns: A JSON type name, or an empty string for an unsupported type.""" if value_type is bool: return "boolean" if value_type is int: return "integer" if value_type is float: return "number" if value_type is str: return "string" if value_type is list: return "array" if value_type is dict: return "object" return "" @classmethod def calculate_metrics(cls, f: engine.Expression, y_true: np.ndarray, y_pred: np.ndarray) -> Dict[str, Any]: """Calculate metrics for precomputed target and prediction arrays. This helper lets code-defined models reuse the symbolic tools' metric definitions. Args: f: Candidate expression used to compute structural complexity. y_true: Observed target values. y_pred: Predicted target values. Returns: Fit, correlation, information-criterion, and complexity metrics.""" return evaluator_utils.regression_metrics(f, y_true, y_pred) def evaluate( self, f: engine.Expression, y: engine.Expression, show_diagnostics: bool = True, fit: bool = False, ) -> Dict[str, Any]: """Evaluate a symbolic prediction against a symbolic target. Metrics, including formula complexity, are supplied by the active evaluator. Residual diagnostics include an error profile, worst samples, and strong residual-variable correlations. Args: f: Symbolic prediction expression. y: Symbolic target expression. show_diagnostics: Whether to include residual diagnostics. fit: Whether to fit parameters on the training split before scoring. Returns: Candidate eligibility and metrics for each available data split.""" if not isinstance(f, engine.Expression) or not isinstance(y, engine.Expression): raise TypeError("f and y must both be sr_harness_engine.Expression instances.") if self.context.target is None: raise ValueError("context.target must be configured before evaluating formulas") evaluator = self.context.evaluator target = self.context.target var_names = {var.name for var in f.iter_preorder() if isinstance(var, engine.Variable)} ineligibility_reasons = [] if y.to_str() != target: ineligibility_reasons.append(f"the left-hand side of the equation is not {target}") if target in var_names: ineligibility_reasons.append(f"the right-hand side of the equation depends on {target}") is_candidate = not ineligibility_reasons isolated_evaluation = getattr(evaluator, "_evaluate_formula_isolated", None) isolated_metrics = None if isolated_evaluation is not None: fitted_f, isolated_metrics, isolated_splits = isolated_evaluation( f, y, self.context, fit=fit, candidate=is_candidate, ) train_context = isolated_splits["train"] validation_context = isolated_splits["validation"] self.context._split_cache = isolated_splits else: train_context = self.context.train_split validation_context = self.context.validation_split fitted_f = fitted_f if isolated_metrics is not None else f if fit and isolated_metrics is None: fitted_f = ( evaluator.fit_candidate(f, train_context) if is_candidate else evaluator.fit(f, y, train_context) ) if not isinstance(fitted_f, engine.Expression): raise TypeError(f"{type(evaluator).__name__}.fit() must return an Expression") if missing := engine.unbound_parameters(fitted_f): raise ValueError(f"unbound parameters: {', '.join(missing)}") def evaluate_split(split_context: AgentContext, *, max_samples: int | None = None): result = isolated_metrics.get("train" if split_context is train_context else "validation") if isolated_metrics is not None else ( evaluator.evaluate_candidate(fitted_f, split_context) if is_candidate else evaluator.evaluate(fitted_f, y, split_context) ) if not isinstance(result, dict): raise TypeError(f"{type(evaluator).__name__}.evaluate() must return a dictionary") metrics = {} for name, value in result.items(): if isinstance(value, (bool, np.bool_)) or not isinstance(value, Real): raise TypeError(f"Evaluator metric {name!r} must be a float or int") metrics[str(name)] = value.item() if isinstance(value, np.generic) else value split = {"metrics": metrics} if show_diagnostics: num_nodes = split_context.num_nodes split["diagnostics"] = self.residual_diagnostics( y_true=np.asarray(y.eval(split_context.data, num_nodes=num_nodes)), y_pred=np.asarray(fitted_f.evaluate(split_context.data, num_nodes=num_nodes)), data=split_context.data, target_expression=y.to_str(), **({"max_samples": max_samples} if max_samples is not None else {}), ) return split data_split_results = {"train": evaluate_split(train_context, max_samples=10)} if validation_context.data: data_split_results["validation"] = evaluate_split(validation_context) for split_name, split_result in data_split_results.items(): metrics = split_result.get("metrics") or {} if not np.isfinite(metrics.get("mse", float("nan"))): ineligibility_reasons.append( f"the formula does not produce a finite MSE on the {split_name} set" ) evaluation = { "formula": fitted_f.to_str(number_format='.8g'), "target_expression": y.to_str(number_format='.8g'), "is_candidate": not ineligibility_reasons, "candidate_ineligibility_reasons": ineligibility_reasons, "data_split_results": data_split_results, } return evaluation def failed_evaluation(self, formula: str = "(None)") -> Dict[str, Any]: """Build the common result shape when a formula-producing backend fails. Args: formula: Formula or placeholder to report. Returns: A non-candidate result with infinite error metrics.""" empty_metrics = { "mse": float("inf"), "rmse": float("inf"), "mae": float("inf"), "mape": float("inf"), "r2": -float("inf"), "aic": float("inf"), "bic": float("inf"), "pearson_r": -float("inf"), "spearman_r": -float("inf"), "complexity": float("inf"), } data_split_results = { 'train': {'metrics': None}, 'validation': {'metrics': None}, } if self.context.target is None: raise ValueError("context.target must be configured before reporting a failed evaluation") data_split_results['train']['metrics'] = empty_metrics if not self.context.validation_split.data: data_split_results.pop('validation') else: data_split_results['validation']['metrics'] = empty_metrics return { "formula": formula, "target_expression": self.context.target, "is_candidate": False, "candidate_ineligibility_reasons": ["no valid formula was produced"], "data_split_results": data_split_results, } @classmethod def format_evaluation_result(cls, result: Dict[str, Any], title: str = "Formula evaluation") -> str: """Format the common formula-evaluation schema for an LLM. Args: result: Evaluation mapping produced by ``evaluate``. title: Heading shown above the formatted result. Returns: Concise formula, metric, warning, and diagnostic text.""" split_results = result["data_split_results"] train_result = split_results["train"] metrics = train_result["metrics"] lhs = result.get("target_expression", "LHS") formula = result["formula"] try: formula = engine.parse(formula).to_str( number_format=cls.FORMULA_DISPLAY_NUMBER_FORMAT ) except Exception: # Some tools use a free-form model description instead of an # SRHarness Engine expression. Preserve those descriptions verbatim. pass lines = [ f"{title}:", f" {lhs} = {formula}", ] for reason in result.get("candidate_ineligibility_reasons", []): lines.append(f" (Note: this formula is not eligible for submission since {reason}.)") validation_result = split_results.get("validation") validation_metrics = validation_result["metrics"] if validation_result else None def three_significant_digits(value: Any) -> str: if isinstance(value, (int, float, np.number)): return format(value, "#.3g").removesuffix(".") return str(value) def metric_pair(name: str) -> str: train_value = three_significant_digits(metrics.get(name, float("nan"))) validation_value = ( three_significant_digits(validation_metrics.get(name, float("nan"))) if validation_metrics else "N/A" ) return f"{train_value} | {validation_value}" lines.extend([ "Fit quality (Train-set | Validation-set):", f" RMSE={metric_pair('rmse')};", f" MAE={metric_pair('mae')};", f" R2={metric_pair('r2')};", f" Formula Complexity={three_significant_digits(metrics.get('complexity', '?'))};", ]) standard_metric_names = { "mse", "rmse", "mae", "mape", "r2", "aic", "bic", "pearson_r", "spearman_r", "complexity", } custom_metric_names = sorted( (set(metrics) | set(validation_metrics or {})) - standard_metric_names ) if custom_metric_names: lines.append("Evaluator-specific metrics (Train-set | Validation-set):") lines.extend( f" {name}={metric_pair(name)};" for name in custom_metric_names ) if validation_metrics: train_rmse = metrics.get("rmse", float("nan")) validation_rmse = validation_metrics.get("rmse", float("nan")) if (np.isfinite(train_rmse) and np.isfinite(validation_rmse) and validation_rmse > max(2.0 * train_rmse, train_rmse + 1e-12)): lines.append( "Warning: validation error is substantially larger than training error; " "the candidate may be overfitting or the validation domain may differ." ) if diagnostics := train_result.get("diagnostics"): samples = diagnostics.get("worst_samples", [])[:10] if samples: input_names = result.get("analyzed_input_expressions") or [ name for name in samples[0]["row"] if name != lhs ] lines.extend([ "Error extremes (Top-10 sorted by |residual|):", f" ({' | '.join([*input_names, lhs, 'residual'])})", ]) for sample in samples: values = [sample["row"].get(name, float("nan")) for name in input_names] values.extend([sample["y_true"], sample["y_pred"] - sample["y_true"]]) lines.append(" " + " | ".join(three_significant_digits(value) for value in values)) if correlations := diagnostics.get("strongest_residual_correlations"): strongest = correlations[0] lines.extend([ "Largest residual correlation among other analyzed variables " "(criterion=max(|Pearson|, |Spearman|)):", f" variable={strongest['variable']}", f" Pearson(residual, {strongest['variable']})={three_significant_digits(strongest['pearson'])};", f" Spearman(residual, {strongest['variable']})={three_significant_digits(strongest['spearman'])};", ]) return "\n".join(lines) @classmethod def residual_diagnostics( cls, y_true: np.ndarray, y_pred: np.ndarray, data: Dict[str, Any] = None, target_expression: str = None, max_samples: int = 3, max_correlations: int = 5, ) -> Dict[str, Any]: """Return compact diagnostics for prediction residuals. Args: y_true: Observed target values. y_pred: Predicted target values. data: Optional variables used to explain residual patterns. target_expression: Target label included in diagnostic rows. max_samples: Maximum number of worst samples to include. max_correlations: Maximum number of residual correlations to include. Returns: Error profile, worst samples, and strongest residual correlations.""" y_true, y_pred = np.broadcast_arrays( np.asarray(y_true, dtype=float), np.asarray(y_pred, dtype=float), ) y_true = y_true.flatten() y_pred = y_pred.flatten() residual = y_pred - y_true absolute_error = np.abs(residual) finite_error = np.isfinite(residual) target_scale = float(np.median(np.abs(y_true[np.isfinite(y_true)]))) if np.any(np.isfinite(y_true)) else 0.0 relative_floor = max(np.finfo(float).eps, target_scale * 1e-12) relative_valid = finite_error & np.isfinite(y_true) & (np.abs(y_true) > relative_floor) relative_error = np.full_like(absolute_error, np.nan, dtype=float) relative_error[relative_valid] = absolute_error[relative_valid] / np.abs(y_true[relative_valid]) finite_absolute = absolute_error[finite_error] finite_relative = relative_error[np.isfinite(relative_error)] error_profile = { "median_signed_residual": float(np.median(residual[finite_error])) if np.any(finite_error) else float("nan"), "p95_absolute_error": float(np.quantile(finite_absolute, 0.95)) if finite_absolute.size else float("nan"), "max_absolute_error": float(np.max(finite_absolute)) if finite_absolute.size else float("nan"), "p95_relative_error": float(np.quantile(finite_relative, 0.95)) if finite_relative.size else float("nan"), "max_relative_error": float(np.max(finite_relative)) if finite_relative.size else float("nan"), "relative_error_defined_ratio": float(np.mean(relative_valid)), } numeric_data = {} for name, values in (data or {}).items(): array = np.asarray(values) if array.size == y_true.size and np.issubdtype(array.dtype, np.number): numeric_data[name] = array.flatten() rank_score = np.where(finite_error, absolute_error, -np.inf) sample_limit = max(0, int(max_samples)) worst_indices = ( np.argsort(rank_score)[-sample_limit:][::-1] if sample_limit else np.array([], dtype=int) ) worst_samples = [] for index in worst_indices: if not np.isfinite(rank_score[index]): continue worst_samples.append({ "index": int(index), "row": {name: float(values[index]) for name, values in numeric_data.items()}, "y_true": float(y_true[index]), "y_pred": float(y_pred[index]), "absolute_error": float(absolute_error[index]), "relative_error": float(relative_error[index]), }) correlation_arrays = dict(numeric_data) if target_expression and target_expression not in correlation_arrays: correlation_arrays[target_expression] = y_true correlations = [] for name, values in correlation_arrays.items(): pearson, spearman = cls.correlation_coefficients(values, residual) strength = max(abs(pearson), abs(spearman)) if np.isfinite(strength): correlations.append({ "variable": name, "pearson": pearson, "spearman": spearman, "strength": strength, }) correlations.sort(key=lambda item: item["strength"], reverse=True) return { "error_profile": error_profile, "worst_samples": worst_samples, "strongest_residual_correlations": correlations[:max(0, int(max_correlations))], } @staticmethod def correlation_coefficients(x: np.ndarray, y: np.ndarray) -> tuple[float, float]: """Return finite-sample Pearson and Spearman coefficients. Undefined correlations, including constant inputs, are returned as NaN. Warnings for these mathematically valid edge cases are suppressed. Args: x: First numeric sample. y: Second numeric sample. Returns: Pearson and Spearman correlation coefficients.""" x = np.asarray(x, dtype=float).flatten() y = np.asarray(y, dtype=float).flatten() finite = np.isfinite(x) & np.isfinite(y) if np.count_nonzero(finite) < 2: return float("nan"), float("nan") x, y = x[finite], y[finite] with np.errstate(invalid="ignore", divide="ignore"), warnings.catch_warnings(): warnings.simplefilter("ignore") pearson = float(np.corrcoef(x, y)[0, 1]) spearman = float(stats.spearmanr(x, y).statistic) return pearson, spearman def is_numeric_array(data: List[Any]) -> bool: """Run the ``is numeric array`` operation. Args: data: Data arrays keyed by variable name. Returns: bool: The operation result. """ arr = np.asarray(data) return np.issubdtype(arr.dtype, np.integer) or np.issubdtype(arr.dtype, np.floating)