lasagnastack.stages.optimize_skill_pipeline.critique_skill

lasagnastack.stages.optimize_skill_pipeline.critique_skill

Stage 3 of the optimize-skill pipeline: critique loop for the generated skill file.

CritiqueSkillStage

CritiqueSkillStage(client: LLMClient | None = None)

Bases: Stage

Iteratively critique and refine the generated skill file.

Source code in src/lasagnastack/stages/optimize_skill_pipeline/critique_skill.py

def __init__(self, client: LLMClient | None = None) -> None:
    self._client = client

run

run(state: PipelineState) -> PipelineState

Run the critique loop and update optimized_skill_text with the result.

Parameters:

Name	Type	Description	Default
state	PipelineState	Pipeline state with optimized_skill_text set.	required

Returns:

Type	Description
PipelineState	Updated state with the approved (or capped) optimized_skill_text.

Source code in src/lasagnastack/stages/optimize_skill_pipeline/critique_skill.py

def run(self, state: PipelineState) -> PipelineState:
    """Run the critique loop and update ``optimized_skill_text`` with the result.

    Args:
        state: Pipeline state with ``optimized_skill_text`` set.

    Returns:
        Updated state with the approved (or capped) ``optimized_skill_text``.
    """
    assert isinstance(state, OptimizeSkillPipelineState)
    skill_text = run(
        state,
        client=self._client,
        max_retries=state.critique_max_retries,
    )
    return dataclasses.replace(state, optimized_skill_text=skill_text)

run

run(
    state: OptimizeSkillPipelineState,
    client: LLMClient | None = None,
    max_retries: int = _MAX_RETRIES_DEFAULT,
) -> str

Critique the generated skill text, revising until approved or the retry cap is hit.

Writes output_dir/skill_critique/iteration_N.json per iteration.

Parameters:

Name	Type	Description	Default
state	OptimizeSkillPipelineState	Pipeline state. draft_cut_lists, draft_reel_style_sheets, and optimized_skill_text must be set.	required
client	LLMClient | None	LLM client to use. Defaults to GeminiClient.	None
max_retries	int	Maximum critique iterations before shipping the current text.	_MAX_RETRIES_DEFAULT

Returns:

Type	Description
str	The approved (or capped) skill text.

Source code in src/lasagnastack/stages/optimize_skill_pipeline/critique_skill.py

def run(
    state: OptimizeSkillPipelineState,
    client: LLMClient | None = None,
    max_retries: int = _MAX_RETRIES_DEFAULT,
) -> str:
    """Critique the generated skill text, revising until approved or the retry cap is hit.

    Writes ``output_dir/skill_critique/iteration_N.json`` per iteration.

    Args:
        state: Pipeline state. ``draft_cut_lists``, ``draft_reel_style_sheets``,
            and ``optimized_skill_text`` must be set.
        client: LLM client to use. Defaults to ``GeminiClient``.
        max_retries: Maximum critique iterations before shipping the current text.

    Returns:
        The approved (or capped) skill text.
    """
    if client is None:
        client = GeminiClient()

    assert state.draft_cut_lists is not None
    assert state.draft_reel_style_sheets is not None
    assert state.optimized_skill_text is not None

    existing_skill = (
        state.skill_path.read_text(encoding="utf-8").strip()
        if state.skill_path and state.skill_path.exists()
        else ""
    )
    examples = _format_examples(state)
    current_skill = state.optimized_skill_text
    previous_issues: list[list[str]] = []

    for iteration in range(max_retries):
        result = _critique_once(
            existing_skill=existing_skill,
            current_skill=current_skill,
            examples=examples,
            client=client,
            iteration=iteration,
            previous_issues=previous_issues,
        )

        critique_path = (
            state.output_dir / "skill_critique" / f"iteration_{iteration}.json"
        )
        io.write_json(result, critique_path)

        if result.verdict == "approved" or result.skill_v2 is None:
            log.info("critique_skill_approved", iteration=iteration)
            return current_skill

        log.info(
            "critique_skill_revising",
            iteration=iteration,
            issues=result.issues,
        )
        previous_issues.append(result.issues)
        current_skill = result.skill_v2

    if max_retries > 0:
        log.warning("critique_skill_cap_reached", max_retries=max_retries)
    return current_skill