lasagnastack.base

lasagnastack.base

Pipeline

Bases: ABC

An ordered sequence of Stages run end-to-end.

Subclasses declare stages. The concrete run() handles MLflow tracking, per-stage spans, confirmation prompts, and pipeline logging. Override the three hook methods to customise observability behaviour without touching the orchestration logic.

run

run(
    state: PipelineState, auto_confirm: bool = False
) -> PipelineState

Run all stages in order, wrapped in MLflow tracking.

Opens an MLflow experiment run and a top-level CHAIN span named "{ClassName}.run" (using the concrete subclass name). Each stage is executed via _run_stage, which opens an additional per-stage CHAIN span. Falls back to uninstrumented execution when MLflow is unreachable or unconfigured.

Parameters:

Name	Type	Description	Default
state	PipelineState	Initial pipeline state.	required
auto_confirm	bool	When True, skip the interactive confirmation prompt between stages.	False

Returns:

Type	Description
PipelineState	Final pipeline state after all stages complete.

Source code in src/lasagnastack/base.py

def run(self, state: PipelineState, auto_confirm: bool = False) -> PipelineState:
    """Run all stages in order, wrapped in MLflow tracking.

    Opens an MLflow experiment run and a top-level CHAIN span named
    ``"{ClassName}.run"`` (using the concrete subclass name). Each stage
    is executed via ``_run_stage``, which opens an additional per-stage
    CHAIN span. Falls back to uninstrumented execution when MLflow is
    unreachable or unconfigured.

    Args:
        state: Initial pipeline state.
        auto_confirm: When ``True``, skip the interactive confirmation
            prompt between stages.

    Returns:
        Final pipeline state after all stages complete.
    """
    run_name = self._mlflow_run_name(state)
    tags = self._mlflow_tags(state)
    span_name = f"{type(self).__name__}.run"

    with _mlflow_run(run_name, tags, span_name):
        state.output_dir.mkdir(parents=True, exist_ok=True)
        log.info(
            "pipeline_start",
            input_dir=str(state.input_dir),
            output_dir=str(state.output_dir),
        )
        for i, stage in enumerate(self.stages):
            state = self._run_stage(stage, state)
            if i < len(self.stages) - 1:
                _confirm(stage.completion_message(state), auto_confirm)

    return state

PipelineState dataclass

PipelineState(
    input_dir: Path,
    output_dir: Path,
    brief_path: Path,
    skill_path: Path | None = None,
    critique_max_retries: int = 2,
)

Bases: ABC

Abstract base carrying configuration shared across all pipeline types.

Stage

Bases: ABC

One step in a Pipeline.

Implement run() to transform the state and completion_message() to provide the text shown in the confirmation prompt before the next stage.

completion_message abstractmethod

completion_message(state: PipelineState) -> str

Human-readable summary shown after this stage completes and before the next.

Source code in src/lasagnastack/base.py

@abstractmethod
def completion_message(self, state: PipelineState) -> str:
    """Human-readable summary shown after this stage completes and before the next."""
    ...

run abstractmethod

run(state: PipelineState) -> PipelineState

Execute this stage. Receives the current state; returns updated state.

Source code in src/lasagnastack/base.py

@abstractmethod
def run(self, state: PipelineState) -> PipelineState:
    """Execute this stage. Receives the current state; returns updated state."""
    ...