Primitives

The seven specialised scientific agent primitives.

primitives

Agent primitives — specialized roles for scientific workflow nodes.

AlgorithmAgent

Bases: Agent

Base class for algorithm and model implementation agents.

Attributes:

Name	Type	Description
assumptions	list[str]	ClassVar list of approximations and model assumptions made by this method.

assumption_summary classmethod

assumption_summary()

Returns a Markdown summary of declared assumptions.

Source code in python/src/science_adk/primitives/algorithm.py

@classmethod
def assumption_summary(cls) -> str:
    """Returns a Markdown summary of declared assumptions."""
    if not cls.assumptions:
        return "No assumptions declared."
    return "\n".join(f"- {a}" for a in cls.assumptions)

CompositeAgent

Bases: Agent

A node that executes an entire nested workflow DAG.

Attributes:

Name	Type	Description
workflow	str	Relative path to the nested workflow.json file.

execute async

execute()

Executes the nested workflow and surfaces its terminal outputs.

Returns:

Type	Description
dict[str, Any]	Dictionary of outputs mapped to declared output ports.

Raises:

Type	Description
AgentError	If the workflow path is missing, not in runtime, or outputs mismatch.

Source code in python/src/science_adk/primitives/composite.py

async def execute(self) -> dict[str, Any]:
    """Executes the nested workflow and surfaces its terminal outputs.

    Returns:
      Dictionary of outputs mapped to declared output ports.

    Raises:
      AgentError: If the workflow path is missing, not in runtime, or
        outputs mismatch.
    """
    from ..runner import run_nested

    if not self.workflow:
        raise AgentError(
            f"{self.name}: composite node has no `workflow` path. "
            "Set it in workflow.json or on the class."
        )
    if self._services is None:
        raise AgentError(
            f"{self.name}: composite nodes can only run within the runtime."
        )

    inputs = {port.name: await self.input(port.name) for port in self.ports.inputs}
    await self.log(f"entering nested workflow {self.workflow}")

    outputs = await run_nested(self.workflow, inputs, self._services)

    declared = {p.name for p in self.ports.outputs}
    missing = sorted(declared - set(outputs))
    if missing:
        raise AgentError(
            f"{self.name}: nested workflow did not produce declared"
            f" output(s): {', '.join(missing)}. Check terminal nodes of"
            f" {self.workflow}."
        )
    return {name: outputs[name] for name in declared}

ConfigAgent

Bases: Agent

Emits configuration settings, hyperparameters, and physical constants.

settings

settings(**defaults)

Merges default values with workflow params overrides.

Parameters:

Name	Type	Description	Default
**defaults	Any	Default configuration key-value pairs.	{}

Returns:

Type	Description
dict[str, Any]	Combined settings dictionary.

Source code in python/src/science_adk/primitives/config.py

def settings(self, **defaults: Any) -> dict[str, Any]:
    """Merges default values with workflow params overrides.

    Args:
      **defaults: Default configuration key-value pairs.

    Returns:
      Combined settings dictionary.
    """
    return {**defaults, **self.params}

DataAgent

Bases: Agent

Produces or ingests dataset payloads for experiment workflows.

Attributes:

Name	Type	Description
source_kind	str	ClassVar string indicating data provenance ('measured', 'simulated', 'synthetic').

check_frame

check_frame(frame, columns, min_rows=1)

Validates that a DataFrame has required columns and minimum row count.

Parameters:

Name	Type	Description	Default
frame	Any	The DataFrame object to validate.	required
columns	list[str]	List of required column names.	required
min_rows	int	Minimum required number of rows.	1

Returns:

Type	Description
Any	The validated DataFrame.

Raises:

Type	Description
AgentError	If columns are missing or row count is insufficient.

Source code in python/src/science_adk/primitives/data.py

def check_frame(self, frame: Any, columns: list[str], min_rows: int = 1) -> Any:
    """Validates that a DataFrame has required columns and minimum row count.

    Args:
      frame: The DataFrame object to validate.
      columns: List of required column names.
      min_rows: Minimum required number of rows.

    Returns:
      The validated DataFrame.

    Raises:
      AgentError: If columns are missing or row count is insufficient.
    """
    actual = list(getattr(frame, "columns", []))
    missing = [c for c in columns if c not in actual]
    if missing:
        raise AgentError(
            f"{self.name}: produced data is missing column(s) "
            f"{', '.join(missing)}. Present:"
            f" {', '.join(map(str, actual)) or 'none'}"
        )
    rows = len(frame)
    if rows < min_rows:
        raise AgentError(
            f"{self.name}: produced only {rows} row(s), expected at least"
            f" {min_rows}."
        )
    return frame

EvaluationAgent

Bases: Agent

Measures the experiment's quantitative outcome.

report

report(metric, value, detail=None, higher_is_better=True)

Constructs standardized evaluation output port dictionary.

Parameters:

Name	Type	Description	Default
metric	str	Name of the evaluated metric.	required
value	float	Quantitative numeric result.	required
detail	dict[str, Any] | None	Optional supporting dictionary of details.	None
higher_is_better	bool	Optimization direction indicator.	True

Returns:

Type	Description
dict[str, Any]	Output port dictionary for execute().

Raises:

Type	Description
AgentError	If the value is non-numeric or NaN.

Source code in python/src/science_adk/primitives/evaluation.py

def report(
    self,
    metric: str,
    value: float,
    detail: dict[str, Any] | None = None,
    higher_is_better: bool = True,
) -> dict[str, Any]:
    """Constructs standardized evaluation output port dictionary.

    Args:
      metric: Name of the evaluated metric.
      value: Quantitative numeric result.
      detail: Optional supporting dictionary of details.
      higher_is_better: Optimization direction indicator.

    Returns:
      Output port dictionary for execute().

    Raises:
      AgentError: If the value is non-numeric or NaN.
    """
    try:
        numeric = float(value)
    except (TypeError, ValueError):
        raise AgentError(
            f"{self.name}: metric {metric!r} must be a number, got {value!r}."
        ) from None

    if numeric != numeric:
        raise AgentError(
            f"{self.name}: metric {metric!r} evaluated to NaN. A NaN result"
            " is a failed computation, not a score."
        )

    payload = dict(detail or {})
    payload["higher_is_better"] = higher_is_better
    return {"metric": metric, "value": numeric, "detail": payload}

TrainingAgent

Bases: Agent

Fits models to data and manages reproducible train/test splits.

Attributes:

Name	Type	Description
seed	int | None	Seed for reproducible random splitting and initialization.

split

split(data, test_fraction=0.2, shuffle=True)

Splits dataset into (train, test) subsets deterministically under seed.

Parameters:

Name	Type	Description	Default
data	Any	Data sequence or DataFrame.	required
test_fraction	float	Fraction of rows allocated to the test set (0 < frac < 1).	0.2
shuffle	bool	Whether to shuffle before splitting.	True

Returns:

Type	Description
tuple[Any, Any]	Tuple of (train_data, test_data).

Raises:

Type	Description
AgentError	If test_fraction is invalid or data has fewer than 2 elements.

Source code in python/src/science_adk/primitives/training.py

def split(
    self,
    data: Any,
    test_fraction: float = 0.2,
    shuffle: bool = True,
) -> tuple[Any, Any]:
    """Splits dataset into (train, test) subsets deterministically under seed.

    Args:
      data: Data sequence or DataFrame.
      test_fraction: Fraction of rows allocated to the test set (0 < frac < 1).
      shuffle: Whether to shuffle before splitting.

    Returns:
      Tuple of (train_data, test_data).

    Raises:
      AgentError: If test_fraction is invalid or data has fewer than 2 elements.
    """
    if not 0.0 < test_fraction < 1.0:
        raise AgentError(
            f"test_fraction must be between 0 and 1, got {test_fraction}"
        )

    total = len(data)
    if total < 2:
        raise AgentError(
            f"{self.name}: need at least 2 rows to split, got {total}."
        )

    indices = list(range(total))
    if shuffle:
        random.Random(self.seed).shuffle(indices)

    cut = max(1, int(round(total * (1.0 - test_fraction))))
    train_idx, test_idx = indices[:cut], indices[cut:]

    if hasattr(data, "iloc"):
        return data.iloc[train_idx], data.iloc[test_idx]
    return [data[i] for i in train_idx], [data[i] for i in test_idx]

seed_everything

seed_everything()

Sets random seed across Python standard library, NumPy, and PyTorch.

Source code in python/src/science_adk/primitives/training.py

def seed_everything(self) -> None:
    """Sets random seed across Python standard library, NumPy, and PyTorch."""
    if self.seed is None:
        return
    random.seed(self.seed)
    try:
        import numpy

        numpy.random.seed(self.seed)
    except ImportError:
        pass
    try:
        import torch

        torch.manual_seed(self.seed)
    except ImportError:
        pass

VisualizationAgent

Bases: Agent

Renders visual plots and figures from results produced upstream.

figure_path

figure_path(name, extension='png')

Reserves a path for a figure file inside this run's data directory.

Parameters:

Name	Type	Description	Default
name	str	Base figure name.	required
extension	str	File extension without leading dot.	'png'

Returns:

Type	Description
Path	Path to the reserved figure file.

Source code in python/src/science_adk/primitives/visualization.py

def figure_path(self, name: str, extension: str = "png") -> Path:
    """Reserves a path for a figure file inside this run's data directory.

    Args:
      name: Base figure name.
      extension: File extension without leading dot.

    Returns:
      Path to the reserved figure file.
    """
    filename = name if name.endswith(f".{extension}") else f"{name}.{extension}"
    return self.artifact(filename)

save_figure

save_figure(figure, name, dpi=150)

Saves a matplotlib figure and returns its run-relative path.

Parameters:

Name	Type	Description	Default
figure	Any	Matplotlib Figure object.	required
name	str	Name of the figure.	required
dpi	int	Resolution in dots per inch.	150

Returns:

Type	Description
str	Run-relative path string.

Source code in python/src/science_adk/primitives/visualization.py

def save_figure(self, figure: Any, name: str, dpi: int = 150) -> str:
    """Saves a matplotlib figure and returns its run-relative path.

    Args:
      figure: Matplotlib Figure object.
      name: Name of the figure.
      dpi: Resolution in dots per inch.

    Returns:
      Run-relative path string.
    """
    path = self.figure_path(name)
    figure.savefig(path, dpi=dpi, bbox_inches="tight")
    try:
        import matplotlib.pyplot

        matplotlib.pyplot.close(figure)
    except ImportError:
        pass
    return f"data/{path.name}"