Sharded Candidate Generation

"I am a query vector, asking a billion items politely to narrow themselves down. They could not all answer at once, so they formed committees, each shard nominating its favorites, and a coordinator counted the votes."

A User Embedding Awaiting Its Shortlist

The previous section built the embeddings that this section consumes. In Section 38.2 we learned how a recommender represents users and items as dense vectors and how the enormous embedding tables are sharded across a parameter-server fleet so that no single machine must hold them all. Those embeddings are not the product; they are the raw material for a decision. A recommendation request arrives, the user tower produces one query vector, and the system must answer a question with a strict latency budget: out of every item in the catalog, which few thousand are even worth a closer look? Answering that question over a billion-item catalog, fast enough to serve a homepage, is the candidate-generation problem, and it is unavoidably a read path over the sharded item embeddings of the prior section.

Recommenders solve this with a funnel. The full catalog enters the top; a cheap retrieval stage cuts it to thousands of candidates; an expensive ranking model (the subject of Section 38.4) then scores only those survivors and orders them. The reduction is dramatic and it has to be, because the ranking model is hundreds of times more costly per item than retrieval. Figure 38.3.1 shows the shape of the whole funnel and the two-tower retrieval that drives its widest stage; the rest of the section unpacks each part.

1. Two-Tower Retrieval as Nearest-Neighbor Search Beginner

The two-tower model is the workhorse of modern candidate generation, and its appeal is entirely structural. One neural network, the user tower, maps a request (the user, their recent history, the time of day, the surface they are on) to a vector $q \in \mathbb{R}^d$. A second network, the item tower, maps each item to a vector $v_i \in \mathbb{R}^d$ in the same space. The predicted affinity between the user and item $i$ is simply their dot product,

and the recommendation for the user is the set of items with the highest scores. The structural payoff is that the item tower does not depend on the user at all. Every item vector can therefore be computed offline, in a batch job, and stored once in an index; at request time only the user tower runs, producing a single $q$. Retrieval then reduces to a geometric question: which item vectors lie nearest to $q$ under the inner product? That is exactly the (maximum inner-product) nearest-neighbor search that Chapter 25 developed in full, now pointed at a recommendation catalog instead of a document corpus.

Formally, candidate generation asks for the top-$k$ set

the $k$ items maximizing the score, out of a catalog of $N$ items. With $N$ in the billions, computing all $N$ dot products on one machine for every request is infeasible at serving latency, which is what forces both approximation and distribution. The approximation, replacing the exact top-$k$ with an approximate nearest-neighbor (ANN) search that inspects only a fraction of the index, is the subject of Chapter 25 and we reuse it here. The distribution, splitting the index across shards, is the subject of the next section.

2. Sharding the Item Index and Gathering Top-k Intermediate

Even with offline item embeddings, one machine cannot hold a billion $d$-dimensional vectors in memory, nor scan them within a serving deadline. The remedy is the partitioning-and-sharding arc the book has followed from Chapter 25: split the item index across $S$ shards, place each shard on its own machine, and let every shard own a disjoint slice of the catalog. A retrieval request then becomes a scatter-gather. The query vector $q$ is scattered to all $S$ shards; each shard runs an ANN search over only its own items and returns its local top-$k$; and a coordinator gathers the $S$ partial lists and merges them.

The merge is correct because top-$k$ decomposes the same way a sum does. If each shard returns the $k$ highest-scoring items it holds, then the global top-$k$ is guaranteed to lie within the union of those $S$ local lists, so

where $\mathrm{TopK}_k^{(s)}$ is the local top-$k$ from shard $s$. No global item can outrank every member of its own shard's local top-$k$ without already being in it, so merging $S$ lists of length $k$ and keeping the best $k$ reproduces the exact global answer. This is the same algebraic fact that made the all-reduce of Chapter 4 exact: the gather here is a reduce whose operator is "merge and keep top-$k$" instead of "add." Selecting the global winners from the per-shard winners is the maximum-selection tie that Section 36.6 used for distributed retrieval in the RAG case study, applied now to a recommendation index.

One subtlety distinguishes the recommendation setting from the textbook reduction. Because each shard runs an approximate nearest-neighbor search, not an exact scan, a shard may miss a true top-$k$ item that it actually holds. The merge is then exact with respect to the candidates the shards surfaced, but the end-to-end retrieval is only approximate, and its quality is measured by recall at $k$,

the fraction of the true top-$k$ that the sharded ANN actually recovered. Production retrieval tunes each shard's search effort and the per-shard $k$ to push recall close to one while staying inside the latency budget. The runnable demo below isolates the merge logic from the approximation: it runs an exact per-shard scan so that the sharded merge provably equals the single-machine answer, letting us verify the decomposition before approximation enters.

import numpy as np

rng = np.random.default_rng(7)
N, d, S, k = 2_000_000, 64, 16, 10   # items, embedding dim, shards, candidates per query

# Two-tower setup: a user tower and an item tower each emit a d-dim embedding.
# Retrieval score = dot product; the candidate set is the top-k items by score.
user_vec = rng.standard_normal(d).astype(np.float32)
item_index = rng.standard_normal((N, d)).astype(np.float32)   # the catalog embeddings

