Shipping LLM Features Safely: Shadow Mode, Feature Flags and Eval Gates in CI

The old path stays in charge

Most LLM features I build replace something that already works: a keyword router, a regex extractor, a person with a spreadsheet. The legacy path is dull, but nobody gets paged for it. So it answers every request until evidence says otherwise, and the evidence comes from three places, each with a gate:

• Offline: a golden dataset scored in CI on every pull request. A prompt or model change cannot merge if it drops below absolute floors or regresses against the committed baseline.
• Shadow: real traffic goes through the new path in the background. Its answer is logged next to the legacy answer and never reaches the user.
• Live: a sticky percentage of tenants gets the new answer, with the legacy answer as the fallback and a kill switch that needs no deploy.

The running example is a ticket triage endpoint where a keyword classifier labels tickets as billing, account, bug or other, and an LLM should take over. The classifier itself is in the structured-output triage article; this one covers the machinery around it. Everything below comes from a reference implementation I wrote and ran for this article, with fake models in place of real providers.

Routing: stable buckets, sticky cohorts, one kill switch

Every request resolves to legacy, shadow or live, decided per tenant rather than per request. If one customer's tickets flip between classifiers minute to minute, you get complaints you cannot reproduce.

BUCKETS = 10_000

def bucket(flag: str, unit_id: str) -> int:
    # sha256, not hash(): str hashing is salted per process, so two workers
    # would put the same tenant in different buckets.
    digest = hashlib.sha256(f"{flag}:{unit_id}".encode()).digest()
    return int.from_bytes(digest[:8], "big") % BUCKETS

class Route(StrEnum):
    LEGACY = "legacy"
    SHADOW = "shadow"
    LIVE = "live"

@dataclass(frozen=True)
class FlagState:
    live_percent: float = 0.0
    shadow_percent: float = 0.0
    kill: bool = False

def route_for(flag: str, unit_id: str, state: FlagState) -> Route:
    if state.kill:
        return Route.LEGACY  # kill also stops shadow traffic and its cost
    b = bucket(flag, unit_id)
    live_cut = state.live_percent * BUCKETS / 100
    if b < live_cut:
        return Route.LIVE
    # Shadow the band just above the live cut: the next cohort to go live.
    if b < live_cut + state.shadow_percent * BUCKETS / 100:
        return Route.SHADOW
    return Route.LEGACY

Four decisions, each preventing a specific failure:

• A cryptographic hash, not `hash()`. Python salts string hashing per process, so hash("tenant-42") differs between two Uvicorn workers. A test starts three subprocesses with different PYTHONHASHSEED values and asserts that they agree on the bucket.
• The flag name is part of the key. Without it, the same unlucky 10% of tenants would receive every experiment you ever run. A test checks that two flags' 10% cohorts overlap by about what chance predicts, not completely.
• Buckets are compared against a cut. Raising live_percent from 5 to 20 only adds tenants. The test asserts that the 5% set is a subset of the 20% set, and that on 20,000 synthetic tenant ids the 20% cut lands between 19% and 21%.
• Shadow is the band above live. The tenants in shadow are exactly the ones who go live at the next step. The shadow data therefore predicts the next step rather than an average over everyone.

The flag state lives in a Redis hash, flag:triage_llm, read through a small FlagStore with a two-second per-process cache. Most of the thinking is about what happens when that read goes wrong:

The cache TTL is the worst-case kill switch latency per worker. Two seconds is a trade I accept: reading on every request adds a Redis round trip and still is not instant under a partition. A fake-clock test checks the kill is invisible at t=1.0 and visible at t=2.1.

Shadow mode that cannot touch the response

The shadow contract: the new path cannot change the response, delay it, or take the process down. Here is the handler:

    @app.middleware("http")
    async def request_context(request: Request, call_next):
        request_id = request.headers.get("x-request-id") or uuid.uuid4().hex
        structlog.contextvars.clear_contextvars()
        structlog.contextvars.bind_contextvars(request_id=request_id)
        response = await call_next(request)
        response.headers["x-request-id"] = request_id
        return response

@app.post("/tickets/triage")
    async def triage(ticket: Ticket) -> TriageResponse:
        legacy = classify_legacy(ticket.text)  # always computed: it is the fallback
        route = route_for(settings.flag_name, ticket.tenant_id, await flags.get())
        structlog.contextvars.bind_contextvars(route=route.value, tenant_id=ticket.tenant_id)

