Model Routing

routing

Difficulty-based model routing for skill runs.

The router asks an agent (running on the harness default model, inside the same sandbox as the real run) to rate a task as low, medium or high difficulty and to write that rating to _run/route.json. The host then maps the rating to a :class:ModelTier (a model id plus a default reasoning effort) and runs the actual skill on it.

This module is pure: it knows nothing about backends or containers. The caller supplies a :class:RunCallable that performs one agent invocation, which keeps :func:classify testable without a sandbox.

Every classifier failure falls back to the caller-supplied default tier, so routing can never make a run worse than an unrouted one.

TIER_NAMES = ('low', 'medium', 'high') module-attribute

Difficulty tiers, from cheapest to strongest.

DEFAULT_CLASSIFIER_MAX_TURNS = 10 module-attribute

Default turn cap for the classifier run (only enforced where the CLI supports it).

ROUTE_FILENAME = 'route.json' module-attribute

File the classifier writes under <work_dir>/_run/.

CLASSIFIER_OUTPUT_FILENAME = 'classifier-output.txt' module-attribute

Raw classifier stream, written under <work_dir>/_run/.

ModelTier(model, effort=None) dataclass

One routing tier: a model id plus the default reasoning effort for it.

effort is harness-specific (see :attr:HarnessModels.efforts). None means "resolve normally": the effort env var, else the registry default_effort. Only the literal override value none (via --effort or the env var) disables the effort flag.

RouteDecision(model, effort, tier, source, reason='') dataclass

Outcome of routing one skill run.

source is "classifier" when the agent's rating was used, "forced" when the caller pinned a tier, and "fallback" when the classifier failed and the default model was used instead. tier is None on fallback.

to_dict()

Return a plain dict (for telemetry and consumers).

Source code in src/agentic_ci/routing.py

def to_dict(self) -> dict:
    """Return a plain dict (for telemetry and consumers)."""
    return asdict(self)

RouteError

Bases: ValueError

Raised when the classifier's route file is missing, malformed, or invalid.

RunCallable

Bases: Protocol

One agent invocation inside an already-prepared sandbox.

route_path(work_dir)

Return the path of the classifier's route file for work_dir.

Source code in src/agentic_ci/routing.py

def route_path(work_dir: Path) -> Path:
    """Return the path of the classifier's route file for *work_dir*."""
    return Path(work_dir) / "_run" / ROUTE_FILENAME

resolve_model_tiers(harness, overrides)

Overlay overrides on the harness default tier map and validate it.

Raises ValueError for an unknown tier name, an empty model id, or an effort value the harness rejects, so configuration errors surface before any container is started.

Source code in src/agentic_ci/routing.py

def resolve_model_tiers(
    harness: Harness, overrides: Mapping[str, ModelTier]
) -> dict[str, ModelTier]:
    """Overlay *overrides* on the harness default tier map and validate it.

    Raises ``ValueError`` for an unknown tier name, an empty model id, or an
    effort value the harness rejects, so configuration errors surface before
    any container is started.
    """
    tiers = dict(harness.default_model_tiers())
    for name, tier in overrides.items():
        if name not in TIER_NAMES:
            raise ValueError(f"Unknown model tier {name!r}; expected one of {TIER_NAMES}")
        tiers[name] = tier
    for name in TIER_NAMES:
        tier = tiers.get(name)
        if tier is None:
            raise ValueError(f"Harness {harness.name} defines no {name!r} model tier")
        if not tier.model or not tier.model.strip():
            raise ValueError(f"Model tier {name!r} has an empty model id")
        # Validates the effort value for this harness; raises ValueError when invalid.
        harness.build_effort_args(tier.effort)
    return tiers

build_classifier_prompt(task_prompt, *, route_file=f'_run/{ROUTE_FILENAME}', max_files=DEFAULT_CLASSIFIER_MAX_TURNS)

Build the prompt that asks the agent to rate task_prompt.

Source code in src/agentic_ci/routing.py

def build_classifier_prompt(
    task_prompt: str,
    *,
    route_file: str = f"_run/{ROUTE_FILENAME}",
    max_files: int = DEFAULT_CLASSIFIER_MAX_TURNS,
) -> str:
    """Build the prompt that asks the agent to rate *task_prompt*."""
    return CLASSIFIER_PROMPT_TEMPLATE.format(
        task_prompt=task_prompt, route_file=route_file, max_files=max_files
    )

load_route(path)

Read the classifier's route file and return (difficulty, reason).

