GEPA agent optimizer -- Pareto-efficient evolutionary optimization.
Uses GEPA's adapter pattern to bridge OpenJarvis traces into GEPA's
evolutionary optimization framework. Outputs TOML config updates
written via AgentConfigEvolver.
Classes
Implements GEPA's adapter protocol for OpenJarvis agents.
Bridges trace data into GEPA's optimization framework via
the assess() and make_reflective_dataset() methods.
Source code in src/openjarvis/learning/agents/gepa_optimizer.py
| def __init__(
self,
trace_store: Any,
agent_name: str,
config: GEPAOptimizerConfig,
) -> None:
self.trace_store = trace_store
self.agent_name = agent_name
self.config = config
self._traces: List[Any] = []
|
Functions
Load traces from the store.
Source code in src/openjarvis/learning/agents/gepa_optimizer.py
| def load_traces(self) -> None:
"""Load traces from the store."""
kwargs: Dict[str, Any] = {"limit": 10_000}
if self.agent_name:
kwargs["agent"] = self.agent_name
self._traces = self.trace_store.list_traces(**kwargs)
|
assess(batch: List[Any], candidate: Dict[str, Any], capture_traces: bool = False) -> Dict[str, Any]
Score a candidate config against a batch of test cases.
Returns a dict with at least 'scores' (list of floats) key.
Source code in src/openjarvis/learning/agents/gepa_optimizer.py
| def assess(
self,
batch: List[Any],
candidate: Dict[str, Any],
capture_traces: bool = False,
) -> Dict[str, Any]:
"""Score a candidate config against a batch of test cases.
Returns a dict with at least 'scores' (list of floats) key.
"""
scores = []
trace_data = []
for item in batch:
query = item if isinstance(item, str) else getattr(item, "query", str(item))
# Find matching traces for this query
matching = [t for t in self._traces if t.query == query]
if matching:
best = max(matching, key=lambda t: t.feedback or 0.0)
scores.append(best.feedback or 0.0)
if capture_traces:
trace_data.append({
"query": query,
"result": best.result,
"feedback": best.feedback,
"outcome": best.outcome,
"steps": [
{
"type": str(s.step_type),
"input": s.input,
"output": s.output,
}
for s in best.steps
],
})
else:
scores.append(0.0)
result: Dict[str, Any] = {"scores": scores}
if capture_traces:
result["traces"] = trace_data
return result
|
make_reflective_dataset(candidate: Dict[str, Any], assessment_batch: List[Any], components_to_update: List[str]) -> List[Dict[str, Any]]
Package trace diagnostics as Actionable Side Information for GEPA.
Source code in src/openjarvis/learning/agents/gepa_optimizer.py
| def make_reflective_dataset(
self,
candidate: Dict[str, Any],
assessment_batch: List[Any],
components_to_update: List[str],
) -> List[Dict[str, Any]]:
"""Package trace diagnostics as Actionable Side Information for GEPA."""
dataset = []
for item in assessment_batch:
query = item if isinstance(item, str) else getattr(item, "query", str(item))
matching = [t for t in self._traces if t.query == query]
if not matching:
continue
best = max(matching, key=lambda t: t.feedback or 0.0)
# Build diagnostic info
tool_calls = []
reasoning_steps = []
errors = []
for step in best.steps:
step_type = str(step.step_type)
if "tool_call" in step_type:
tool_calls.append(step.input.get("tool", "unknown"))
if "generate" in step_type:
reasoning_steps.append(str(step.output)[:200])
is_error = (
step.output
and isinstance(step.output, dict)
and "error" in step.output
)
if is_error:
errors.append(step.output["error"])
dataset.append({
"query": query,
"result": best.result,
"feedback": best.feedback,
"outcome": best.outcome,
"tool_calls": tool_calls,
"reasoning_steps": reasoning_steps,
"errors": errors,
"components_to_update": components_to_update,
})
return dataset
|
Optimize agent configs using GEPA evolutionary optimization.
| PARAMETER |
DESCRIPTION |
config
|
GEPAOptimizerConfig controlling optimization parameters.
TYPE:
GEPAOptimizerConfig
|
Source code in src/openjarvis/learning/agents/gepa_optimizer.py
| def __init__(self, config: GEPAOptimizerConfig) -> None:
self.config = config
|
Functions
optimize(trace_store: Any) -> Dict[str, Any]
Run GEPA optimization on traces from the store.
- Load traces and build the GEPA adapter
- Define the search space (agent config fields)
- Run GEPA's evolutionary optimization
- Extract best candidate as TOML updates
- Write via AgentConfigEvolver if config_dir is set
Source code in src/openjarvis/learning/agents/gepa_optimizer.py
| def optimize(self, trace_store: Any) -> Dict[str, Any]:
"""Run GEPA optimization on traces from the store.
1. Load traces and build the GEPA adapter
2. Define the search space (agent config fields)
3. Run GEPA's evolutionary optimization
4. Extract best candidate as TOML updates
5. Write via AgentConfigEvolver if config_dir is set
"""
kwargs: Dict[str, Any] = {"limit": 10_000}
if self.config.agent_filter:
kwargs["agent"] = self.config.agent_filter
traces = trace_store.list_traces(**kwargs)
if len(traces) < self.config.min_traces:
return {
"status": "skipped",
"reason": (
f"only {len(traces)} traces, "
f"min_traces={self.config.min_traces}"
),
}
if not HAS_GEPA:
return {
"status": "error",
"reason": (
"gepa not installed"
" (pip install 'openjarvis[learning-gepa]')"
),
}
agent_name = self.config.agent_filter or "default"
adapter = OpenJarvisGEPAAdapter(trace_store, agent_name, self.config)
adapter.load_traces()
try:
best_candidate = self._run_gepa(adapter, traces)
except Exception as exc:
logger.warning("GEPA optimization failed: %s", exc)
return {"status": "error", "reason": str(exc)}
config_updates = self._to_config_updates(best_candidate)
if self.config.config_dir:
self._write_configs(agent_name, config_updates)
return {
"status": "completed",
"traces_used": len(traces),
"config_updates": config_updates,
}
|