if route is Route.LIVE:
            try:
                async with asyncio.timeout(settings.live_timeout_s):
                    label = await classify_with_llm(
                        model, ticket.text, model_id=settings.model_id, path="live")
                return TriageResponse(category=label.category, source="llm")
            except Exception as exc:
                log.warning("live_fallback", error_type=type(exc).__name__)
                return TriageResponse(category=legacy, source="legacy_fallback")

if route is Route.SHADOW:
            shadow.submit(
                lambda: classify_with_llm(
                    model, ticket.text, model_id=settings.model_id, path="shadow"),
                legacy=legacy,
            )
        return TriageResponse(category=legacy, source="legacy")

The legacy answer is computed first on every route. In shadow it is the answer; in live it is the fallback, so a provider timeout returns a correct, if less clever, label instead of a 500. source records which path answered. Ticket is a frozen Pydantic model, so the shadow closure cannot mutate the input the legacy path used.

The shadow work goes to a ShadowRunner rather than FastAPI's BackgroundTasks. I checked the difference with a test. With BackgroundTasks, the ASGI call for the request does not complete until the task does. A hung background task therefore holds the request cycle open with nothing to time it out; under httpx.ASGITransport the client's post() does not return until it finishes. A separate asyncio task with its own timeout has neither problem:

    def submit(self, candidate: Callable[[], Awaitable[TriageLabel]], *, legacy: str) -> bool:
        if len(self._tasks) >= self._max_inflight:
            log.info("shadow_skipped", reason="max_inflight")
            return False
        # create_task copies the current contextvars, so request_id comes along.
        task = asyncio.create_task(self._run(candidate, legacy))
        self._tasks.add(task)
        task.add_done_callback(self._tasks.discard)
        return True

async def _run(self, candidate: Callable[[], Awaitable[TriageLabel]], legacy: str) -> None:
        started = time.perf_counter()
        try:
            async with asyncio.timeout(self._timeout_s):
                label = await candidate()
        except TimeoutError:
            log.warning("shadow_timeout", timeout_s=self._timeout_s)
            return
        except Exception as exc:  # the shadow path is allowed to be broken
            log.warning("shadow_error", error_type=type(exc).__name__)
            return
        log.info(
            "shadow_compare",
            legacy=legacy,
            candidate=label.category,
            agree=label.category == legacy,
            latency_ms=round((time.perf_counter() - started) * 1000, 1),
        )

The details matter here:

• `self._tasks` keeps strong references. The event loop holds only weak references to tasks, so a fire-and-forget task can be garbage-collected mid-flight.
• `max_inflight` caps the damage. When the provider slows down, extra requests skip the shadow and log shadow_skipped rather than piling load on a struggling dependency.
• `except Exception` is deliberate here and only here. CancelledError is a BaseException, so shutdown can still cancel the task. On shutdown, the lifespan hook calls drain(), which waits a few seconds for in-flight comparisons and then cancels the rest.
• The candidate is a zero-argument factory. When the shadow is skipped, no coroutine is created, so none is left un-awaited.
• Shadow must be side-effect free. If the new path writes data or calls tools, it needs a dry-run mode, or you do everything twice.

One limit no timeout can fix: if something in the shadow path makes a blocking synchronous call, it stalls the event loop for every request. The timeout only fires when the loop gets control back. Use async clients in the shadow path, or wrap the call in asyncio.to_thread.

Proving it with a shadow path that hangs

I do not trust "the response is unaffected" until a test makes the shadow path as bad as it can be. The model here is a RunnableLambda that sleeps for an hour, behind the same ainvoke interface ChatOpenAI exposes:

async def hang(_messages):
    await asyncio.sleep(3600)

# ...

async def test_hanging_shadow_never_delays_or_changes_the_response(logs):
    client, shadow = build(RunnableLambda(hang), FlagState(shadow_percent=100))
    async with client:
        started = time.perf_counter()
        response = await client.post("/tickets/triage", json=BILLING,
                                     headers={"x-request-id": "req-hang"})
        elapsed = time.perf_counter() - started

assert response.status_code == 200
    assert response.json() == {"category": "billing", "source": "legacy"}
    assert elapsed < 0.1  # well under the 0.3 s shadow timeout
    assert shadow.inflight == 1  # still running after the user got their answer