Raises :class:RouteError when the file is missing, is a symlink, is not a JSON object, or names a difficulty outside :data:TIER_NAMES.

Source code in src/agentic_ci/routing.py

def load_route(path: Path) -> tuple[str, str]:
    """Read the classifier's route file and return ``(difficulty, reason)``.

    Raises :class:`RouteError` when the file is missing, is a symlink, is not
    a JSON object, or names a difficulty outside :data:`TIER_NAMES`.
    """
    path = Path(path)
    if path.is_symlink():
        raise RouteError(f"route file is a symlink: {path}")
    if not path.is_file():
        raise RouteError(f"route file not found: {path}")
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        raise RouteError(f"route file unreadable or not JSON: {exc}") from exc
    if not isinstance(data, dict):
        raise RouteError("route file must contain a JSON object")
    difficulty = str(data.get("difficulty", "")).strip().lower()
    if difficulty not in TIER_NAMES:
        raise RouteError(f"route file difficulty {data.get('difficulty')!r} not in {TIER_NAMES}")
    reason = data.get("reason", "")
    if not isinstance(reason, str):
        reason = ""
    return difficulty, reason[:_REASON_MAX_LENGTH]

forced_route(tier, tiers)

Return a decision that pins tier without running the classifier.

Source code in src/agentic_ci/routing.py

def forced_route(tier: str, tiers: Mapping[str, ModelTier]) -> RouteDecision:
    """Return a decision that pins *tier* without running the classifier."""
    if tier not in tiers:
        raise ValueError(f"Unknown model tier {tier!r}; expected one of {TIER_NAMES}")
    chosen = tiers[tier]
    return RouteDecision(
        model=chosen.model,
        effort=chosen.effort,
        tier=tier,
        source="forced",
        reason="tier forced by caller",
    )

classify(run, *, work_dir, task_prompt, tiers, classifier_model, classifier_effort, classifier_args, fallback, prompt_builder=None)

Run the classifier and map its rating to a :class:RouteDecision.

run performs one agent invocation. The classifier prompt is built by prompt_builder (default :func:build_classifier_prompt) from task_prompt. Any failure (exception, non-zero exit, missing or invalid route file) logs a warning and returns a "fallback" decision using fallback; this function never raises on classifier failure.

Source code in src/agentic_ci/routing.py

def classify(
    run: RunCallable,
    *,
    work_dir: Path,
    task_prompt: str,
    tiers: Mapping[str, ModelTier],
    classifier_model: str,
    classifier_effort: str | None,
    classifier_args: list[str],
    fallback: ModelTier,
    prompt_builder: Callable[[str], str] | None = None,
) -> RouteDecision:
    """Run the classifier and map its rating to a :class:`RouteDecision`.

    *run* performs one agent invocation. The classifier prompt is built by
    *prompt_builder* (default :func:`build_classifier_prompt`) from
    *task_prompt*. Any failure (exception, non-zero exit, missing or invalid
    route file) logs a warning and returns a ``"fallback"`` decision using
    *fallback*; this function never raises on classifier failure.
    """
    work_dir = Path(work_dir)
    run_dir = work_dir / "_run"
    run_dir.mkdir(parents=True, exist_ok=True)
    route_file = route_path(work_dir)
    if route_file.is_symlink() or route_file.exists():
        route_file.unlink()

    prompt = (prompt_builder or build_classifier_prompt)(task_prompt)
    output_file = run_dir / CLASSIFIER_OUTPUT_FILENAME

    failure: str | None = None
    try:
        rc = run(
            prompt,
            model=classifier_model,
            effort=classifier_effort,
            extra_args=list(classifier_args) or None,
            output_file=output_file,
        )
    except Exception as exc:
        failure = f"classifier raised: {exc}"
    else:
        if rc != 0:
            failure = f"classifier exited with code {rc}"

    if failure is None:
        try:
            difficulty, reason = load_route(route_file)
        except RouteError as exc:
            failure = str(exc)
        else:
            chosen = tiers[difficulty]
            log.info(
                "Routed to tier %s (model=%s, effort=%s): %s",
                difficulty,
                chosen.model,
                chosen.effort,
                reason,
            )
            return RouteDecision(
                model=chosen.model,
                effort=chosen.effort,
                tier=difficulty,
                source="classifier",
                reason=reason,
            )

    log.warning("Model routing fell back to %s: %s", fallback.model, failure)
    return RouteDecision(
        model=fallback.model,
        effort=fallback.effort,
        tier=None,
        source="fallback",
        reason=failure[:_REASON_MAX_LENGTH],
    )