feat(model_group): Thompson-sample intra-group live routing

contextual_orchestrator/model_group.py

@@ -23,11 +23,19 @@
   degenerates gracefully -- members without any observation share one
   identical neutral score, so ordering falls back to the caller's static
   ranking until real evidence exists.
+- **Live selection** (:meth:`ModelGroupRouter.sampled_ranked_member_ids`) draws
+  one Thompson sample (Thompson, 1933) per member from its own Beta(alpha,
+  beta) posterior instead of comparing the posterior mean, so traffic keeps
+  probabilistically exploring every credible member in proportion to
+  remaining uncertainty rather than concentrating on whichever member
+  currently leads the point estimate (Chapelle & Li, 2011; Agrawal & Goyal,
+  2012). Admin/report reads keep the deterministic mean-based order.
 """
 
 from __future__ import annotations
 
 import math
+import random
 import re
 import threading
 import time
@@ -84,6 +92,7 @@ def __init__(
         min_latency_seconds: float = MIN_ROUTING_LATENCY_SECONDS,
         prior_resolver: Callable[[str], tuple[float, float]] | None = None,
         clock: Callable[[], float] = time.monotonic,
+        rng: random.Random | None = None,
     ) -> None:
         if not 0 < ewma_gain <= 1:
             raise ValueError("ewma_gain must be within (0, 1]")
@@ -93,6 +102,11 @@ def __init__(
         self._min_latency_seconds = float(min_latency_seconds)
         self._prior_resolver = prior_resolver
         self._clock = clock
+        # A literal default (`rng: random.Random = random.Random()`) would be
+        # evaluated once at def time -- one shared, unsynchronized instance
+        # reused by every router built without an explicit rng=. The
+        # None-sentinel gives each router its own private generator instead.
+        self._rng = rng if rng is not None else random.Random()
         self._lock = threading.Lock()
         # member_id -> {"alpha", "beta", "ewma", "ewma_tps"}; ewma/ewma_tps are
         # None until the first observation of each kind arrives.
@@ -274,18 +288,64 @@ def member_score(self, member_id: str) -> float:
     def member_observation_count(self, member_id: str) -> int:
         """Total completed attempts recorded for one member (success + failure)."""
         with self._lock:
-            state = self._members.get(member_id)
-            if state is None:
-                return 0
-            alpha = float(state["alpha"]) - float(state.get("prior_alpha", BETA_PRIOR_SUCCESS_COUNT))
-            beta = float(state["beta"]) - float(state.get("prior_beta", BETA_PRIOR_FAILURE_COUNT))
-            return int(max(alpha, 0.0)) + int(max(beta, 0.0))
+            return self._observation_count_locked(member_id)
 
     def ranked_member_ids(self, member_ids: list[str] | tuple[str, ...]) -> list[str]:
         """Order member ids best-first by measured score, preserving input ties."""
         scored = {member_id: self.member_score(member_id) for member_id in member_ids}
         return sorted(member_ids, key=lambda member_id: -scored[member_id])
 
+    def sampled_ranked_member_ids(
+        self, member_ids: list[str] | tuple[str, ...]
+    ) -> list[str]:
+        """Order member ids best-first by one live Thompson-sampled draw per member.
+
+        Same physical quantity as :meth:`ranked_member_ids` -- P(success | data)
+        divided by EWMA latency -- except a member with at least one real
+        observation (:meth:`member_observation_count` > 0) contributes one draw
+        from its own Beta(alpha, beta) posterior (Thompson, 1933) instead of the
+        posterior mean, so a group's traffic keeps probabilistically exploring
+        every credible member in proportion to genuine remaining uncertainty
+        rather than letting whichever member currently leads the point estimate
+        absorb all subsequent traffic -- the correct, intended behavior at the
+        thin per-member observation counts this free-tier gateway actually runs
+        at (Chapelle & Li, 2011; Agrawal & Goyal, 2012).
+
+        A member still at the shared prior (no real observation yet) keeps its
+        exact :data:`UNOBSERVED_MEMBER_SCORE`, unsampled -- the "no evidence ->
+        caller's static order survives untouched" contract holds identically to
+        :meth:`ranked_member_ids`.
+
+        Uses this router's own injected ``rng`` -- a plain, non-cryptographic
+        :class:`random.Random` (a routing weight, not a security boundary).
+        """
+        scored: dict[str, float] = {}
+        with self._lock:
+            for member_id in member_ids:
+                state = self._members.get(member_id)
+                if state is None or self._observation_count_locked(member_id) == 0:
+                    scored[member_id] = UNOBSERVED_MEMBER_SCORE
+                    continue
+                alpha = float(state["alpha"])
+                beta = float(state["beta"])
+                stability_sample = (
+                    self._rng.betavariate(alpha, beta)
+                    if alpha > 0.0 and beta > 0.0
+                    # Degenerate Beta shape (an operator/prior_resolver
+                    # pseudo-count of exactly 0.0 -- update_prior already
+                    # permits this): the distribution collapses to a point
+                    # mass, so fall back to the same closed-form mean
+                    # _score_locked uses, instead of raising out of
+                    # betavariate() (`ValueError: alpha and beta must be > 0`).
+                    else alpha / (alpha + beta)
+                )
+                ewma = state["ewma"]
+                latency = (
+                    1.0 if ewma is None else max(float(ewma), self._min_latency_seconds)
+                )
+                scored[member_id] = stability_sample / latency
+        return sorted(member_ids, key=lambda member_id: -scored[member_id])
+
     def member_report(self, member_id: str) -> dict[str, float | int | None]:
         """One member's measured evidence row for admin/analytics surfaces."""
         with self._lock:
