From 6210f887626b3555b2b26033c54b0961baee8d29 Mon Sep 17 00:00:00 2001 From: Deep Kumar Singh Kushwah Date: Sun, 7 Jun 2026 17:33:26 +0530 Subject: [PATCH 1/2] =?UTF-8?q?feat(experiments):=20shadow=20mode=20?= =?UTF-8?q?=E2=80=94=20observe-only=20candidate=20fan-out=20(Step=205.7b)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Run a candidate retriever alongside the served control path on a sampled fraction of live queries — observe-only — and feed both variants' outcome (mean retrieval score) into the 5.7a ABExperimentTracker, so the dashboard compares them on real traffic before anyone serves the candidate. - ShadowRunner (rag_gateway.experiments): scheduled as a FastAPI BackgroundTask so the served response is never delayed; degrade-open; deterministic request_id-hash sampling; skipped on a retrieval-cache hit. - Candidate is any SupportsRoute; the config build makes a RetrievalRouter differing only in shadow_candidate RRF weights; production injects one via build_app(shadow_runner=...). Same read_chunk PDP, event-only observability. - cfg.experiments gains shadow_enabled / shadow_sample_rate / shadow_experiment / shadow_candidate; doubly opt-in (enabled AND shadow_enabled). ragctl shadow. - 27 tests; ruff / mypy --strict (292) / RAG001 / log + policy gates green; rag.schema regenerated (no dist/schemas or openapi change). Docs: ADR-0032 (5.7b update), reference/experiments.md, architecture/ab-shadow-mode.md. Co-Authored-By: Claude Opus 4.7 --- TRACKER.md | 17 +- apps/gateway/src/rag_gateway/app.py | 17 ++ apps/gateway/src/rag_gateway/experiments.py | 163 ++++++++++++++ apps/gateway/src/rag_gateway/query.py | 94 +++++++- apps/gateway/src/rag_gateway/wiring.py | 46 ++++ apps/gateway/tests/test_shadow_mode.py | 238 ++++++++++++++++++++ dist/rag.schema.json | 48 +++- dist/rag.schema.yaml | 56 ++++- docs/README.md | 1 + docs/adr/ADR-0032-ab-testing-shadow-mode.md | 29 +++ docs/architecture/ab-shadow-mode.md | 92 ++++++++ docs/reference/experiments.md | 49 +++- packages/config/src/rag_config/schema.py | 28 +++ packages/ragctl/src/ragctl/main.py | 104 +++++++++ packages/ragctl/tests/test_shadow.py | 22 ++ tests/config/test_experiments_config.py | 45 ++++ 16 files changed, 1030 insertions(+), 19 deletions(-) create mode 100644 apps/gateway/src/rag_gateway/experiments.py create mode 100644 apps/gateway/tests/test_shadow_mode.py create mode 100644 docs/architecture/ab-shadow-mode.md create mode 100644 packages/ragctl/tests/test_shadow.py diff --git a/TRACKER.md b/TRACKER.md index a9dfc23..6171a77 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.7b — Shadow mode** — fan out N% of live queries to a candidate retriever (observe-only) and feed the A/B tracker. Final Phase-5 step, delivered in slices 5.7a–d. | +| **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. | **Recently shipped** +- **5.7b** ✅ Shadow mode — observe-only candidate fan-out (`ShadowRunner`, background task) feeding the A/B tracker - **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) - **5.5** ✅ Drift monitors — `rag-drift`, five PSI / mean-drop monitors — [#136](https://github.com/officialCodeWork/AgentContextOS/pull/136) @@ -523,7 +524,7 @@ | 5.6f | — Cross-links + close-out | ✅ | [#142](https://github.com/officialCodeWork/AgentContextOS/pull/142) | | 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 | ⏳ | — | +| 5.7b | — Shadow mode | ✅ | _pending_ | | 5.7c | — A/B routing | ⏳ | — | | 5.7d | — Console + close-out | ⏳ | — | @@ -612,9 +613,17 @@ - `cfg.experiments` (opt-in, since A/B routing can change responses); wired inert by default; **observe-only** — no query-path change yet (shadow / routing are 5.7b/c); `ragctl experiments`; 17 tests - [ADR-0032](docs/adr/ADR-0032-ab-testing-shadow-mode.md), [reference/experiments.md](docs/reference/experiments.md) -#### 5.7b — Shadow mode ⏳ +#### 5.7b — Shadow mode ✅ _pending_ -- **Next up.** Fan out N% of live queries to a candidate retriever (observe-only) and feed the A/B tracker +- New `ShadowRunner` (`rag_gateway.experiments`) runs a **candidate** retriever alongside the served **control** path on a sampled fraction of live queries — observe-only — and feeds both variants' `outcome_metric` (mean retrieval score) into the 5.7a `ABExperimentTracker` +- **Never delays the response:** scheduled as a FastAPI `BackgroundTask` (runs after the response is sent); **degrade-open** (`experiment.shadow_failed`); skipped on a retrieval-cache hit +- **Deterministic** `request_id`-hash sampling; candidate = any `SupportsRoute` (config builds a `RetrievalRouter` differing only in `shadow_candidate` RRF weights; production injects one via `build_app(shadow_runner=…)`); same `read_chunk` PDP — no coverage-linter entry +- `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 ⏳ + +- **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 ## Phase 6 — Governance & Tenancy (Weeks 28–34) ⏳ diff --git a/apps/gateway/src/rag_gateway/app.py b/apps/gateway/src/rag_gateway/app.py index 8d8873d..e9910c1 100644 --- a/apps/gateway/src/rag_gateway/app.py +++ b/apps/gateway/src/rag_gateway/app.py @@ -105,7 +105,9 @@ FallbackChain, FallbackConfig, HybridRetriever, + HybridWeights, RetrievalRouter, + RouterConfig, SemanticEmbeddingCache, ) from rag_webhooks import build_default_dispatcher, ingest_completed_event @@ -166,6 +168,7 @@ def build_default_understanding() -> QueryUnderstandingPipeline: def build_default_retrieval_router( *, breaker_registry: BreakerRegistry | None = None, + base_weights: HybridWeights | None = None, ) -> RetrievalRouter: """Noop-backed HybridRetriever + RetrievalRouter. @@ -173,6 +176,10 @@ def build_default_retrieval_router( shape's preferred path; production wiring substitutes real pgvector / Elasticsearch / Neo4j backends. + ``base_weights`` overrides the per-source RRF weights (before shape bias) — + used to build a *candidate* router for shadow mode (Step 5.7b) that differs + from the control only in its fusion balance. + When ``breaker_registry`` is supplied (Step 4.4), each backend is wrapped in a circuit breaker drawn from the registry, so a failing backend is isolated (open breaker → ``CircuitOpenError`` → ``HybridRetriever`` drops @@ -192,9 +199,11 @@ def build_default_retrieval_router( keyword_backend=keyword, graph_backend=graph, ) + config = RouterConfig(base_weights=base_weights) if base_weights is not None else None return RetrievalRouter( hybrid=hybrid, embedder=NoopEmbedder(dimension=8), + config=config, ) @@ -356,6 +365,7 @@ def build_app( drift_registry: Any | None = None, cost_tracker: Any | None = None, experiment_tracker: Any | None = None, + shadow_runner: Any | None = None, enable_cors: bool = True, default_tenant_id: TenantId | None = None, ) -> FastAPI: @@ -548,6 +558,13 @@ def build_app( # by ``GET /v1/status/experiments``. ``None`` in the plain ``build_app``. app.state.experiment_tracker = experiment_tracker + # Shadow-mode fan-out (Step 5.7b) — runs a candidate retriever observe-only + # on a sample of live queries and feeds the experiment tracker, from a + # background task so the served response is never delayed. ``None`` (inert) + # in the plain ``build_app``; the config-driven wiring builds it from + # ``cfg.experiments`` when ``shadow_enabled``. + app.state.shadow_runner = shadow_runner + # 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/experiments.py b/apps/gateway/src/rag_gateway/experiments.py new file mode 100644 index 0000000..faefbe2 --- /dev/null +++ b/apps/gateway/src/rag_gateway/experiments.py @@ -0,0 +1,163 @@ +"""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 + (``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). +""" + +from __future__ import annotations + +import hashlib +from typing import Any + +from rag_core import get_logger +from rag_core.types import RequestContext +from rag_observability.experiments import ABExperimentTracker + +_log = get_logger(__name__) + +__all__ = ["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. +_HASH_DENOM = 0xFFFFFFFF + + +def outcome_metric(chunk_refs: list[Any]) -> float: + """Scalar retrieval-quality outcome for a result set: the mean ref score. + + Higher is better; an empty result set scores ``0.0``. This is the same + retrieval-score signal the Step 5.5 drift monitors observe — a backend-only + proxy that needs no served answer (the candidate is never shown to a user). + Richer outcomes (latency, recall proxies, per-shape buckets) are future work. + """ + scores = [float(s) for r in chunk_refs if (s := getattr(r, "score", None)) is not None] + if not scores: + return 0.0 + return sum(scores) / len(scores) + + +def _sampled(request_id: str, sample_rate: float) -> bool: + """Deterministic in ``request_id`` — hash → ``[0, 1)`` vs ``sample_rate``.""" + if sample_rate <= 0.0: + return False + if sample_rate >= 1.0: + return True + digest = hashlib.sha256(request_id.encode("utf-8")).hexdigest()[:8] + return (int(digest, 16) / _HASH_DENOM) < sample_rate + + +class ShadowRunner: + """Runs a candidate retriever in shadow and feeds the experiment tracker. + + Holds a candidate :class:`~rag_retrieval.router.SupportsRoute` (the + alternative retrieval config), the :class:`ABExperimentTracker` to feed, the + sample rate, and the experiment + variant labels. :meth:`should_sample` + decides inclusion; :meth:`run` does the observe-only candidate retrieval and + records both variants' :func:`outcome_metric`. + + 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 = "shadow", + 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 + + def should_sample(self, request_id: str) -> bool: + """Whether this ``request_id`` is in the shadow sample (deterministic).""" + return _sampled(request_id, self._sample_rate) + + async def run( + self, + ctx: RequestContext, + *, + text: str, + expansion_terms: dict[str, list[str]], + hyde_vector: list[float] | None, + corpus_ids: list[Any] | None, + top_k: int, + control_refs: list[Any], + ) -> None: + """Observe-only: retrieve with the candidate, record both outcome metrics. + + ``control_refs`` is the result set the served path already produced, so + the control's metric needs no re-retrieval; only the candidate runs here. + Degrade-open — any failure logs ``experiment.shadow_failed`` and returns. + """ + try: + _decision, candidate_refs = await self._candidate.route( + ctx, + text=text, + expansion_terms=expansion_terms, + hyde_vector=hyde_vector, + corpus_ids=corpus_ids, + top_k=top_k, + ) + control_value = outcome_metric(control_refs) + candidate_value = outcome_metric(candidate_refs) + self._tracker.observe(self._experiment, self._control, control_value) + self._tracker.observe(self._experiment, self._candidate_variant, candidate_value) + _log.info( + "experiment.shadow_complete", + extra={ + "event_kind": "experiment.shadow_complete", + "tenant_id": str(ctx.tenant_id), + "experiment": self._experiment, + "control_value": control_value, + "candidate_value": candidate_value, + "candidate_results_n": len(candidate_refs), + }, + ) + except (KeyboardInterrupt, SystemExit): + raise + except Exception as exc: + _log.warning( + "experiment.shadow_failed", + extra={ + "event_kind": "experiment.shadow_failed", + "tenant_id": str(ctx.tenant_id), + "experiment": self._experiment, + "error": repr(exc), + }, + ) diff --git a/apps/gateway/src/rag_gateway/query.py b/apps/gateway/src/rag_gateway/query.py index b2e7091..b3a952a 100644 --- a/apps/gateway/src/rag_gateway/query.py +++ b/apps/gateway/src/rag_gateway/query.py @@ -30,7 +30,7 @@ from collections.abc import Sequence from typing import Any -from fastapi import APIRouter, HTTPException, Request +from fastapi import APIRouter, BackgroundTasks, HTTPException, Request from rag_core import get_logger from rag_core.errors import ( ACLDeniedError, @@ -448,7 +448,9 @@ def make_query_router() -> APIRouter: }, summary="Full RAG query: retrieve → rerank → pack → optional LLM answer", ) - async def query(request: Request, body: QueryRequest) -> QueryResponse: + async def query( + request: Request, body: QueryRequest, background: BackgroundTasks + ) -> QueryResponse: """End-to-end RAG query. Runs the Phase 2 pipeline in sequence — understanding, @@ -456,7 +458,7 @@ async def query(request: Request, body: QueryRequest) -> QueryResponse: returns the combined evidence envelope. See [docs/reference/gateway.md](../../../../docs/reference/gateway.md#post-v1query). """ - return await _handle_query(request, body) + return await _handle_query(request, body, background) @router.post( "/v1/retrieve", @@ -469,13 +471,15 @@ async def query(request: Request, body: QueryRequest) -> QueryResponse: }, summary="Retrieval-only: understanding + router; no rerank / pack / LLM", ) - async def retrieve(request: Request, body: RetrieveRequest) -> RetrieveResponse: + async def retrieve( + request: Request, body: RetrieveRequest, background: BackgroundTasks + ) -> RetrieveResponse: """Retrieval-only subset of ``/v1/query``. Useful when the caller brings their own reranker / packer / LLM and just wants the policy-checked, fused chunk-ref list. """ - return await _handle_retrieve(request, body) + return await _handle_retrieve(request, body, background) @router.get( "/v1/query/{request_id}/trace", @@ -501,7 +505,9 @@ async def query_trace(request_id: str, request: Request) -> QueryTraceResponse: # --------------------------------------------------------------------------- # Handlers # --------------------------------------------------------------------------- -async def _handle_query(request: Request, body: QueryRequest) -> QueryResponse: +async def _handle_query( + request: Request, body: QueryRequest, background: BackgroundTasks +) -> QueryResponse: ctx = _build_ctx_from_body( request, tenant_id_str=str(body.tenant_id), @@ -534,6 +540,7 @@ async def _handle_query(request: Request, body: QueryRequest) -> QueryResponse: decision: Any = None chunk_refs: list[Any] = [] hyde_vector_for_drift: list[float] | None = None + shadow_expansion: dict[str, list[str]] = {} cache_hit = False if cache is not None: cached_refs = await cache.get_best( @@ -555,6 +562,7 @@ async def _handle_query(request: Request, body: QueryRequest) -> QueryResponse: with timings.stage("understand"): understood = await deps.understanding.understand(ctx, body.query) hyde_vector_for_drift = understood.hyde_vector + shadow_expansion = dict(understood.expansion_terms) # Routing — corpus selection (when wired) + backend execute. with timings.stage("route"): @@ -686,6 +694,23 @@ async def _handle_query(request: Request, body: QueryRequest) -> QueryResponse: corpus_ids=list(body.corpus_ids), ) + # Shadow mode (Step 5.7b) — on a sampled fraction of freshly-retrieved + # queries, fan out to the candidate retriever in the background and feed + # the experiment tracker. No-op when no runner is wired; never delays + # this response. Skipped on a cache hit (the control did not retrieve). + if not cache_hit: + _maybe_schedule_shadow( + deps, + background, + ctx, + text=body.query, + expansion_terms=shadow_expansion, + hyde_vector=hyde_vector_for_drift, + corpus_ids=list(body.corpus_ids) or None, + top_k=body.top_k, + control_refs=chunk_refs, + ) + return QueryResponse( request_id=ctx.request_id, query=body.query, @@ -701,7 +726,9 @@ async def _handle_query(request: Request, body: QueryRequest) -> QueryResponse: ) -async def _handle_retrieve(request: Request, body: RetrieveRequest) -> RetrieveResponse: +async def _handle_retrieve( + request: Request, body: RetrieveRequest, background: BackgroundTasks +) -> RetrieveResponse: ctx = _build_ctx_from_body( request, tenant_id_str=str(body.tenant_id), @@ -812,6 +839,20 @@ async def _handle_retrieve(request: Request, body: RetrieveRequest) -> RetrieveR corpus_ids=list(body.corpus_ids), ) + # Shadow mode (Step 5.7b) — observe-only candidate fan-out on a sample of + # freshly-retrieved queries (background; never delays this response). + _maybe_schedule_shadow( + deps, + background, + ctx, + text=body.query, + expansion_terms=dict(understood.expansion_terms), + hyde_vector=understood.hyde_vector, + corpus_ids=list(body.corpus_ids) or None, + top_k=body.top_k, + control_refs=chunk_refs, + ) + return RetrieveResponse( request_id=ctx.request_id, query=body.query, @@ -891,6 +932,41 @@ def _observe_query_drift( registry.observe(DriftMetric.faithfulness, grounded) +def _maybe_schedule_shadow( + deps: _GatewayDeps, + background: BackgroundTasks, + ctx: RequestContext, + *, + text: str, + expansion_terms: dict[str, list[str]], + hyde_vector: list[float] | None, + corpus_ids: list[Any] | None, + top_k: int, + control_refs: list[Any], +) -> None: + """Schedule an observe-only shadow run (Step 5.7b) when the query is sampled. + + No-op when no runner is wired (the default) or the request is not in the + shadow sample. The candidate retrieval runs from a FastAPI background task, + so the served response — already built by the caller — is never delayed. + Only called on the retrieval path (not on a cache hit), so the control and + candidate are compared on the same freshly-retrieved query. + """ + runner = deps.shadow_runner + if runner is None or not runner.should_sample(str(ctx.request_id)): + return + background.add_task( + runner.run, + ctx, + text=text, + expansion_terms=expansion_terms, + hyde_vector=hyde_vector, + corpus_ids=corpus_ids, + top_k=top_k, + control_refs=control_refs, + ) + + async def _handle_query_trace(request: Request, request_id: str) -> QueryTraceResponse: """Serve ``GET /v1/query/{id}/trace`` — signed provenance + captured spans. @@ -1020,6 +1096,7 @@ class _GatewayDeps: "guard_enabled", "provenance_recorder", "drift_registry", + "shadow_runner", ) def __init__( @@ -1038,6 +1115,7 @@ def __init__( guard_enabled: bool, provenance_recorder: Any, drift_registry: Any, + shadow_runner: Any, ) -> None: self.understanding = understanding self.router = router @@ -1052,6 +1130,7 @@ def __init__( self.guard_enabled = guard_enabled self.provenance_recorder = provenance_recorder self.drift_registry = drift_registry + self.shadow_runner = shadow_runner @classmethod def from_request(cls, request: Request) -> _GatewayDeps: @@ -1070,6 +1149,7 @@ def from_request(cls, request: Request) -> _GatewayDeps: guard_enabled=getattr(state, "guard_enabled", False), provenance_recorder=getattr(state, "provenance_recorder", None), drift_registry=getattr(state, "drift_registry", None), + shadow_runner=getattr(state, "shadow_runner", None), ) diff --git a/apps/gateway/src/rag_gateway/wiring.py b/apps/gateway/src/rag_gateway/wiring.py index 60e1711..b39db72 100644 --- a/apps/gateway/src/rag_gateway/wiring.py +++ b/apps/gateway/src/rag_gateway/wiring.py @@ -88,6 +88,8 @@ if TYPE_CHECKING: from fastapi import FastAPI + from rag_gateway.experiments import ShadowRunner + _log = get_logger(__name__) _TERM_WEIGHTS_KEY = "term_weights" @@ -498,6 +500,43 @@ def build_experiment_tracker_from_config(cfg: RagConfig) -> ABExperimentTracker ) +def build_shadow_runner_from_config( + cfg: RagConfig, tracker: ABExperimentTracker | None +) -> ShadowRunner | None: + """Build the shadow-mode runner from ``cfg.experiments`` (Step 5.7b). + + Returns ``None`` unless experiments **and** shadow mode are enabled and a + ``tracker`` exists to feed. Builds a candidate :class:`RetrievalRouter` over + fresh noop backends with the configured candidate fusion weights, so the + feature works end-to-end from config; production injects a candidate over the + real backends via ``build_app(shadow_runner=...)``. + """ + ec = cfg.experiments + if not ec.enabled or not ec.shadow_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 ShadowRunner + + sc = ec.shadow_candidate + candidate = build_default_retrieval_router( + base_weights=HybridWeights( + vector=sc.vector_weight, + keyword=sc.keyword_weight, + graph=sc.graph_weight, + ) + ) + return ShadowRunner( + candidate_router=candidate, + tracker=tracker, + sample_rate=ec.shadow_sample_rate, + experiment=ec.shadow_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. @@ -593,6 +632,13 @@ def build_app_from_config(cfg: RagConfig, **overrides: Any) -> FastAPI: if "experiment_tracker" not in overrides: overrides["experiment_tracker"] = build_experiment_tracker_from_config(cfg) + # Shadow mode (Step 5.7b) — observe-only candidate fan-out fed into the same + # tracker. ``None`` unless experiments + shadow are both enabled in config. + if "shadow_runner" not in overrides: + overrides["shadow_runner"] = build_shadow_runner_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_shadow_mode.py b/apps/gateway/tests/test_shadow_mode.py new file mode 100644 index 0000000..20f9a16 --- /dev/null +++ b/apps/gateway/tests/test_shadow_mode.py @@ -0,0 +1,238 @@ +"""Tests for shadow mode — observe-only candidate fan-out (Step 5.7b). + +Covers the pure :class:`ShadowRunner` (sampling, the outcome metric, observing +both variants, degrade-open) and its gateway integration (a sampled ``/v1/query`` +feeds the experiment tracker from a background task; inert by default; the +config-driven build wires it only when ``shadow_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.errors import RetrievalError +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 ShadowRunner, outcome_metric +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 = "c1", 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 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) + + +class _FailingCandidate: + async def route( + self, ctx: RequestContext, **kwargs: Any + ) -> tuple[RoutingDecision, list[ChunkRef]]: + raise RetrievalError("candidate backend down") + + +# --------------------------------------------------------------------------- +# outcome_metric +# --------------------------------------------------------------------------- +def test_outcome_metric_empty_is_zero() -> None: + assert outcome_metric([]) == 0.0 + + +def test_outcome_metric_is_mean_of_scores() -> None: + assert outcome_metric([_ref(0.2), _ref(0.8)]) == 0.5 + + +def test_outcome_metric_ignores_missing_scores() -> None: + class _NoScore: + pass + + assert outcome_metric([_ref(0.6), _NoScore()]) == 0.6 # type: ignore[list-item] + + +# --------------------------------------------------------------------------- +# sampling +# --------------------------------------------------------------------------- +def test_sampling_bounds() -> None: + tracker = ABExperimentTracker() + never = ShadowRunner(candidate_router=_StubCandidate([]), tracker=tracker, sample_rate=0.0) + always = ShadowRunner(candidate_router=_StubCandidate([]), tracker=tracker, sample_rate=1.0) + assert never.should_sample("anything") is False + assert always.should_sample("anything") is True + + +def test_sampling_is_deterministic_in_request_id() -> None: + runner = ShadowRunner( + candidate_router=_StubCandidate([]), tracker=ABExperimentTracker(), sample_rate=0.5 + ) + assert runner.should_sample("req-abc") == runner.should_sample("req-abc") + + +def test_sampling_rate_splits_traffic() -> None: + runner = ShadowRunner( + candidate_router=_StubCandidate([]), tracker=ABExperimentTracker(), sample_rate=0.5 + ) + sampled = sum(runner.should_sample(f"req-{i}") for i in range(400)) + # Deterministic hash → roughly half; allow a wide band so it never flakes. + assert 120 < sampled < 280 + + +# --------------------------------------------------------------------------- +# ShadowRunner.run +# --------------------------------------------------------------------------- +async def test_run_observes_both_variants() -> None: + tracker = ABExperimentTracker(min_samples=2) + runner = ShadowRunner( + candidate_router=_StubCandidate([_ref(0.9), _ref(0.7)]), + tracker=tracker, + sample_rate=1.0, + experiment="exp1", + ) + await runner.run( + _ctx(), + text="q", + expansion_terms={}, + hyde_vector=None, + corpus_ids=None, + top_k=10, + control_refs=[_ref(0.4), _ref(0.6)], + ) + assert tracker.samples("exp1", "control") == [0.5] + assert tracker.samples("exp1", "candidate") == [0.8] + + +async def test_run_degrade_open_on_candidate_failure() -> None: + tracker = ABExperimentTracker() + runner = ShadowRunner(candidate_router=_FailingCandidate(), tracker=tracker, experiment="exp1") + # Must not raise — observe-only, degrade-open. + await runner.run( + _ctx(), + text="q", + expansion_terms={}, + hyde_vector=None, + corpus_ids=None, + top_k=10, + control_refs=[_ref(0.5)], + ) + # Neither variant recorded (the pair is atomic per run). + assert tracker.experiments() == [] + + +# --------------------------------------------------------------------------- +# Gateway integration +# --------------------------------------------------------------------------- +def _query(client: TestClient, request_id: str = "shadow-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 test_shadow_inert_by_default() -> None: + app = build_app() + assert app.state.shadow_runner is None + resp = _query(TestClient(app)) + assert resp.status_code == 200, resp.text + + +def test_shadow_feeds_tracker_on_sampled_query() -> None: + tracker = ABExperimentTracker(min_samples=2) + runner = ShadowRunner( + candidate_router=_StubCandidate([_ref(0.8)]), + tracker=tracker, + sample_rate=1.0, + experiment="shadow", + ) + client = TestClient(build_app(experiment_tracker=tracker, shadow_runner=runner)) + resp = _query(client) + assert resp.status_code == 200, resp.text + # The background shadow task ran before the TestClient call returned. + assert tracker.samples("shadow", "candidate") == [0.8] + assert len(tracker.samples("shadow", "control")) == 1 + # ...and the dashboard now sees the experiment. + body = client.get("/v1/status/experiments").json() + assert any(e["experiment"] == "shadow" for e in body["experiments"]) + + +def test_shadow_skipped_when_rate_zero() -> None: + tracker = ABExperimentTracker() + runner = ShadowRunner( + candidate_router=_StubCandidate([_ref(0.8)]), + tracker=tracker, + sample_rate=0.0, + experiment="shadow", + ) + resp = _query(TestClient(build_app(experiment_tracker=tracker, shadow_runner=runner))) + assert resp.status_code == 200, resp.text + assert tracker.experiments() == [] + + +# --------------------------------------------------------------------------- +# Config-driven wiring +# --------------------------------------------------------------------------- +def test_build_from_config_wires_shadow_when_enabled() -> None: + cfg = RagConfig( + experiments=ExperimentsConfig(enabled=True, shadow_enabled=True, shadow_sample_rate=1.0) + ) + app = build_app_from_config(cfg) + assert app.state.experiment_tracker is not None + assert app.state.shadow_runner is not None + + +def test_build_from_config_inert_when_shadow_disabled() -> None: + # Experiments on, shadow off → tracker built, runner not. + cfg = RagConfig(experiments=ExperimentsConfig(enabled=True, shadow_enabled=False)) + app = build_app_from_config(cfg) + assert app.state.experiment_tracker is not None + assert app.state.shadow_runner is None + + +def test_build_from_config_inert_when_experiments_disabled() -> None: + app = build_app_from_config(RagConfig()) + assert app.state.shadow_runner is None diff --git a/dist/rag.schema.json b/dist/rag.schema.json index 9cfe4d0..f97644a 100644 --- a/dist/rag.schema.json +++ b/dist/rag.schema.json @@ -541,7 +541,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.", + "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.", "properties": { "enabled": { "default": false, @@ -576,6 +576,26 @@ "default": "candidate", "title": "Candidate Variant", "type": "string" + }, + "shadow_enabled": { + "default": false, + "title": "Shadow Enabled", + "type": "boolean" + }, + "shadow_sample_rate": { + "default": 0.1, + "maximum": 1.0, + "minimum": 0.0, + "title": "Shadow Sample Rate", + "type": "number" + }, + "shadow_experiment": { + "default": "shadow", + "title": "Shadow Experiment", + "type": "string" + }, + "shadow_candidate": { + "$ref": "#/$defs/ShadowCandidateConfig" } }, "title": "ExperimentsConfig", @@ -1142,6 +1162,32 @@ "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 88a28cf..3ffa20f 100644 --- a/dist/rag.schema.yaml +++ b/dist/rag.schema.yaml @@ -505,7 +505,13 @@ $defs: \ cap.\n* ``min_samples`` — below this many samples on *either* side a comparison\n\ \ reports ``insufficient_data`` rather than a verdict.\n* ``confidence`` —\ \ confidence level for the interval / significance test.\n* ``control_variant``\ - \ / ``candidate_variant`` — the two variant labels the\n dashboard compares." + \ / ``candidate_variant`` — the two variant labels the\n dashboard compares.\n\ + \nShadow mode (Step 5.7b) — *observe-only* candidate fan-out:\n\n* ``shadow_enabled``\ + \ — turn on the shadow fan-out (requires ``enabled``).\n* ``shadow_sample_rate``\ + \ — 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." properties: enabled: default: false @@ -535,6 +541,22 @@ $defs: default: candidate title: Candidate Variant type: string + shadow_enabled: + default: false + title: Shadow Enabled + type: boolean + shadow_sample_rate: + default: 0.1 + maximum: 1.0 + minimum: 0.0 + title: Shadow Sample Rate + type: number + shadow_experiment: + default: shadow + title: Shadow Experiment + type: string + shadow_candidate: + $ref: '#/$defs/ShadowCandidateConfig' title: ExperimentsConfig type: object FallbackConfig: @@ -1049,6 +1071,38 @@ $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/docs/README.md b/docs/README.md index b052968..ea7514d 100644 --- a/docs/README.md +++ b/docs/README.md @@ -51,6 +51,7 @@ | [latency.md](architecture/latency.md) | Latency gate, load test & profiling (Step 4.6): the gateway-**overhead** budget (p99 ≤ 30 ms, noop backends) vs the end-to-end target; the server-side `gateway.request_duration_ms` as the gate signal; in-process / sequential / single-runner / retry-tolerant rationale (incl. the GC-on finding); in-process harness vs cProfile profiler (own-time sort, warmup outside the profiler) vs Locust; invariants + deferred + reviewer checklist | | [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 | | [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 7b59db3..fac0a46 100644 --- a/docs/adr/ADR-0032-ab-testing-shadow-mode.md +++ b/docs/adr/ADR-0032-ab-testing-shadow-mode.md @@ -69,5 +69,34 @@ response a user receives, so the whole feature is strictly opt-in. - One scalar metric per experiment at a time; multi-metric experiments + sequential testing (peeking correction) are future work. +## Update — Step 5.7b (shadow mode, landed) + +Shadow fan-out is now implemented (the second slice). The query-path integration +followed these decisions, consistent with the Phase-5 observe-only features: + +- **Background, never inline.** `ShadowRunner` (`rag_gateway.experiments`) runs the + candidate from a FastAPI `BackgroundTask`, so the served (control) response is + returned first and its latency is untouched. `/v1/query` + `/v1/retrieve` + schedule it; it is skipped on a retrieval-cache hit (the control did not + retrieve, so there is nothing comparable). +- **Degrade-open + event-only.** A candidate failure logs `experiment.shadow_failed` + and is swallowed; success logs `experiment.shadow_complete`. No per-call span — + the candidate's own `retrieve.route` span already traces the work. +- **Deterministic sampling.** Inclusion is `hash(request_id) < sample_rate`, not an + RNG draw, so it is reproducible and testable. +- **Backend-only outcome.** The candidate is never served, so the comparison uses + `outcome_metric` (mean retrieval score) — the same signal the drift monitors + observe — not a served-answer metric. +- **Candidate via fusion weights.** The config build constructs a second + `RetrievalRouter` differing only in `shadow_candidate` RRF weights; production + injects a candidate over real backends via `build_app(shadow_runner=…)`. Both + route through the same `read_chunk` PDP, so no coverage-linter entry is needed. +- **Doubly opt-in.** Runs only when `cfg.experiments.enabled` **and** + `shadow_enabled`; even then it is observe-only and cannot change a response. + (A/B *routing*, which can, remains the separately-gated 5.7c.) + +See [architecture/ab-shadow-mode.md](../architecture/ab-shadow-mode.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 diff --git a/docs/architecture/ab-shadow-mode.md b/docs/architecture/ab-shadow-mode.md new file mode 100644 index 0000000..1225998 --- /dev/null +++ b/docs/architecture/ab-shadow-mode.md @@ -0,0 +1,92 @@ +# A/B testing & shadow mode — architecture + +How the gateway compares two retrieval configs on live traffic without risking +the served response. 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. It ships 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.7b (shadow mode)**. The pieces it builds on (the pure +`analyze_ab_experiment` and the `ABExperimentTracker`) are described in +[reference/experiments.md](../reference/experiments.md). + +## The shadow path + +``` +POST /v1/query + → understand → route → (rerank → pack → generate) [served: control] + → build + return the response ← user is unblocked here + └─ BackgroundTask (only if sampled): + candidate.route(same query) [observe-only: candidate] + tracker.observe(exp, "control", outcome(control_refs)) + tracker.observe(exp, "candidate", outcome(candidate_refs)) +``` + +`ShadowRunner` (`rag_gateway.experiments`) owns the background step; +`/v1/query` and `/v1/retrieve` schedule it via `_maybe_schedule_shadow`. + +## Key decisions + +- **Background task, not inline.** The candidate retrieval is real async work, so + running it inline would add latency to the sampled requests. Scheduling it as a + FastAPI `BackgroundTask` means it runs *after* the response is sent — the served + path's latency is provably untouched. (Starlette runs background tasks within + the same ASGI call, so they remain deterministically testable.) +- **Degrade-open.** `ShadowRunner.run` traps every exception, logs + `experiment.shadow_failed`, and returns. The response was already sent, so a + flaky candidate backend can never surface to the user — the same reliability + invariant the Phase-4 features carry into Phase 5. +- **Deterministic sampling.** Inclusion is `hash(request_id) / 2³² < sample_rate`, + not an RNG draw: reproducible, testable, and free of per-request global state. + `rate ≤ 0` is off; `rate ≥ 1` is always. +- **A backend-only outcome metric.** The candidate is never shown to a user, so + the comparison cannot use a served-answer signal (satisfaction, click-through). + `outcome_metric` is the mean retrieval score of the result set — the same signal + the Step 5.5 drift monitors observe — so both variants are scored identically + with no ground truth. Latency / recall-proxy outcomes are a future extension. +- **Skipped on a cache hit.** When the control is served from the retrieval cache + it did not retrieve, so there is nothing to compare; shadowing only fires on the + fresh-retrieval path, keeping control and candidate apples-to-apples. +- **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 + (`shadow_candidate`); production injects a candidate over the real backends via + `build_app(shadow_runner=…)`. + +## 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 shadow exactly as on the control. The PolicyEngine coverage + linter needs no new allowlist entry — `ShadowRunner` makes no governed SPI call + of its own. +- **Event-only.** Like the cost / quota / breaker features, there is no per-call + span; observability is the PII-free `experiment.shadow_complete` / + `experiment.shadow_failed` structured-log events (counts + the experiment id, + never query or answer text) plus the pull-able tracker windows. + +## Inert by default + +`build_app` leaves `shadow_runner` unset. `build_app_from_config` builds it only +when `cfg.experiments.enabled` **and** `cfg.experiments.shadow_enabled` and a +tracker exists to feed — and even then it is observe-only, so turning it on cannot +change a single served response. (A/B *routing*, which can, is the opt-in 5.7c.) + +## Extension points + +- **Outcome metric** — swap `outcome_metric` for latency, a recall proxy, or + per-shape buckets; the tracker holds whatever scalar you observe. +- **Candidate topology** — inject any `SupportsRoute` as `shadow_runner`'s + candidate (a different reranker/packer chain, a remote gateway, …). +- **Multiple experiments** — run several `ShadowRunner`s (distinct `experiment` + ids) feeding one tracker; the dashboard analyzes each independently. diff --git a/docs/reference/experiments.md b/docs/reference/experiments.md index c3060f3..66e6b20 100644 --- a/docs/reference/experiments.md +++ b/docs/reference/experiments.md @@ -3,8 +3,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 (observe-only); shadow fan-out (5.7b) and -A/B routing (5.7c) feed it later. +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. ## Overview @@ -61,6 +62,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 ``` ### Configuration (`cfg.experiments`) @@ -72,6 +74,40 @@ ragctl experiments --lift 0.25 # feed demo control/candidate samples, print | `min_samples` | `30` | Below this on either side → `insufficient_data`. | | `confidence` | `0.95` | Confidence level for the interval / significance. | | `control_variant` / `candidate_variant` | `control` / `candidate` | The two compared labels. | +| `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 mode (Step 5.7b) + +**Shadow mode** runs a *candidate* retriever alongside the served (*control*) +path on a sample of live queries — observe-only, never affecting the response — +and feeds both variants' outcome into the tracker, so the dashboard can compare +them on real traffic before anyone serves the candidate. + +- **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. It is skipped on a retrieval-cache hit (the control did + not retrieve, so there is nothing comparable to measure). +- **Degrade-open.** A candidate failure logs `experiment.shadow_failed` and is + swallowed — the response was already sent, so shadowing can never break a query. +- **Deterministic sampling.** Whether a query is shadowed is a pure hash of its + `request_id` against `shadow_sample_rate`, so the decision is reproducible (and + testable) with no RNG in the hot path. +- **The outcome metric** is the mean retrieval score of the result set + (`outcome_metric`, higher = better) — the same backend-only signal the drift + monitors use, needing no served answer (the candidate is never shown). Richer + metrics (latency, recall proxies) are a future extension. +- **The candidate** is any `SupportsRoute`. The config-driven build constructs a + second `RetrievalRouter` whose RRF fusion weights come from `shadow_candidate`, + so an operator can shadow-test an alternative balance (e.g. vector-heavy) + against the live config; production injects a candidate over the real backends + via `build_app(shadow_runner=…)`. + +`ShadowRunner` (in `rag_gateway.experiments`) ties these together; `/v1/query` and +`/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). ## Internals @@ -81,13 +117,14 @@ ragctl experiments --lift 0.25 # feed demo control/candidate samples, print 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` unset; - `build_app_from_config` builds it from `cfg.experiments` only when enabled. +- **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`. ## Extension points -- **Feed it** — 5.7b shadow / 5.7c routing call `tracker.observe(experiment, - variant, outcome)` from the query path with the per-query metric. +- **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. - **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 d6edfea..d431047 100644 --- a/packages/config/src/rag_config/schema.py +++ b/packages/config/src/rag_config/schema.py @@ -738,6 +738,21 @@ 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). + + 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). + """ + + vector_weight: Annotated[float, Field(ge=0.0)] = 1.0 + keyword_weight: Annotated[float, Field(ge=0.0)] = 1.0 + graph_weight: Annotated[float, Field(ge=0.0)] = 1.0 + + class ExperimentsConfig(_StrictBase): """A/B testing & shadow-mode knobs (Step 5.7). @@ -754,6 +769,15 @@ class ExperimentsConfig(_StrictBase): * ``confidence`` — confidence level for the interval / significance test. * ``control_variant`` / ``candidate_variant`` — the two variant labels the dashboard compares. + + Shadow mode (Step 5.7b) — *observe-only* candidate fan-out: + + * ``shadow_enabled`` — turn on the shadow fan-out (requires ``enabled``). + * ``shadow_sample_rate`` — fraction ``[0, 1]`` of live queries shadowed + (deterministic per ``request_id``); the candidate runs in the background + 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. """ enabled: bool = False @@ -762,6 +786,10 @@ class ExperimentsConfig(_StrictBase): confidence: Annotated[float, Field(gt=0.0, lt=1.0)] = 0.95 control_variant: str = "control" candidate_variant: str = "candidate" + 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) class RagConfig(_StrictBase): diff --git a/packages/ragctl/src/ragctl/main.py b/packages/ragctl/src/ragctl/main.py index 7d8122e..596485a 100644 --- a/packages/ragctl/src/ragctl/main.py +++ b/packages/ragctl/src/ragctl/main.py @@ -2872,6 +2872,110 @@ def experiments( typer.echo(f" p={r.p_value:.4f} → {verdict} at {r.confidence:.0%}") +@app.command("shadow") +def shadow( + queries: int = typer.Option(80, "--queries", "-n", help="Simulated sampled queries."), + lift: float = typer.Option( + 0.3, "--lift", help="Mean retrieval-score uplift of the candidate over control." + ), +) -> None: + """Drive shadow mode (Step 5.7b) end-to-end against in-process stubs. + + Builds a :class:`~rag_gateway.experiments.ShadowRunner` over a candidate that + scores ``--lift`` above the control, runs ``--queries`` *observe-only* shadow + passes (each feeds both variants into the tracker), and prints the A/B + comparison the dashboard (``GET /v1/status/experiments``) would show. No + external services, no credentials. + + Example:: + + ragctl shadow --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 ShadowRunner + 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)) + runner = ShadowRunner( + candidate_router=_Candidate(), tracker=tracker, sample_rate=1.0, experiment="shadow" + ) + + async def _drive() -> None: + for i in range(queries): + await runner.run( + _ctx(i), + text="demo query", + expansion_terms={}, + hyde_vector=None, + corpus_ids=None, + top_k=10, + control_refs=_refs(0.5, i), + ) + + asyncio.run(_drive()) + r = analyze_ab_experiment( + tracker.samples("shadow", tracker.control), + tracker.samples("shadow", tracker.candidate), + experiment="shadow", + metric="retrieval_score", + confidence=tracker.confidence, + min_samples=tracker.min_samples, + ) + verdict = "SIGNIFICANT" if r.significant else "not significant" + typer.echo(f"\nShadow A/B — {r.experiment} ({r.metric}): {r.status}") + typer.echo("─" * 64) + 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/packages/ragctl/tests/test_shadow.py b/packages/ragctl/tests/test_shadow.py new file mode 100644 index 0000000..aadac31 --- /dev/null +++ b/packages/ragctl/tests/test_shadow.py @@ -0,0 +1,22 @@ +"""Tests for ``ragctl shadow`` — the Step 5.7b shadow-mode smoke.""" + +from __future__ import annotations + +from ragctl.main import app +from typer.testing import CliRunner + +runner = CliRunner() + + +def test_shadow_default_run_prints_comparison() -> None: + result = runner.invoke(app, ["shadow"]) + assert result.exit_code == 0, result.output + assert "Shadow A/B" in result.output + for token in ("control", "candidate", "lift=", "CI=", "p="): + assert token in result.output + + +def test_shadow_positive_lift_is_significant() -> None: + result = runner.invoke(app, ["shadow", "--lift", "0.3", "-n", "80"]) + assert result.exit_code == 0, result.output + assert "SIGNIFICANT" in result.output diff --git a/tests/config/test_experiments_config.py b/tests/config/test_experiments_config.py index cb8fa55..7ed34d7 100644 --- a/tests/config/test_experiments_config.py +++ b/tests/config/test_experiments_config.py @@ -38,3 +38,48 @@ def test_confidence_bounds() -> None: def test_rejects_unknown_field() -> None: with pytest.raises(ValidationError): ExperimentsConfig(unknown=True) # type: ignore[call-arg] + + +def test_shadow_defaults() -> None: + ec = RagConfig(version="1").experiments + assert ec.shadow_enabled is False # requires experiments.enabled too + assert ec.shadow_sample_rate == pytest.approx(0.1) + assert ec.shadow_experiment == "shadow" + assert ec.shadow_candidate.vector_weight == pytest.approx(1.0) + assert ec.shadow_candidate.keyword_weight == pytest.approx(1.0) + assert ec.shadow_candidate.graph_weight == pytest.approx(1.0) + + +def test_shadow_override_from_mapping() -> None: + cfg = RagConfig.model_validate( + { + "version": "1", + "experiments": { + "enabled": True, + "shadow_enabled": True, + "shadow_sample_rate": 0.25, + "shadow_experiment": "router_v2", + "shadow_candidate": {"vector_weight": 1.5, "keyword_weight": 0.5}, + }, + } + ) + ec = cfg.experiments + assert ec.shadow_enabled is True + assert ec.shadow_sample_rate == pytest.approx(0.25) + assert ec.shadow_experiment == "router_v2" + assert ec.shadow_candidate.vector_weight == pytest.approx(1.5) + assert ec.shadow_candidate.keyword_weight == pytest.approx(0.5) + assert ec.shadow_candidate.graph_weight == pytest.approx(1.0) + + +def test_shadow_sample_rate_bounds() -> None: + for bad in (-0.1, 1.1): + with pytest.raises(ValidationError): + ExperimentsConfig(shadow_sample_rate=bad) + + +def test_shadow_candidate_weights_nonnegative() -> None: + from rag_config.schema import ShadowCandidateConfig + + with pytest.raises(ValidationError): + ShadowCandidateConfig(vector_weight=-1.0) From 2a9e8ce0bdf7939a7733c3ffea99d4bb9e4f361b Mon Sep 17 00:00:00 2001 From: Deep Kumar Singh Kushwah Date: Sun, 7 Jun 2026 17:38:26 +0530 Subject: [PATCH 2/2] chore(tracker): sync PR #145 link for Step 5.7b Co-Authored-By: Claude Opus 4.7 --- TRACKER.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/TRACKER.md b/TRACKER.md index 6171a77..5e8325c 100644 --- a/TRACKER.md +++ b/TRACKER.md @@ -20,7 +20,7 @@ **Recently shipped** -- **5.7b** ✅ Shadow mode — observe-only candidate fan-out (`ShadowRunner`, background task) feeding the A/B tracker +- **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) - **5.5** ✅ Drift monitors — `rag-drift`, five PSI / mean-drop monitors — [#136](https://github.com/officialCodeWork/AgentContextOS/pull/136) @@ -524,7 +524,7 @@ | 5.6f | — Cross-links + close-out | ✅ | [#142](https://github.com/officialCodeWork/AgentContextOS/pull/142) | | 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 | ✅ | _pending_ | +| 5.7b | — Shadow mode | ✅ | [#145](https://github.com/officialCodeWork/AgentContextOS/pull/145) | | 5.7c | — A/B routing | ⏳ | — | | 5.7d | — Console + close-out | ⏳ | — | @@ -613,7 +613,7 @@ - `cfg.experiments` (opt-in, since A/B routing can change responses); wired inert by default; **observe-only** — no query-path change yet (shadow / routing are 5.7b/c); `ragctl experiments`; 17 tests - [ADR-0032](docs/adr/ADR-0032-ab-testing-shadow-mode.md), [reference/experiments.md](docs/reference/experiments.md) -#### 5.7b — Shadow mode ✅ _pending_ +#### 5.7b — Shadow mode ✅ [#145](https://github.com/officialCodeWork/AgentContextOS/pull/145) - New `ShadowRunner` (`rag_gateway.experiments`) runs a **candidate** retriever alongside the served **control** path on a sampled fraction of live queries — observe-only — and feeds both variants' `outcome_metric` (mean retrieval score) into the 5.7a `ABExperimentTracker` - **Never delays the response:** scheduled as a FastAPI `BackgroundTask` (runs after the response is sent); **degrade-open** (`experiment.shadow_failed`); skipped on a retrieval-cache hit