File size: 1,567 Bytes
dfb775d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 | """A/B traffic split — canary vs live.
Pure-Python `Splitter`: deterministic per-request seed → canary or live based
on `cfg.canary_pct`. The actual traffic injection happens in the operator
FastAPI app's chat handler (which calls `Splitter.pick(req)` to decide which
slot's backend to route to).
"""
from __future__ import annotations
import hashlib
import random
from pydantic import BaseModel, ConfigDict, Field
class AbConfig(BaseModel):
model_config = ConfigDict(extra="forbid")
canary_pct: float = Field(default=0.05, ge=0.0, le=1.0)
auto_rollback_threshold: float = Field(default=0.02, ge=0.0)
class Splitter:
"""Decides per-request which slot serves traffic."""
def __init__(self, cfg: AbConfig | None = None) -> None:
self.cfg = cfg or AbConfig()
def pick(self, request_id: str | None = None) -> str:
"""Return `'canary'` or `'live'`.
If `request_id` is provided we hash it (stable across retries); else
we use random.random() (stateless).
"""
if self.cfg.canary_pct <= 0.0:
return "live"
if request_id is None:
r = random.random()
else:
digest = hashlib.blake2s(request_id.encode("utf-8"), digest_size=8).digest()
r = int.from_bytes(digest, "big") / float(2**64)
return "canary" if r < self.cfg.canary_pct else "live"
def serve_split(cfg: AbConfig) -> Splitter:
"""Construct the Splitter for an active A/B run."""
return Splitter(cfg)
__all__ = ["AbConfig", "Splitter", "serve_split"]
|