@@ -301,6 +361,25 @@ def snapshot(self) -> dict[str, dict[str, float | int | None]]:
     def _ensure_locked(self, member_id: str) -> dict[str, float | None]:
         return self._members.setdefault(member_id, self._blank_state(member_id))
 
+    @staticmethod
+    def _outcome_count(total: float, prior: float) -> int:
+        """Recover an integer observed-outcome count from floating prior mass."""
+        return int(round(max(total - prior, 0.0)))
+
+    def _observation_count_locked(self, member_id: str) -> int:
+        state = self._members.get(member_id)
+        if state is None:
+            return 0
+        alpha = self._outcome_count(
+            float(state["alpha"]),
+            float(state.get("prior_alpha", BETA_PRIOR_SUCCESS_COUNT)),
+        )
+        beta = self._outcome_count(
+            float(state["beta"]),
+            float(state.get("prior_beta", BETA_PRIOR_FAILURE_COUNT)),
+        )
+        return alpha + beta
+
     def _score_locked(self, member_id: str) -> float:
         state = self._members.get(member_id)
         if state is None:
@@ -342,7 +421,13 @@ def _report_locked(self, member_id: str) -> dict[str, float | int | None]:
             "max_observed_rpm": self._max_observed_rpm.get(member_id, 0),
             "max_observed_tpm": self._max_observed_tpm.get(member_id, 0),
             "rate_observation_window_seconds": int(RATE_OBSERVATION_WINDOW_SECONDS),
-            "success_count": int(alpha - float(state.get("prior_alpha", BETA_PRIOR_SUCCESS_COUNT))),
-            "failure_count": int(beta - float(state.get("prior_beta", BETA_PRIOR_FAILURE_COUNT))),
+            "success_count": self._outcome_count(
+                alpha,
+                float(state.get("prior_alpha", BETA_PRIOR_SUCCESS_COUNT)),
+            ),
+            "failure_count": self._outcome_count(
+                beta,
+                float(state.get("prior_beta", BETA_PRIOR_FAILURE_COUNT)),
+            ),
             "score": round(self._score_locked(member_id), 9),
         }

contextual_orchestrator/orchestrator.py

