diff --git a/graqle/__version__.py b/graqle/__version__.py index 74b892b7..15783290 100644 --- a/graqle/__version__.py +++ b/graqle/__version__.py @@ -5,4 +5,4 @@ # constraints: none # ── /graqle:intelligence ── -__version__ = "0.77.0" +__version__ = "0.78.0" diff --git a/graqle/backends/__init__.py b/graqle/backends/__init__.py index 7b2b7d61..88b205e2 100644 --- a/graqle/backends/__init__.py +++ b/graqle/backends/__init__.py @@ -22,6 +22,12 @@ def __getattr__(name: str): if name == "GeminiBackend": from graqle.backends.gemini import GeminiBackend return GeminiBackend + if name == "BackendFallbackChain": + from graqle.backends.fallback import BackendFallbackChain + return BackendFallbackChain + if name == "BackendRaceChain": # ADR-240 D1 + from graqle.backends.race import BackendRaceChain + return BackendRaceChain if name in ("create_provider_backend", "PROVIDER_PRESETS", "get_provider_names"): from graqle.backends import providers diff --git a/graqle/backends/race.py b/graqle/backends/race.py new file mode 100644 index 00000000..9089f17c --- /dev/null +++ b/graqle/backends/race.py @@ -0,0 +1,143 @@ +"""BackendRaceChain — race multiple backends, take the first (or best-of-N). + +ADR-240 D1: the SDK had FALLBACK (sequential) and PARALLEL fan-out (BackendPool, +wait-for-ALL) but no RACE / first-to-finish / best-of-N primitive. This adds one. + +Mirrors BackendFallbackChain exactly (subclasses BaseBackend, takes +list[BaseBackend], empty→ValueError, all-fail→RuntimeError) so it is a drop-in +ModelBackend anywhere a backend is expected — the agent behind AutonomousExecutor, +a TaskRouter target, a BackendPool panelist, or nested in a BackendFallbackChain. + +Two modes: + - scorer=None (default): FIRST-TO-FINISH — return the first successful result, + cancel the losers (stops paying for slower providers). + - scorer set: BEST-OF-N — await all, return the highest-scored result + (deterministic tie-break: score, then original backend order). + +Uses ONLY BaseBackend.generate — no per-provider logic. Generic asyncio hedge +pattern; contains none of the patented debate/clearance machinery. +""" +from __future__ import annotations + +import asyncio +import logging +from typing import Callable + +from graqle.backends.base import BaseBackend, GenerateResult + +logger = logging.getLogger("graqle.backends.race") + + +class BackendRaceChain(BaseBackend): + """Race backends concurrently; return the first success or the best-of-N. + + Usage: + # first-to-finish (hedged request) — take whichever provider answers first + race = BackendRaceChain([AnthropicBackend(...), OpenAIBackend(...)]) + result = await race.generate("prompt") + + # best-of-N — run both, keep the higher-scored answer + race = BackendRaceChain([a, b], scorer=lambda r: len(r.text)) + result = await race.generate("prompt") + """ + + def __init__( + self, + backends: list[BaseBackend], + *, + timeout_s: float | None = None, + scorer: Callable[[GenerateResult], float] | None = None, + ) -> None: + if not backends: + raise ValueError("BackendRaceChain requires at least one backend") + self._backends = backends + self._timeout_s = timeout_s + self._scorer = scorer + self._last_used: str = "" + self._last_cost_per_1k: float = backends[0].cost_per_1k_tokens + + async def generate( + self, + prompt: str, + *, + max_tokens: int = 512, + temperature: float = 0.3, + stop: list[str] | None = None, + ) -> GenerateResult: + async def _run(b: BaseBackend) -> tuple[str, float, GenerateResult]: + coro = b.generate( + prompt, max_tokens=max_tokens, temperature=temperature, stop=stop + ) + if self._timeout_s is not None: + coro = asyncio.wait_for(coro, timeout=self._timeout_s) + return b.name, b.cost_per_1k_tokens, await coro + + tasks = {asyncio.ensure_future(_run(b)) for b in self._backends} + + if self._scorer is None: + # ── FIRST-TO-FINISH ───────────────────────────────────────────── + errors: list[BaseException] = [] + pending = tasks + while pending: + done, pending = await asyncio.wait( + pending, return_when=asyncio.FIRST_COMPLETED + ) + for t in done: + exc = t.exception() + if exc is None: + name, cpk, result = t.result() + self._last_used, self._last_cost_per_1k = name, cpk + await self._cancel(pending) # stop the losers + if errors: + logger.info( + "Race won by %s after %d failure(s)", name, len(errors) + ) + return result + errors.append(exc) # this one failed — keep waiting for a winner + self._raise_all_failed(errors) + + # ── BEST-OF-N ─────────────────────────────────────────────────────── + done, _ = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED) + scored: list[tuple[float, str, float, GenerateResult]] = [] + errors = [] + for t in done: + exc = t.exception() + if exc is None: + name, cpk, result = t.result() + scored.append((self._scorer(result), name, cpk, result)) + else: + errors.append(exc) + if not scored: + self._raise_all_failed(errors) + # deterministic tie-break: highest score, then earliest in the original list + order = {b.name: i for i, b in enumerate(self._backends)} + best = max(scored, key=lambda s: (s[0], -order.get(s[1], 0))) + _, self._last_used, self._last_cost_per_1k, result = best + return result + + async def _cancel(self, pending: set) -> None: + for t in pending: + t.cancel() + if pending: + # drain cancellations so no "Task was destroyed but pending" warnings + await asyncio.gather(*pending, return_exceptions=True) + + def _raise_all_failed(self, errors: list[BaseException]) -> None: + summary = "; ".join(f"{type(e).__name__}: {e}" for e in errors) + raise RuntimeError( + f"All {len(self._backends)} backends failed: {summary}" + ) + + @property + def name(self) -> str: + mode = "race" if self._scorer is None else "best_of" + return f"{mode}:[{' | '.join(b.name for b in self._backends)}]" + + @property + def cost_per_1k_tokens(self) -> float: + # cost of the winning backend; before any run, defaults to the first. + return self._last_cost_per_1k + + @property + def last_used(self) -> str: + return self._last_used diff --git a/graqle/routing.py b/graqle/routing.py index 3553b532..f7006c5e 100644 --- a/graqle/routing.py +++ b/graqle/routing.py @@ -35,6 +35,7 @@ import logging import os from dataclasses import dataclass, field +from enum import Enum from typing import Any logger = logging.getLogger("graqle.routing") @@ -489,3 +490,94 @@ def to_config(self) -> dict[str, Any]: if self.default_model: config["default_model"] = self.default_model return config + + +# --------------------------------------------------------------------------- +# ADR-240 D2 — CostAwareRouter: automatic cost/latency selection over a +# candidate set given a task-difficulty signal. TaskRouter (above) stays +# ADVISORY/opt-in and is UNTOUCHED — this is a separate, additive runtime +# selector. Provider-agnostic: it ranks candidates by their RELATIVE +# cost_per_1k_tokens (the one capability signal every BaseBackend exposes), +# so no per-model capability database is needed. The CALLER supplies the +# difficulty (from task length, plan-vs-code step type, CR-B09 retry flags). +# --------------------------------------------------------------------------- + +class Difficulty(str, Enum): + """Task difficulty tier the caller passes to CostAwareRouter.select().""" + + SIMPLE = "simple" # cheap/fast tier is enough + MODERATE = "moderate" + HARD = "hard" # force a strong (pricier) candidate + + +_DIFFICULTY_MIN_TIER = { + Difficulty.SIMPLE: 0, + Difficulty.MODERATE: 1, + Difficulty.HARD: 2, +} + + +def _assign_tiers(ranked: list) -> dict[str, int]: + """Assign 0..2 capability tiers by terciles of the cost-ascending order. + + Cheapest third → tier 0 (weakest), dearest third → tier 2 (strongest). + Deterministic given the candidate list. + """ + n = len(ranked) + tiers: dict[str, int] = {} + for i, b in enumerate(ranked): + if n <= 1: + tiers[b.name] = 0 + else: + # position 0..n-1 → tier 0/1/2 by thirds + frac = i / (n - 1) + tiers[b.name] = 0 if frac < 1 / 3 else (1 if frac < 2 / 3 else 2) + return tiers + + +@dataclass +class CostAwareRouter: + """Auto-select the cheapest backend that meets a task's difficulty. + + select() never returns None: HARD with only cheap candidates still returns + the strongest available, and ceilings never eliminate the last option. + Tie-break is deterministic (cost, latency_hint, name) so it is unit-testable + with MockBackend. + """ + + max_cost_per_1k: float | None = None + max_latency_ms: float | None = None + + def select( + self, + candidates: list, + difficulty: Difficulty, + *, + latency_hint: dict[str, float] | None = None, + ) -> Any: + if not candidates: + raise ValueError("CostAwareRouter.select requires at least one candidate") + + # 1. rank by cost ascending (stable on name) → assign capability tiers + ranked = sorted(candidates, key=lambda b: (b.cost_per_1k_tokens, b.name)) + tier_of = _assign_tiers(ranked) + min_tier = _DIFFICULTY_MIN_TIER[difficulty] + + # 2. eligible = candidates whose tier meets the difficulty; if none + # qualifies, take the strongest (dearest) available. + eligible = [b for b in ranked if tier_of[b.name] >= min_tier] + if not eligible: + eligible = [ranked[-1]] + + # 3. apply optional cost ceiling, but never empty the set + if self.max_cost_per_1k is not None: + filtered = [b for b in eligible if b.cost_per_1k_tokens <= self.max_cost_per_1k] + eligible = filtered or eligible + + # 4. pick cheapest meeting difficulty; deterministic tie-break + hints = latency_hint or {} + + def _key(b: Any) -> tuple: + return (b.cost_per_1k_tokens, hints.get(b.name, float("inf")), b.name) + + return min(eligible, key=_key) diff --git a/pyproject.toml b/pyproject.toml index 926dd101..4f2a1478 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,12 +4,12 @@ build-backend = "hatchling.build" [project] name = "graqle" -# V-ADR239-STAGE2: native version bump (G4 config, CG-14 gate can't resolve cross-repo -# relative path — established native-edit precedent per prior V-CR-G1-NATIVE bump). -# 0.77.0 = ADR-239 Stage 2: AutonomousExecutor CheckpointProtocol + run_tests (one -# engine for git AND cloud consumers; backward-compatible). 0.76.0 = ADR-225 G1 -# multi-tenant memory service. 0.75.1 = ADR-220-A R2 Studio fail-closed. -version = "0.77.0" +# V-ADR240-DELTA: native version bump (G4 config, CG-14 native-edit precedent). +# 0.78.0 = ADR-240 D1+D2: BackendRaceChain (first-to-finish/best-of-N) + +# CostAwareRouter (auto cost/latency selection) — dual-provider autonomy; +# additive, backward-compatible, no IP concern. 0.77.0 = ADR-239 Stage 2 +# (CheckpointProtocol + run_tests). 0.76.0 = ADR-225 G1 multi-tenant memory. +version = "0.78.0" description = "Give your AI tools architecture-aware reasoning. Build a knowledge graph from any codebase — dependency analysis, impact analysis, governed AI answers with confidence scores. Works with Claude Code, Cursor, VS Code Copilot. 14 LLM backends, fully offline capable." readme = "README.md" license = {text = "Apache-2.0"} diff --git a/tests/test_backends/test_adr240_race_and_router.py b/tests/test_backends/test_adr240_race_and_router.py new file mode 100644 index 00000000..bf0bad0d --- /dev/null +++ b/tests/test_backends/test_adr240_race_and_router.py @@ -0,0 +1,141 @@ +""" +ADR-240 D1+D2: BackendRaceChain (first-to-finish / best-of-N) + CostAwareRouter +(auto cost/latency selection). Deterministic tests via MockBackend. +""" +from __future__ import annotations + +import asyncio + +import pytest + +from graqle.backends import BackendRaceChain +from graqle.backends.base import GenerateResult +from graqle.backends.mock import MockBackend +from graqle.routing import CostAwareRouter, Difficulty + + +# ── fakes ───────────────────────────────────────────────────────────────────── + +class _TimedMock(MockBackend): + def __init__(self, name: str, delay: float, text: str, *, fail: bool = False): + super().__init__() + self._n, self._d, self._t, self._fail = name, delay, text, fail + + @property + def name(self) -> str: + return self._n + + async def generate(self, prompt, **kw) -> GenerateResult: + await asyncio.sleep(self._d) + if self._fail: + raise RuntimeError(f"{self._n} failed") + return GenerateResult(text=self._t, model=self._n) + + +class _CostMock(MockBackend): + def __init__(self, name: str, cost: float): + super().__init__() + self._n, self._c = name, cost + + @property + def name(self) -> str: + return self._n + + @property + def cost_per_1k_tokens(self) -> float: + return self._c + + +# ── D1: BackendRaceChain ────────────────────────────────────────────────────── + +def test_race_requires_at_least_one_backend(): + with pytest.raises(ValueError): + BackendRaceChain([]) + + +def test_race_first_to_finish_returns_fastest(): + fast = _TimedMock("fast", 0.02, "FAST") + slow = _TimedMock("slow", 0.50, "SLOW") + r = asyncio.run(BackendRaceChain([slow, fast]).generate("hi")) + assert r.text == "FAST" + + +def test_race_first_to_finish_skips_a_failing_backend(): + boom = _TimedMock("boom", 0.01, "-", fail=True) + ok = _TimedMock("ok", 0.05, "OK") + r = asyncio.run(BackendRaceChain([boom, ok]).generate("hi")) + assert r.text == "OK" + + +def test_race_all_fail_raises_runtimeerror(): + a = _TimedMock("a", 0.01, "-", fail=True) + b = _TimedMock("b", 0.02, "-", fail=True) + with pytest.raises(RuntimeError, match="All 2 backends failed"): + asyncio.run(BackendRaceChain([a, b]).generate("hi")) + + +def test_best_of_n_picks_highest_score_deterministically(): + a = _TimedMock("a", 0.01, "short") + b = _TimedMock("b", 0.01, "a much longer answer") + r = asyncio.run( + BackendRaceChain([a, b], scorer=lambda g: len(g.text)).generate("hi") + ) + assert r.text == "a much longer answer" + + +def test_best_of_n_tie_breaks_on_original_order(): + a = _TimedMock("a", 0.01, "same") + b = _TimedMock("b", 0.01, "same") + r = asyncio.run( + BackendRaceChain([a, b], scorer=lambda g: len(g.text)).generate("hi") + ) + assert r.model == "a", "equal scores tie-break to earliest in the list" + + +def test_race_is_a_backend_droppable_anywhere(): + from graqle.backends.base import BaseBackend + from graqle.core.types import ModelBackend + race = BackendRaceChain([_TimedMock("a", 0.01, "x")]) + assert isinstance(race, BaseBackend) + assert isinstance(race, ModelBackend) + assert race.name.startswith("race:[") + + +# ── D2: CostAwareRouter ─────────────────────────────────────────────────────── + +def test_router_requires_candidates(): + with pytest.raises(ValueError): + CostAwareRouter().select([], Difficulty.SIMPLE) + + +def test_router_simple_picks_cheapest(): + cheap, mid, strong = _CostMock("cheap", 0.001), _CostMock("mid", 0.01), _CostMock("strong", 0.06) + picked = CostAwareRouter().select([strong, mid, cheap], Difficulty.SIMPLE) + assert picked.name == "cheap" + + +def test_router_hard_forces_strongest(): + cheap, mid, strong = _CostMock("cheap", 0.001), _CostMock("mid", 0.01), _CostMock("strong", 0.06) + picked = CostAwareRouter().select([cheap, mid, strong], Difficulty.HARD) + assert picked.name == "strong" + + +def test_router_hard_with_only_cheap_returns_strongest_available(): + a, b = _CostMock("a", 0.001), _CostMock("b", 0.002) + picked = CostAwareRouter().select([a, b], Difficulty.HARD) + assert picked.name == "b", "never None — the dearest available is returned" + + +def test_router_cost_ceiling_never_empties(): + cheap, strong = _CostMock("cheap", 0.001), _CostMock("strong", 0.06) + # a ceiling below even the cheapest must still return something + picked = CostAwareRouter(max_cost_per_1k=0.0001).select([cheap, strong], Difficulty.SIMPLE) + assert picked.name in ("cheap", "strong") + + +def test_router_tie_breaks_on_latency_then_name(): + a, b = _CostMock("a", 0.01), _CostMock("b", 0.01) # equal cost + picked = CostAwareRouter().select( + [a, b], Difficulty.SIMPLE, latency_hint={"a": 200.0, "b": 50.0} + ) + assert picked.name == "b", "equal cost → lower latency wins"