await asyncio.sleep(0.4)
    [timeout] = events(logs, "shadow_timeout")
    assert timeout["request_id"] == "req-hang" and shadow.inflight == 0
    [call] = events(logs, "llm_call")
    assert call["outcome"] == "cancelled" and call["path"] == "shadow"

The key assertion is shadow.inflight == 1 after the response arrived: the user got an answer while the new path was still stuck. Sibling tests cover the other failures:

• a model that raises RuntimeError produces a shadow_error log and an unchanged 200;
• a model that answers in prose instead of JSON logs outcome=invalid_output and still does not affect the user;
• five requests against a hung model with max_inflight=2 leave exactly two tasks running and log three skips;
• a hung model on the live route falls back to legacy inside the 0.2-second live timeout;
• the kill switch produces no llm_call events at all.

A last test uses real Redis. It sets live_percent to 100 and sees an llm answer, then sets kill to 1 and sees legacy on the next request.

Every LLM call is attributable

When shadow agreement drops, the first question is "what changed?", and the logs must answer it. Every model call goes through one function that logs one event in a finally block, so failures are logged as reliably as successes:

@dataclass(frozen=True)
class PromptSpec:
    name: str
    version: str
    system: str

@property
    def sha(self) -> str:
        # Catches "edited the text, forgot to bump the version".
        return hashlib.sha256(self.system.encode()).hexdigest()[:12]

# ...

async def classify_with_llm(model: Runnable, text: str, *, model_id: str, path: str,
                            prompt: PromptSpec = TRIAGE_PROMPT) -> TriageLabel:
    started = time.perf_counter()
    outcome, usage = "error", None
    try:
        message = await model.ainvoke(build_messages(prompt, text))
        usage = getattr(message, "usage_metadata", None)
        label = parse_label(message.content)
        outcome = "ok"
        return label
    except ValidationError:
        outcome = "invalid_output"
        raise
    except (asyncio.CancelledError, TimeoutError):
        outcome = "cancelled"
        raise
    finally:
        # request_id, route and tenant_id arrive via structlog contextvars.
        log.info(
            "llm_call",
            path=path,
            prompt_name=prompt.name,
            prompt_version=prompt.version,
            prompt_sha=prompt.sha,
            model_id=model_id,
            outcome=outcome,
            latency_ms=round((time.perf_counter() - started) * 1000, 1),
            usage=usage,
        )

Here is one real line from the reference app, trimmed: {"event": "llm_call", "path": "shadow", "prompt_version": "triage-v3", "prompt_sha": "72b65b76a9d8", "model_id": "gpt-4.1-mini-2025-04-14", "outcome": "ok", "request_id": "req-7f3a", "tenant_id": "t-1"}. The shadow_compare line for the same request carries the same request_id.

The request id reaches the shadow task without any extra work. asyncio.create_task copies the current contextvars context, and structlog's merge_contextvars reads it. The test asserts it rather than assuming it.

A version string is a promise; a hash is a fact. When someone edits the prompt and forgets to bump the version, the hash still changes in the logs and invalidates the eval recordings in the next section. The model id is a dated snapshot from config, never an alias that can point at a different model next month.

In production, the model is a single line of configuration in main.py, which I did not run for this article: ChatOpenAI(model=settings.model_id, temperature=0, max_retries=1). Retries belong on the live path's budget, not stacked beneath it. The idempotency and retries article covers that in detail.

An offline eval gate: golden set, recordings, baseline

Shadow data describes today's traffic; it does not stop tomorrow's prompt edit from making things worse. That is the CI gate's job, and I design it around three rules.

It runs offline. CI does not call the provider. Instead, evals/record.py runs the production prompt against each golden example and commits the raw outputs to evals/recordings/triage.jsonl, along with the prompt version, prompt hash and model id. The gate scores those committed outputs. This makes it deterministic, free and runnable on forks with no secrets. The golden set has 24 labelled tickets, including an ambiguous "error when I try to pay" case and a prompt-injection attempt labelled other. For this article I produced the recordings with a scripted fake model, so the pipeline runs end to end without a key. In a real project the same script runs against the pinned model.

It refuses stale recordings. If the prompt hash or model id in the recordings does not match the code, scoring fails before computing anything. Otherwise a pull request could change the prompt and pass by re-scoring last week's outputs.

It uses the production parser. An output that fails parse_label in production counts as invalid here too.

