ace_optimizer

ace_optimizer

ACE agent optimizer — context evolution via the ACE Generator / Reflector / Curator triad.

ACE (ace-agent/ace <https://github.com/ace-agent/ace>_) optimizes agent context — an annotated natural-language playbook of strategies the agent reads at inference time — rather than mutating prompts (DSPy) or evolving prompt populations (GEPA). The output is a textual playbook with entries like::

[str-00001] helpful=5 harmful=0 :: When the user asks for a unit
                                   conversion, prefer the exact
                                   rational form before rounding.

The wrapper adapts OpenJarvis traces into ACE's train_samples / val_samples / test_samples shape, builds a minimal DataProcessor from the trace feedback signal, runs ACE in offline mode, and writes the resulting final_playbook.txt as a sidecar overlay under ~/.openjarvis/learning/ace/<task>/ for the agent runtime to pick up on next start.

ACE is not on PyPI as of v1.0.1. The learning-ace extra installs from the upstream git repo; if the import fails we surface a status="error" with the install hint.

Classes

ACEAgentOptimizer

ACEAgentOptimizer(config: ACEOptimizerConfig)

Optimize an agent's playbook context using ACE.

PARAMETER	DESCRIPTION
config	:class:ACEOptimizerConfig controlling models, run length, and output location. TYPE: ACEOptimizerConfig

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

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

Functions

optimize

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

Run ACE offline-mode optimization on traces from the store.

Returns a status dict mirroring DSPy / GEPA shape: {"status": "completed" | "skipped" | "error", ...}.

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

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

    Returns a status dict mirroring DSPy / GEPA shape:
    ``{"status": "completed" | "skipped" | "error", ...}``.
    """
    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_ACE:
        return {
            "status": "error",
            "reason": (
                "ace not installed (pip install "
                "'openjarvis[learning-ace]')"
            ),
        }

    samples = _traces_to_samples(traces)
    if len(samples) < self.config.min_traces:
        return {
            "status": "skipped",
            "reason": (
                f"only {len(samples)} usable samples after filtering "
                f"(min={self.config.min_traces})"
            ),
        }

    train, val, test = _split_samples(samples)
    save_dir = Path(
        self.config.save_dir or _default_save_dir(self.config.task_name)
    )
    save_dir.mkdir(parents=True, exist_ok=True)

    try:
        ace_system = ace.ACE(  # type: ignore[union-attr]
            api_provider=self.config.api_provider,
            generator_model=self.config.generator_model,
            reflector_model=self.config.reflector_model,
            curator_model=self.config.curator_model,
            max_tokens=self.config.max_tokens,
        )
        ace_system.run(
            mode="offline",
            train_samples=train,
            val_samples=val,
            test_samples=test,
            data_processor=_TraceDataProcessor(),
            config={
                "num_epochs": self.config.num_epochs,
                "max_num_rounds": self.config.max_num_rounds,
                "eval_steps": self.config.eval_steps,
                "playbook_token_budget": self.config.playbook_token_budget,
                "task_name": self.config.task_name,
                "save_dir": str(save_dir),
                "api_provider": self.config.api_provider,
            },
        )
    except Exception as exc:
        logger.warning("ACE optimization failed: %s", exc)
        return {"status": "error", "reason": str(exc)}

    playbook_path = save_dir / "final_playbook.txt"
    if not playbook_path.exists():
        # ACE completed but didn't write the expected artifact — surface
        # so the user can diagnose without trusting a silent zero.
        return {
            "status": "error",
            "reason": (
                f"ACE finished without writing {playbook_path}; "
                "check ACE logs in save_dir"
            ),
        }

    playbook = playbook_path.read_text(encoding="utf-8")
    return {
        "status": "completed",
        "traces_used": len(traces),
        "samples_used": len(samples),
        "playbook_path": str(playbook_path),
        "playbook_chars": len(playbook),
        "save_dir": str(save_dir),
    }