From 96339295f54ca7538839f99fa1749538c55cf9b2 Mon Sep 17 00:00:00 2001 From: Deep Kumar Singh Kushwah Date: Sun, 7 Jun 2026 18:34:13 +0530 Subject: [PATCH] =?UTF-8?q?feat(experiments):=20A/B=20routing=20=E2=80=94?= =?UTF-8?q?=20serve=20the=20candidate=20to=20a=20fraction=20of=20users=20(?= =?UTF-8?q?Step=205.7c)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add ABRouter (rag_gateway.experiments) — deterministic, request_id-hashed variant assignment that *serves* the candidate retrieval config to the assigned fraction of users (the first 5.7 slice that can change a response). Wired into the served /v1/query + /v1/retrieve path: the candidate arm routes inline via _route_query (not degrade-open — a candidate failure surfaces as 502, never a silent fallback that would corrupt the experiment), the retrieval cache is partitioned by variant so control/candidate never share an entry, and the served arm's outcome is recorded once per fresh retrieval. Responses carry a new frozen ExperimentAssignment (experiment/variant/ is_candidate) on QueryResponse/RetrieveResponse (REST-only — gRPC proto mirror deferred like corpus_decision). Gated separately by cfg.experiments.routing_* (independent of shadow_enabled); ShadowCandidateConfig generalised to the shared CandidateConfig. ragctl ab drives assign -> serve -> record -> analyze. dist/schemas + dist/openapi + dist/rag.schema regenerated. All gates green (ruff, mypy --strict 292 files, RAG001, schema/openapi-drift, proto-compat, policy-coverage); ~25 new tests. Co-Authored-By: Claude Opus 4.7 --- TRACKER.md | 17 +- apps/gateway/src/rag_gateway/app.py | 8 + apps/gateway/src/rag_gateway/cache_keys.py | 23 +- apps/gateway/src/rag_gateway/experiments.py | 144 +++++++++--- apps/gateway/src/rag_gateway/query.py | 111 ++++++++- apps/gateway/src/rag_gateway/wiring.py | 50 ++++- apps/gateway/tests/test_ab_routing.py | 235 ++++++++++++++++++++ apps/gateway/tests/test_cache_keys.py | 21 ++ dist/openapi.json | 44 ++++ dist/openapi.yaml | 47 ++++ dist/rag.schema.json | 76 ++++--- dist/rag.schema.yaml | 96 +++++--- dist/schemas/ExperimentAssignment.json | 24 ++ dist/schemas/QueryResponse.json | 35 +++ dist/schemas/RetrieveResponse.json | 35 +++ docs/README.md | 1 + docs/adr/ADR-0032-ab-testing-shadow-mode.md | 47 ++++ docs/architecture/ab-routing.md | 109 +++++++++ docs/reference/experiments.md | 56 ++++- packages/config/src/rag_config/schema.py | 30 ++- packages/core/src/rag_core/__init__.py | 2 + packages/core/src/rag_core/gateway_types.py | 25 +++ packages/core/src/rag_core/gen_schemas.py | 3 + packages/ragctl/src/ragctl/main.py | 112 ++++++++++ tests/config/test_experiments_config.py | 44 +++- tests/contract/test_grpc_proto_compat.py | 10 +- 26 files changed, 1279 insertions(+), 126 deletions(-) create mode 100644 apps/gateway/tests/test_ab_routing.py create mode 100644 dist/schemas/ExperimentAssignment.json create mode 100644 docs/architecture/ab-routing.md diff --git a/TRACKER.md b/TRACKER.md index 5e8325c..14da970 100644 --- a/TRACKER.md +++ b/TRACKER.md @@ -16,10 +16,11 @@ | **Last updated** | 2026-06-07 | | **Current phase** | Phase 5 — Eval & Observability (**6 / 7 steps**) | | **Overall** | **63 / 84 steps** — Phases 0–4 complete | -| **Next action** | **Step 5.7c — A/B routing** — deterministic variant assignment that actually *serves* the candidate to a fraction of users (the first slice that can change a response). Final Phase-5 step, slices 5.7a–d. | +| **Next action** | **Step 5.7d — Console + close-out** — admin-console A/B card surfacing `GET /v1/status/experiments` (lift + CI per experiment), plus Phase-5 close-out. Final slice of the final Phase-5 step (5.7a–d). | **Recently shipped** +- **5.7c** ✅ A/B routing — `ABRouter` deterministically *serves* the candidate to a fraction of users (variant-partitioned cache, `ExperimentAssignment` response tag) — [#146](https://github.com/officialCodeWork/AgentContextOS/pull/146) - **5.7b** ✅ Shadow mode — observe-only candidate fan-out (`ShadowRunner`, background task) feeding the A/B tracker — [#145](https://github.com/officialCodeWork/AgentContextOS/pull/145) - **5.7a** ✅ A/B analyzer + experiment tracker + `GET /v1/status/experiments` dashboard — [#143](https://github.com/officialCodeWork/AgentContextOS/pull/143) - **5.6** ✅ Status & Metrics GUI (full build) — drift/feedback/cost cards, query-trace viewer, regression bisector, Grafana dashboards, cross-links — [#137–#142](https://github.com/officialCodeWork/AgentContextOS/pull/142) @@ -525,7 +526,7 @@ | 5.7 | A/B testing & shadow mode | 🚧 | — | | 5.7a | — A/B analyzer + tracker + dashboard | ✅ | [#143](https://github.com/officialCodeWork/AgentContextOS/pull/143) | | 5.7b | — Shadow mode | ✅ | [#145](https://github.com/officialCodeWork/AgentContextOS/pull/145) | -| 5.7c | — A/B routing | ⏳ | — | +| 5.7c | — A/B routing | ✅ | [#146](https://github.com/officialCodeWork/AgentContextOS/pull/146) | | 5.7d | — Console + close-out | ⏳ | — | ### 5.1 — Per-query tracing & provenance ✅ [#132](https://github.com/officialCodeWork/AgentContextOS/pull/132) @@ -621,9 +622,14 @@ - `cfg.experiments` gains `shadow_enabled` / `shadow_sample_rate` / `shadow_experiment` / `shadow_candidate`; doubly opt-in (`enabled` **and** `shadow_enabled`); `ragctl shadow`; 27 tests - [ADR-0032](docs/adr/ADR-0032-ab-testing-shadow-mode.md), [reference/experiments.md](docs/reference/experiments.md), [architecture/ab-shadow-mode.md](docs/architecture/ab-shadow-mode.md) -#### 5.7c — A/B routing ⏳ +#### 5.7c — A/B routing ✅ [#146](https://github.com/officialCodeWork/AgentContextOS/pull/146) -- **Next up.** Deterministic variant assignment that serves the candidate to a fraction of users (the first slice that can change a response) and tags it +- New `ABRouter` (`rag_gateway.experiments`, beside `ShadowRunner`) — deterministic `hash(request_id) < routing_sample_rate` variant assignment that **serves** the candidate retrieval config to the assigned fraction of users (the first slice that can change a response); `assign` / `is_candidate` / `candidate` / `record` / `tag` +- **Served-path wiring** in `/v1/query` + `/v1/retrieve`: the candidate arm is routed **inline** via `_route_query` (one retrieval, no added latency); **not degrade-open** (a candidate failure → 502, never a silent fallback to control that would corrupt the experiment); the served arm's `outcome_metric` is recorded once per **fresh** retrieval (skipped on a cache hit) +- **Variant-partitioned retrieval cache** — `compute_plan_hash` / `compute_params_hash` fold in the assigned `variant`, so control and candidate never share a cache entry (byte-identical key when routing is off) +- New frozen **`ExperimentAssignment`** (`experiment` / `variant` / `is_candidate`) on the `QueryResponse` / `RetrieveResponse` `experiment` field (additive, `dist/schemas` + `dist/openapi` regenerated; REST-only — gRPC proto mirror deferred like `corpus_decision`); served variant also on the `gateway.query_complete` log +- Same `read_chunk` PDP site (no coverage-linter entry); gated separately by `cfg.experiments.routing_enabled` (requires `enabled`) — independent of `shadow_enabled` — since routing can change a response; `ShadowCandidateConfig` generalised to `CandidateConfig` (shared by `shadow_candidate` + new `routing_candidate`); `build_app(ab_router=…)` inject seam; `ragctl ab` drives assign → serve → record → analyze; ~25 new tests; all gates green (ruff, mypy --strict, RAG001, schema/openapi-drift, proto-compat, policy-coverage) +- [ADR-0032](docs/adr/ADR-0032-ab-testing-shadow-mode.md) (5.7c update), [reference/experiments.md](docs/reference/experiments.md), [architecture/ab-routing.md](docs/architecture/ab-routing.md) ## Phase 6 — Governance & Tenancy (Weeks 28–34) ⏳ @@ -787,6 +793,9 @@ Complete log of every PR. Routine Dependabot bumps are grouped; everything else | [#141](https://github.com/officialCodeWork/AgentContextOS/pull/141) | 2026-06-05 | feat(observability): drift + cost Prometheus metrics + Grafana dashboard (Step 5.6e) | | [#142](https://github.com/officialCodeWork/AgentContextOS/pull/142) | 2026-06-05 | feat(admin-ui): observability cross-links + Step 5.6 close-out (5.6f) | | [#143](https://github.com/officialCodeWork/AgentContextOS/pull/143) | 2026-06-05 | feat(experiments): A/B analyzer + tracker + dashboard (Step 5.7a) | +| [#144](https://github.com/officialCodeWork/AgentContextOS/pull/144) | 2026-06-05 | docs(tracker): restructure for readability + complete PR history | +| [#145](https://github.com/officialCodeWork/AgentContextOS/pull/145) | 2026-06-05 | feat(experiments): shadow mode — observe-only candidate fan-out (Step 5.7b) | +| [#146](https://github.com/officialCodeWork/AgentContextOS/pull/146) | 2026-06-07 | feat(experiments): A/B routing — serve the candidate to a fraction of users (Step 5.7c) | | #78–#80, #116–#118 | Open | Dependabot bumps — awaiting merge | | #81 | Closed | Dependabot bump — superseded | diff --git a/apps/gateway/src/rag_gateway/app.py b/apps/gateway/src/rag_gateway/app.py index e9910c1..522fa95 100644 --- a/apps/gateway/src/rag_gateway/app.py +++ b/apps/gateway/src/rag_gateway/app.py @@ -366,6 +366,7 @@ def build_app( cost_tracker: Any | None = None, experiment_tracker: Any | None = None, shadow_runner: Any | None = None, + ab_router: Any | None = None, enable_cors: bool = True, default_tenant_id: TenantId | None = None, ) -> FastAPI: @@ -565,6 +566,13 @@ def build_app( # ``cfg.experiments`` when ``shadow_enabled``. app.state.shadow_runner = shadow_runner + # A/B routing (Step 5.7c) — deterministically assigns each query to a variant + # and *serves* the candidate config to the assigned fraction of users (the + # first experiment slice that can change a response), feeding the same + # tracker. ``None`` (inert) in the plain ``build_app``; the config-driven + # wiring builds it from ``cfg.experiments`` when ``routing_enabled``. + app.state.ab_router = ab_router + # Outbound webhooks (Step 3.9) — the subscription registry + the delivery # dispatcher behind ``/v1/webhooks/subscriptions`` and the # ``ingest.completed`` emission. Defaults: in-memory store + an HTTP diff --git a/apps/gateway/src/rag_gateway/cache_keys.py b/apps/gateway/src/rag_gateway/cache_keys.py index 5d92585..0593534 100644 --- a/apps/gateway/src/rag_gateway/cache_keys.py +++ b/apps/gateway/src/rag_gateway/cache_keys.py @@ -35,9 +35,16 @@ def compute_plan_hash( corpus_ids: list[str], top_k: int, filters: dict[str, Any], + variant: str | None = None, ) -> str: - """Return a stable SHA-256 hex digest over the retrieval-determining inputs.""" - payload = { + """Return a stable SHA-256 hex digest over the retrieval-determining inputs. + + ``variant`` partitions the cache by A/B-routing arm (Step 5.7c): the control + and candidate configs retrieve different chunks for the same query, so they + must never share a cache entry. Omitted (``None``) when A/B routing is off, + which leaves the key byte-identical to the pre-5.7c hash. + """ + payload: dict[str, Any] = { "t": tenant_id, "p": principal_id, "q": query, @@ -45,6 +52,8 @@ def compute_plan_hash( "k": top_k, "f": filters, } + if variant is not None: + payload["v"] = variant blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) return hashlib.sha256(blob.encode("utf-8")).hexdigest() @@ -56,20 +65,28 @@ def compute_params_hash( corpus_ids: list[str], top_k: int, filters: dict[str, Any], + variant: str | None = None, ) -> str: """SHA-256 over every retrieval-determining input *except the query text*. This is the L2 semantic-cache ``scope``: a similar query may reuse a prior result only when these parameters match exactly, so the semantic tier can never serve a result built with a different top_k / filter / corpus / ACL. + + ``variant`` partitions the scope by A/B-routing arm (Step 5.7c) for the same + reason as :func:`compute_plan_hash` — the semantic tier must not serve a + control result to a candidate-assigned request (or vice versa). Omitted when + routing is off, leaving the scope byte-identical to the pre-5.7c value. """ - payload = { + payload: dict[str, Any] = { "t": tenant_id, "p": principal_id, "c": sorted(corpus_ids), "k": top_k, "f": filters, } + if variant is not None: + payload["v"] = variant blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) return hashlib.sha256(blob.encode("utf-8")).hexdigest() diff --git a/apps/gateway/src/rag_gateway/experiments.py b/apps/gateway/src/rag_gateway/experiments.py index faefbe2..6cdc0e6 100644 --- a/apps/gateway/src/rag_gateway/experiments.py +++ b/apps/gateway/src/rag_gateway/experiments.py @@ -1,31 +1,41 @@ -"""Shadow-mode A/B fan-out (Step 5.7b). - -On a sampled fraction of live queries, run a **candidate** retriever alongside -the served (**control**) path — *observe-only*, never affecting the response — -and feed both variants' outcome metric into the -:class:`~rag_observability.experiments.ABExperimentTracker`, which the dashboard -(``GET /v1/status/experiments``) compares with the pure -``analyze_ab_experiment`` (lift + confidence interval, Step 5.7a). - -Design (mirrors the Phase-5 observe-only features — drift, cost): - -* **Never alters the request.** The candidate retrieval is scheduled as a - FastAPI ``BackgroundTask``, so it runs *after* the response is sent — the - served path's latency is untouched. -* **Degrade-open.** Any failure in the candidate retrieval is logged +"""Shadow-mode fan-out (Step 5.7b) + A/B routing (Step 5.7c). + +Two ways the gateway compares a **candidate** retrieval config against the live +**control** on real traffic, both feeding the +:class:`~rag_observability.experiments.ABExperimentTracker` the dashboard +(``GET /v1/status/experiments``) reads through the pure ``analyze_ab_experiment`` +(lift + confidence interval, Step 5.7a): + +* :class:`ShadowRunner` (5.7b) is **observe-only** — it runs the candidate + *alongside* the served control on a sampled fraction of queries, in the + background, and never affects the response. +* :class:`ABRouter` (5.7c) is **live routing** — it deterministically assigns + each query to one variant and, when that variant is the candidate, **serves** + the candidate's retrieval to the user. This is the first slice that can change + which config a user gets, so it is separately opt-in (``routing_enabled``). + +Shared design (mirrors the Phase-5 observe-only features — drift, cost): + +* **Deterministic assignment.** Whether a query is shadowed / which variant it is + routed to is a pure hash of its ``request_id`` against the sample rate — no RNG + in the hot path, so the decision is reproducible and testable. +* **Degrade-open (shadow).** A candidate failure in the *shadow* path is logged (``experiment.shadow_failed``) and swallowed; the response was already sent. -* **Deterministic sampling.** Whether a query is shadowed is a pure hash of its - ``request_id`` against the sample rate, so the decision is reproducible and - testable (no RNG in the hot path). - -The candidate is any :class:`~rag_retrieval.router.SupportsRoute` (a second -:class:`~rag_retrieval.router.RetrievalRouter` with different fusion weights is -the config-driven default; production injects one over the real backends). -A/B *routing* — actually serving the candidate to a fraction of users — is the -next slice (5.7c); this slice is strictly observe-only. - -See [docs/reference/experiments.md](../../../../docs/reference/experiments.md) -and [docs/architecture/ab-shadow-mode.md](../../../../docs/architecture/ab-shadow-mode.md). + A/B *routing* serves the candidate inline, so a candidate failure surfaces as a + normal retrieval error (the variant is the served path) — there is nothing to + degrade *to* without silently swapping the user back to control mid-experiment. +* **One ``read_chunk`` PDP site.** The candidate is any + :class:`~rag_retrieval.router.SupportsRoute` and routes through the same + ``HybridRetriever.retrieve`` (the canonical policy site), so ACLs are enforced + on the candidate exactly as on the control — no coverage-linter entry needed. + +The candidate is a second :class:`~rag_retrieval.router.RetrievalRouter` with +different fusion weights in the config-driven default; production injects one over +the real backends via ``build_app(shadow_runner=…)`` / ``build_app(ab_router=…)``. + +See [docs/reference/experiments.md](../../../../docs/reference/experiments.md), +[docs/architecture/ab-shadow-mode.md](../../../../docs/architecture/ab-shadow-mode.md), +and [docs/architecture/ab-routing.md](../../../../docs/architecture/ab-routing.md). """ from __future__ import annotations @@ -34,12 +44,13 @@ from typing import Any from rag_core import get_logger +from rag_core.gateway_types import ExperimentAssignment from rag_core.types import RequestContext from rag_observability.experiments import ABExperimentTracker _log = get_logger(__name__) -__all__ = ["ShadowRunner", "outcome_metric"] +__all__ = ["ABRouter", "ShadowRunner", "outcome_metric"] # Largest value of an 8-hex-digit digest prefix — the denominator that maps a # hash into the half-open unit interval ``[0, 1)`` for sampling. @@ -161,3 +172,80 @@ async def run( "error": repr(exc), }, ) + + +class ABRouter: + """Deterministic A/B variant assignment that *serves* the candidate (Step 5.7c). + + Unlike :class:`ShadowRunner` (observe-only, background), ``ABRouter`` is on + the served path and can change the response: :meth:`assign` deterministically + hashes the ``request_id`` to a variant, and when that variant is the + candidate the gateway routes retrieval through :attr:`candidate` (a second + :class:`~rag_retrieval.router.SupportsRoute`) instead of the control router — + so the configured fraction of users is actually served the candidate config. + + The runner stays a thin policy object: it *decides* the variant, *holds* the + candidate router, *records* the served variant's :func:`outcome_metric` into + the tracker, and *tags* the response. The query handler owns the wiring (it + is the one place that knows the served control path), so ``ABRouter`` makes no + governed SPI call of its own — the candidate's own ``read_chunk`` PDP site + enforces ACLs. Stateless beyond its dependencies, so one instance is shared + across requests (the tracker it feeds is itself thread-safe). + """ + + def __init__( + self, + *, + candidate_router: Any, # SupportsRoute — Any to match the gateway's deps style. + tracker: ABExperimentTracker, + sample_rate: float = 0.1, + experiment: str = "ab", + control_variant: str = "control", + candidate_variant: str = "candidate", + ) -> None: + self._candidate = candidate_router + self._tracker = tracker + self._sample_rate = max(0.0, min(1.0, sample_rate)) + self._experiment = experiment + self._control = control_variant + self._candidate_variant = candidate_variant + + @property + def experiment(self) -> str: + return self._experiment + + @property + def candidate(self) -> Any: + """The candidate :class:`~rag_retrieval.router.SupportsRoute` to serve.""" + return self._candidate + + def assign(self, request_id: str) -> str: + """Return the variant label for ``request_id`` (deterministic). + + The candidate is served to the ``sample_rate`` fraction of requests whose + ``request_id`` hashes into ``[0, sample_rate)``; everyone else gets the + control. ``rate <= 0`` assigns everyone to control (routing a no-op); + ``rate >= 1`` assigns everyone to the candidate. + """ + return self._candidate_variant if _sampled(request_id, self._sample_rate) else self._control + + def is_candidate(self, variant: str) -> bool: + """Whether ``variant`` is the candidate arm (the one served the alternative).""" + return variant == self._candidate_variant + + def record(self, variant: str, chunk_refs: list[Any]) -> None: + """Record the served ``variant``'s :func:`outcome_metric` into the tracker. + + Called once per freshly-retrieved query (the served arm only), so the + per-``(experiment, variant)`` windows the dashboard analyses fill from + real served traffic. An O(1) append — never raises into the request. + """ + self._tracker.observe(self._experiment, variant, outcome_metric(chunk_refs)) + + def tag(self, variant: str) -> ExperimentAssignment: + """Build the :class:`ExperimentAssignment` stamped on the response.""" + return ExperimentAssignment( + experiment=self._experiment, + variant=variant, + is_candidate=self.is_candidate(variant), + ) diff --git a/apps/gateway/src/rag_gateway/query.py b/apps/gateway/src/rag_gateway/query.py index b3a952a..eb3a643 100644 --- a/apps/gateway/src/rag_gateway/query.py +++ b/apps/gateway/src/rag_gateway/query.py @@ -42,6 +42,7 @@ ) from rag_core.gateway_types import ( Answer, + ExperimentAssignment, GatewayError, QueryRequest, QueryResponse, @@ -362,8 +363,12 @@ async def _generate_answer( return answer -def _plan_hash_for(body: RetrieveRequest | QueryRequest) -> str: - """Stable retrieval-cache plan hash from a request's determining inputs.""" +def _plan_hash_for(body: RetrieveRequest | QueryRequest, variant: str | None = None) -> str: + """Stable retrieval-cache plan hash from a request's determining inputs. + + ``variant`` partitions the cache by A/B-routing arm (Step 5.7c) so the + control and candidate never share an entry; ``None`` leaves the pre-5.7c key. + """ return compute_plan_hash( tenant_id=str(body.tenant_id), principal_id=str(body.principal_id), @@ -371,17 +376,23 @@ def _plan_hash_for(body: RetrieveRequest | QueryRequest) -> str: corpus_ids=[str(c) for c in body.corpus_ids], top_k=body.top_k, filters=dict(body.filters), + variant=variant, ) -def _scope_for(body: RetrieveRequest | QueryRequest) -> str: - """L2 semantic-cache scope — the request params excluding the query text.""" +def _scope_for(body: RetrieveRequest | QueryRequest, variant: str | None = None) -> str: + """L2 semantic-cache scope — the request params excluding the query text. + + ``variant`` partitions the scope by A/B-routing arm (Step 5.7c), mirroring + :func:`_plan_hash_for`; ``None`` leaves the pre-5.7c scope. + """ return compute_params_hash( tenant_id=str(body.tenant_id), principal_id=str(body.principal_id), corpus_ids=[str(c) for c in body.corpus_ids], top_k=body.top_k, filters=dict(body.filters), + variant=variant, ) @@ -394,6 +405,7 @@ async def _route_query( hyde_vector: list[float] | None, corpus_ids: list[Any] | None, top_k: int, + variant: str | None = None, ) -> tuple[Any, Any, list[Any]]: """Run routing, preferring the corpus router when one is wired. @@ -401,7 +413,26 @@ async def _route_query( :class:`~rag_retrieval.corpus_router.CorpusRouter` is installed (the pre-3.5 default), ``corpus_decision`` is ``None`` and retrieval goes straight through the :class:`~rag_retrieval.router.RetrievalRouter`. + + A/B routing (Step 5.7c): when an :class:`~rag_gateway.experiments.ABRouter` + is wired and ``variant`` is its *candidate* arm, retrieval is served by the + candidate router instead — a whole-pipeline alternative, so it bypasses the + corpus router exactly like the shadow candidate (``corpus_decision`` is + ``None``). The candidate routes through the same ``HybridRetriever.retrieve`` + PDP site, so ACLs are enforced identically. Every other case (control arm, + or no A/B routing) is the unchanged served path. """ + ab = deps.ab_router + if ab is not None and variant is not None and ab.is_candidate(variant): + decision, chunk_refs = await ab.candidate.route( + ctx, + text=text, + expansion_terms=expansion_terms, + hyde_vector=hyde_vector, + corpus_ids=corpus_ids, + top_k=top_k, + ) + return None, decision, chunk_refs if deps.corpus_router is not None: corpus_decision, decision, chunk_refs = await deps.corpus_router.route( ctx, @@ -529,12 +560,18 @@ async def _handle_query( span.set_attribute("rag.gateway.generate", body.generate) timings = _Timings() + # A/B routing (Step 5.7c) — deterministically assign this request to a + # variant *before* the cache lookup, so the cache is partitioned per arm + # (control and candidate never share an entry) and the served config is + # fixed for the whole request. ``None`` when no ABRouter is wired. + variant = _assign_variant(deps, str(ctx.request_id)) + # Retrieval cache (Step 4.1). A hit supplies the chunk refs and skips # understanding + routing; rerank / pack / generate still run over the # cached refs (those stages are not part of the retrieval cache). cache = deps.retrieval_cache - plan_hash = _plan_hash_for(body) - scope = _scope_for(body) + plan_hash = _plan_hash_for(body, variant) + scope = _scope_for(body, variant) corpus_version = await resolve_corpus_version(ctx, deps.corpus_store, list(body.corpus_ids)) corpus_decision: Any = None decision: Any = None @@ -565,6 +602,8 @@ async def _handle_query( shadow_expansion = dict(understood.expansion_terms) # Routing — corpus selection (when wired) + backend execute. + # When A/B routing assigned the candidate arm, ``_route_query`` + # serves the candidate config instead (Step 5.7c). with timings.stage("route"): corpus_decision, decision, chunk_refs = await _route_query( ctx, @@ -574,6 +613,7 @@ async def _handle_query( hyde_vector=understood.hyde_vector, corpus_ids=list(body.corpus_ids) or None, top_k=body.top_k, + variant=variant, ) except (RetrievalError, AuthError, RateLimitError, RagError): # Domain errors propagate to the FastAPI exception handlers @@ -594,6 +634,10 @@ async def _handle_query( value=chunk_refs, ) + # A/B routing (Step 5.7c) — record the served arm's outcome into the + # experiment tracker, once per fresh retrieval (not on a cache hit). + _record_ab_outcome(deps, variant, chunk_refs) + # Optional rerank (returns hydrated Chunks). chunks: list[Chunk] = [] if body.rerank and chunk_refs: @@ -665,6 +709,9 @@ async def _handle_query( "citations_n": len(citations), "answer_generated": answer is not None, "elapsed_ms": total_ms, + # A/B routing (Step 5.7c) — which variant served this query (when + # routing is active); ``None`` otherwise. PII-free: a label only. + "experiment_variant": variant, }, ) @@ -723,6 +770,7 @@ async def _handle_query( answer=answer, timings=timings.to_model(), trace=ctx.trace, + experiment=_ab_tag(deps, variant), ) @@ -747,12 +795,16 @@ async def _handle_retrieve( span.set_attribute("rag.gateway.query_chars", len(body.query)) timings = _Timings() + # A/B routing (Step 5.7c) — assign a variant before the cache lookup so + # the cache is partitioned per arm (see ``_handle_query``). + variant = _assign_variant(deps, str(ctx.request_id)) + # Retrieval cache (Step 4.1). A hit serves cached chunk refs and # skips understanding + routing entirely; the decision is flagged # ``served_from_cache``. cache = deps.retrieval_cache - plan_hash = _plan_hash_for(body) - scope = _scope_for(body) + plan_hash = _plan_hash_for(body, variant) + scope = _scope_for(body, variant) corpus_version = await resolve_corpus_version(ctx, deps.corpus_store, list(body.corpus_ids)) if cache is not None: cached_refs = await cache.get_best( @@ -786,6 +838,7 @@ async def _handle_retrieve( chunks=cached_refs, timings=timings.to_model(), trace=ctx.trace, + experiment=_ab_tag(deps, variant), ) span.set_attribute("rag.gateway.cache_hit", False) @@ -801,6 +854,7 @@ async def _handle_retrieve( hyde_vector=understood.hyde_vector, corpus_ids=list(body.corpus_ids) or None, top_k=body.top_k, + variant=variant, ) except (RetrievalError, AuthError, RateLimitError, RagError): raise @@ -817,6 +871,9 @@ async def _handle_retrieve( scope=scope, ) + # A/B routing (Step 5.7c) — record the served arm's outcome (fresh path). + _record_ab_outcome(deps, variant, chunk_refs) + total_ms = (time.monotonic() - start) * 1000.0 timings.total_ms = total_ms span.set_attribute("rag.gateway.results_n", len(chunk_refs)) @@ -861,6 +918,7 @@ async def _handle_retrieve( chunks=chunk_refs, timings=timings.to_model(), trace=ctx.trace, + experiment=_ab_tag(deps, variant), ) @@ -967,6 +1025,39 @@ def _maybe_schedule_shadow( ) +def _assign_variant(deps: _GatewayDeps, request_id: str) -> str | None: + """Assign this request to an A/B variant (Step 5.7c), or ``None`` when off. + + Deterministic in ``request_id`` (no RNG), computed once at the top of the + handler — before the retrieval-cache lookup, so the cache is partitioned per + variant and the served arm is fixed for the whole request. ``None`` when no + :class:`~rag_gateway.experiments.ABRouter` is wired (the default). + """ + ab = deps.ab_router + return ab.assign(request_id) if ab is not None else None + + +def _record_ab_outcome(deps: _GatewayDeps, variant: str | None, chunk_refs: list[Any]) -> None: + """Record the served variant's outcome into the experiment tracker (Step 5.7c). + + Called only on the fresh-retrieval path (not on a cache hit), so each served + retrieval contributes one sample to its arm's window — the same + apples-to-apples discipline shadow mode uses. No-op when A/B routing is off. + """ + ab = deps.ab_router + if ab is not None and variant is not None: + ab.record(variant, chunk_refs) + + +def _ab_tag(deps: _GatewayDeps, variant: str | None) -> ExperimentAssignment | None: + """Build the response's A/B-assignment tag (Step 5.7c), or ``None`` when off.""" + ab = deps.ab_router + if ab is None or variant is None: + return None + tag: ExperimentAssignment = ab.tag(variant) + return tag + + async def _handle_query_trace(request: Request, request_id: str) -> QueryTraceResponse: """Serve ``GET /v1/query/{id}/trace`` — signed provenance + captured spans. @@ -1097,6 +1188,7 @@ class _GatewayDeps: "provenance_recorder", "drift_registry", "shadow_runner", + "ab_router", ) def __init__( @@ -1116,6 +1208,7 @@ def __init__( provenance_recorder: Any, drift_registry: Any, shadow_runner: Any, + ab_router: Any, ) -> None: self.understanding = understanding self.router = router @@ -1131,6 +1224,7 @@ def __init__( self.provenance_recorder = provenance_recorder self.drift_registry = drift_registry self.shadow_runner = shadow_runner + self.ab_router = ab_router @classmethod def from_request(cls, request: Request) -> _GatewayDeps: @@ -1150,6 +1244,7 @@ def from_request(cls, request: Request) -> _GatewayDeps: provenance_recorder=getattr(state, "provenance_recorder", None), drift_registry=getattr(state, "drift_registry", None), shadow_runner=getattr(state, "shadow_runner", None), + ab_router=getattr(state, "ab_router", None), ) diff --git a/apps/gateway/src/rag_gateway/wiring.py b/apps/gateway/src/rag_gateway/wiring.py index b39db72..a34a216 100644 --- a/apps/gateway/src/rag_gateway/wiring.py +++ b/apps/gateway/src/rag_gateway/wiring.py @@ -88,7 +88,7 @@ if TYPE_CHECKING: from fastapi import FastAPI - from rag_gateway.experiments import ShadowRunner + from rag_gateway.experiments import ABRouter, ShadowRunner _log = get_logger(__name__) @@ -537,6 +537,46 @@ def build_shadow_runner_from_config( ) +def build_ab_router_from_config( + cfg: RagConfig, tracker: ABExperimentTracker | None +) -> ABRouter | None: + """Build the A/B-routing runner from ``cfg.experiments`` (Step 5.7c). + + Returns ``None`` unless experiments **and** A/B routing are enabled and a + ``tracker`` exists to feed. Builds a candidate :class:`RetrievalRouter` over + fresh noop backends with the configured ``routing_candidate`` fusion weights, + so the feature works end-to-end from config; production injects a candidate + over the real backends via ``build_app(ab_router=...)``. + + Separately gated from shadow mode because A/B routing *serves* the candidate + — it can change which response a user receives. + """ + ec = cfg.experiments + if not ec.enabled or not ec.routing_enabled or tracker is None: + return None + from rag_retrieval import HybridWeights + + from rag_gateway.app import build_default_retrieval_router + from rag_gateway.experiments import ABRouter + + rc = ec.routing_candidate + candidate = build_default_retrieval_router( + base_weights=HybridWeights( + vector=rc.vector_weight, + keyword=rc.keyword_weight, + graph=rc.graph_weight, + ) + ) + return ABRouter( + candidate_router=candidate, + tracker=tracker, + sample_rate=ec.routing_sample_rate, + experiment=ec.routing_experiment, + control_variant=ec.control_variant, + candidate_variant=ec.candidate_variant, + ) + + def build_app_from_config(cfg: RagConfig, **overrides: Any) -> FastAPI: """Build the gateway app with a config-driven corpus store + router. @@ -639,6 +679,14 @@ def build_app_from_config(cfg: RagConfig, **overrides: Any) -> FastAPI: cfg, overrides.get("experiment_tracker") ) + # A/B routing (Step 5.7c) — live variant assignment that *serves* the + # candidate to a fraction of users, fed into the same tracker. ``None`` + # unless experiments + routing are both enabled in config. + if "ab_router" not in overrides: + overrides["ab_router"] = build_ab_router_from_config( + cfg, overrides.get("experiment_tracker") + ) + # Phase-5 quality/cost signals as Prometheus metrics (Step 5.6e) — expose the # drift report + per-tenant cost verdicts as OTel observable gauges on the # same pipeline that carries ``rag.spi.*`` so Grafana can graph them. The diff --git a/apps/gateway/tests/test_ab_routing.py b/apps/gateway/tests/test_ab_routing.py new file mode 100644 index 0000000..7c553f5 --- /dev/null +++ b/apps/gateway/tests/test_ab_routing.py @@ -0,0 +1,235 @@ +"""Tests for A/B routing — live variant assignment that serves the candidate (Step 5.7c). + +Covers the pure :class:`ABRouter` (deterministic assignment, the candidate +accessor, recording the served outcome, the response tag) and its gateway +integration (a candidate-assigned ``/v1/query`` is *served* the candidate's +result and tagged; the control arm is served the control and tagged; inert by +default; the config-driven build wires it only when ``routing_enabled``). +""" + +from __future__ import annotations + +from typing import Any + +from fastapi.testclient import TestClient +from rag_config import RagConfig +from rag_config.schema import ExperimentsConfig +from rag_core.gateway_types import ExperimentAssignment +from rag_core.types import ( + ChunkId, + ChunkRef, + Principal, + PrincipalId, + PrincipalKind, + QueryShape, + RequestContext, + RequestId, + RoutingDecision, + TenantId, + TraceContext, +) +from rag_gateway import build_app +from rag_gateway.experiments import ABRouter +from rag_gateway.wiring import build_app_from_config +from rag_observability import ABExperimentTracker + +_DECISION = RoutingDecision( + shape=QueryShape.SEMANTIC, use_vector=True, use_keyword=True, use_graph=False +) + + +def _ref(score: float, chunk_id: str = "cand-1", tenant: str = "t1") -> ChunkRef: + return ChunkRef(chunk_id=ChunkId(chunk_id), tenant_id=TenantId(tenant), score=score) + + +def _ctx(request_id: str = "r1") -> RequestContext: + tenant = TenantId("t1") + return RequestContext( + request_id=RequestId(request_id), + tenant_id=tenant, + principal=Principal( + id=PrincipalId("p1"), + kind=PrincipalKind.service, + display_name="p1", + tenant_id=tenant, + ), + trace=TraceContext(), + ) + + +class _StubCandidate: + """A minimal ``SupportsRoute`` returning a fixed, recognisable result set.""" + + def __init__(self, refs: list[ChunkRef]) -> None: + self._refs = refs + self.calls = 0 + + async def route( + self, ctx: RequestContext, **kwargs: Any + ) -> tuple[RoutingDecision, list[ChunkRef]]: + self.calls += 1 + return _DECISION, list(self._refs) + + +def _ab(tracker: ABExperimentTracker, candidate: Any, *, sample_rate: float) -> ABRouter: + return ABRouter( + candidate_router=candidate, + tracker=tracker, + sample_rate=sample_rate, + experiment="ab", + ) + + +# --------------------------------------------------------------------------- +# ABRouter — assignment +# --------------------------------------------------------------------------- +def test_assign_bounds() -> None: + tracker = ABExperimentTracker() + never = _ab(tracker, _StubCandidate([]), sample_rate=0.0) + always = _ab(tracker, _StubCandidate([]), sample_rate=1.0) + assert never.assign("anything") == "control" + assert always.assign("anything") == "candidate" + + +def test_assign_is_deterministic_in_request_id() -> None: + ab = _ab(ABExperimentTracker(), _StubCandidate([]), sample_rate=0.5) + assert ab.assign("req-abc") == ab.assign("req-abc") + + +def test_assign_splits_traffic() -> None: + ab = _ab(ABExperimentTracker(), _StubCandidate([]), sample_rate=0.5) + candidate_n = sum(ab.assign(f"req-{i}") == "candidate" for i in range(400)) + # Deterministic hash → roughly half; wide band so it never flakes. + assert 120 < candidate_n < 280 + + +def test_is_candidate() -> None: + ab = _ab(ABExperimentTracker(), _StubCandidate([]), sample_rate=0.5) + assert ab.is_candidate("candidate") is True + assert ab.is_candidate("control") is False + + +# --------------------------------------------------------------------------- +# ABRouter — record + tag +# --------------------------------------------------------------------------- +def test_record_feeds_tracker_with_outcome() -> None: + tracker = ABExperimentTracker(min_samples=2) + ab = _ab(tracker, _StubCandidate([]), sample_rate=1.0) + ab.record("candidate", [_ref(0.9), _ref(0.7)]) + assert tracker.samples("ab", "candidate") == [0.8] # mean retrieval score + + +def test_tag_describes_assignment() -> None: + ab = _ab(ABExperimentTracker(), _StubCandidate([]), sample_rate=0.5) + cand = ab.tag("candidate") + ctrl = ab.tag("control") + assert isinstance(cand, ExperimentAssignment) + assert cand.experiment == "ab" and cand.variant == "candidate" and cand.is_candidate is True + assert ctrl.variant == "control" and ctrl.is_candidate is False + + +# --------------------------------------------------------------------------- +# Gateway integration +# --------------------------------------------------------------------------- +def _query(client: TestClient, request_id: str = "ab-req") -> Any: + return client.post( + "/v1/query", + headers={"X-Request-Id": request_id}, + json={"tenant_id": "t1", "principal_id": "p1", "query": "what is RAG?"}, + ) + + +def _retrieve(client: TestClient, request_id: str = "ab-req") -> Any: + return client.post( + "/v1/retrieve", + headers={"X-Request-Id": request_id}, + json={"tenant_id": "t1", "principal_id": "p1", "query": "what is RAG?"}, + ) + + +def test_routing_inert_by_default() -> None: + app = build_app() + assert app.state.ab_router is None + resp = _query(TestClient(app)) + assert resp.status_code == 200, resp.text + # No experiment tag when routing is off. + assert resp.json()["experiment"] is None + + +def test_candidate_arm_is_served_and_tagged() -> None: + tracker = ABExperimentTracker(min_samples=2) + candidate = _StubCandidate([_ref(0.8, chunk_id="cand-1")]) + ab = _ab(tracker, candidate, sample_rate=1.0) # everyone → candidate + client = TestClient(build_app(experiment_tracker=tracker, ab_router=ab)) + + body = _query(client).json() + # The candidate actually served the response (its ref is in the output). + assert candidate.calls == 1 + assert [c["chunk_id"] for c in body["chunk_refs"]] == ["cand-1"] + # ...and the response carries the assignment tag. + assert body["experiment"]["variant"] == "candidate" + assert body["experiment"]["is_candidate"] is True + # ...and the served outcome is recorded under the candidate arm. + assert tracker.samples("ab", "candidate") == [0.8] + + +def test_control_arm_does_not_call_candidate_and_is_tagged() -> None: + tracker = ABExperimentTracker(min_samples=2) + candidate = _StubCandidate([_ref(0.8)]) + ab = _ab(tracker, candidate, sample_rate=0.0) # everyone → control + client = TestClient(build_app(experiment_tracker=tracker, ab_router=ab)) + + body = _query(client).json() + # The candidate was never invoked — the control (default noop) served it. + assert candidate.calls == 0 + assert body["experiment"]["variant"] == "control" + assert body["experiment"]["is_candidate"] is False + # The control arm records its (empty noop → 0.0) outcome. + assert tracker.samples("ab", "control") == [0.0] + assert tracker.samples("ab", "candidate") == [] + + +def test_retrieve_endpoint_serves_and_tags_candidate() -> None: + tracker = ABExperimentTracker(min_samples=2) + candidate = _StubCandidate([_ref(0.6, chunk_id="cand-r")]) + ab = _ab(tracker, candidate, sample_rate=1.0) + client = TestClient(build_app(experiment_tracker=tracker, ab_router=ab)) + + body = _retrieve(client).json() + assert candidate.calls == 1 + assert [c["chunk_id"] for c in body["chunks"]] == ["cand-r"] + assert body["experiment"]["variant"] == "candidate" + assert tracker.samples("ab", "candidate") == [0.6] + + +def test_dashboard_sees_served_experiment() -> None: + tracker = ABExperimentTracker(min_samples=2) + ab = _ab(tracker, _StubCandidate([_ref(0.8)]), sample_rate=1.0) + client = TestClient(build_app(experiment_tracker=tracker, ab_router=ab)) + _query(client) + dash = client.get("/v1/status/experiments").json() + assert any(e["experiment"] == "ab" for e in dash["experiments"]) + + +# --------------------------------------------------------------------------- +# Config-driven wiring +# --------------------------------------------------------------------------- +def test_build_from_config_wires_routing_when_enabled() -> None: + cfg = RagConfig( + experiments=ExperimentsConfig(enabled=True, routing_enabled=True, routing_sample_rate=1.0) + ) + app = build_app_from_config(cfg) + assert app.state.experiment_tracker is not None + assert app.state.ab_router is not None + + +def test_build_from_config_inert_when_routing_disabled() -> None: + cfg = RagConfig(experiments=ExperimentsConfig(enabled=True, routing_enabled=False)) + app = build_app_from_config(cfg) + assert app.state.experiment_tracker is not None + assert app.state.ab_router is None + + +def test_build_from_config_inert_when_experiments_disabled() -> None: + app = build_app_from_config(RagConfig()) + assert app.state.ab_router is None diff --git a/apps/gateway/tests/test_cache_keys.py b/apps/gateway/tests/test_cache_keys.py index a86ebc5..c1660d3 100644 --- a/apps/gateway/tests/test_cache_keys.py +++ b/apps/gateway/tests/test_cache_keys.py @@ -54,6 +54,27 @@ def test_params_hash_changes_with_top_k() -> None: assert compute_params_hash(**params) != compute_params_hash(**{**params, "top_k": 5}) +def test_variant_partitions_plan_hash() -> None: + # A/B routing (Step 5.7c): control and candidate arms must not share an entry. + control = compute_plan_hash(**_BASE, variant="control") + candidate = compute_plan_hash(**_BASE, variant="candidate") + assert control != candidate + + +def test_variant_none_is_backward_compatible() -> None: + # Omitting the variant leaves the key byte-identical to the pre-5.7c hash. + assert compute_plan_hash(**_BASE, variant=None) == compute_plan_hash(**_BASE) + params = {k: v for k, v in _BASE.items() if k != "query"} + assert compute_params_hash(**params, variant=None) == compute_params_hash(**params) + + +def test_variant_partitions_params_hash() -> None: + params = {k: v for k, v in _BASE.items() if k != "query"} + control = compute_params_hash(**params, variant="control") + candidate = compute_params_hash(**params, variant="candidate") + assert control != candidate + + def test_cache_served_decision_is_flagged() -> None: decision = cache_served_decision() assert decision.reason == "served_from_cache" diff --git a/dist/openapi.json b/dist/openapi.json index 2e219e8..37af8da 100644 --- a/dist/openapi.json +++ b/dist/openapi.json @@ -1654,6 +1654,30 @@ "title": "EmbeddingsUsage", "type": "object" }, + "ExperimentAssignment": { + "description": "Which A/B variant served this response (Step 5.7c).\n\nPresent on a :class:`QueryResponse` / :class:`RetrieveResponse` only when\nA/B *routing* is active and assigned the request to an experiment — the\ndeterministic, ``request_id``-hashed slice that can actually change which\nretrieval config a user gets (shadow mode, Step 5.7b, never tags a response).\n``None`` otherwise (routing off, or the request was not in the experiment).\n\nCarries no query or answer text — only the experiment id and the assigned\nvariant label, so the tag is safe to log and surface. ``is_candidate``\nflags whether the *candidate* (alternative) config served the response, so a\nconsumer can split traffic without re-deriving it from the variant name.", + "properties": { + "experiment": { + "title": "Experiment", + "type": "string" + }, + "is_candidate": { + "default": false, + "title": "Is Candidate", + "type": "boolean" + }, + "variant": { + "title": "Variant", + "type": "string" + } + }, + "required": [ + "experiment", + "variant" + ], + "title": "ExperimentAssignment", + "type": "object" + }, "ExperimentsStatusResponse": { "description": "A/B experiment comparisons (Step 5.7).\n\nOne :class:`~rag_core.eval.ABAnalysisResult` per running experiment, each\ncomparing the control variant against the candidate on the collected metric.\nEmpty when experiments are disabled or none have been observed.", "properties": { @@ -2968,6 +2992,16 @@ "decision": { "$ref": "#/components/schemas/RoutingDecision" }, + "experiment": { + "anyOf": [ + { + "$ref": "#/components/schemas/ExperimentAssignment" + }, + { + "type": "null" + } + ] + }, "packed": { "anyOf": [ { @@ -3361,6 +3395,16 @@ "decision": { "$ref": "#/components/schemas/RoutingDecision" }, + "experiment": { + "anyOf": [ + { + "$ref": "#/components/schemas/ExperimentAssignment" + }, + { + "type": "null" + } + ] + }, "query": { "title": "Query", "type": "string" diff --git a/dist/openapi.yaml b/dist/openapi.yaml index 4ea7ecd..a154299 100644 --- a/dist/openapi.yaml +++ b/dist/openapi.yaml @@ -1431,6 +1431,45 @@ components: type: integer title: EmbeddingsUsage type: object + ExperimentAssignment: + description: 'Which A/B variant served this response (Step 5.7c). + + + Present on a :class:`QueryResponse` / :class:`RetrieveResponse` only when + + A/B *routing* is active and assigned the request to an experiment — the + + deterministic, ``request_id``-hashed slice that can actually change which + + retrieval config a user gets (shadow mode, Step 5.7b, never tags a response). + + ``None`` otherwise (routing off, or the request was not in the experiment). + + + Carries no query or answer text — only the experiment id and the assigned + + variant label, so the tag is safe to log and surface. ``is_candidate`` + + flags whether the *candidate* (alternative) config served the response, so + a + + consumer can split traffic without re-deriving it from the variant name.' + properties: + experiment: + title: Experiment + type: string + is_candidate: + default: false + title: Is Candidate + type: boolean + variant: + title: Variant + type: string + required: + - experiment + - variant + title: ExperimentAssignment + type: object ExperimentsStatusResponse: description: 'A/B experiment comparisons (Step 5.7). @@ -2573,6 +2612,10 @@ components: - type: 'null' decision: $ref: '#/components/schemas/RoutingDecision' + experiment: + anyOf: + - $ref: '#/components/schemas/ExperimentAssignment' + - type: 'null' packed: anyOf: - $ref: '#/components/schemas/PackedContext' @@ -2954,6 +2997,10 @@ components: - type: 'null' decision: $ref: '#/components/schemas/RoutingDecision' + experiment: + anyOf: + - $ref: '#/components/schemas/ExperimentAssignment' + - type: 'null' query: title: Query type: string diff --git a/dist/rag.schema.json b/dist/rag.schema.json index f97644a..235aa14 100644 --- a/dist/rag.schema.json +++ b/dist/rag.schema.json @@ -177,6 +177,32 @@ "title": "CacheProvider", "type": "string" }, + "CandidateConfig": { + "additionalProperties": false, + "description": "Fusion weights for a candidate retriever (Steps 5.7b / 5.7c).\n\nThe candidate runs the same retrieval backends as the served (control) path\nbut with these per-source RRF weights, so an operator can compare an\nalternative fusion balance (e.g. vector-heavy) against the live config \u2014\nobserve-only under shadow mode (``shadow_candidate``) or served to a fraction\nof users under A/B routing (``routing_candidate``). Defaults are neutral\n``1.0`` (an operator sets the candidate to something distinct to learn\nanything).", + "properties": { + "vector_weight": { + "default": 1.0, + "minimum": 0.0, + "title": "Vector Weight", + "type": "number" + }, + "keyword_weight": { + "default": 1.0, + "minimum": 0.0, + "title": "Keyword Weight", + "type": "number" + }, + "graph_weight": { + "default": 1.0, + "minimum": 0.0, + "title": "Graph Weight", + "type": "number" + } + }, + "title": "CandidateConfig", + "type": "object" + }, "ConfigVersion": { "enum": [ "1" @@ -541,7 +567,7 @@ }, "ExperimentsConfig": { "additionalProperties": false, - "description": "A/B testing & shadow-mode knobs (Step 5.7).\n\nWhen ``enabled`` the gateway collects a per-query outcome metric per\n``(experiment, variant)`` and serves ``GET /v1/status/experiments``, which\ncompares the **control** against the **candidate** variant with the pure\n``analyze_ab_experiment`` (lift + confidence interval). **Disabled by\ndefault** \u2014 A/B routing (Step 5.7c) can change which response a user gets, so\nexperiments are strictly opt-in.\n\n* ``window_size`` \u2014 per-``(experiment, variant)`` rolling sample cap.\n* ``min_samples`` \u2014 below this many samples on *either* side a comparison\n reports ``insufficient_data`` rather than a verdict.\n* ``confidence`` \u2014 confidence level for the interval / significance test.\n* ``control_variant`` / ``candidate_variant`` \u2014 the two variant labels the\n dashboard compares.\n\nShadow mode (Step 5.7b) \u2014 *observe-only* candidate fan-out:\n\n* ``shadow_enabled`` \u2014 turn on the shadow fan-out (requires ``enabled``).\n* ``shadow_sample_rate`` \u2014 fraction ``[0, 1]`` of live queries shadowed\n (deterministic per ``request_id``); the candidate runs in the background\n so the served response is never delayed.\n* ``shadow_experiment`` \u2014 the experiment id the shadow samples land under.\n* ``shadow_candidate`` \u2014 the candidate retriever's fusion weights.", + "description": "A/B testing & shadow-mode knobs (Step 5.7).\n\nWhen ``enabled`` the gateway collects a per-query outcome metric per\n``(experiment, variant)`` and serves ``GET /v1/status/experiments``, which\ncompares the **control** against the **candidate** variant with the pure\n``analyze_ab_experiment`` (lift + confidence interval). **Disabled by\ndefault** \u2014 A/B routing (Step 5.7c) can change which response a user gets, so\nexperiments are strictly opt-in.\n\n* ``window_size`` \u2014 per-``(experiment, variant)`` rolling sample cap.\n* ``min_samples`` \u2014 below this many samples on *either* side a comparison\n reports ``insufficient_data`` rather than a verdict.\n* ``confidence`` \u2014 confidence level for the interval / significance test.\n* ``control_variant`` / ``candidate_variant`` \u2014 the two variant labels the\n dashboard compares.\n\nShadow mode (Step 5.7b) \u2014 *observe-only* candidate fan-out:\n\n* ``shadow_enabled`` \u2014 turn on the shadow fan-out (requires ``enabled``).\n* ``shadow_sample_rate`` \u2014 fraction ``[0, 1]`` of live queries shadowed\n (deterministic per ``request_id``); the candidate runs in the background\n so the served response is never delayed.\n* ``shadow_experiment`` \u2014 the experiment id the shadow samples land under.\n* ``shadow_candidate`` \u2014 the candidate retriever's fusion weights.\n\nA/B routing (Step 5.7c) \u2014 *live* variant assignment that **serves** the\ncandidate to a fraction of users (the first slice that can change a response,\nso it is gated separately from shadow):\n\n* ``routing_enabled`` \u2014 turn on A/B routing (requires ``enabled``).\n* ``routing_sample_rate`` \u2014 fraction ``[0, 1]`` of requests served the\n *candidate* config (deterministic per ``request_id``); the rest get control.\n* ``routing_experiment`` \u2014 the experiment id the served samples land under.\n* ``routing_candidate`` \u2014 the candidate retriever's fusion weights.", "properties": { "enabled": { "default": false, @@ -595,7 +621,27 @@ "type": "string" }, "shadow_candidate": { - "$ref": "#/$defs/ShadowCandidateConfig" + "$ref": "#/$defs/CandidateConfig" + }, + "routing_enabled": { + "default": false, + "title": "Routing Enabled", + "type": "boolean" + }, + "routing_sample_rate": { + "default": 0.1, + "maximum": 1.0, + "minimum": 0.0, + "title": "Routing Sample Rate", + "type": "number" + }, + "routing_experiment": { + "default": "ab", + "title": "Routing Experiment", + "type": "string" + }, + "routing_candidate": { + "$ref": "#/$defs/CandidateConfig" } }, "title": "ExperimentsConfig", @@ -1162,32 +1208,6 @@ "title": "SemanticCacheConfig", "type": "object" }, - "ShadowCandidateConfig": { - "additionalProperties": false, - "description": "Fusion weights for the shadow candidate retriever (Step 5.7b).\n\nThe candidate runs the same retrieval backends as the served (control) path\nbut with these per-source RRF weights, so an operator can shadow-test an\nalternative fusion balance (e.g. vector-heavy) against the live config on a\nsample of real traffic \u2014 observe-only. Defaults are neutral ``1.0`` (an\noperator sets the candidate to something distinct to learn anything).", - "properties": { - "vector_weight": { - "default": 1.0, - "minimum": 0.0, - "title": "Vector Weight", - "type": "number" - }, - "keyword_weight": { - "default": 1.0, - "minimum": 0.0, - "title": "Keyword Weight", - "type": "number" - }, - "graph_weight": { - "default": 1.0, - "minimum": 0.0, - "title": "Graph Weight", - "type": "number" - } - }, - "title": "ShadowCandidateConfig", - "type": "object" - }, "StorageConfig": { "additionalProperties": false, "properties": { diff --git a/dist/rag.schema.yaml b/dist/rag.schema.yaml index 3ffa20f..83ad8c9 100644 --- a/dist/rag.schema.yaml +++ b/dist/rag.schema.yaml @@ -147,6 +147,42 @@ $defs: - valkey title: CacheProvider type: string + CandidateConfig: + additionalProperties: false + description: 'Fusion weights for a candidate retriever (Steps 5.7b / 5.7c). + + + The candidate runs the same retrieval backends as the served (control) path + + but with these per-source RRF weights, so an operator can compare an + + alternative fusion balance (e.g. vector-heavy) against the live config — + + observe-only under shadow mode (``shadow_candidate``) or served to a fraction + + of users under A/B routing (``routing_candidate``). Defaults are neutral + + ``1.0`` (an operator sets the candidate to something distinct to learn + + anything).' + properties: + vector_weight: + default: 1.0 + minimum: 0.0 + title: Vector Weight + type: number + keyword_weight: + default: 1.0 + minimum: 0.0 + title: Keyword Weight + type: number + graph_weight: + default: 1.0 + minimum: 0.0 + title: Graph Weight + type: number + title: CandidateConfig + type: object ConfigVersion: enum: - '1' @@ -511,7 +547,15 @@ $defs: \ — fraction ``[0, 1]`` of live queries shadowed\n (deterministic per ``request_id``);\ \ the candidate runs in the background\n so the served response is never delayed.\n\ * ``shadow_experiment`` — the experiment id the shadow samples land under.\n\ - * ``shadow_candidate`` — the candidate retriever's fusion weights." + * ``shadow_candidate`` — the candidate retriever's fusion weights.\n\nA/B routing\ + \ (Step 5.7c) — *live* variant assignment that **serves** the\ncandidate to\ + \ a fraction of users (the first slice that can change a response,\nso it is\ + \ gated separately from shadow):\n\n* ``routing_enabled`` — turn on A/B routing\ + \ (requires ``enabled``).\n* ``routing_sample_rate`` — fraction ``[0, 1]`` of\ + \ requests served the\n *candidate* config (deterministic per ``request_id``);\ + \ the rest get control.\n* ``routing_experiment`` — the experiment id the served\ + \ samples land under.\n* ``routing_candidate`` — the candidate retriever's fusion\ + \ weights." properties: enabled: default: false @@ -556,7 +600,23 @@ $defs: title: Shadow Experiment type: string shadow_candidate: - $ref: '#/$defs/ShadowCandidateConfig' + $ref: '#/$defs/CandidateConfig' + routing_enabled: + default: false + title: Routing Enabled + type: boolean + routing_sample_rate: + default: 0.1 + maximum: 1.0 + minimum: 0.0 + title: Routing Sample Rate + type: number + routing_experiment: + default: ab + title: Routing Experiment + type: string + routing_candidate: + $ref: '#/$defs/CandidateConfig' title: ExperimentsConfig type: object FallbackConfig: @@ -1071,38 +1131,6 @@ $defs: type: integer title: SemanticCacheConfig type: object - ShadowCandidateConfig: - additionalProperties: false - description: 'Fusion weights for the shadow candidate retriever (Step 5.7b). - - - The candidate runs the same retrieval backends as the served (control) path - - but with these per-source RRF weights, so an operator can shadow-test an - - alternative fusion balance (e.g. vector-heavy) against the live config on a - - sample of real traffic — observe-only. Defaults are neutral ``1.0`` (an - - operator sets the candidate to something distinct to learn anything).' - properties: - vector_weight: - default: 1.0 - minimum: 0.0 - title: Vector Weight - type: number - keyword_weight: - default: 1.0 - minimum: 0.0 - title: Keyword Weight - type: number - graph_weight: - default: 1.0 - minimum: 0.0 - title: Graph Weight - type: number - title: ShadowCandidateConfig - type: object StorageConfig: additionalProperties: false properties: diff --git a/dist/schemas/ExperimentAssignment.json b/dist/schemas/ExperimentAssignment.json new file mode 100644 index 0000000..18780c6 --- /dev/null +++ b/dist/schemas/ExperimentAssignment.json @@ -0,0 +1,24 @@ +{ + "description": "Which A/B variant served this response (Step 5.7c).\n\nPresent on a :class:`QueryResponse` / :class:`RetrieveResponse` only when\nA/B *routing* is active and assigned the request to an experiment \u2014 the\ndeterministic, ``request_id``-hashed slice that can actually change which\nretrieval config a user gets (shadow mode, Step 5.7b, never tags a response).\n``None`` otherwise (routing off, or the request was not in the experiment).\n\nCarries no query or answer text \u2014 only the experiment id and the assigned\nvariant label, so the tag is safe to log and surface. ``is_candidate``\nflags whether the *candidate* (alternative) config served the response, so a\nconsumer can split traffic without re-deriving it from the variant name.", + "properties": { + "experiment": { + "title": "Experiment", + "type": "string" + }, + "variant": { + "title": "Variant", + "type": "string" + }, + "is_candidate": { + "default": false, + "title": "Is Candidate", + "type": "boolean" + } + }, + "required": [ + "experiment", + "variant" + ], + "title": "ExperimentAssignment", + "type": "object" +} diff --git a/dist/schemas/QueryResponse.json b/dist/schemas/QueryResponse.json index 2c460c9..2e3a457 100644 --- a/dist/schemas/QueryResponse.json +++ b/dist/schemas/QueryResponse.json @@ -448,6 +448,30 @@ "title": "CorpusScore", "type": "object" }, + "ExperimentAssignment": { + "description": "Which A/B variant served this response (Step 5.7c).\n\nPresent on a :class:`QueryResponse` / :class:`RetrieveResponse` only when\nA/B *routing* is active and assigned the request to an experiment \u2014 the\ndeterministic, ``request_id``-hashed slice that can actually change which\nretrieval config a user gets (shadow mode, Step 5.7b, never tags a response).\n``None`` otherwise (routing off, or the request was not in the experiment).\n\nCarries no query or answer text \u2014 only the experiment id and the assigned\nvariant label, so the tag is safe to log and surface. ``is_candidate``\nflags whether the *candidate* (alternative) config served the response, so a\nconsumer can split traffic without re-deriving it from the variant name.", + "properties": { + "experiment": { + "title": "Experiment", + "type": "string" + }, + "variant": { + "title": "Variant", + "type": "string" + }, + "is_candidate": { + "default": false, + "title": "Is Candidate", + "type": "boolean" + } + }, + "required": [ + "experiment", + "variant" + ], + "title": "ExperimentAssignment", + "type": "object" + }, "GuardAction": { "description": "What the hallucination guard does with an unfaithful answer (Step 4.3).\n\nANNOTATE\n Observe only: attach per-claim verdicts to the response and emit the\n ``guard.claim_blocked`` event, but return the answer text unchanged.\n The behaviour-neutral default \u2014 safe to enable in production for\n monitoring before any enforcement.\nREDACT\n Drop the unsupported claims from the answer, returning the grounded\n remainder.\nBLOCK\n Withhold the whole answer when any claim is unsupported, replacing it\n with a refusal notice.", "enum": [ @@ -846,6 +870,17 @@ }, "trace": { "$ref": "#/$defs/TraceContext" + }, + "experiment": { + "anyOf": [ + { + "$ref": "#/$defs/ExperimentAssignment" + }, + { + "type": "null" + } + ], + "default": null } }, "required": [ diff --git a/dist/schemas/RetrieveResponse.json b/dist/schemas/RetrieveResponse.json index aa66e02..ea59b27 100644 --- a/dist/schemas/RetrieveResponse.json +++ b/dist/schemas/RetrieveResponse.json @@ -132,6 +132,30 @@ "title": "CorpusScore", "type": "object" }, + "ExperimentAssignment": { + "description": "Which A/B variant served this response (Step 5.7c).\n\nPresent on a :class:`QueryResponse` / :class:`RetrieveResponse` only when\nA/B *routing* is active and assigned the request to an experiment \u2014 the\ndeterministic, ``request_id``-hashed slice that can actually change which\nretrieval config a user gets (shadow mode, Step 5.7b, never tags a response).\n``None`` otherwise (routing off, or the request was not in the experiment).\n\nCarries no query or answer text \u2014 only the experiment id and the assigned\nvariant label, so the tag is safe to log and surface. ``is_candidate``\nflags whether the *candidate* (alternative) config served the response, so a\nconsumer can split traffic without re-deriving it from the variant name.", + "properties": { + "experiment": { + "title": "Experiment", + "type": "string" + }, + "variant": { + "title": "Variant", + "type": "string" + }, + "is_candidate": { + "default": false, + "title": "Is Candidate", + "type": "boolean" + } + }, + "required": [ + "experiment", + "variant" + ], + "title": "ExperimentAssignment", + "type": "object" + }, "QueryShape": { "description": "Coarse classification of an incoming query, used by the router.\n\nStep 2.10. The router maps an :class:`~rag_query.types.UnderstoodQuery`\nonto a shape and uses it to bias backend selection + per-backend\ntop-k. Shape detection is deliberately heuristic \u2014 the cost of a\nwrong call is small (an unused backend's results are simply ignored\nafter RRF fusion) and a richer classifier (LLM-based, learned) can\nbe slotted behind the same enum later.\n\nValues\n------\nKEYWORD_HEAVY\n Short, term-like query (no question form, \u2264 3 tokens, no HyDE\n signal). Keyword path is primary; vector path is optional.\nSEMANTIC\n Natural-language question or HyDE-augmented query. All three\n backends are useful; vector + keyword in fusion is the typical\n sweet spot.\nENTITY_RICH\n Query whose glossary expansion / decomposition surfaces multiple\n named entities \u2014 graph + keyword get a boost; vector still runs.\nMIXED\n Default fall-through when no specific shape rule fires. All\n wired backends run with equal weight.", "enum": [ @@ -351,6 +375,17 @@ }, "trace": { "$ref": "#/$defs/TraceContext" + }, + "experiment": { + "anyOf": [ + { + "$ref": "#/$defs/ExperimentAssignment" + }, + { + "type": "null" + } + ], + "default": null } }, "required": [ diff --git a/docs/README.md b/docs/README.md index ea7514d..b7a56c3 100644 --- a/docs/README.md +++ b/docs/README.md @@ -52,6 +52,7 @@ | [per-query-tracing.md](architecture/per-query-tracing.md) | Per-query tracing & provenance (Step 5.1): stable span attributes via the one `span_from_trace_context` choke point (`rag.schema_version` on every span) + the `telemetry_attrs` registry/contract; HMAC-signed `ProvenanceRecord` (privacy-by-default hashes, degrade-open recording); the `TraceCollector` read-side span processor grouping by `rag.trace_id`; tenant-isolated `GET /v1/query/{id}/trace`; inert-by-default wiring; invariants + reviewer checklist | | [golden-set-eval.md](architecture/golden-set-eval.md) | Offline golden-set eval harness (Step 5.2): why the noop backends give a real signal (`NoopKeywordStore` token overlap + a deterministic `HashingEmbedder` for the zero-vector `NoopEmbedder`) fused via the real `HybridRetriever`; corpus/golden-set co-design (5 domains × 20 concepts × 3 passages, `chunk_id == source_uri`, committed + drift-gated); metric definitions/edge cases (nDCG, dependency-free `lexical_faithfulness` vs RAGAS); additive report types + self-contained HTML; relationship to the Step 5.3 gate; reviewer checklist | | [ab-shadow-mode.md](architecture/ab-shadow-mode.md) | A/B testing & shadow mode (Step 5.7b): run a candidate retriever observe-only on a sampled fraction of live queries and feed the `ABExperimentTracker`; `ShadowRunner` scheduled as a FastAPI `BackgroundTask` (served-path latency untouched) + degrade-open + deterministic `request_id`-hash sampling; backend-only `outcome_metric` (mean retrieval score, the drift signal, since the candidate is never served); skipped on a cache hit; candidate = any `SupportsRoute` (config builds a `RetrievalRouter` differing only in RRF fusion weights; production injects one); same `read_chunk` PDP, event-only (`experiment.shadow_*`), inert + opt-in by default; key decisions + extension points | +| [ab-routing.md](architecture/ab-routing.md) | A/B routing (Step 5.7c): deterministically assign each query to a variant and *serve* the candidate config to the `routing_sample_rate` fraction (the first slice that can change a response); `ABRouter` decides/records/tags while the `/v1/query` + `/v1/retrieve` handlers own the served-path wiring; served-not-background + not-degrade-open (candidate failure → 502, no silent fallback) + variant-partitioned retrieval cache + one outcome per fresh retrieval; `ExperimentAssignment` wire tag on the response (REST-only for now); same `read_chunk` PDP, gated separately (`routing_enabled`) from shadow, inert + opt-in by default; governance + extension points | | [drift-monitors.md](architecture/drift-monitors.md) | Drift monitors (Step 5.5): the five monitors (query distribution / embedding PSI / retrieval score / citation-clickthrough / faithfulness) comparing a reference vs current window; two statistics (PSI for distributions, mean-drop for rate/score) over one scalar-window `DriftMonitor`; infra-scoped `DriftMonitorRegistry` fed via `observe` from the query + feedback paths (cheap O(1) appends, sparse signals report `insufficient_data`); detection on dashboard-poll `evaluate()` with transition-edge `drift.detected` (structured event + the Step 3.9 webhook); observe-only / inert-by-default / rebaseline; `GET /v1/status/drift`; invariants + extension points + reviewer checklist | | [cost-anomaly.md](architecture/cost-anomaly.md) | Cost anomaly (Step 5.6c): per-tenant spend-spike detection — a `CostTracker` (in `rag-observability`, beside the metrics/log/trace read-side) keeping a bounded rolling window of recent per-request token costs per tenant, with a two-gate (ratio + z-score, z relaxed on a flat baseline) tri-state verdict; scale-free on tokens so it's decoupled from quota pricing; fed O(1) from `record_request_usage` independent of quotas; pull-based `GET /v1/status/cost` (no per-request span/event); inert-by-default; invariants + extension points + reviewer checklist | | [online-feedback.md](architecture/online-feedback.md) | Online feedback & implicit signals (Step 5.4): the `POST /v1/feedback` → `FeedbackRecorder` (normalise score → PII-redact comment → tenant-scoped `FeedbackStore.put` → `feedback.recorded`, degrade-open) flow + the `GET /v1/status/feedback` → `aggregate_feedback` → `FeedbackStats` dashboard; one polymorphic `signal` enum for explicit + implicit; redact-don't-hash (vs provenance); `[-1,1]` score normalisation; body identity; event-only observability; inert-by-default wiring; tenant-isolation / no-PII / degrade-open invariants; reviewer checklist | diff --git a/docs/adr/ADR-0032-ab-testing-shadow-mode.md b/docs/adr/ADR-0032-ab-testing-shadow-mode.md index fac0a46..5413f89 100644 --- a/docs/adr/ADR-0032-ab-testing-shadow-mode.md +++ b/docs/adr/ADR-0032-ab-testing-shadow-mode.md @@ -97,6 +97,53 @@ followed these decisions, consistent with the Phase-5 observe-only features: See [architecture/ab-shadow-mode.md](../architecture/ab-shadow-mode.md). +## Update — Step 5.7c (A/B routing, landed) + +A/B *routing* is now implemented (the third slice) — the first slice that can +change which response a user gets. It follows these decisions: + +- **`ABRouter` decides; the handler serves.** `ABRouter` (`rag_gateway.experiments`, + beside `ShadowRunner`) `assign(request_id)`s a variant by the same deterministic + `hash(request_id) < sample_rate` rule shadow uses, holds the candidate + `SupportsRoute`, `record`s the served arm's `outcome_metric`, and `tag`s the + response. The REST query handler owns the wiring: when the candidate arm is + assigned, `_route_query` serves `ab.candidate.route(...)` *inline* instead of + the control router (so the user actually gets the candidate config). +- **Served, not background.** Unlike shadow's `BackgroundTask`, routing is on the + served path — that is the point. Only the assigned arm runs (one retrieval, no + added latency vs. a normal query); the other arm is simply not executed. +- **Not degrade-open.** Because the candidate *is* the served path, a candidate + failure surfaces as a normal `RetrievalError` (→ 502) rather than being + swallowed — silently swapping the user back to control mid-experiment would + corrupt the comparison and hide a broken candidate. +- **Variant-partitioned cache.** The retrieval cache key (`plan_hash` + `scope`) + gains the assigned `variant`, so control and candidate never share an entry + (a candidate result can never be served to a control request, or vice versa). + Omitted when routing is off, leaving the pre-5.7c key byte-identical. +- **One outcome per fresh retrieval.** The served arm's outcome is recorded only + on the fresh-retrieval path (skipped on a cache hit), the same apples-to-apples + discipline shadow uses, so repeated cached queries don't over-weight a window. +- **Tag on the wire.** A new frozen `ExperimentAssignment` (`experiment` / + `variant` / `is_candidate`) joins `rag_core.gateway_types` and is set on the + `QueryResponse` / `RetrieveResponse` `experiment` field (additive, defaulted + `None`; in `dist/schemas` + `dist/openapi`). It is REST-only for now — the gRPC + `RagService` predates A/B routing, so (like `corpus_decision`) the proto mirror + is deferred. +- **Same `read_chunk` PDP site + separate gate.** The candidate routes through + the same `HybridRetriever.retrieve`, so ACLs are enforced identically and the + coverage linter needs no new entry. Routing is gated by `routing_enabled` + (requires `enabled`) — independent of `shadow_enabled` — because it, unlike + shadow, can change a response. + +`ShadowCandidateConfig` was generalised to `CandidateConfig` (shared by +`shadow_candidate` + the new `routing_candidate`); `cfg.experiments` gains +`routing_enabled` / `routing_sample_rate` / `routing_experiment` / +`routing_candidate`. `ragctl ab` drives the full assign → serve → record → analyze +flow against stubs. The console surface is the remaining slice (5.7d). + +See [architecture/ab-routing.md](../architecture/ab-routing.md). + ## See also - [reference/experiments.md](../reference/experiments.md) — API + config + endpoint - [architecture/ab-shadow-mode.md](../architecture/ab-shadow-mode.md) — shadow-mode design +- [architecture/ab-routing.md](../architecture/ab-routing.md) — A/B-routing design diff --git a/docs/architecture/ab-routing.md b/docs/architecture/ab-routing.md new file mode 100644 index 0000000..4027c80 --- /dev/null +++ b/docs/architecture/ab-routing.md @@ -0,0 +1,109 @@ +# A/B routing — architecture + +How the gateway *serves* a candidate retrieval config to a fraction of live +traffic and measures it against the control. Decision record: +[ADR-0032](../adr/ADR-0032-ab-testing-shadow-mode.md). Public API + config: +[reference/experiments.md](../reference/experiments.md). + +## Overview + +Step 5.7 lets an operator ask *"is config B better than the live config A?"* on +real traffic, in vertical slices: + +| Slice | What | Touches the served response? | +|-------|------|------------------------------| +| 5.7a | analyzer + sample collector + `GET /v1/status/experiments` | no | +| 5.7b | shadow mode — run a candidate observe-only, feed the collector | no | +| **5.7c** | **A/B routing** — actually serve the candidate to a fraction of users | **yes** | +| 5.7d | console surface + close-out | — | + +This doc covers **5.7c (A/B routing)** — the first slice that can change which +response a user receives. It builds on the pure `analyze_ab_experiment` and the +`ABExperimentTracker` from 5.7a (see +[reference/experiments.md](../reference/experiments.md)) and is the served-path +sibling of [shadow mode](ab-shadow-mode.md). + +## The routed path + +``` +POST /v1/query + → assign variant = hash(request_id) < routing_sample_rate ? candidate : control + → cache lookup, keyed by (plan_hash, scope) INCLUDING the variant + → understand → route: + control arm: the normal RetrievalRouter / CorpusRouter + candidate arm: ABRouter.candidate.route(...) ← served to the user + → record the served arm's outcome (fresh retrievals only) + → (rerank → pack → generate) + → return the response, tagged with the ExperimentAssignment +``` + +`ABRouter` (`rag_gateway.experiments`) owns the *decision* (`assign`), *holds* the +candidate, *records* the served outcome, and *tags* the response; the query +handler owns the *wiring* (it is the one place that knows the served control +path). `/v1/query` and `/v1/retrieve` both route through it. + +## Key decisions + +- **Served, not background.** Shadow mode runs its candidate in a + `BackgroundTask` precisely so it can't affect the response; A/B routing does the + opposite *on purpose* — the assigned candidate retrieval *is* the served path. + Only one arm runs per request, so a routed query costs the same as a normal one. +- **Deterministic assignment.** A request is assigned by + `hash(request_id) / 2³² < routing_sample_rate`, not an RNG draw: reproducible, + testable, sticky per request, and free of per-request global state. `rate ≤ 0` + sends everyone to control (routing is a no-op); `rate ≥ 1` sends everyone to the + candidate. +- **Not degrade-open.** Because the candidate is the served path, a candidate + failure surfaces as a normal `RetrievalError` (→ 502) rather than being trapped. + Silently falling back to control mid-experiment would both corrupt the A/B + comparison and mask a broken candidate. (Shadow mode, which never serves, stays + degrade-open.) +- **Variant-partitioned cache.** The retrieval-cache key folds in the assigned + variant (`compute_plan_hash` / `compute_params_hash` gain a `variant`), so the + control and candidate arms keep separate cache partitions — a candidate result + can never be served to a control-assigned request, or vice versa. With routing + off the `variant` is omitted and the key is byte-identical to before. +- **One outcome per fresh retrieval.** The served arm's `outcome_metric` (mean + retrieval score, the same backend-only signal shadow + drift use) is recorded + only when retrieval actually ran — skipped on a cache hit — so a frequently + cached query doesn't repeatedly stuff the same value into its window. Over many + distinct requests both arms' windows fill, and the dashboard analyses them. +- **Candidate = a `SupportsRoute`.** Reusing the structural router contract means + a `RetrievalRouter` (or `FallbackChain`) drops in unchanged. The config build + makes a second `RetrievalRouter` differing only in RRF fusion weights + (`routing_candidate`); production injects a candidate over the real backends via + `build_app(ab_router=…)`. +- **A wire tag.** The response carries a frozen `ExperimentAssignment` + (`experiment` / `variant` / `is_candidate`) so a caller (or a downstream + evaluator) knows which arm it got. It is PII-free — labels only. + +## Governance & observability + +- **No new PDP call.** The candidate routes through the same + `HybridRetriever.retrieve` (the canonical `read_chunk` policy site), so ACLs are + enforced on the candidate exactly as on the control. The PolicyEngine coverage + linter needs no new allowlist entry — `ABRouter` makes no governed SPI call of + its own. +- **Event-light.** There is no dedicated per-query routing event; the served + variant rides on the existing `gateway.query_complete` log (`experiment_variant`, + a label) plus the response tag and the pull-able tracker windows. + +## Opt-in by default + +`build_app` leaves `ab_router` unset. `build_app_from_config` builds it only when +`cfg.experiments.enabled` **and** `cfg.experiments.routing_enabled` and a tracker +exists to feed. Routing is gated *separately* from shadow because it — unlike +shadow — can change a served response; both can feed one tracker under distinct +`experiment` ids. + +## Extension points + +- **Outcome metric** — swap `outcome_metric` for latency, a served-answer + satisfaction signal (now available, since the candidate *is* served), or + per-shape buckets. +- **Candidate topology** — inject any `SupportsRoute` as the `ab_router`'s + candidate (a different reranker/packer chain, a remote gateway, …). +- **Assignment key** — `assign` hashes `request_id`; a sticky *per-user* split + (hash a stable principal id instead) is a drop-in change. +- **More variants** — the tracker holds any number of `(experiment, variant)` + windows; a multi-arm split is a generalisation of the binary `assign`. diff --git a/docs/reference/experiments.md b/docs/reference/experiments.md index 66e6b20..269816d 100644 --- a/docs/reference/experiments.md +++ b/docs/reference/experiments.md @@ -4,8 +4,9 @@ Compare two configs on live traffic — measure a candidate variant and decide with statistics whether it beats the control. See [ADR-0032](../adr/ADR-0032-ab-testing-shadow-mode.md). Slice **5.7a** delivers the analyzer + sample collector + dashboard; slice **5.7b** adds **shadow mode** — an -observe-only candidate fan-out that *feeds* the collector from live traffic. A/B -routing (5.7c) — actually serving the candidate to a fraction of users — is next. +observe-only candidate fan-out that *feeds* the collector from live traffic; slice +**5.7c** adds **A/B routing** — deterministically *serving* the candidate to a +fraction of users (the first slice that can change a response). ## Overview @@ -63,6 +64,7 @@ Empty when experiments are disabled or none have been observed. ```bash ragctl experiments --lift 0.25 # feed demo control/candidate samples, print lift + CI ragctl shadow --lift 0.25 # drive the real ShadowRunner end-to-end, print the A/B verdict +ragctl ab --rate 0.5 --lift 0.25 # drive the real ABRouter — assign, serve, record, analyze ``` ### Configuration (`cfg.experiments`) @@ -77,7 +79,11 @@ ragctl shadow --lift 0.25 # drive the real ShadowRunner end-to-end, pri | `shadow_enabled` | `false` | Turn on the shadow fan-out (requires `enabled`). | | `shadow_sample_rate` | `0.1` | Fraction `[0, 1]` of live queries shadowed (deterministic per `request_id`). | | `shadow_experiment` | `shadow` | Experiment id the shadow samples land under. | -| `shadow_candidate.{vector,keyword,graph}_weight` | `1.0` | The candidate retriever's RRF fusion weights. | +| `shadow_candidate.{vector,keyword,graph}_weight` | `1.0` | The shadow candidate retriever's RRF fusion weights. | +| `routing_enabled` | `false` | Turn on A/B routing — *serves* the candidate (requires `enabled`). | +| `routing_sample_rate` | `0.1` | Fraction `[0, 1]` of requests *served* the candidate (deterministic per `request_id`); the rest get control. | +| `routing_experiment` | `ab` | Experiment id the served samples land under. | +| `routing_candidate.{vector,keyword,graph}_weight` | `1.0` | The A/B-routing candidate retriever's RRF fusion weights. | ## Shadow mode (Step 5.7b) @@ -109,6 +115,39 @@ them on real traffic before anyone serves the candidate. `/v1/retrieve` schedule it when one is wired and the request is in the sample. See [architecture/ab-shadow-mode.md](../architecture/ab-shadow-mode.md). +## A/B routing (Step 5.7c) + +**A/B routing** deterministically assigns each query to a variant and, for the +`routing_sample_rate` fraction assigned to the *candidate*, **serves** the +candidate retrieval config to the user — the first slice that changes which +response a user gets, so it is gated separately (`routing_enabled`) from shadow. + +- **Serves, not observes.** Only the assigned arm runs: the control arm takes the + normal served path; the candidate arm is served `ABRouter.candidate.route(...)` + *inline*. A routed query costs the same as a normal one (one retrieval). +- **Deterministic assignment.** The variant is `hash(request_id) < routing_sample_rate ? + candidate : control` — reproducible, sticky per request, testable. `rate ≤ 0` + routes everyone to control (a no-op); `rate ≥ 1` routes everyone to the candidate. +- **Variant-partitioned cache.** The retrieval-cache key includes the assigned + variant, so the control and candidate arms never share a cached result. +- **Records the served arm.** Each fresh retrieval records its served outcome + (mean retrieval score) under its arm; over many requests both windows fill and + `GET /v1/status/experiments` analyses them — now from *served* traffic. +- **Tags the response.** `QueryResponse.experiment` / `RetrieveResponse.experiment` + carry an `ExperimentAssignment` (`experiment` / `variant` / `is_candidate`) when + routing assigned the request; `None` otherwise. +- **Not degrade-open.** The candidate is the served path, so a candidate failure + surfaces as a normal retrieval error rather than silently swapping to control. + +```python +r = client.post("/v1/query", json={"tenant_id": "t1", "principal_id": "p1", "query": "…"}) +r.json()["experiment"] # {"experiment": "ab", "variant": "candidate", "is_candidate": true} | null +``` + +`ABRouter` (in `rag_gateway.experiments`) decides + records + tags; the `/v1/query` +and `/v1/retrieve` handlers own the served-path wiring. See +[architecture/ab-routing.md](../architecture/ab-routing.md). + ## Internals - **Normal-approximation Welch test.** With `min_samples ≥ 30` the CLT makes a @@ -117,14 +156,15 @@ them on real traffic before anyone serves the candidate. numpy/scipy. `significant` is whether the CI on the mean difference excludes 0. - **Decoupled by design.** The tracker holds samples; the analyzer is pure; the gateway endpoint composes them. `rag_observability` never imports `rag_config`. -- **Opt-in + inert.** `build_app` leaves `experiment_tracker` + `shadow_runner` - unset; `build_app_from_config` builds the tracker when `enabled` and the shadow - runner only when `enabled` **and** `shadow_enabled`. +- **Opt-in + inert.** `build_app` leaves `experiment_tracker` + `shadow_runner` + + `ab_router` unset; `build_app_from_config` builds the tracker when `enabled`, the + shadow runner only when `enabled` **and** `shadow_enabled`, and the A/B router + only when `enabled` **and** `routing_enabled`. ## Extension points -- **Feed it** — shadow mode (5.7b, shipped) calls `tracker.observe(...)` from a - background task; A/B routing (5.7c) will feed it from the served path. +- **Feed it** — shadow mode (5.7b) calls `tracker.observe(...)` from a background + task; A/B routing (5.7c) feeds it from the served path via `ABRouter.record`. - **Swap the test** — replace the normal approximation in `analyze_ab_experiment` with an exact t-test (scipy) or a sequential test without changing the result shape or the endpoint. diff --git a/packages/config/src/rag_config/schema.py b/packages/config/src/rag_config/schema.py index d431047..94536bd 100644 --- a/packages/config/src/rag_config/schema.py +++ b/packages/config/src/rag_config/schema.py @@ -738,14 +738,16 @@ class CostConfig(_StrictBase): dollars_per_1k_tokens: Annotated[float, Field(ge=0.0)] = 0.0 -class ShadowCandidateConfig(_StrictBase): - """Fusion weights for the shadow candidate retriever (Step 5.7b). +class CandidateConfig(_StrictBase): + """Fusion weights for a candidate retriever (Steps 5.7b / 5.7c). The candidate runs the same retrieval backends as the served (control) path - but with these per-source RRF weights, so an operator can shadow-test an - alternative fusion balance (e.g. vector-heavy) against the live config on a - sample of real traffic — observe-only. Defaults are neutral ``1.0`` (an - operator sets the candidate to something distinct to learn anything). + but with these per-source RRF weights, so an operator can compare an + alternative fusion balance (e.g. vector-heavy) against the live config — + observe-only under shadow mode (``shadow_candidate``) or served to a fraction + of users under A/B routing (``routing_candidate``). Defaults are neutral + ``1.0`` (an operator sets the candidate to something distinct to learn + anything). """ vector_weight: Annotated[float, Field(ge=0.0)] = 1.0 @@ -778,6 +780,16 @@ class ExperimentsConfig(_StrictBase): so the served response is never delayed. * ``shadow_experiment`` — the experiment id the shadow samples land under. * ``shadow_candidate`` — the candidate retriever's fusion weights. + + A/B routing (Step 5.7c) — *live* variant assignment that **serves** the + candidate to a fraction of users (the first slice that can change a response, + so it is gated separately from shadow): + + * ``routing_enabled`` — turn on A/B routing (requires ``enabled``). + * ``routing_sample_rate`` — fraction ``[0, 1]`` of requests served the + *candidate* config (deterministic per ``request_id``); the rest get control. + * ``routing_experiment`` — the experiment id the served samples land under. + * ``routing_candidate`` — the candidate retriever's fusion weights. """ enabled: bool = False @@ -789,7 +801,11 @@ class ExperimentsConfig(_StrictBase): shadow_enabled: bool = False shadow_sample_rate: Annotated[float, Field(ge=0.0, le=1.0)] = 0.1 shadow_experiment: str = "shadow" - shadow_candidate: ShadowCandidateConfig = Field(default_factory=ShadowCandidateConfig) + shadow_candidate: CandidateConfig = Field(default_factory=CandidateConfig) + routing_enabled: bool = False + routing_sample_rate: Annotated[float, Field(ge=0.0, le=1.0)] = 0.1 + routing_experiment: str = "ab" + routing_candidate: CandidateConfig = Field(default_factory=CandidateConfig) class RagConfig(_StrictBase): diff --git a/packages/core/src/rag_core/__init__.py b/packages/core/src/rag_core/__init__.py index 9636282..1e58cb7 100644 --- a/packages/core/src/rag_core/__init__.py +++ b/packages/core/src/rag_core/__init__.py @@ -76,6 +76,7 @@ Answer, Corpus, CorpusList, + ExperimentAssignment, GatewayError, QueryRequest, QueryResponse, @@ -351,6 +352,7 @@ "Answer", "Corpus", "CorpusList", + "ExperimentAssignment", "GatewayError", "QueryRequest", "QueryResponse", diff --git a/packages/core/src/rag_core/gateway_types.py b/packages/core/src/rag_core/gateway_types.py index e1e0eaa..65f52c0 100644 --- a/packages/core/src/rag_core/gateway_types.py +++ b/packages/core/src/rag_core/gateway_types.py @@ -257,6 +257,28 @@ class Answer(BaseModel): # --------------------------------------------------------------------------- # Response shapes # --------------------------------------------------------------------------- +class ExperimentAssignment(BaseModel): + """Which A/B variant served this response (Step 5.7c). + + Present on a :class:`QueryResponse` / :class:`RetrieveResponse` only when + A/B *routing* is active and assigned the request to an experiment — the + deterministic, ``request_id``-hashed slice that can actually change which + retrieval config a user gets (shadow mode, Step 5.7b, never tags a response). + ``None`` otherwise (routing off, or the request was not in the experiment). + + Carries no query or answer text — only the experiment id and the assigned + variant label, so the tag is safe to log and surface. ``is_candidate`` + flags whether the *candidate* (alternative) config served the response, so a + consumer can split traffic without re-deriving it from the variant name. + """ + + model_config = {"frozen": True} + + experiment: str + variant: str + is_candidate: bool = False + + class QueryResponse(BaseModel): """``POST /v1/query`` response envelope. @@ -298,6 +320,7 @@ class QueryResponse(BaseModel): answer: Answer | None = None timings: StageTimings = Field(default_factory=StageTimings) trace: TraceContext = Field(default_factory=TraceContext) + experiment: ExperimentAssignment | None = None class RetrieveResponse(BaseModel): @@ -318,6 +341,7 @@ class RetrieveResponse(BaseModel): chunks: list[ChunkRef] = Field(default_factory=list) timings: StageTimings = Field(default_factory=StageTimings) trace: TraceContext = Field(default_factory=TraceContext) + experiment: ExperimentAssignment | None = None class QueryTraceResponse(BaseModel): @@ -441,6 +465,7 @@ class GatewayError(BaseModel): "Answer", "Corpus", "CorpusList", + "ExperimentAssignment", "FeedbackAck", "FeedbackRequest", "GatewayError", diff --git a/packages/core/src/rag_core/gen_schemas.py b/packages/core/src/rag_core/gen_schemas.py index 758c5cb..eed1d9c 100644 --- a/packages/core/src/rag_core/gen_schemas.py +++ b/packages/core/src/rag_core/gen_schemas.py @@ -28,6 +28,7 @@ Answer, Corpus, CorpusList, + ExperimentAssignment, FeedbackAck, FeedbackRequest, GatewayError, @@ -197,6 +198,8 @@ StageTimings, Answer, GatewayError, + # A/B routing assignment tag (Step 5.7c). + ExperimentAssignment, # Online feedback wire types (Step 5.4). FeedbackRequest, FeedbackAck, diff --git a/packages/ragctl/src/ragctl/main.py b/packages/ragctl/src/ragctl/main.py index 596485a..a64f57b 100644 --- a/packages/ragctl/src/ragctl/main.py +++ b/packages/ragctl/src/ragctl/main.py @@ -2976,6 +2976,118 @@ async def _drive() -> None: typer.echo(f" p={r.p_value:.4f} → {verdict} at {r.confidence:.0%}") +@app.command("ab") +def ab( + queries: int = typer.Option(200, "--queries", "-n", help="Simulated live queries."), + rate: float = typer.Option( + 0.5, "--rate", help="Fraction of queries served the candidate (the rest get control)." + ), + lift: float = typer.Option( + 0.3, "--lift", help="Mean retrieval-score uplift of the candidate over control." + ), +) -> None: + """Drive A/B routing (Step 5.7c) end-to-end against in-process stubs. + + Builds an :class:`~rag_gateway.experiments.ABRouter` over a candidate that + scores ``--lift`` above the control, then for each of ``--queries`` simulated + requests deterministically assigns a variant: the candidate arm is *served* + the candidate's retrieval (and recorded), the control arm is served the + control. Prints the realised traffic split and the A/B comparison the + dashboard (``GET /v1/status/experiments``) would show. No services, no creds. + + Example:: + + ragctl ab --rate 0.5 --lift 0.25 + """ + import asyncio + + from rag_config.eval import analyze_ab_experiment + from rag_core.types import ( + ChunkId, + ChunkRef, + Principal, + PrincipalId, + PrincipalKind, + QueryShape, + RequestContext, + RequestId, + RoutingDecision, + TenantId, + TraceContext, + ) + from rag_gateway.experiments import ABRouter + from rag_observability import ABExperimentTracker + + tenant = TenantId("demo") + decision = RoutingDecision( + shape=QueryShape.SEMANTIC, use_vector=True, use_keyword=True, use_graph=False + ) + + def _ctx(i: int) -> RequestContext: + return RequestContext( + request_id=RequestId(f"q{i}"), + tenant_id=tenant, + principal=Principal( + id=PrincipalId("ragctl"), + kind=PrincipalKind.service, + display_name="ragctl", + tenant_id=tenant, + ), + trace=TraceContext(), + ) + + def _refs(base: float, i: int) -> list[ChunkRef]: + jitter = 0.02 * (i % 5 - 2) + return [ChunkRef(chunk_id=ChunkId("c1"), tenant_id=tenant, score=base + jitter)] + + class _Candidate: + async def route( + self, ctx: RequestContext, **kwargs: object + ) -> tuple[RoutingDecision, list[ChunkRef]]: + i = int(str(ctx.request_id)[1:]) + return decision, _refs(0.5 + lift, i) + + tracker = ABExperimentTracker(min_samples=min(30, queries)) + candidate = _Candidate() + router = ABRouter( + candidate_router=candidate, tracker=tracker, sample_rate=rate, experiment="ab" + ) + + async def _drive() -> None: + for i in range(queries): + ctx = _ctx(i) + variant = router.assign(str(ctx.request_id)) + if router.is_candidate(variant): + # A/B routing serves the candidate's retrieval to this arm. + _decision, refs = await router.candidate.route(ctx) + else: + # The control arm is served the live (control) retrieval. + refs = _refs(0.5, i) + router.record(variant, refs) + + asyncio.run(_drive()) + n_candidate = len(tracker.samples("ab", "candidate")) + r = analyze_ab_experiment( + tracker.samples("ab", tracker.control), + tracker.samples("ab", tracker.candidate), + experiment="ab", + metric="retrieval_score", + confidence=tracker.confidence, + min_samples=tracker.min_samples, + ) + verdict = "SIGNIFICANT" if r.significant else "not significant" + served_pct = n_candidate / queries if queries else 0.0 + typer.echo(f"\nA/B routing — {r.experiment} ({r.metric}): {r.status}") + typer.echo("─" * 64) + typer.echo(f" served candidate to {served_pct:.0%} of {queries} queries (target {rate:.0%})") + typer.echo(f" control n={r.n_a:<4} mean={r.mean_a:.4f}") + typer.echo(f" candidate n={r.n_b:<4} mean={r.mean_b:.4f}") + typer.echo( + f" lift={r.lift:+.1%} diff={r.diff:+.4f} CI=[{r.ci_lower:+.4f}, {r.ci_upper:+.4f}]" + ) + typer.echo(f" p={r.p_value:.4f} → {verdict} at {r.confidence:.0%}") + + @app.command("perf") def perf( requests: int = typer.Option( diff --git a/tests/config/test_experiments_config.py b/tests/config/test_experiments_config.py index 7ed34d7..a55fec6 100644 --- a/tests/config/test_experiments_config.py +++ b/tests/config/test_experiments_config.py @@ -78,8 +78,46 @@ def test_shadow_sample_rate_bounds() -> None: ExperimentsConfig(shadow_sample_rate=bad) -def test_shadow_candidate_weights_nonnegative() -> None: - from rag_config.schema import ShadowCandidateConfig +def test_candidate_weights_nonnegative() -> None: + from rag_config.schema import CandidateConfig with pytest.raises(ValidationError): - ShadowCandidateConfig(vector_weight=-1.0) + CandidateConfig(vector_weight=-1.0) + + +def test_routing_defaults() -> None: + ec = RagConfig(version="1").experiments + assert ec.routing_enabled is False # requires experiments.enabled too + assert ec.routing_sample_rate == pytest.approx(0.1) + assert ec.routing_experiment == "ab" + assert ec.routing_candidate.vector_weight == pytest.approx(1.0) + assert ec.routing_candidate.keyword_weight == pytest.approx(1.0) + assert ec.routing_candidate.graph_weight == pytest.approx(1.0) + + +def test_routing_override_from_mapping() -> None: + cfg = RagConfig.model_validate( + { + "version": "1", + "experiments": { + "enabled": True, + "routing_enabled": True, + "routing_sample_rate": 0.5, + "routing_experiment": "router_v2", + "routing_candidate": {"vector_weight": 2.0, "graph_weight": 0.0}, + }, + } + ) + ec = cfg.experiments + assert ec.routing_enabled is True + assert ec.routing_sample_rate == pytest.approx(0.5) + assert ec.routing_experiment == "router_v2" + assert ec.routing_candidate.vector_weight == pytest.approx(2.0) + assert ec.routing_candidate.graph_weight == pytest.approx(0.0) + assert ec.routing_candidate.keyword_weight == pytest.approx(1.0) + + +def test_routing_sample_rate_bounds() -> None: + for bad in (-0.1, 1.1): + with pytest.raises(ValidationError): + ExperimentsConfig(routing_sample_rate=bad) diff --git a/tests/contract/test_grpc_proto_compat.py b/tests/contract/test_grpc_proto_compat.py index c40f6b8..ddd37e9 100644 --- a/tests/contract/test_grpc_proto_compat.py +++ b/tests/contract/test_grpc_proto_compat.py @@ -51,8 +51,14 @@ # corpus router; mirroring CorpusRoutingDecision onto the proto is # deferred to the gRPC corpus-routing pass, so it is REST/MCP-only for # now and intentionally absent from the proto messages. - "QueryResponse": {"corpus_decision"}, - "RetrieveResponse": {"corpus_decision"}, + # + # experiment (Step 5.7c) is the A/B-routing assignment tag, set on the REST + # /v1/query + /v1/retrieve responses by the REST handler. A/B routing is + # wired into that handler only; the gRPC RagService predates it, so + # mirroring ExperimentAssignment onto the proto is deferred to a future + # gRPC pass and the field is REST-only for now. + "QueryResponse": {"corpus_decision", "experiment"}, + "RetrieveResponse": {"corpus_decision", "experiment"}, # guard (Step 4.3) is the hallucination-guard verdict surfaced on the REST # /v1/query answer + the OpenAI chat response. The gRPC RagService # (Step 3.2) predates the guard; mirroring GuardResult onto the proto is