def score(golden: list[dict], recordings: list[dict], *, model_id: str,
          prompt: PromptSpec = TRIAGE_PROMPT) -> dict:
    by_id = {r["example_id"]: r for r in recordings}
    missing = sorted({g["id"] for g in golden} - by_id.keys())
    stale = sorted(r["example_id"] for r in recordings
                   if (r["prompt_sha"], r["model_id"]) != (prompt.sha, model_id))
    if missing or stale:
        # Scoring old outputs against a new prompt would pass a change nobody tested.
        raise StaleRecordings(f"missing={missing} stale={stale}: re-run evals/record.py")

per_example, invalid = {}, 0
    for example in golden:
        try:
            predicted = parse_label(by_id[example["id"]]["output"]).category
        except ValidationError:  # same parser as production, same failure
            predicted, invalid = None, invalid + 1
        per_example[example["id"]] = predicted == example["expected"]
    # ... accuracy, invalid_rate, recall per category; returns metrics + per_example

def regressions(baseline: dict, report: dict, tolerance: float = 0.0):
    worse = {}
    for name, value in report["metrics"].items():
        old = baseline["metrics"].get(name)
        if old is None:
            worse[name] = ("missing from baseline", value)
            continue
        drop = value - old if name in LOWER_IS_BETTER else old - value
        if drop > tolerance:
            worse[name] = (old, value)
    flipped = sorted(i for i, ok in baseline["per_example"].items()
                     if ok and not report["per_example"].get(i, False))
    return worse, flipped

The baseline is evals/baseline.json, written by python -m evals.gate --write-baseline and committed. It stores the metrics and the pass or fail result of every example. Changing it is a reviewed diff in the same pull request as the prompt change, so "we accepted a lower billing recall" is a decision with a name attached.

The pytest gate applies two kinds of check:

pytestmark = pytest.mark.eval

# Absolute floors: the feature is not shippable below these, whatever the baseline says.
FLOORS = {
    "accuracy": 0.85,
    "recall_billing": 0.80,
    "recall_account": 0.80,
    "recall_bug": 0.80,
    "recall_other": 0.60,
}
CEILINGS = {"invalid_rate": 0.05}

@pytest.fixture(scope="module")
def report() -> dict:
    return current_report()  # raises StaleRecordings if prompt or model changed

@pytest.mark.parametrize(("metric", "floor"), FLOORS.items())
def test_metric_above_floor(report, metric, floor):
    assert report["metrics"][metric] >= floor

# ... same shape for CEILINGS

def test_no_regression_against_baseline(report):
    baseline = json.loads(BASELINE.read_text())
    worse, flipped = regressions(baseline, report, tolerance=0.0)
    assert not worse, f"metrics regressed vs baseline: {worse}"
    assert not flipped, f"examples that passed on baseline now fail: {flipped}"

Floors alone let quality drift down to the floor one harmless-looking change at a time. Baseline comparison alone lets a bad baseline become the standard. You need both.

I also check individual examples, not just the mean. One test swaps one right answer for a wrong one and one wrong answer for a right one. Accuracy is identical, but the gate still names bill-001 as newly failing and flags recall_billing.

On 24 examples, one flip moves accuracy by about four points. That is why the tolerance is zero and the report lists example ids. The reviewer looks at the ticket, not at a decimal.

Other tests check the gate itself. Editing one sentence of the prompt raises StaleRecordings, and so does changing the model id. A gate that has never been seen failing is decoration. If you would rather not maintain this yourself, my evaluation-driven development test suite packages the same idea. It takes golden JSONL datasets, deterministic lexical metrics and an edd-eval compare command with a --tolerance option. The command exits 1 on a regression, 0 otherwise. For retrieval-specific metrics, see the retrieval evals article.

CI: what runs on every PR, and what needs secrets

The workflow has three jobs. Only the third one touches a secret, and it never runs on pull requests.

jobs:
  unit:
    runs-on: ubuntu-latest
    services:
      redis:
        image: redis:7
        ports: ["6379:6379"]
        options: --health-cmd "redis-cli ping" --health-interval 2s --health-retries 10
    env:
      REDIS_URL: redis://localhost:6379/0
    steps:
      # ... checkout, setup-python 3.12, pip install -r requirements.txt
      - run: pytest -m "not eval" -q

