skill_optimizer

skill_optimizer

SkillOptimizer — per-skill DSPy/GEPA optimization wrapper (Plan 2A).

Buckets traces by skill name, runs the underlying optimizer on each skill's bucket, and writes the result as a sidecar overlay file in ~/.diapason/learning/skills/<skill-name>/optimized.toml.

The actual DSPy/GEPA invocation is done in _run_dspy / _run_gepa, which are isolated for easy mocking in tests. In Plan 2A these are deliberately minimal — they call the existing optimizer modules and extract the description + few-shot examples. Plan 2B will measure the impact via benchmarks.

Classes

SkillOptimizationResult dataclass

SkillOptimizationResult(
    skill_name: str,
    status: str,
    trace_count: int = 0,
    overlay_path: Optional[Path] = None,
    error: str = "",
)

Result of optimizing a single skill.

SkillOptimizer

SkillOptimizer(
    *,
    min_traces_per_skill: int = 20,
    optimizer: str = "dspy",
)

Per-skill optimization wrapper around DSPyAgentOptimizer / GEPA.

PARAMETER	DESCRIPTION
min_traces_per_skill	Minimum trace count for a skill to be eligible for optimization. TYPE: int DEFAULT: 20
optimizer	"dspy" or "gepa". Determines which underlying optimizer is called inside _run_dspy / _run_gepa. TYPE: str DEFAULT: 'dspy'

Source code in src/diapason/learning/agents/skill_optimizer.py

def __init__(
    self,
    *,
    min_traces_per_skill: int = 20,
    optimizer: str = "dspy",
) -> None:
    self._min_traces = min_traces_per_skill
    self._optimizer = optimizer

Methods:

optimize

optimize(
    trace_store: Any,
    skill_manager: SkillManager,
    *,
    overlay_dir: Optional[Path] = None,
) -> Dict[str, SkillOptimizationResult]

Run the per-skill optimization loop.

Returns a dict mapping skill name to result. Writes overlay TOML files for each skill that produced output.

Source code in src/diapason/learning/agents/skill_optimizer.py

def optimize(
    self,
    trace_store: Any,
    skill_manager: SkillManager,
    *,
    overlay_dir: Optional[Path] = None,
) -> Dict[str, SkillOptimizationResult]:
    """Run the per-skill optimization loop.

    Returns a dict mapping skill name to result.  Writes overlay TOML
    files for each skill that produced output.
    """
    if overlay_dir is None:
        # Try config first; fall back to the default tree.
        try:
            from diapason.core.config import load_config

            cfg = load_config()
            cfg_dir = getattr(
                getattr(cfg.learning, "skills", None),
                "overlay_dir",
                None,
            )
            if cfg_dir:
                overlay_dir = Path(cfg_dir).expanduser()
        except Exception:
            pass
        if overlay_dir is None:
            overlay_dir = Path(
                str(get_config_dir() / "learning" / "skills")
            ).expanduser()
    overlay_dir = Path(overlay_dir).expanduser()

    traces = trace_store.list_traces(limit=10000)
    buckets = self._bucket_traces_by_skill(traces)

    results: Dict[str, SkillOptimizationResult] = {}
    for skill_name, skill_traces in buckets.items():
        if len(skill_traces) < self._min_traces:
            results[skill_name] = SkillOptimizationResult(
                skill_name=skill_name,
                status="skipped",
                trace_count=len(skill_traces),
            )
            continue

        try:
            if self._optimizer == "gepa":
                output = self._run_gepa(skill_name, skill_traces)
            else:
                output = self._run_dspy(skill_name, skill_traces)
        except Exception as exc:
            LOGGER.warning("Skill optimizer failed for '%s': %s", skill_name, exc)
            results[skill_name] = SkillOptimizationResult(
                skill_name=skill_name,
                status="error",
                trace_count=len(skill_traces),
                error=str(exc),
            )
            continue

        overlay = SkillOverlay(
            skill_name=skill_name,
            optimizer=self._optimizer,
            optimized_at=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
            trace_count=len(skill_traces),
            description=output.description,
            few_shot=list(output.few_shot),
        )
        path = write_overlay(overlay, overlay_dir)
        results[skill_name] = SkillOptimizationResult(
            skill_name=skill_name,
            status="optimized",
            trace_count=len(skill_traces),
            overlay_path=path,
        )

    return results

Functions: