Production blueprint: DynamoDB + OpenSearch

Chapter 27 graded a prototype. This chapter does two jobs: it re-audits that work with a production eye (what must be swapped, and for what), and it lays out a concrete AWS design for the feed itself, with the split you will actually run: user state and dedup state in DynamoDB, article content and vector search in OpenSearch. It ends with a phased implementation plan.

The production audit

Every artifact in Chapter 27 was chosen to be readable and runnable on a laptop. Here is what each one becomes when real traffic hits it:

Two pieces of general advice before the design. First, decide the failure policy for every soft dependency now: if the judge, the verdict cache, or the image hasher is down, the feed still serves, and ambiguous pairs default to keep both (showing a duplicate is a smaller failure than showing an empty feed). Second, make the embedder version part of every stored vector's identity. Profile vectors and article vectors must live in the same space; an upgrade means a new OpenSearch index, a backfill, an alias flip, and a plan for profiles (either re-embed recent history or let the EMA rebuild over the next few clicks).

The architecture

feed in ──► ingest Lambda ──► embed + phash + simhash ──► story clustering
                                     │                    (LSH; gray pairs ──► judge batch, nightly)
                                     ▼
                              DynamoDB `articles` (system of record)
                                     │  zero-ETL / streams
                                     ▼
                              OpenSearch `news-v1` (content + kNN index)

