Skip to content

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 ~/.openjarvis/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/openjarvis/learning/agents/skill_optimizer.py 
60
61
62
63
64
65
66
67
def __init__(
    self,
    *,
    min_traces_per_skill: int = 20,
    optimizer: str = "dspy",
) -> None:
    self._min_traces = min_traces_per_skill
    self._optimizer = optimizer
Functions
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/openjarvis/learning/agents/skill_optimizer.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
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 openjarvis.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(
                "~/.openjarvis/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