Skip to content

workspace

agentic_ml.core.workspace

On-disk layout of a research run.

A research directory looks like::

<research_dir>/
    run.json            # run configuration (for reproducibility)
    leaderboard.csv     # metric table, the audit of every trial
    _dataset.joblib     # the research dataset, saved once
    session/            # Strands SnapshotSessionManager state (conversation history)
    trial_1/
        model.py
        helpers.py      # optional
        description.txt # the agent's natural-language description of the trial
        error.txt       # only present if the trial failed
    trial_2/
        ...

Workspace 

Workspace(root: str | Path)

Owns paths and file operations for a single research run.

Source code in src/agentic_ml/core/workspace.py 
34
35
36
37
def __init__(self, root: str | Path) -> None:
    # Resolved eagerly: trials are evaluated in a subprocess with a different cwd, so a
    # relative root would silently break dataset/trial lookups inside the sandbox.
    self.root = Path(root).resolve()

save_dataset 

save_dataset(x_data: DataFrame, y_data: Series) -> None

Persist the research dataset once so trials can reload it in the subprocess.

Source code in src/agentic_ml/core/workspace.py 
54
55
56
def save_dataset(self, x_data: pd.DataFrame, y_data: pd.Series) -> None:
    """Persist the research dataset once so trials can reload it in the subprocess."""
    joblib.dump((x_data, y_data), self.dataset_path)

next_trial_id 

next_trial_id() -> str

Return the next unused trial_N id.

Source code in src/agentic_ml/core/workspace.py 
61
62
63
64
65
66
67
68
def next_trial_id(self) -> str:
    """Return the next unused ``trial_N`` id."""
    numbers = [
        int(match.group(1))
        for child in self.root.glob("trial_*")
        if child.is_dir() and (match := _TRIAL_RE.match(child.name))
    ]
    return f"trial_{max(numbers, default=0) + 1}"

write_trial 

write_trial(
    trial_id: str, model_py: str, helpers_py: str = ""
) -> Path

Write the agent-provided code for a trial and return its directory.

Source code in src/agentic_ml/core/workspace.py 
73
74
75
76
77
78
79
80
def write_trial(self, trial_id: str, model_py: str, helpers_py: str = "") -> Path:
    """Write the agent-provided code for a trial and return its directory."""
    directory = self.trial_dir(trial_id)
    directory.mkdir(parents=True, exist_ok=True)
    (directory / "model.py").write_text(model_py)
    if helpers_py.strip():
        (directory / "helpers.py").write_text(helpers_py)
    return directory

write_error 

write_error(trial_id: str, error: str) -> None

Persist the failure reason so a failed trial can be debugged after the fact.

Source code in src/agentic_ml/core/workspace.py 
85
86
87
def write_error(self, trial_id: str, error: str) -> None:
    """Persist the failure reason so a failed trial can be debugged after the fact."""
    (self.trial_dir(trial_id) / "error.txt").write_text(error)

copy_trial_artifacts 

copy_trial_artifacts(
    trial_id: str, destination: str | Path
) -> Path

Copy a trial's source files into destination (used when exporting a model).

Source code in src/agentic_ml/core/workspace.py 
 97
 98
 99
100
101
102
103
def copy_trial_artifacts(self, trial_id: str, destination: str | Path) -> Path:
    """Copy a trial's source files into ``destination`` (used when exporting a model)."""
    destination = Path(destination)
    destination.mkdir(parents=True, exist_ok=True)
    for name, content in self.read_trial_code(trial_id).items():
        (destination / name).write_text(content)
    return destination