eval-gate:
    # Offline by design: scores committed recordings. No secrets in this job.
    runs-on: ubuntu-latest
    steps:
      # ... checkout, setup-python 3.12, pip install -r requirements.txt
      - run: pytest -m eval -q
      - if: always()
        run: python -m evals.gate > eval-metrics.json
      - if: always()
        uses: actions/upload-artifact@v4
        with: { name: eval-metrics, path: eval-metrics.json }

rerecord:
    # NEEDS SECRETS. Manual only; a human reviews the diff and commits it.
    if: github.event_name == 'workflow_dispatch' && inputs.rerecord
    runs-on: ubuntu-latest
    environment: llm-recording  # holds OPENAI_API_KEY, requires approval
    steps:
      # ... checkout, setup-python 3.12, pip install -r requirements.txt
      - run: python -m evals.record
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
      - run: pytest -m eval -q || true  # report only; the PR run is the real gate
      - uses: actions/upload-artifact@v4
        with: { name: new-recordings, path: evals/recordings/ }

Triggers are pull_request, pushes to main, and workflow_dispatch with a boolean rerecord input. I did not run it on GitHub; the full file passes actionlint 1.7.12, and both pytest commands pass locally against Redis with OPENAI_API_KEY unset (19 unit tests, 7 eval tests). The secret sits in an environment with required reviewers, so fork pull requests cannot read it.

Re-recording is manual on purpose: even at temperature 0 providers are not fully deterministic, and re-recording on every run would make the gate a coin flip. The intended flow is:

1. Change the prompt.
2. Watch the eval job fail with StaleRecordings.
3. Dispatch rerecord.
4. Commit the new recordings, and the baseline if you accept the change.
5. Let the pull request run pass the gate on those exact outputs.

Rollback without a deploy

The runbook is short because the design does most of the work:

1. Stop the new path: redis-cli -u "$REDIS_URL" HSET flag:triage_llm kill 1. Each worker picks this up within the flag cache TTL, two seconds here. Live traffic returns to legacy, and shadow calls stop too, so provider spend drops at the same time.
2. Confirm in the logs: no llm_call events with path=live, and source=legacy in responses. If flag_read_failed appears, workers cannot see Redis and are running on the last known good state; deploy with the config default instead, which has nothing live.
3. Find what changed. Group recent llm_call events by prompt_sha and model_id, and compare shadow_compare agreement before and after.
4. If a prompt or model change caused it, git revert the merge. Prompt, recordings and baseline changed together, so they revert together.
5. Add every ticket that caused the incident to golden.jsonl with its correct label. Re-record, then return to an earlier stage: redis-cli -u "$REDIS_URL" HSET flag:triage_llm kill 0 live_percent 0 shadow_percent 10.
6. Widen again only after shadow agreement and the gate both hold.

Shadow disagreements are also the cheapest source of new golden examples: I sample them, label them by hand and add them, so the gate learns from real traffic. The same loop applies to adversarial inputs, covered in the red-teaming article.

Checklist

• Rollout decisions use a sha256 bucket of flag:unit_id, never hash(). A test proves agreement across processes.
• Percentage increases only add tenants. Shadow targets the next cohort to go live.
• The kill switch stops both live and shadow traffic. Its worst-case latency equals the flag cache TTL, and that number is written down.
• Malformed flags fail towards legacy. A Redis outage keeps the last known good state rather than flipping it.
• The legacy answer is computed first and is the fallback for every live error or timeout.
• Shadow runs in its own asyncio task with a timeout, a strong reference, an in-flight cap and a drain on shutdown. It does not use BackgroundTasks.
• A test makes the shadow path hang and raise, and asserts the response is unchanged and returned while the shadow task is still running.
• Every LLM call logs request_id, prompt_version, prompt_sha, model_id, path, outcome and latency from a finally block.
• Model ids are dated snapshots in config, not aliases.
• The eval gate scores committed recordings offline, uses the production parser, and refuses recordings whose prompt hash or model id do not match.
• The gate has absolute floors and a zero-tolerance per-example comparison with a committed baseline, and a test proves it fails when it should.
• Secrets exist only in a manually dispatched, approval-gated re-record job.

Tested with Python 3.12.9, FastAPI 0.141.1, Starlette 1.7.0, Pydantic 2.13.5, structlog 26.1.0, langchain-core 1.6.5, redis-py 8.1.0 against Redis 7.2, httpx 0.28.1, pytest 9.1.1, pytest-asyncio 1.4.0 and actionlint 1.7.12 on 2026-09-27.