@@ -6958,17 +6958,27 @@ def _refine_partition(self, partition: list[ModelAgent], role: str) -> list[Mode
             for sub_partition in (eligible, excluded):
                 by_id = {member.id: member for member in sub_partition}
                 ordered.extend(
-                    by_id[member_id] for member_id in self._measured_member_order(list(by_id))
+                    by_id[member_id]
+                    for member_id in self._measured_member_order(list(by_id), sample=True)
                 )
         return ordered
 
-    def _measured_member_order(self, member_ids: list[str]) -> list[str]:
+    def _measured_member_order(
+        self, member_ids: list[str], *, sample: bool = False
+    ) -> list[str]:
         """Order same-declaration members by measured evidence, quality first.
 
         Evidence ladder: judged-answer observations (real-time fast-mlsirm
         verdicts) govern when any member has them; otherwise the transport
         throughput/stability ledger decides; with no evidence at all the
         caller's input order survives untouched. No synthetic scores.
+
+        ``sample=True`` is for live serving selection only: it asks the chosen
+        router for one Thompson-sampled draw per member with real evidence
+        (:meth:`ModelGroupRouter.sampled_ranked_member_ids`) instead of
+        comparing posterior means. The default ``sample=False`` is the exact
+        prior deterministic behavior, kept for every admin/report reader
+        (:meth:`get_model_group`) unchanged.
         """
         judged_quality = any(
             self._quality_router.member_observation_count(member_id) > 0
@@ -6983,6 +6993,8 @@ def _measured_member_order(self, member_ids: list[str]) -> list[str]:
                     judged_quality,
                     router.member_score(member_id),
                 )
+        if sample:
+            return router.sampled_ranked_member_ids(member_ids)
         return router.ranked_member_ids(member_ids)
 
     def _psychometric_order(

docs/papers/thompson-sampling-model-group-routing.md

@@ -0,0 +1,36 @@
+# Thompson sampling for model-group routing
+
+Status: research traceability for `ModelGroupRouter.sampled_ranked_member_ids` on PR #1034.
+
+## Decision boundary
+
+Live serving uses posterior sampling to avoid deterministic winner-take-all selection after measured outcomes exist. Administrative and reporting reads remain deterministic. The implementation draws from each observed member's Beta posterior and combines that draw with the repository's measured latency term; it does not change the stored posterior or fabricate observations.
+
+The cited literature supports the probability-matching / exploration-exploitation mechanism, but its guarantees must not be overstated. Thompson (1933) supplies the original posterior probability-matching construction. Chapelle and Li (2011) provide empirical evidence that Thompson sampling is a competitive baseline on simulated and real bandit data. Agrawal and Goyal (2012) prove logarithmic expected regret for stochastic Bernoulli multi-armed bandits under their analyzed Thompson-sampling setting.
+
+`contextual-orchestrator` does **not** implement exactly the reward model in Agrawal and Goyal (2012): routing ranks a Beta stability sample divided by an EWMA latency estimate. Therefore the COLT regret bound is supporting evidence for posterior sampling of Bernoulli stability, not a proof of regret or optimality for the repository's composite successful-responses-per-second score. Any such claim requires a separate model and acceptance experiment.
+
+## Code traceability
+
+| Research concept | Repository implementation | Acceptance evidence |
+|---|---|---|
+| Posterior probability matching | `contextual_orchestrator/model_group.py::ModelGroupRouter.sampled_ranked_member_ids` | Seeded sampling tests in `tests/test_model_group.py` |
+| Bernoulli success/failure posterior | `observe_success`, `observe_failure`, per-member `alpha` / `beta` | Existing model-group posterior/report tests |
+| Deterministic reporting vs stochastic live selection | `ranked_member_ids` vs `sampled_ranked_member_ids`; `_measured_member_order(sample=...)` | Admin/report callers keep `sample=False`; serving refinement requests `sample=True` |
+| Integer completed-outcome invariant under fractional prior refresh | `_outcome_count` shared by live observation-count and report paths | RED `c8223a3e4f10c2c48396359abdbad7520b84e920`; GREEN `40a5cc41e4e10525e6fdb73f4a7b19682a083e24` |
+
+## Risks and follow-up acceptance
+
+- The theory cited here does not validate the latency-normalized composite score. Measure routing regret / successful responses per second on right-cleared traffic or an approved replay before making an optimality claim.
+- Members with no real outcomes currently retain the repository's neutral static score rather than receiving a sampled prior draw. This preserves the existing no-evidence ordering contract but is a repository-specific cold-start policy, not a consequence of the cited Thompson-sampling theory. A future change must test starvation and first-observation acquisition explicitly.
+- Prior pseudo-counts may be fractional while completed outcomes are integral. Binary floating-point subtraction can drift below an integer; PR #1034 therefore recovers the domain-invariant completed-outcome count before deciding whether a member is observed.
+
+PDFs are not vendored solely for this change because redistribution permission is not assumed. The primary publication pages below are the traceable sources.
+
+## References
+
+Agrawal, S., & Goyal, N. (2012). Analysis of Thompson sampling for the multi-armed bandit problem. In S. Mannor, N. Srebro, & R. C. Williamson (Eds.), *Proceedings of the 25th Annual Conference on Learning Theory* (Vol. 23, pp. 39.1–39.26). Proceedings of Machine Learning Research. https://proceedings.mlr.press/v23/agrawal12.html
+
+Chapelle, O., & Li, L. (2011). An empirical evaluation of Thompson sampling. In J. Shawe-Taylor, R. Zemel, P. Bartlett, F. Pereira, & K. Q. Weinberger (Eds.), *Advances in Neural Information Processing Systems 24*. https://papers.nips.cc/paper/2011/hash/e53a0a2978c28872a4505bdb51db06dc-Abstract.html
+
+Thompson, W. R. (1933). On the likelihood that one unknown probability exceeds another in view of the evidence of two samples. *Biometrika, 25*(3–4), 285–294. https://doi.org/10.1093/biomet/25.3-4.285

tests/test_model_group.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import random
 
 import pytest
 
@@ -139,6 +140,10 @@ def test_orchestrator_resolves_group_alias_and_reorders_only_its_members() -> No
 
     assert canonical_group_name("shared-reasoning-model") == "shared_reasoning_model"
     assert orchestrator._requested_agent("shared-reasoning-model") == high
+    # Live serving now samples from each measured member's posterior
+    # (sampled_ranked_member_ids); a fixed seed keeps this assertion
+    # deterministic while still exercising the sampled path.
+    orchestrator._group_router._rng = random.Random(0)
     orchestrator._group_router.observe_failure(high.id)
     orchestrator._group_router.observe_success(low.id, 0.1)
     assert orchestrator._requested_agent("shared_reasoning_model") == low
@@ -271,6 +276,7 @@ def test_explicit_group_alias_routes_plain_completion_to_measured_member() -> No
     first = ModelAgent("provider_one_model", "vendor-one/model-a", group_name="shared_reasoning_model")
     second = ModelAgent("provider_two_model", "vendor-two/model-b", group_name="shared_reasoning_model")
     orchestrator = TaskOrchestrator([first, second])
+    orchestrator._group_router._rng = random.Random(0)
     orchestrator._group_router.observe_failure(first.id)
     orchestrator._group_router.observe_success(second.id, 0.1)
 
@@ -340,7 +346,64 @@ def test_group_selects_measured_member_for_every_model_capability(capability: st
     first = ModelAgent("first_member", "provider/first", tags=(tag,), group_name="shared_model")
     second = ModelAgent("second_member", "provider/second", tags=(tag,), group_name="shared_model")
     orchestrator = TaskOrchestrator([first, second])
+    orchestrator._group_router._rng = random.Random(0)
     orchestrator._group_router.observe_failure(first.id)
     orchestrator._group_router.observe_success(second.id, 0.1)
 
     assert orchestrator.select_capability_agent(capability, "shared-model") == second
+
+
+def test_sampled_ranking_leaves_unobserved_members_at_the_shared_prior_score() -> None:
+    """Neither member has a real observation, so both stay at the identical
+    ``UNOBSERVED_MEMBER_SCORE`` -- unsampled -- and the caller's input order
+    survives untouched, exactly like ``ranked_member_ids``.
+    """
+    router = ModelGroupRouter(rng=random.Random(0))
+    assert router.sampled_ranked_member_ids(["member_b", "member_a"]) == [
+        "member_b",
+        "member_a",
+    ]
+
+
+def test_sampled_ranking_falls_back_to_the_posterior_mean_for_a_degenerate_zero_prior() -> None:
+    """``update_prior`` permits ``prior_alpha``/``prior_beta`` == 0.0. A member
+    whose resulting ``alpha`` (or ``beta``) lands at exactly 0.0 with a real
+    observation recorded must not raise out of ``random.betavariate(0.0, x)``
+    (``ValueError: alpha and beta must be > 0``); it falls back to the same
+    closed-form posterior mean ``_score_locked`` uses.
+    """
+    router = ModelGroupRouter(rng=random.Random(0))
+    router.register_member("degenerate_member")
+    router.update_prior("degenerate_member", 0.0, 0.0)
+    router.observe_failure("degenerate_member")  # alpha stays 0.0, beta becomes 1.0
+
+    report = router.member_report("degenerate_member")
+    assert report["success_count"] == 0 and report["failure_count"] == 1
+    assert report["success_posterior_mean"] == 0.0
+
+    # Posterior mean 0.0 ranks strictly below an unobserved sibling's neutral
+    # UNOBSERVED_MEMBER_SCORE (0.5) -- proof the fallback ran instead of
+    # raising or silently dropping the member.
+    ranked = router.sampled_ranked_member_ids(["degenerate_member", "unobserved_sibling"])
+    assert ranked == ["unobserved_sibling", "degenerate_member"]
+
+
+def test_sampled_ranking_prefers_but_does_not_monopolize_the_better_member() -> None:
+    """Authoritative statistical contract for :meth:`ModelGroupRouter.sampled_ranked_member_ids`.
+
+    With one observation each (Beta(1, 2) for the loser, Beta(2, 1) for the
+    winner, equal latency), the true win probability for the better member is
+    exactly 5/6 ~= 0.833 (Thompson, 1933; the early-stage exploration behavior
+    Agrawal & Goyal 2012 prove drives Thompson Sampling's near-optimal regret).
+    0.65 leaves a wide margin against a false failure while still catching a
+    broken (e.g. uniform-random or reverted-to-argmax) implementation.
+    """
+    router = ModelGroupRouter(rng=random.Random(100))
+    router.observe_failure("loser")  # alpha=1, beta=2
+    router.observe_success("winner", 1.0)  # alpha=2, beta=1; same latency floor as loser's
+    wins = sum(
+        router.sampled_ranked_member_ids(["loser", "winner"])[0] == "winner"
+        for _ in range(200)
+    )
+    assert wins / 200 >= 0.65
+    assert wins < 200  # exploration, not permanent monopolization