# Exact (single-machine) top-k: score every item, take the k highest.
scores_full = item_index @ user_vec
exact_topk = np.argsort(-scores_full)[:k]

# Sharded ANN: partition the item index across S shards. Each shard scores ONLY
# its own items and returns its local top-k. A gather (reduce) step merges the
# S local top-k lists and keeps the global top-k. This is scatter-gather.
shard_bounds = np.array_split(np.arange(N), S)
gathered_ids, gathered_scores = [], []
for ids in shard_bounds:                       # one "worker" per shard
    s = item_index[ids] @ user_vec             # local scores
    local = ids[np.argsort(-s)[:k]]            # local top-k item ids
    gathered_ids.append(local)
    gathered_scores.append(scores_full[local])

cand_ids = np.concatenate(gathered_ids)        # S * k merged candidates
cand_scores = np.concatenate(gathered_scores)
merged_topk = cand_ids[np.argsort(-cand_scores)[:k]]

# Verify the sharded merge equals the exact top-k (as a set).
match = set(merged_topk.tolist()) == set(exact_topk.tolist())
recall = len(set(merged_topk.tolist()) & set(exact_topk.tolist())) / k

print("items N              :", f"{N:,}")
print("shards S             :", S)
print("candidates per shard :", k, "-> gathered :", len(cand_ids))
print("merged == exact set  :", match)
print("recall@%d            :" % k, f"{recall:.3f}")
print("funnel               :", f"{N:,} items -> {len(cand_ids)} gathered -> {k} candidates")
print("reduction factor     :", f"{N // k:,}x")

items N              : 2,000,000
shards S             : 16
candidates per shard : 10 -> gathered : 160
merged == exact set  : True
recall@10            : 1.000
funnel               : 2,000,000 items -> 160 gathered -> 10 candidates
reduction factor     : 200,000x

import faiss, numpy as np
# build one IVF-PQ sub-index per shard, then wrap them as a sharded index
shards = [faiss.index_factory(d, "IVF4096,PQ32") for _ in range(S)]
for sub, vecs in zip(shards, item_shards):
    sub.train(vecs); sub.add(vecs)
index = faiss.IndexShards(d, threaded=True)      # fan-out + gather, built in
for sub in shards: index.add_shard(sub)
scores, ids = index.search(user_vec[None, :], k)  # one call: scatter, search, merge

3. Blending Multiple Candidate Sources Intermediate

A single two-tower retriever, however well sharded, is rarely enough on its own. The learned embedding space is good at "items semantically similar to what this user likes," but it systematically underweights a few things a good homepage needs: brand-new items the towers have barely seen, items surging in global popularity, and items reachable only through the relationship graph rather than the content space. Production candidate generation therefore runs several retrievers in parallel and unions their outputs. A typical blend combines the ANN two-tower source with a popularity source (the globally trending items, cheap to maintain), a recency source (items added or updated in the last hours), and a graph-based source.

The graph-based source is where Chapter 13 returns. Co-engagement defines a bipartite user-item graph; candidates can be generated by random walks from the user's recently engaged items, or by reading a node's precomputed graph-neighbor list, both of which surface items that share no obvious content similarity yet are strongly co-consumed. Because the graph itself is partitioned across machines (the distributed graph-ML problem of Chapter 13), this source is itself a distributed read path, gathered alongside the ANN shards. The blend is a union followed by deduplication: if $C_r$ is the candidate set from source $r$, the combined pool is $\bigcup_r C_r$, distinct, and the downstream ranker of Section 38.4 is responsible for scoring this heterogeneous pool on a common scale. Retrieval's job is recall (do not miss anything good); precision is the ranker's job.

4. Freshness: Getting New Items Into the Index Fast Advanced

The recency source in the example is a patch over a deeper structural problem: the item index is, by construction, a snapshot. Item embeddings are computed in an offline batch job and frozen into the ANN shards, so an item that did not exist when the last batch ran is invisible to two-tower retrieval no matter how relevant it would be. For a static document corpus this is tolerable; for a marketplace, a news feed, or a short-video platform where the most valuable items are minutes old, a stale index is a broken product. Freshness, the latency between an item appearing and an item becoming retrievable, is therefore a first-class scaling requirement of candidate generation, not an afterthought.

The standard architecture splits the index into two tiers. A large base index holds the bulk of the catalog and is rebuilt on a slow cadence (hours to a day) by the offline item tower. A small, mutable real-time index sits beside it, receives new and updated item embeddings within seconds of the item appearing, and is searched in parallel with the base shards; its results are gathered into the same merge. Periodically the real-time index is folded into the next base rebuild and reset. This mirrors the log-structured merge pattern of Chapter 9: a fast-write tier absorbs the stream while a slow-rebuild tier holds the steady state, and queries read both. The cost of freshness is that the real-time tier must embed items on the fly, often with a lighter item tower, and that the gather now spans heterogeneous index types, but the alternative, a catalog where new items cannot be recommended for hours, is usually unacceptable.