Skip to content

Adapters

recsys/adapters/ is the only place IO lives. Each adapter implements a port Protocol, so the pure core never imports a driver.

flowchart LR
    subgraph ports["Port (Protocol)"]
        p1["EventSource"]
        p2["ContentStore"]
        p3["UserModelStore"]
        p4["EmbeddingModel"]
    end
    subgraph impl["adapters/"]
        a1["RedisEventBuffer"]
        a2["QdrantContentStore"]
        a3["RedisUserModelStore"]
        a4["FastEmbedModel"]
        a5["rudderstack.normalize_*"]
    end
    p1 -.-> a1
    p2 -.-> a2
    p3 -.-> a3
    p4 -.-> a4
    a5 -->|produces| ev["InteractionEvent"]

rudderstack.py, pure normalizer (no infra)

The boundary translator: raw RudderStack payload -> canonical InteractionEvent.

Function Does
normalize_content_id(raw_id) 'content_1234' -> '1234'; bridges string IDs to Qdrant int points.
normalize_event(raw) One track payload -> InteractionEvent (or None if unusable).
normalize_events(raws) Map many, drop None, sort by timestamp.

This is pure and infra-free on purpose, it is the shared contract every event source must satisfy (one contract test asserts identical InteractionEvent from equivalent raw across all source adapters).

qdrant_store.py, ContentStore

QdrantContentStore(client, collection_name):

Method Does
get(ids) Payloads -> dict[id, Content].
get_vectors(ids) Embeddings -> dict[id, Vector] (handles named vectors).
search_vector(vector, *, limit) Top-k similar -> Candidates with cosine scores.
search_tags(tag_keys, *, limit) Match any facet:label via Filter(should=...) over the tag_labels KEYWORD index -> Candidates.

Tag recall is a filter, not a ranker, graded tag ranking happens in pure score_tag. _pid converts numeric string IDs to ints for Qdrant.

redis_store.py, hot path stores

flowchart LR
    subgraph buffer["RedisEventBuffer (sorted set, score=ts)"]
        ap["append(event)<br/>auto-prune > window_days"]
        fe["fetch_events(user_id)"]
    end
    subgraph model["RedisUserModelStore (JSON + TTL)"]
        gs["get_signals(user_id)"]
        sv["save_signals(signals)"]
    end
  • RedisEventBuffer: per-user sorted set keyed by timestamp, window_days=30 default, auto-prunes old events on append. Implements EventSource plus .append().
  • RedisUserModelStore: materialized UserSignals as JSON, ttl_secondsā‰ˆ7d. The fast read on the serve path. Also exposes iter_signals() for cohort-wide stats.
  • RedisImpressionStore: request_id then {content_id: feature_vector} (JSON, TTL'd). The short-lived join store for online bandit updates: a reward event echoing the request_id looks up the served context here, then consume(request_id, content_id) drops it so a redelivered reward cannot double-count (idempotent). Implements the ImpressionStore port; test fake is InMemoryImpressionStore.

All three Redis stores take a key_prefix so the composition root can isolate tenants ({tenant}:umodel / evt / imp); see Multi-tenancy.

Event log, impressions, and config

Three adapters added alongside the hot-path stores.

event_log.py, durable Parquet log

The permanent training/eval record (Redis is ephemeral). Two append-only datasets under EVENT_LOG_DIR:

Dataset Holds
date=YYYY-MM-DD/part-*.parquet ingested InteractionEvents (the reward record), including request_id
served/date=YYYY-MM-DD/part-*.parquet recommendations served: user, ranked items with feature vectors, distractor, request_id
  • NullEventLog: no-op when EVENT_LOG_DIR is unset.
  • ParquetEventLog(base_dir): append(events) partitions by day and writes immutable parts; log_served(record) appends one served-impression row (the items list is JSON-stringified so the row stays flat). Joining the two datasets on (request_id, content_id) yields (context, action, reward) tuples for the bandit. pyarrow is imported lazily.

redis_store.py -> RedisImpressionStore (ImpressionStore)

Covered above: the request_id then features store that feeds online bandit updates, TTL'd and consume-on-use.

config_store.py, runtime RecConfig override

Holds an optional runtime override applied on top of the env/default RecConfig, so recsys params change live (via PUT /config and the settings page) without a redeploy.

  • RedisConfigStore(client, key=...): get / set / clear a JSON override under a per-tenant key ({tenant}:recsys:config). deep_merge layers a partial patch onto the baseline. Redis is the live override layer; the env/configmap stays the source of truth, so if Redis is flushed the config reverts to the baseline.
  • InMemoryConfigStore: dev/test fallback (no Redis).

fastembed_model.py, EmbeddingModel

FastEmbedModel(model_name="sentence-transformers/all-MiniLM-L6-v2") -> dim property + encode(text) -> Vector. Used where the system must embed text at runtime (e.g. cold-start query text). Test fake is InMemoryEmbeddingModel (deterministic hash buckets).

Infra guarding

Adapters import their drivers lazily / behind guards; the composition root only instantiates a real adapter when its env var (REDIS_URL, QDRANT_API_URL) is set, else falls back to a fake. Result: the whole pipeline runs offline with zero infra.

Full auto-generated reference

Code reference -> Recsys package.