`lasagnastack.optimize_skill_pipeline`¶

lasagnastack.optimize_skill_pipeline ¶

Three-stage pipeline that analyses human-edited CapCut drafts and produces an improved skill file.

OptimizeSkillPipeline ¶

OptimizeSkillPipeline(
    optimize_client: LLMClient | None = None,
    critique_client: LLMClient | None = None,
)

Bases: Pipeline

Trainset → improved skill file pipeline.

Stage 1 (IngestDraftsStage): parse every CapCut draft in the trainset folder into (CutList, ReelStyleSheet) pairs. Stage 2 (OptimizeSkillStage): call the LLM with the existing skill + examples to generate a draft improved skill file. Stage 3 (CritiqueSkillStage): iteratively critique and revise the draft skill until approved or the retry cap is reached.

Initialise the pipeline with optional per-stage LLM clients.

Parameters:

Name	Type	Description	Default
`optimize_client`	`LLMClient \| None`	LLM client for Stage 2 (optimize).	`None`
`critique_client`	`LLMClient \| None`	LLM client for Stage 3 (critique).	`None`

Source code in src/lasagnastack/optimize_skill_pipeline.py

def __init__(
    self,
    optimize_client: LLMClient | None = None,
    critique_client: LLMClient | None = None,
) -> None:
    """Initialise the pipeline with optional per-stage LLM clients.

    Args:
        optimize_client: LLM client for Stage 2 (optimize).
        critique_client: LLM client for Stage 3 (critique).
    """
    self._optimize_client = optimize_client
    self._critique_client = critique_client

stages `property` ¶

stages: list[Stage]

Return the three pipeline stages in execution order.

run ¶

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

Run all three stages, then write the approved skill file to disk.

Parameters:

Name	Type	Description	Default
`state`	`OptimizeSkillPipelineState`	Initial pipeline state. `input_dir` must point at the trainset folder; `skill_path` (optional) points at the existing skill file to improve.	required
`auto_confirm`	`bool`	When `True`, skip the interactive prompts between stages.	`False`

Returns:

Type	Description
`OptimizeSkillPipelineState`	Final pipeline state with `optimized_skill_text` set.

Source code in src/lasagnastack/optimize_skill_pipeline.py

def run(  # type: ignore[override]
    self,
    state: OptimizeSkillPipelineState,
    auto_confirm: bool = False,
) -> OptimizeSkillPipelineState:
    """Run all three stages, then write the approved skill file to disk.

    Args:
        state: Initial pipeline state. ``input_dir`` must point at the
            trainset folder; ``skill_path`` (optional) points at the
            existing skill file to improve.
        auto_confirm: When ``True``, skip the interactive prompts between stages.

    Returns:
        Final pipeline state with ``optimized_skill_text`` set.
    """
    state = super().run(state, auto_confirm=auto_confirm)
    assert isinstance(state, OptimizeSkillPipelineState)

    if state.optimized_skill_text:
        out_path = state.output_dir / _OUTPUT_FILENAME
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text(state.optimized_skill_text, encoding="utf-8")
        log.info("optimize_skill_written", path=str(out_path))
        print(f"\nOptimized skill file saved to: {out_path}")

    return state

build_state ¶

build_state(
    trainset_dir: Path,
    output_dir: Path,
    skill_path: Path | None,
    critique_max_retries: int,
) -> OptimizeSkillPipelineState

Construct an OptimizeSkillPipelineState suitable for OptimizeSkillPipeline.

brief_path is set to skill_path (the skill being optimised acts as the primary input document, which doubles as the MLflow run identifier). When no existing skill is provided a sentinel path is used so the pipeline can still start from scratch.

Parameters:

Name	Type	Description	Default
`trainset_dir`	`Path`	Folder containing CapCut draft subfolders.	required
`output_dir`	`Path`	Where to write `optimized_skill.md` and critique JSONs.	required
`skill_path`	`Path \| None`	Path to an existing skill file to improve, or `None`.	required
`critique_max_retries`	`int`	Maximum critique iterations.	required

Returns:

Type	Description
`OptimizeSkillPipelineState`	A fully initialised `OptimizeSkillPipelineState`.

Source code in src/lasagnastack/optimize_skill_pipeline.py

def build_state(
    trainset_dir: Path,
    output_dir: Path,
    skill_path: Path | None,
    critique_max_retries: int,
) -> OptimizeSkillPipelineState:
    """Construct an OptimizeSkillPipelineState suitable for OptimizeSkillPipeline.

    ``brief_path`` is set to ``skill_path`` (the skill being optimised acts as
    the primary input document, which doubles as the MLflow run identifier).
    When no existing skill is provided a sentinel path is used so the pipeline
    can still start from scratch.

    Args:
        trainset_dir: Folder containing CapCut draft subfolders.
        output_dir: Where to write ``optimized_skill.md`` and critique JSONs.
        skill_path: Path to an existing skill file to improve, or ``None``.
        critique_max_retries: Maximum critique iterations.

    Returns:
        A fully initialised ``OptimizeSkillPipelineState``.
    """
    brief_path = skill_path if skill_path else (trainset_dir / "skill.md")
    return OptimizeSkillPipelineState(
        input_dir=trainset_dir,
        output_dir=output_dir,
        brief_path=brief_path,
        skill_path=skill_path,
        critique_max_retries=critique_max_retries,
    )

lasagnastack.optimize_skill_pipeline¶