conductor

conductor

ConductorAgent — static-DAG planner (Sakana AI, arXiv 2512.04388).

Stage-1 inference-only repro. The paper's trained Qwen2.5-7B conductor is not released; we substitute a strong zero-shot cloud planner (default Opus) and run the same plan-then-execute machinery.

Pipeline per task:

1. Plan — the conductor reads the question + numbered worker pool and emits three lists (model_id, subtasks, access_list) in JSON, up to 5 steps.
2. Execute — for each step i: build the worker prompt from subtasks[i] + the concatenated prior (subtask, output) messages selected by access_list[i]; call worker model_id[i]; the final answer is the output of the last step.

On plan parse failure: retry once with a stricter "JSON only" prompt; on second failure, fall back to a single call to the strongest available worker (last in the pool by convention).

Workers come from cfg["workers"] or a sensible default pool (local Qwen if vLLM is up, plus Opus 4.7 and gpt-5-mini).

Hybrid harness result: conductor-swebenchverified-opusplan-30 = 0.367 acc / $0.22 per task — +10pp vs baseline-cloud at ~15× cheaper.

Ported from hybrid-local-cloud-compute/adapters/conductor_adapter.py.

Classes

ConductorAgent

ConductorAgent(engine: InferenceEngine, model: str, *, local_model: Optional[str] = None, local_endpoint: Optional[str] = None, cloud_endpoint: str = 'anthropic', cfg: Optional[Dict[str, Any]] = None, bus: Optional[Any] = None, temperature: Optional[float] = None, max_tokens: Optional[int] = None)

Bases: LocalCloudAgent

Plan-then-execute static DAG over a worker pool. See module docstring.

Source code in src/openjarvis/agents/hybrid/_base.py

def __init__(
    self,
    engine: InferenceEngine,
    model: str,
    *,
    local_model: Optional[str] = None,
    local_endpoint: Optional[str] = None,
    cloud_endpoint: str = "anthropic",
    cfg: Optional[Dict[str, Any]] = None,
    bus: Optional[Any] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
) -> None:
    super().__init__(
        engine,
        model,
        bus=bus,
        temperature=temperature,
        max_tokens=max_tokens,
    )
    self._cloud_model = model
    self._cloud_endpoint = (cloud_endpoint or "anthropic").lower()
    self._local_model = local_model
    self._local_endpoint = local_endpoint
    self._cfg: Dict[str, Any] = dict(cfg or {})

Functions