dspy_optimizer

dspy_optimizer

DSPy agent optimizer -- programmatic pipeline optimization.

Wraps an agent's reasoning pipeline as a DSPy Module and optimizes it end-to-end using DSPy teleprompters. Outputs TOML-compatible config updates written via AgentConfigEvolver.

Classes

DSPyAgentOptimizer

DSPyAgentOptimizer(config: DSPyOptimizerConfig)

Optimize agent configs using DSPy teleprompters.

PARAMETER	DESCRIPTION
config	DSPyOptimizerConfig controlling optimizer type and parameters. TYPE: DSPyOptimizerConfig

Source code in src/openjarvis/learning/agents/dspy_optimizer.py

def __init__(self, config: DSPyOptimizerConfig) -> None:
    self.config = config

Functions

optimize

optimize(trace_store: Any) -> Dict[str, Any]

Run DSPy optimization on traces from the store.

1. Extract traces and convert to DSPy Examples
2. Build a DSPy Module mirroring the agent pipeline
3. Run the configured teleprompter
4. Extract optimized parameters as TOML updates
5. Write via AgentConfigEvolver if config_dir is set

Source code in src/openjarvis/learning/agents/dspy_optimizer.py

def optimize(self, trace_store: Any) -> Dict[str, Any]:
    """Run DSPy optimization on traces from the store.

    1. Extract traces and convert to DSPy Examples
    2. Build a DSPy Module mirroring the agent pipeline
    3. Run the configured teleprompter
    4. Extract optimized parameters as TOML updates
    5. Write via AgentConfigEvolver if config_dir is set
    """
    # Get traces
    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_DSPY:
        return {
            "status": "error",
            "reason": (
                "dspy not installed "
                "(pip install 'openjarvis[learning-dspy]')"
            ),
        }

    try:
        optimized = self._run_dspy_optimization(traces)
    except Exception as exc:
        logger.warning("DSPy optimization failed: %s", exc)
        return {"status": "error", "reason": str(exc)}

    config_updates = self._to_config_updates(optimized)

    # Write configs if config_dir is set
    if self.config.config_dir:
        self._write_configs(config_updates)

    return {
        "status": "completed",
        "traces_used": len(traces),
        "config_updates": config_updates,
    }