clicks ──► stream ──► EMA Lambda ──► DynamoDB `profiles` (one item per user)
                                     DynamoDB `seen`     (TTL'd impressions)

request ──► read profile (DynamoDB, ~5 ms)
        ──► kNN query + filters + collapse by story_id (OpenSearch, ~30 ms)
        ──► residue screen + cached verdicts (DynamoDB `verdicts`, in-process)
        ──► rank ──► slate ──► sampled rubric scoring (async, off the hot path)

Don't be confused: DynamoDB stores vectors; it cannot search them. There is no nearest-neighbor query in a key-value store, and scanning a table to compute cosines is the $O(n)$ trap. The split is: DynamoDB is the system of record and the per-user state store (point reads by key, single-digit milliseconds), OpenSearch is the derived index that answers "nearest 200 unseen articles to this vector". AWS ships a zero-ETL integration (DynamoDB streams into an OpenSearch ingestion pipeline) precisely for this write-once, sync-automatically pattern.

Vectors in DynamoDB

The reason DynamoDB fits the EMA profile is mathematical, not operational. Chapter 5's taste vector is a decayed average over the full history, which looks like it needs a history scan per update. It does not. Keep three fields per user, fold each click in constant time,

$$ S \leftarrow S \cdot e^{-\lambda(t - t_\text{last})} + e, \qquad W \leftarrow W \cdot e^{-\lambda(t - t_\text{last})} + 1, \qquad t_\text{last} \leftarrow t, $$

and read the profile as $S/W$. Better still, $S/W$ is decay-invariant: decaying both to any later read time multiplies numerator and denominator by the same factor, which cancels. A profile written on Monday is exactly right on Friday with no maintenance job. scripts/profile_store.py proves both claims and shows the byte layout of the item:

#!/usr/bin/env python3
"""The EMA profile as a single DynamoDB item: O(1) updates, no history scan.

Chapter 5 defines the taste vector as a decayed average over the FULL click
history:

    taste(now) = sum_i w_i * e_i / sum_i w_i,   w_i = exp(-lambda * (now - t_i))

Recomputing that on every click means fetching the whole history. This lab
shows the production form: keep only (S, W, t_last) and fold each click in,

    decay = exp(-lambda * (t - t_last))
    S <- S * decay + e        # decayed numerator
    W <- W * decay + 1        # decayed denominator
    t_last <- t

then read taste = S / W. Two facts make this a perfect fit for a key-value
store like DynamoDB:

  1. the update touches ONE item (read, three-line math, conditional write);
  2. S / W is decay-invariant: decaying to any later read time multiplies S
     and W by the same factor, which cancels. Nothing goes stale at rest.

The lab proves both numerically, then packs the state into the exact byte
layout you would store (float32 Binary attribute).

Run:  python scripts/profile_store.py
"""
from __future__ import annotations

import numpy as np

DIM = 384                      # e.g. sentence-transformers all-MiniLM-L6-v2
HALF_LIFE_H = 72.0
LAM = np.log(2) / (HALF_LIFE_H * 3600.0)


def batch_profile(history, now):
    """Chapter 5's formula: full scan over (t, embedding) pairs."""
    w = np.array([np.exp(-LAM * (now - t)) for t, _ in history])
    E = np.stack([e for _, e in history])
    return (w[:, None] * E).sum(0) / w.sum()


def incremental_update(state, t, e):
    """The O(1) fold: what the click-stream Lambda runs per event."""
    S, W, t_last = state
    decay = np.exp(-LAM * (t - t_last)) if t_last is not None else 0.0
    return (S * decay + e, W * decay + 1.0, t)


def pack(S, W, t_last, embedder="minilm-l6-v2@384"):
    """The DynamoDB item: vector as a float32 Binary attribute."""
    return {
        "PK": "USER#U106",
        "vec": S.astype("<f4").tobytes(),   # 384 * 4 = 1536 bytes
        "wsum": float(W),
        "t_last": int(t_last),
        "embedder": embedder,               # vectors are useless without this
    }


def main():
    rng = np.random.default_rng(0)

    # 40 clicks over 14 days, unit embeddings
    t0 = 1_700_000_000
    ts = np.sort(rng.integers(0, 14 * 86400, size=40)) + t0
    history = []
    for t in ts:
        e = rng.normal(size=DIM)
        history.append((float(t), e / np.linalg.norm(e)))

    # fold the stream, one click at a time
    state = (np.zeros(DIM), 0.0, None)
    for t, e in history:
        state = incremental_update(state, t, e)
    S, W, t_last = state

    now = t_last
    print(f"history: {len(history)} clicks over 14 days, {DIM}-dim embeddings")
    diff = np.abs(S / W - batch_profile(history, now)).max()
    print(f"incremental S/W vs batch full-scan:      max |diff| = {diff:.1e}")

    week_later = now + 7 * 86400
    drift = np.abs(batch_profile(history, week_later) - S / W).max()
    print(f"batch profile read 7 days later:         max |diff| = {drift:.1e}"
          "   (S/W is decay-invariant)")

    item = pack(S, W, t_last)
    print("\nDynamoDB item layout:")
    for k, v in item.items():
        shown = f"{len(v)} bytes ({DIM} x float32)" if isinstance(v, bytes) else v
        print(f"  {k:<9} {shown}")

    S_back = np.frombuffer(item["vec"], dtype="<f4").astype(np.float64)
    err = np.abs(S_back / item["wsum"] - S / W).max()
    print(f"\nfloat32 roundtrip error on the profile:  max |diff| = {err:.1e}")


if __name__ == "__main__":
    main()

$ python scripts/profile_store.py
history: 40 clicks over 14 days, 384-dim embeddings
incremental S/W vs batch full-scan:      max |diff| = 1.0e-17
batch profile read 7 days later:         max |diff| = 1.7e-17   (S/W is decay-invariant)

DynamoDB item layout:
  PK        USER#U106
  vec       1536 bytes (384 x float32)
  wsum      12.817213725993117
  t_last    1701174210
  embedder  minilm-l6-v2@384

float32 roundtrip error on the profile:  max |diff| = 1.2e-09

A 384-d float32 vector is 1,536 bytes against DynamoDB's 400 KB item limit: vectors are small. The full table design:

The click-stream Lambda is the only writer to profiles, and it guards against concurrent clicks with an optimistic lock (follow-along; requires boto3 and an AWS account, output not shown):

import boto3

table = boto3.resource("dynamodb").Table("profiles")

def apply_click(user_id, emb, t):
    item = table.get_item(Key={"PK": f"USER#{user_id}"}).get("Item")
    S, W, t_last, ver = unpack(item)              # np.frombuffer + floats
    decay = math.exp(-LAM * (t - t_last))
    S, W = S * decay + emb, W * decay + 1.0
    table.put_item(
        Item={"PK": f"USER#{user_id}", "vec": S.astype("<f4").tobytes(),
              "wsum": Decimal(str(W)), "t_last": int(t),
              "embedder": EMBEDDER_VER, "ver": ver + 1},
        ConditionExpression="ver = :v",           # optimistic lock
        ExpressionAttributeValues={":v": ver},
    )                                             # on failure: re-read, retry

Use on-demand capacity mode until traffic is predictable, and let DynamoDB TTL do the retention work: profiles of users idle for a year, impressions older than 30 days, and articles past their news lifetime all delete themselves.

Content and search in OpenSearch

OpenSearch holds the article documents and the kNN index, so one query returns displayable cards directly. The index mapping (follow-along):

PUT /news-v1
{
  "settings": { "index.knn": true },
  "mappings": { "properties": {
    "title":        { "type": "text" },
    "abstract":     { "type": "text" },
    "subcategory":  { "type": "keyword" },
    "published_at": { "type": "date" },
    "image_url":    { "type": "keyword" },
    "image_phash":  { "type": "unsigned_long" },
    "story_id":     { "type": "keyword" },
    "embedder":     { "type": "keyword" },
    "embedding":    { "type": "knn_vector", "dimension": 384,
                      "method": { "name": "hnsw", "engine": "lucene",
                                  "space_type": "cosinesimil" } }
  } }
}

The hnsw method is the graph index from the companion HNSW book; story_id is the dedup key the next section produces. The serve-time query does three of the rubric's jobs in one round trip: taste matching (kNN from the DynamoDB profile vector), freshness (a date filter), and duplicate exclusion (field collapse on story_id, one card per story). Follow-along; the exact kNN filter syntax varies a little by OpenSearch version and engine:

POST /news-v1/_search
{
  "size": 50,
  "query": { "knn": { "embedding": {
      "vector": [/* 384 floats read from the profiles table */],
      "k": 200,
      "filter": { "bool": {
        "filter":   [ { "range": { "published_at": { "gte": "now-72h" } } } ],
        "must_not": [ { "terms": { "story_id": [/* recently seen stories */] } } ]
      } }
  } } },
  "collapse": { "field": "story_id" }
}

Two operational notes. Name indexes with a version (news-v1) and serve through an alias: an embedder upgrade becomes create news-v2, backfill from the DynamoDB articles table, flip the alias, delete the old index. And bound the must_not list (the last few hundred seen stories is plenty); the seen table remains the exact record.

Excluding duplicates from results

Duplicates get excluded at three layers, in decreasing order of leverage:

1. Ingest: assign story_id once per article. All the expensive thinking happens here, offline, exactly once.
2. Query: collapse on story_id. OpenSearch returns one card per story for free. This is the line that actually removes duplicates from results.
3. Serve: the Chapter 27 residue screen. A cheap in-process check over the final k cards (title Jaccard, pHash Hamming, cached verdicts) catches whatever crossed cluster boundaries. It calls no models.

Layer 1 is the new machinery. Comparing each incoming article to a million-article catalog cannot be $O(n^2)$, and the standard fix is SimHash + locality-sensitive hashing: hash every title to 64 bits such that similar token sets land a few bits apart, bucket the catalog by 8-bit bands, and score only bucket collisions. Tight matches merge automatically (union-find); the gray zone goes to the judge queue; everything else is distinct. scripts/story_clusters.py runs the whole pipeline on the chapter's case list:

#!/usr/bin/env python3
"""Ingest-time story clustering: SimHash + LSH bands + union-find.

Chapter 27 screened duplicates per slate: 45 pairs for 10 cards, fine at
request time. A production catalog needs the opposite shape: decide "which
story is this?" ONCE per article at ingest, store the answer as `story_id`,
and let the query layer collapse on it. Comparing each new article to a
million existing ones is the part that must not be O(n^2), and the standard
fix is locality-sensitive hashing:

  1. SimHash every title into 64 bits (token hashes vote per bit position;
     the sign of each vote becomes the bit). Similar token sets differ in
     few bits.
  2. Split the 64 bits into 8 bands of 8 bits and bucket articles by each
     band value. Near-duplicates almost surely collide in SOME band, so
     candidates come from bucket collisions, not from all pairs.
  3. For candidates: Hamming distance <= MERGE joins the same story
     (union-find); the gray zone up to JUDGE goes to the LLM judge queue.

Everything is deterministic (token hashes come from md5, not Python's
salted hash()), so ingest is reproducible.

Run:  python scripts/story_clusters.py
"""
from __future__ import annotations

import hashlib
import itertools

MERGE = 4    # Hamming <= 4: same story, merge without asking
JUDGE = 16   # 4 < Hamming <= 16 on an LSH candidate: send the pair to the judge

CORPUS = [
    ("N1", "Lakers edge Celtics 102-99 in overtime thriller"),
    ("N2", "Lakers beat Celtics in 102-99 overtime thriller"),      # case B
    ("N3", "Lakers edge Celtics 102-99 in overtime thriller"),      # case A
    ("N4", "Warriors edge Suns 118-115 in overtime thriller"),      # case D
    ("N5", "Fed raises interest rates for third time this year"),
    ("N6", "Borrowing costs climb as central bank hikes again"),    # case E
    ("N7", "NBA playoff race tightens after week of upsets"),
]


def token_hash(tok: str) -> int:
    return int.from_bytes(hashlib.md5(tok.encode()).digest()[:8], "big")


def simhash(title: str) -> int:
    votes = [0] * 64
    for tok in set(title.lower().split()):
        h = token_hash(tok)
        for b in range(64):
            votes[b] += 1 if (h >> b) & 1 else -1
    return sum(1 << b for b in range(64) if votes[b] > 0)


def hamming(a: int, b: int) -> int:
    return (a ^ b).bit_count()


def bands(sig: int, n=8, width=8):
    return [(i, (sig >> (i * width)) & ((1 << width) - 1)) for i in range(n)]


class UnionFind:
    def __init__(self, keys):
        self.p = {k: k for k in keys}

    def find(self, k):
        while self.p[k] != k:
            self.p[k] = self.p[self.p[k]]
            k = self.p[k]
        return k

    def union(self, a, b):
        self.p[self.find(b)] = self.find(a)


def main():
    sigs = {nid: simhash(t) for nid, t in CORPUS}
    titles = dict(CORPUS)

    # LSH: bucket by band value; collisions are the only pairs we ever score
    buckets = {}
    for nid, sig in sigs.items():
        for band in bands(sig):
            buckets.setdefault(band, []).append(nid)
    candidates = {tuple(sorted(p))
                  for ids in buckets.values() if len(ids) > 1
                  for p in itertools.combinations(ids, 2)}

    n_all = len(CORPUS) * (len(CORPUS) - 1) // 2
    print(f"catalog: {len(CORPUS)} articles, {n_all} possible pairs, "
          f"{len(candidates)} LSH candidate pair(s)\n")

    # pass 1: auto-merge the obvious duplicates
    uf = UnionFind(sigs)
    print("candidate  hamming  decision")
    print("-" * 46)
    gray = []
    for a, b in sorted(candidates):
        d = hamming(sigs[a], sigs[b])
        if d <= MERGE:
            uf.union(a, b)
            decision = "merge (same story)"
        elif d <= JUDGE:
            gray.append((a, b))
            decision = "-> judge queue"
        else:
            decision = "distinct (chance band collision)"
        print(f" {a},{b}     {d:4d}    {decision}")

    # pass 2: key the judge queue by STORY pair, so merged members
    # (N1 == N3) do not ask the same question twice
    judge_queue = {tuple(sorted((uf.find(a), uf.find(b)))) for a, b in gray}

    clusters = {}
    for nid in sigs:
        clusters.setdefault(uf.find(nid), []).append(nid)
    print("\nstory_id assignment (stored on the article + OpenSearch doc):")
    for root, members in sorted(clusters.items()):
        print(f"  story:{root}  <- {', '.join(sorted(members))}")

    print("\njudge queue, deduped by story pair (batched nightly, cached):")
    for a, b in sorted(judge_queue):
        print(f"  story:{a} vs story:{b}: '{titles[a]}' / '{titles[b]}'")

    d14 = hamming(sigs["N1"], sigs["N4"])
    d56 = hamming(sigs["N5"], sigs["N6"])
    print(f"\nnotes: N1 vs N4 (case D template lookalike) hamming = {d14}: no"
          "\nband collision and past the judge cutoff, so no judge spend on it."
          f"\nN5 vs N6 (case E paraphrase, zero shared tokens) hamming = {d56}:"
          "\nSimHash cannot see paraphrases; the embedding-neighbor check"
          "\nat ingest is what routes case E to the judge.")


if __name__ == "__main__":
    main()

$ python scripts/story_clusters.py
catalog: 7 articles, 21 possible pairs, 4 LSH candidate pair(s)

candidate  hamming  decision
----------------------------------------------
 N1,N2        8    -> judge queue
 N1,N3        0    merge (same story)
 N2,N3        8    -> judge queue
 N2,N5       22    distinct (chance band collision)

story_id assignment (stored on the article + OpenSearch doc):
  story:N1  <- N1, N3
  story:N2  <- N2
  story:N4  <- N4
  story:N5  <- N5
  story:N6  <- N6
  story:N7  <- N7

judge queue, deduped by story pair (batched nightly, cached):
  story:N1 vs story:N2: 'Lakers edge Celtics 102-99 in overtime thriller' / 'Lakers beat Celtics in 102-99 overtime thriller'

notes: N1 vs N4 (case D template lookalike) hamming = 17: no
band collision and past the judge cutoff, so no judge spend on it.
N5 vs N6 (case E paraphrase, zero shared tokens) hamming = 32:
SimHash cannot see paraphrases; the embedding-neighbor check
at ingest is what routes case E to the judge.

Read the economics: 21 possible pairs became 4 scored candidates and one judge call, and that call is keyed by story pair, so its verdict is cached forever in the verdicts table. The case D lookalike cost nothing. The case E paraphrase is SimHash-blind, which is why ingest also runs one kNN query against the fresh-article window and sends high-cosine, low-token-overlap neighbors to the same judge queue. When the judge answers same_story for N1 vs N2, the ingest job merges the clusters and rewrites story_id on the losing article (one DynamoDB update, one OpenSearch partial update), and the collapse layer hides the duplicate from every future query.

For the judge itself, production means the Chapter 27 script with three upgrades: send the nightly queue through the Message Batches API (half price, no rate-limit pressure), put the rubric system prompt behind prompt caching, and write every verdict to verdicts with the judge model id and prompt hash, so a prompt change is visible as a new cache generation rather than a silent behavior shift.

Why not send every image to the LLM?

A fair question: a vision model can look at two thumbnails and decide, so why keep dHash at all? Because a hash and a judge answer different questions ("same photo?" vs. "same meaning?"), and the hash has properties no per-pair model call can have:

The per-image fingerprint is the structural advantage: an LLM verdict is a function of a pair, so there is nothing to precompute, index, or cache per image, and deduping one new article against a million-article catalog would cost a million calls. There is also a governance angle: hashing runs inside your VPC, while thumbnails may be licensed wire-service content you cannot ship to a third party, and any image containing rendered text is an injection surface for the judge.

What the hash cannot do is semantics. It only detects "same pixels, roughly" (cases A and C: recompressed, brightened, lightly cropped copies), is blind to two different photos of one event and to flips, heavy crops, and watermarks, and it false-positives on near-uniform graphics (solid backgrounds, scoreboard templates that differ only in the digits). So the escalation ladder mirrors the text side exactly:

dHash / pHash  ──►  image embedding kNN (CLIP-tier)  ──►  vision judge
per image, free     per image, ANN-indexable, semantic     per pair, batched,
decides A and C     catches same-scene-different-photo     coherence + gray zone

Each layer feeds only its leftovers to the next. The rule that falls out of the whole chapter: never pay a model to answer a question XOR can answer.

Design decisions: must, should, consider

The blueprint above made specific choices. This section separates the non-negotiables from the defaults and the scale-dependent options, in RFC-2119 spirit, so you can tell which deviations are fine and which are production incidents waiting to happen.

Must (non-negotiable, whatever the stack):

Should (strong defaults; deviate only with a written reason):

Consider (worth a design discussion once scale or product demands it):

• Vector quantization (float16 or int8): halves or quarters storage and I/O; starts mattering around tens of millions of vectors, not before.
• The CLIP tier from the ladder above, once the vision judge queue is the cost line that hurts.
• A read cache (DAX or ElastiCache) in front of hot article items; trending stories make hot keys.
• Multiple profiles per user: a short-half-life vector for "right now" and a long one for stable taste, blended at query time; also per-surface half-lives (push notifications want fresher than the homepage).
• Storyline vs. event clustering: an ongoing saga ("day 3 of the trial") is many events one storyline; decide which granularity story_id means, because collapse hides whichever one you pick.
• Majority-vote judging only for pairs that flip across runs; spending three votes on every pair triples cost for noise you mostly do not have.
• Multi-region: DynamoDB global tables + a second OpenSearch domain; buy it when the availability math says so, not for launch.
• Daily index rollover (ISM policies) so news retention is a delete of old indexes instead of per-document deletes.

The implementation plan

Ordering rationale: the golden set comes first because every threshold in phases 1 through 3 is tuned against it; serving comes before judge operations because the fail-open policy means the feed must already be correct (if slightly duplicate-prone) with the judge switched off.

The ops checklist

• Alarms on the flag rate (deterministic layer), the judge overturn rate (how often the judge disagrees with the auto-merge layer), the verdict cache hit rate, and the collapse count per query. Each one drifting is an early warning from a different subsystem.
• Budgets: a hard monthly cap on judge spend, enforced by the batch job's queue length, with fail-open (keep both) past the cap.
• Retention and deletion: profiles are behavioral data. TTL inactive users, and wire the account-deletion API to delete the profiles and seen items (the GDPR path must exist before launch, not after).
• Locale: SimHash tokenization, judge prompts, and thresholds are all per-language. Do not reuse English thresholds on agglutinative languages.
• Property tests in CI: incremental EMA equals batch recompute; clustering is order-independent for the same daily batch; a verdict cache hit never calls the API.

Tools shortlist for the pieces this book did from scratch: sentence-transformers (embeddings), imagehash + Pillow (pHash), datasketch (MinHash LSH, if you outgrow SimHash), spaCy (NER at ingest), opensearch-py and boto3 (clients), Anthropic Message Batches + prompt caching (judge economics), MLflow (scorecards next to Chapter 19's metrics).

And that is the blueprint: vectors in DynamoDB, articles in OpenSearch, and duplicates gone before the query returns. One chapter remains, and it is the assembly manual: the whole rubric framework run end to end on a worked example, from a user's click history to a repaired slate, with the validation methods that earn it the right to be trusted. 👉