Spaces:
Sleeping
Sleeping
File size: 1,678 Bytes
116524e | 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 | """CheckpointStep — periodically saves the skillbook to disk."""
from __future__ import annotations
import logging
from pathlib import Path
from ..core.skillbook import Skillbook
from ..core.context import ACEStepContext
logger = logging.getLogger(__name__)
class CheckpointStep:
"""Save the skillbook to disk at a configurable interval.
Optional tail step appended by factory methods when ``checkpoint_dir``
is provided.
Stateless — uses ``ctx.global_sample_index`` for interval logic.
Saves both a numbered checkpoint and a ``latest.json`` that is
always overwritten with the most recent state.
"""
requires: frozenset[str] = frozenset({"global_sample_index"})
provides: frozenset[str] = frozenset()
def __init__(
self,
directory: str | Path,
skillbook: Skillbook,
*,
interval: int = 10,
) -> None:
self.directory = Path(directory)
self.skillbook = skillbook
self.interval = interval
def __call__(self, ctx: ACEStepContext) -> ACEStepContext:
if ctx.global_sample_index % self.interval != 0:
return ctx
self.directory.mkdir(parents=True, exist_ok=True)
numbered = self.directory / f"checkpoint_{ctx.global_sample_index}.json"
latest = self.directory / "latest.json"
self.skillbook.save_to_file(str(numbered))
self.skillbook.save_to_file(str(latest))
logger.info(
"CheckpointStep: saved checkpoint at sample %d → %s",
ctx.global_sample_index,
numbered,
)
return ctx
|