Skip to content

Code Reference, Recsys package

Auto-generated from source by mkdocstrings. Signatures, type annotations, fields, and docstrings are rendered directly from ai_engine.recsys.

Contracts

Models

models

Tag pydantic-model

Bases: BaseModel

One expert tag on a piece of content. facet is a taxonomy dimension (e.g. 'theme_what', 'person_who.age_group'); label the value.

Fields:

  • facet (str)
  • label (str)
  • weight (float)

Content pydantic-model

Bases: BaseModel

Normalized item. Supersedes the loose Qdrant payload dicts.

This is the ONE place raw payload shape is interpreted (see qdrant_store._payload_to_content): consumers — endpoints and the dashboard — read these fields and never parse payloads themselves (docs/debt-payload-scatter.md D1).

Fields:

  • id (str)
  • content_type (ContentType)
  • title (str)
  • text (str)
  • tags (list[Tag])
  • word_count (int)
  • has_image (bool)
  • lat (Optional[float])
  • lon (Optional[float])
  • image_url (Optional[str])
  • public_url (Optional[str])
  • years (list[int])

InteractionEvent pydantic-model

Bases: BaseModel

One thing a visitor did, in a source-agnostic shape.

Every event source (RudderStack, PostHog, Postgres) is normalized into this single model (see ai_engine.recsys.adapters.rudderstack.normalize_events), so the rest of the engine never sees a raw payload. The event string says what happened, and different events fill different fields (the rest stay None/empty):

  • A view (CONTENT_VIEW_ENDED) fills content_id, dwell_seconds, end_reason, request_id, and impressions.
  • A survey / identify (SURVEY_SUBMITTED, IDENTIFY) fills survey_answers.
  • A search / lookup (CONTENT_LOOKUP) fills query_text and maybe clicked_id.

build_user_signals folds a visitor's list of these into their user model.

Fields:

user_id pydantic-field 

user_id: str

Stable visitor id (RudderStack userId or anonymousId).

event pydantic-field 

event: str

What happened, e.g. CONTENT_VIEW_ENDED, SURVEY_SUBMITTED, IDENTIFY, CONTENT_LOOKUP.

ts pydantic-field 

ts: datetime

UTC timestamp of the interaction (used for recency decay + ordering).

session_id pydantic-field 

session_id: Optional[str] = None

Browser/app session, for sequence grouping.

request_id pydantic-field 

request_id: Optional[str] = None

The rec-response id echoed back on a resulting view, joins a reward to the exact impression (its served feature vector) for bandit training.

content_id pydantic-field 

content_id: Optional[str] = None

The item this event is about (source prefix like 'content_1234' stripped to '1234').

dwell_seconds pydantic-field 

dwell_seconds: Optional[float] = None

Seconds spent on the content (explicit, or computed from a start/end pair).

end_reason pydantic-field 

end_reason: Optional[EndReason] = None

How a view ended (next_button/link/close_button/abandon) → a completion score.

query_text pydantic-field 

query_text: Optional[str] = None

Search or lookup text, for CONTENT_LOOKUP events.

clicked_id pydantic-field 

clicked_id: Optional[str] = None

Item clicked from a result/rec list.

impressions pydantic-field 

impressions: list[str]

Other item ids shown alongside but not engaged → treated as soft negatives.

survey_answers pydantic-field 

survey_answers: dict

question_id → answer (str / list for multi-select / float rating). Presurvey + personalization.

raw pydantic-field 

raw: dict

The untransformed source payload, kept for debugging/audit.

UserSignals pydantic-model

Bases: BaseModel

The user model: everything the recommender needs about one visitor.

Built by ai_engine.recsys.signals.signal_builder.build_user_signals from the visitor's events (plus survey demographics), and read at serve time by the Recommender. Two shapes of signal live here: tag signals (tag_affinity / tag_aversion, matched by score_tag) and vector signals (taste_vector / recency_vector, matched by the embedding scorers). A visitor with no positive engagement yet is "cold" (see is_cold), and recommendations lean on their survey tags until browsing warms the model up.

Fields:

user_id pydantic-field 

user_id: str

The visitor this model describes.

positives pydantic-field 

positives: dict[str, float]

content_id -> recency-decayed positive strength, for items the visitor engaged with well. Seeds the taste vector and tag affinity.

negatives pydantic-field 

negatives: dict[str, float]

content_id -> recency-decayed penalty, from disliked views and shown-but-ignored impressions (soft negatives).

viewed pydantic-field 

viewed: list[str]

Every content_id the visitor has seen (any outcome). Used to exclude already-seen items from recommendations.

recent_views pydantic-field 

recent_views: list[str]

content_ids ordered most-recent-first, giving the model sequence awareness (the recency signal).

tag_affinity pydantic-field 

tag_affinity: dict[str, float]

'facet:label' -> [0,1] interest weight, blended from survey answers and engaged content. The main signal score_tag matches against content tags.

tag_aversion pydantic-field 

tag_aversion: dict[str, float]

'facet:label' -> [0,1] penalty weight, from the themes of content the visitor disliked. Applied as a negative in fusion.

taste_vector pydantic-field 

taste_vector: Optional[Vector] = None

L2-normalized centroid of liked items' embeddings (the whole-history semantic taste). None until the visitor has a positive.

recency_vector pydantic-field 

recency_vector: Optional[Vector] = None

Embedding of the most-recent viewed item, powering the 'more like what you just read' signal.

behavior pydantic-field 

behavior: dict

Engagement summary stats (n_views, completion_rate, depth, ...). Used for persona explanations, not for scoring.

demographics pydantic-field 

demographics: dict

Raw survey demographics, stored for inspection. The affinity they seed lives in tag_affinity. E.g. {"age": 60, "gender": "female", "nationality": "france", "province": "Drenthe", "personal_connection": "descendant"}.

is_cold property 

is_cold: bool

True until the visitor has at least one positively-engaged item. Cold visitors get survey-led / diverse recommendations; warm ones get taste-vector-led ones.

Recommendation pydantic-model

Bases: BaseModel

The served result: the ranked items, the strategy that produced them (warm / cold), and a diagnostics dict (pool size, generators used, filter, ranking mode, distractor placement) for the inspector and explanations.

Fields:

  • user_id (str)
  • items (list[ScoredCandidate])
  • strategy (str)
  • diagnostics (dict)

VisitorType pydantic-model

Bases: BaseModel

Falk (2009) visit-identity classification of the visitor.

Fields:

  • type (str)
  • confidence (float)
  • rationale (str)
  • scores (dict[str, float])

PersonaExplanation pydantic-model

Bases: BaseModel

Structured, evidence-backed explanation derived purely from UserSignals + content taxonomy. Deterministic; summary is the optional verbalized prose.

Fields:

  • user_id (str)
  • is_cold (bool)
  • interests (list[Interest])
  • aversions (list[Interest])
  • engagement_style (str)
  • experience_preference (str)
  • visitor_type (Optional[VisitorType])
  • trajectory (list[str])
  • demographics (dict)
  • behavior (dict)
  • summary (Optional[str])

Enums

enums

EndReason

Bases: str, Enum

How a content view ended (RudderStack CONTENT_VIEW_ENDED.details.reason).

Config

config

EngagementWeights pydantic-model

Bases: BaseModel

How much each behavioral signal contributes to a viewed item's engagement strength.

The four weights are combined into a single strength in [-1, 1] per item, which then decides whether the view counts as positive, neutral, or negative.

Fields:

dwell pydantic-field 

dwell: float = 0.4

Weight of how long the visitor stayed, relative to the item's estimated reading time.

completion pydantic-field 

completion: float = 0.3

Weight of how the view ended (finishing the item vs abandoning it).

revisit pydantic-field 

revisit: float = 0.2

Weight of coming back to the same item more than once.

survey pydantic-field 

survey: float = 0.1

Weight of an explicit rating (e.g. a 1 to 5 star), when the visitor gives one.

FusionWeights pydantic-model

Bases: BaseModel

How much each scorer contributes to the final fused score. Each scorer returns [0,1].

Tag-first policy: tag matching is the dominant signal so recommendations track the visitor's stated (survey) and browsed interests legibly. The semantic embedding signal is kept below tag: it adds coverage for cold/unseen content, it doesn't lead.

Fields:

tag pydantic-field 

tag: float = 0.45

Dominant signal: overlap between the visitor's tag interests and the content's tags (survey + browsing).

semantic pydantic-field 

semantic: float = 0.2

Similarity to the visitor's overall taste vector (the centroid of everything they liked).

recency pydantic-field 

recency: float = 0.05

Similarity to the item the visitor viewed most recently.

aversion pydantic-field 

aversion: float = -0.3

Overlap with disliked themes. Negative, so it pushes matching content down.

geo pydantic-field 

geo: float = 0.2

Physical proximity to the request's location. Scored only when a location is given.

RecConfig pydantic-model

Bases: BaseModel

All tunable settings for the recommender in one typed place.

Passing a config is how tests and tenants pin behavior. Every value has a sensible default, so RecConfig() is a complete, working configuration; override individual fields (or via RECSYS_* environment variables) to change behavior.

Fields:

engagement pydantic-field 

engagement: EngagementWeights

Weights for turning a raw view into an engagement strength.

fusion pydantic-field 

fusion: FusionWeights

Weights for combining the per-scorer signals into the final ranking score.

reading_speed_wps pydantic-field 

reading_speed_wps: float = 4.2

Assumed reading speed in words per second (~250 wpm), used to estimate how long an item should take to read.

img_extra_time pydantic-field 

img_extra_time: float = 1.3

Extra seconds added to the reading-time estimate for an item that has an image.

dwell_cap_ratio pydantic-field 

dwell_cap_ratio: float = 2.0

Caps dwell / estimated-reading-time at this ratio before normalizing, so one very long view can't dominate.

positive_threshold pydantic-field 

positive_threshold: float = 0.3

Engagement strength at or above this counts as a positive (liked) view.

negative_threshold pydantic-field 

negative_threshold: float = -0.05

Engagement strength at or below this counts as a negative (disliked) view; between the two thresholds is neutral.

half_life_days pydantic-field 

half_life_days: float = 14.0

Half-life in days for time decay: a signal's weight halves every this many days, so recent behavior counts for more.

soft_negative_weight pydantic-field 

soft_negative_weight: float = 0.3

Penalty for an item that was shown to the visitor but never engaged (a 'soft negative').

pool_per_generator pydantic-field 

pool_per_generator: int = 30

How many candidate items each generator (semantic, tag, geo, ...) contributes before ranking.

final_limit pydantic-field 

final_limit: int = 10

Number of items returned in a recommendation list.

mmr_lambda pydantic-field 

mmr_lambda: float = 0.7

Relevance vs diversity trade-off in MMR reranking: 1.0 is pure relevance, 0.0 is pure diversity.

geo_scale_m pydantic-field 

geo_scale_m: float = 300.0

Distance scale in metres for geo proximity: score = exp(-distance / scale), roughly a camp-sized falloff.

geo_radius_m pydantic-field 

geo_radius_m: float = 1000.0

Default radius in metres used when a geo filter is requested.

filter_reshow_when_exhausted pydantic-field 

filter_reshow_when_exhausted: bool = True

When a location filter runs out of unseen content, re-show its already-seen items instead of returning empty (never leaks outside the filter).

distractor_enabled pydantic-field 

distractor_enabled: bool = True

Whether to inject one deliberately off-profile item for novelty / exploration.

distractor_strategy pydantic-field 

distractor_strategy: str = 'max_dissimilar'

How the distractor is chosen: 'max_dissimilar', 'unexplored_theme', or 'random'.

distractor_probability pydantic-field 

distractor_probability: float = 1.0

Chance of injecting the distractor on a given request (1.0 = always, 0.35 = occasional).

distractor_slots pydantic-field 

distractor_slots: list[int] = [3, 4]

Candidate 1-based positions where the distractor may land; one is picked at random.

cold_start_min_positives pydantic-field 

cold_start_min_positives: int = 1

Positive views needed before a visitor is considered warm (the live check is simply whether they have any positives).

tag_engagement_trust_k pydantic-field 

tag_engagement_trust_k: float = 5.0

Cold-start pace for blending survey vs browsing tags: eng_trust = n_positive / (n_positive + k). Larger k means browsing takes over more slowly.

tag_match_topk pydantic-field 

tag_match_topk: int = 6

score_tag normalizes by the visitor's strongest this-many tag affinities, so a few strong interests aren't diluted by many weak ones.

ranking_mode pydantic-field 

ranking_mode: str = 'static'

Ranking policy: 'static' weighted fusion, or a learned 'bandit' whose starting point is the fusion weights.

bandit_alpha pydantic-field 

bandit_alpha: float = 0.3

Bandit exploration strength (the UCB bonus). 0 means greedy / exploit only.

bandit_ridge pydantic-field 

bandit_ridge: float = 1.0

Bandit prior strength: how tightly the learned weights start pinned to the fusion weights.

bandit_explore pydantic-field 

bandit_explore: bool = True

Whether to add the UCB exploration bonus when serving with the bandit.

bandit_online pydantic-field 

bandit_online: bool = False

Whether to update the bandit live as reward events arrive, instead of offline batch training.

Ports

ports

EmbeddingModel

Bases: Protocol

Text -> vector. Real impl = fastembed; test fake = deterministic.

EventSource

Bases: Protocol

A source of one visitor's interaction history, as canonical events.

This is a port: anything that can return a visitor's InteractionEvents can back it, and the recommender only depends on this interface, never on a concrete backend. The adapter behind it does the normalization (see ai_engine.recsys.adapters.rudderstack.normalize_events), so downstream logic never sees source-specific payload shapes.

The online serving model. The engine serves recommendations online (documented on the "Online Serving Model" page): app events flow app -> RudderStack -> /api/ingest webhook, where each event is normalized and pushed into a hot buffer (a Redis sorted set), and the user model is rebuilt from that buffer on every event. A recommendation is then a fast read of the already-built model, with no rebuild on the request path. In this setup EventSource is that hot buffer, and fetch_events returns the visitor's recent events from it. (The contrast is an offline/batch setup, where the same interface is instead backed by a data warehouse query. PostHog is only the analytics sink; it is never read at serve time.)

fetch_events 

fetch_events(user_id: str) -> list[InteractionEvent]

Return one visitor's recent interaction history as canonical events.

Parameters:

Name Type Description Default
user_id str

The visitor whose events to return.

 required

Returns:

Type Description
list[InteractionEvent]

That visitor's recent InteractionEvents, oldest first.
Example 
buffer.fetch_events("u-42")   # online: Redis EventBuffer; tests: an in-memory fake
# [InteractionEvent(user_id="u-42", event="SURVEY_SUBMITTED", ...),
#  InteractionEvent(user_id="u-42", event="CONTENT_VIEW_ENDED", content_id="5567", ...)]
Source code in ai-engine/src/ai_engine/recsys/contracts/ports.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def fetch_events(self, user_id: str) -> list[InteractionEvent]:
    """Return one visitor's recent interaction history as canonical events.

    Args:
        user_id: The visitor whose events to return.

    Returns:
        That visitor's recent `InteractionEvent`s, oldest first.

    Example:
        ```python
        buffer.fetch_events("u-42")   # online: Redis EventBuffer; tests: an in-memory fake
        # [InteractionEvent(user_id="u-42", event="SURVEY_SUBMITTED", ...),
        #  InteractionEvent(user_id="u-42", event="CONTENT_VIEW_ENDED", content_id="5567", ...)]
        ```
    """
    ...

DemographicsProvider

Bases: Protocol

Supplies a user's survey demographics (age/gender/nationality) for the cold-start tag bridge. Source is pluggable: Postgres visitor table, survey events, or a static map. Returns {} when unknown.

UserModelStore

Bases: Protocol

Materialized user model (UserSignals) for online serving.

The ingestion webhook updates this on each event so a rec request is a fast read, not a rebuild. The in-memory fake / recompute-backed impl make this a drop-in: swap to Redis without touching the recommender.

ImpressionStore

Bases: Protocol

Short-lived store of the FEATURE VECTORS we served, keyed by request_id, so a later reward event (CONTENT_VIEW echoing that request_id) can be joined back to the exact context for an ONLINE bandit update. TTL'd; not durable (the Parquet log is).

ContentStore

Bases: Protocol

Content structure + vectors (Qdrant). Test fake = in-memory.

Signals

engagement

engagement

Pure engagement scoring. No IO. Input = plain numbers, output = float/enum.

These functions are the easiest thing to validate: feed known numbers, assert the behavior the design promises (longer dwell -> higher, abandon -> negative, ...).

estimate_reading_time 

estimate_reading_time(word_count: int, has_image: bool, cfg: RecConfig) -> float

Seconds a typical visitor needs to consume this content.

Source code in ai-engine/src/ai_engine/recsys/signals/engagement.py 
22
23
24
25
26
27
def estimate_reading_time(word_count: int, has_image: bool, cfg: RecConfig) -> float:
    """Seconds a typical visitor needs to consume this content."""
    base = word_count / cfg.reading_speed_wps if cfg.reading_speed_wps > 0 else 0.0
    if has_image:
        base += cfg.img_extra_time
    return base

engagement_strength 

engagement_strength(*, dwell_seconds: Optional[float], est_reading_time: float, end_reason: Optional[EndReason], visits: int, survey_rating: Optional[float], cfg: RecConfig) -> float

Continuous engagement in roughly [-1, 1]. Weighted blend of behavioral signals.

Source code in ai-engine/src/ai_engine/recsys/signals/engagement.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def engagement_strength(
    *,
    dwell_seconds: Optional[float],
    est_reading_time: float,
    end_reason: Optional[EndReason],
    visits: int,
    survey_rating: Optional[float],
    cfg: RecConfig,
) -> float:
    """Continuous engagement in roughly [-1, 1]. Weighted blend of behavioral signals."""
    w = cfg.engagement
    completion = _COMPLETION.get(end_reason, 0.0)
    strength = (
        w.dwell * _dwell_ratio(dwell_seconds, est_reading_time, cfg)
        + w.completion * completion
        + w.revisit * _revisit(visits)
        + w.survey * _survey(survey_rating)
    )
    return strength

classify_outcome 

classify_outcome(strength: float, cfg: RecConfig) -> Outcome

Bucket a continuous engagement strength into a discrete outcome.

strength >= positive_threshold → positive; <= negative_threshold → negative; anything in between → neutral (ignored by the user model).

Source code in ai-engine/src/ai_engine/recsys/signals/engagement.py 
74
75
76
77
78
79
80
81
82
83
84
def classify_outcome(strength: float, cfg: RecConfig) -> Outcome:
    """Bucket a continuous engagement strength into a discrete outcome.

    `strength >= positive_threshold` → positive; `<= negative_threshold` → negative;
    anything in between → neutral (ignored by the user model).
    """
    if strength >= cfg.positive_threshold:
        return Outcome.positive
    if strength <= cfg.negative_threshold:
        return Outcome.negative
    return Outcome.neutral

signal_builder

signal_builder

Pure construction of the USER MODEL (UserSignals) from events + content structure.

events (+ content tags/vectors) -> UserSignals. No IO: the caller fetches content and vectors and passes them in. now is passed in too, so the function is fully deterministic and testable.

ViewAggregate dataclass 

ViewAggregate(content_id: str, dwell_seconds: Optional[float] = None, visits: int = 0, end_reason: Optional[EndReason] = None, last_ts: Optional[datetime] = None, survey_rating: Optional[float] = None)

All views of one content folded together.

aggregate_views 

aggregate_views(events: Sequence[InteractionEvent]) -> dict[str, ViewAggregate]

Group events by content_id and pair start/end into dwell.

Robust to the online case (start and end arrive as separate webhook events) and to sources that already carry dwell_seconds on the end event.

Source code in ai-engine/src/ai_engine/recsys/signals/signal_builder.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def aggregate_views(events: Sequence[InteractionEvent]) -> dict[str, ViewAggregate]:
    """Group events by content_id and pair start/end into dwell.

    Robust to the online case (start and end arrive as separate webhook events) and to
    sources that already carry dwell_seconds on the end event.
    """
    by_content: dict[str, list[InteractionEvent]] = {}
    for e in events:
        if e.content_id is None:
            continue
        by_content.setdefault(e.content_id, []).append(e)

    out: dict[str, ViewAggregate] = {}
    for cid, evs in by_content.items():
        agg = ViewAggregate(content_id=cid)
        starts = [e for e in evs if e.event == _VIEW_START]
        ends = [e for e in evs if e.event == _VIEW_END]
        agg.visits = max(len(starts), 1)

        explicit = [e.dwell_seconds for e in evs if e.dwell_seconds is not None]
        if explicit:
            agg.dwell_seconds = max(explicit)
        elif starts and ends:
            span = max(e.ts for e in ends) - min(e.ts for e in starts)
            agg.dwell_seconds = max(span.total_seconds(), 0.0)

        if ends:
            last_end = max(ends, key=lambda e: e.ts)
            agg.end_reason = last_end.end_reason

        agg.last_ts = max(e.ts for e in evs)

        rating_evs = [
            e for e in evs
            if isinstance(e.survey_answers, dict) and "rating" in e.survey_answers
        ]
        if rating_evs:                       # most-recent rating wins (by ts, not list order)
            latest_rating = max(rating_evs, key=lambda e: e.ts)
            agg.survey_rating = float(latest_rating.survey_answers["rating"])

        out[cid] = agg
    return out

build_user_signals 

build_user_signals(*, user_id: str, events: Sequence[InteractionEvent], contents: dict[str, Content], vectors: dict[str, Vector], now: datetime, cfg: RecConfig, demographics: Optional[dict] = None) -> UserSignals

Fold events + content structure into the user model.

Source code in ai-engine/src/ai_engine/recsys/signals/signal_builder.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def build_user_signals(
    *,
    user_id: str,
    events: Sequence[InteractionEvent],
    contents: dict[str, Content],
    vectors: dict[str, Vector],
    now: datetime,
    cfg: RecConfig,
    demographics: Optional[dict] = None,
) -> UserSignals:
    """Fold events + content structure into the user model."""
    aggs = aggregate_views(events)

    positives: dict[str, float] = {}
    negatives: dict[str, float] = {}
    eng_affinity: dict[str, float] = {}      # tag affinity from BROWSING (engaged content)
    survey_affinity: dict[str, float] = {}   # tag affinity from the PRESURVEY (stated + demographics)
    tag_aversion: dict[str, float] = {}

    engaged_ids = set(aggs.keys())
    dwell_ratios: list[float] = []
    completions = revisits = 0

    for cid, agg in aggs.items():
        content = contents.get(cid)
        est = estimate_reading_time(
            content.word_count if content else 0,
            content.has_image if content else False,
            cfg,
        )
        strength = engagement_strength(
            dwell_seconds=agg.dwell_seconds,
            est_reading_time=est,
            end_reason=agg.end_reason,
            visits=agg.visits,
            survey_rating=agg.survey_rating,
            cfg=cfg,
        )
        outcome = classify_outcome(strength, cfg)
        decay = _decay(agg.last_ts, now, cfg.half_life_days)

        dwell_ratios.append(_dwell_ratio(agg.dwell_seconds, est, cfg))
        if agg.end_reason == EndReason.next_button:
            completions += 1
        if agg.visits > 1:
            revisits += 1

        if outcome == Outcome.positive:
            positives[cid] = max(strength, 0.0) * decay
            if content:
                for tag in content.tags:
                    if tag.facet == "place_where.camp_areas":
                        continue    # AR areas are a proximity filter, never a taste signal
                    eng_affinity[tag.key] = eng_affinity.get(tag.key, 0.0) + positives[cid] * tag.weight
        elif outcome == Outcome.negative:
            negatives[cid] = abs(strength) * decay
            if content:                       # the THEMES of disliked content -> aversion
                for tag in content.tags:
                    if tag.facet == "place_where.camp_areas":
                        continue
                    tag_aversion[tag.key] = tag_aversion.get(tag.key, 0.0) + negatives[cid] * tag.weight

    # soft negatives: shown in an impression set but never engaged
    for e in events:
        for imp in e.impressions:
            if imp not in engaged_ids and imp not in positives:
                pen = cfg.soft_negative_weight * _decay(e.ts, now, cfg.half_life_days)
                negatives[imp] = max(negatives.get(imp, 0.0), pen)

    # survey + identify events -> demographics + person_who/persona affinity
    from ..survey import DEMOGRAPHIC_EVENTS, survey_affinity as _survey_answers_affinity, extract_demographics
    survey_demo: dict = {}
    merged_answers: dict = {}
    for e in events:                     # chronological: a re-answer REPLACES the old one
        if e.event in DEMOGRAPHIC_EVENTS and e.survey_answers:
            merged_answers.update(e.survey_answers)
            survey_demo.update(extract_demographics(e.survey_answers))
    # affinity from the MERGED latest answers, not summed per event: an early
    # placeholder identify (age child, nationality unknown) must not survive the
    # real survey, and answering twice must not double a weight
    if merged_answers:
        for key, w in _survey_answers_affinity(merged_answers).items():
            survey_affinity[key] = w

    # explicit demographic affinity (cold-start seed; person_who facets)
    demographics = {**survey_demo, **(demographics or {})}
    if demographics:
        for key, w in _demographic_affinity(demographics).items():
            survey_affinity[key] = survey_affinity.get(key, 0.0) + w

    # taste vector = weighted centroid of positively-engaged content vectors
    taste_vector: Optional[list[float]] = None
    acc: Optional[list[float]] = None
    for cid, w in positives.items():
        v = vectors.get(cid)
        if not v:
            continue
        if acc is None:
            acc = [0.0] * len(v)
        for i, x in enumerate(v):
            acc[i] += w * x
    if acc is not None and any(acc):
        taste_vector = _normalize_unit(acc)

    # Blend survey (stated) and engagement (browsed) tag affinities into one signal.
    # Each side is canonicalized (lowercased, to merge case variants) and max-normalized
    # to [0,1] SEPARATELY, so a handful of survey seeds are never washed out by large
    # engagement magnitudes. Engagement then ramps in by cold-start trust:
    #     eng_trust = n_positive / (n_positive + tag_engagement_trust_k)
    # On cold start (n_positive == 0) eng_trust == 0, so survey/demographic tags fully
    # drive recommendations; as the visitor browses, engagement grows to compete. Survey
    # stays at full scale throughout. Result is still ONE tag_affinity dict for score_tag.
    def _fold_norm(d: dict[str, float]) -> dict[str, float]:
        folded: dict[str, float] = {}
        for k, v in d.items():
            folded[k.lower()] = folded.get(k.lower(), 0.0) + v
        mx = max(folded.values(), default=0.0)
        return {k: v / mx for k, v in folded.items()} if mx > 0 else folded

    survey_norm = _fold_norm(survey_affinity)
    eng_norm = _fold_norm(eng_affinity)
    n_pos = len(positives)
    eng_trust = (
        n_pos / (n_pos + cfg.tag_engagement_trust_k)
        if cfg.tag_engagement_trust_k > 0 else 1.0
    )

    tag_affinity: dict[str, float] = dict(survey_norm)
    for k, v in eng_norm.items():
        tag_affinity[k] = tag_affinity.get(k, 0.0) + eng_trust * v
    # renormalize the blend to [0, 1] so score_tag's contract holds
    mxb = max(tag_affinity.values(), default=0.0)
    if mxb > 0:
        tag_affinity = {k: v / mxb for k, v in tag_affinity.items()}

    # same fold + normalize for aversion (negatively-engaged themes)
    folded_av: dict[str, float] = {}
    for k, v in tag_aversion.items():
        folded_av[k.lower()] = folded_av.get(k.lower(), 0.0) + v
    tag_aversion = folded_av
    if tag_aversion:
        mxa = max(tag_aversion.values())
        if mxa > 0:
            tag_aversion = {k: v / mxa for k, v in tag_aversion.items()}

    # sequence: order viewed content by most-recent interaction first
    ordered = sorted(aggs.items(), key=lambda kv: (kv[1].last_ts or now), reverse=True)
    recent_views = [cid for cid, _ in ordered]
    recency_vector = vectors.get(recent_views[0]) if recent_views else None

    # engagement summary (depth / completion / pace): evidence for persona explanations
    n_views = len(aggs)
    behavior = {
        "n_views": n_views,
        "n_positive": len(positives),
        "n_negative": len(negatives),
        "avg_dwell_ratio": round(sum(dwell_ratios) / n_views, 4) if n_views else 0.0,
        "completion_rate": round(completions / n_views, 4) if n_views else 0.0,
        "revisit_rate": round(revisits / n_views, 4) if n_views else 0.0,
        "depth": round(len(positives) / n_views, 4) if n_views else 0.0,
    }

    return UserSignals(
        user_id=user_id,
        positives=positives,
        negatives=negatives,
        viewed=sorted(aggs.keys()),          # full view history (any outcome) for dedup
        recent_views=recent_views,           # sequence awareness
        tag_affinity=tag_affinity,
        tag_aversion=tag_aversion,
        taste_vector=taste_vector,
        recency_vector=recency_vector,
        behavior=behavior,
        demographics=demographics or {},
    )

Survey

survey

survey

Survey -> user-model mapping (the persona).

Survey answers become tag-affinity in the SAME taxonomy the content is tagged with, so the recommender's score_tag does persona<->content tag similarity. Supports the kwb research survey (survey:kwb:survey: q:age/q:gender/q:nationality + q:personalization_*) and the demographic onboarding quiz (age_group/gender/nationality).

Personalization questions are explicit preferences: their answer VALUE must be the canonical taxonomy label (e.g. "Forced Labor") so it matches content tags 1:1.

canon_demo_value 

canon_demo_value(field: str, value) -> Optional[str]

One canonical token per answer meaning, or None for placeholder junk. Empty/junk gender collapses to 'no_answer' (a real survey outcome); junk in any other field is dropped from distributions entirely.

Source code in ai-engine/src/ai_engine/recsys/survey.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def canon_demo_value(field: str, value) -> Optional[str]:
    """One canonical token per answer meaning, or None for placeholder junk.
    Empty/junk gender collapses to 'no_answer' (a real survey outcome); junk in
    any other field is dropped from distributions entirely."""
    v = str(value or "").strip().lower().rstrip(".")
    v = v.replace("'", "").replace("’", "")
    v = _re.sub(r"[\s\-.]+", "_", v).strip("_") if v else ""
    # values with no letter/digit at all (zero-width chars, dashes, emoji …) render
    # as blank labels: treat them exactly like an empty answer
    if not v or v in _DEMO_JUNK or not _re.search(r"[a-z0-9]", v):
        return "no_answer" if field == "gender" else None
    out = _DEMO_CANON.get(field, {}).get(v, v)
    if field == "gender" and out not in _GENDER_CANONICAL:
        return "other"
    return out

demo_label 

demo_label(field: str, value) -> str

Human label for a canonical demographic value. The server owns both the semantics (canon_demo_value) and the wording, so the dashboard renders labels verbatim instead of re-deriving them (docs/debt-payload-scatter.md D3).

Source code in ai-engine/src/ai_engine/recsys/survey.py 
180
181
182
183
184
185
186
187
188
189
def demo_label(field: str, value) -> str:
    """Human label for a canonical demographic value. The server owns both the
    semantics (canon_demo_value) and the wording, so the dashboard renders labels
    verbatim instead of re-deriving them (docs/debt-payload-scatter.md D3)."""
    v = str(value or "")
    if field == "age":
        if v.startswith("under_"):
            return "<" + "".join(ch for ch in v if ch.isdigit())
        return v.replace("_plus", "+").replace("_", "–")
    return v.replace("_", " ").strip().title()

extract_demographics 

extract_demographics(answers: dict) -> dict

Pull the demographic fields out of raw survey answers into a flat dict.

Reads the five demographic fields (age, gender, nationality, province, personal_connection) from a survey or identify payload and returns them as plain {field: value}. Each field can arrive under several question ids or trait keys (for example age as q:age, age_group, or age); the first id that is present wins, and for a multi-select answer the first value is taken.

The values are kept raw (for example "55_64", not the taxonomy label "age 55-64"). This function is only for storing and inspecting who the visitor is (it populates UserSignals.demographics). Turning demographics into weighted taxonomy tags for matching is a separate step, survey_affinity / _demographic_affinity.

Parameters:

Name Type Description Default
answers dict

question_id -> answer from a survey/identify event.

 required

Returns:

Type Description
dict

{field: value} containing only the demographic fields that were answered
dict

(empty dict if none are present).
Example 
extract_demographics({
    "q:age": "55_64",
    "q:gender": "female",
    "nationality": "france",       # from an identify trait
})
# {"age": "55_64", "gender": "female", "nationality": "france"}
Source code in ai-engine/src/ai_engine/recsys/survey.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def extract_demographics(answers: dict) -> dict:
    """Pull the demographic fields out of raw survey answers into a flat dict.

    Reads the five demographic fields (age, gender, nationality, province,
    personal_connection) from a survey or identify payload and returns them as plain
    `{field: value}`. Each field can arrive under several question ids or trait keys (for
    example age as `q:age`, `age_group`, or `age`); the first id that is present wins, and
    for a multi-select answer the first value is taken.

    The values are kept **raw** (for example `"55_64"`, not the taxonomy label
    `"age 55-64"`). This function is only for storing and inspecting who the visitor is
    (it populates `UserSignals.demographics`). Turning demographics into weighted taxonomy
    tags for matching is a separate step, `survey_affinity` / `_demographic_affinity`.

    Args:
        answers: `question_id -> answer` from a survey/identify event.

    Returns:
        `{field: value}` containing only the demographic fields that were answered
        (empty dict if none are present).

    Example:
        ```python
        extract_demographics({
            "q:age": "55_64",
            "q:gender": "female",
            "nationality": "france",       # from an identify trait
        })
        # {"age": "55_64", "gender": "female", "nationality": "france"}
        ```
    """
    out = {}
    for field, qids in (("age", _AGE_QIDS), ("gender", _GENDER_QIDS),
                        ("nationality", _NAT_QIDS), ("province", _PROVINCE_QIDS),
                        ("personal_connection", _CONN_QIDS)):
        vals = _vals(answers, *qids)
        if vals:
            out[field] = vals[0]
    # email: PII-guarded to a pure yes/no flag — did the visitor leave an address?
    # The address itself NEVER enters the stored demographics.
    for k, v in answers.items():
        if "email" in str(k).lower():
            sv = str(_clean(v) or "").strip().lower()
            if sv in ("", "no", "nee", "false", "none", "null"):
                out["email_shared"] = "no"
            elif "@" in sv or sv in ("yes", "ja", "true"):
                out["email_shared"] = "yes"
            else:
                out["email_shared"] = "no"
            break
    return out

split_survey_answers 

split_survey_answers(answers: dict) -> dict

Group raw survey answers for the holistic visitor profile.

The app sends every question TWICE: a plain key with the human display text ("knowledge_level": "Oneens") and a q:-prefixed key with the canonical code ("q:knowledge_level": "2"). Both used to render as separate duplicate rows; now each base question appears once, preferring the human display value. Email answers are dropped entirely (PII — the email_shared flag covers them), and zero-width characters are stripped from values.

Source code in ai-engine/src/ai_engine/recsys/survey.py 
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
def split_survey_answers(answers: dict) -> dict:
    """Group raw survey answers for the holistic visitor profile.

    The app sends every question TWICE: a plain key with the human display text
    ("knowledge_level": "Oneens") and a q:-prefixed key with the canonical code
    ("q:knowledge_level": "2"). Both used to render as separate duplicate rows;
    now each base question appears once, preferring the human display value.
    Email answers are dropped entirely (PII — the email_shared flag covers them),
    and zero-width characters are stripped from values."""
    demo_qids = set(_AGE_QIDS) | set(_GENDER_QIDS) | set(_NAT_QIDS) | set(_PROVINCE_QIDS) | set(_CONN_QIDS)
    pers_qids = set(_PERSONALIZATION)

    def cleanv(v):
        if isinstance(v, list):
            return [cleanv(x) for x in v]
        return str(v).replace("​", "").strip()

    # dedup q:/plain pairs onto the base name. The q:-prefixed key carries the
    # localized display text ("q:prior_visit": "Nee."), the plain key the
    # canonical code ("prior_visit": "no"). The dashboard is English, so the
    # CANONICAL value wins and is mapped to an English label; localized free
    # text (open questions) has no code counterpart and passes through.
    merged: dict = {}
    for k, v in (answers or {}).items():
        if v is None or v == "":
            continue
        is_q = str(k).startswith("q:")
        base = k[2:] if is_q else str(k)
        if "email" in base.lower():
            continue
        if not is_q or base not in merged:
            cv = cleanv(v)
            if isinstance(cv, list):
                cv = [x for x in cv if x != ""]
            if cv == "" or cv == []:      # answer was only whitespace / zero-width junk
                continue
            merged[base] = cv

    demographic: dict = {}
    personalization: dict = {}
    background: dict = {}
    feedback: dict = {}
    for base, v in merged.items():
        qids = {base, "q:" + base}
        if qids & demo_qids:
            demographic[base] = v
        elif qids & pers_qids or base.startswith("personalization_"):
            personalization[base] = v
        elif base in _FEEDBACK_BASE:
            feedback[base] = _answer_label(base, v)
        else:
            background[base] = _answer_label(base, v)
    return {"demographic": demographic, "personalization": personalization,
            "background": background, "feedback": feedback,
            "other": {**background, **feedback}}   # legacy shape for old clients

survey_affinity 

survey_affinity(answers: dict) -> dict[str, float]

Turn raw survey answers into weighted taxonomy tags (the visitor's persona).

Emits {"facet:label": weight} keyed in the SAME taxonomy the content is tagged with, so score_tag can match persona against content directly. Two kinds of answer contribute, at deliberately different weights:

  • Demographics (age, gender, nationality, NL province) map to person_who.* facets at modest weights: age 0.5, gender 0.3, nationality 0.4, province 0.5. A "core country" is an origin the collection tags content for specifically (CORE_COUNTRIES = Netherlands, Germany, Poland); a nationality outside that set has no country tag of its own, so it also emits the International rollup (0.3) and matches content tagged for that non-core complement.
  • Personalization preferences (theme / interest / area) are what the visitor explicitly picked, so they get the strongest weight (1.0). The answer value IS the taxonomy label, run through _canonical_label so spelling/separator variants (e.g. "forced labour") still line up with the content label ("Forced Labor").

Multi-select answers emit one key per selected value. These land in the survey side of tag_affinity; build_user_signals blends them with engagement so they dominate on cold start (see ai_engine.recsys.signals.signal_builder.build_user_signals).

Parameters:

Name Type Description Default
answers dict

question_id -> answer from a survey/identify event (values may be scalars or lists; entity-id values like a:age:55_64 are cleaned first).

 required

Returns:

Type Description
dict[str, float]

{"facet:label": weight}. Empty if no recognized questions are present.
Example 
survey_affinity({
    "q:age": "55_64",
    "q:gender": "female",
    "q:nationality": "france",              # not a core country
    "q:personalization_theme": "forced labour",
})
# {
#     'person_who.age_group:age 55-64': 0.5,
#     'person_who.gender_and_age:female': 0.3,
#     'person_who.city_village_country:From: France': 0.4,
#     'person_who.city_village_country:International': 0.3,
#     'theme_what:Forced Labor': 1.0,
# }
Source code in ai-engine/src/ai_engine/recsys/survey.py 
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
def survey_affinity(answers: dict) -> dict[str, float]:
    """Turn raw survey answers into weighted taxonomy tags (the visitor's persona).

    Emits `{"facet:label": weight}` keyed in the SAME taxonomy the content is tagged
    with, so `score_tag` can match persona against content directly. Two kinds of answer
    contribute, at deliberately different weights:

    - **Demographics** (age, gender, nationality, NL province) map to `person_who.*`
      facets at modest weights: age 0.5, gender 0.3, nationality 0.4, province 0.5. A
      "core country" is an origin the collection tags content for specifically
      (`CORE_COUNTRIES` = Netherlands, Germany, Poland); a nationality outside that set
      has no country tag of its own, so it also emits the `International` rollup (0.3) and
      matches content tagged for that non-core complement.
    - **Personalization preferences** (theme / interest / area) are what the visitor
      explicitly picked, so they get the strongest weight (1.0). The answer value IS the
      taxonomy label, run through `_canonical_label` so spelling/separator variants
      (e.g. "forced labour") still line up with the content label ("Forced Labor").

    Multi-select answers emit one key per selected value. These land in the survey side
    of `tag_affinity`; `build_user_signals` blends them with engagement so they dominate
    on cold start (see `ai_engine.recsys.signals.signal_builder.build_user_signals`).

    Args:
        answers: `question_id -> answer` from a survey/identify event (values may be
            scalars or lists; entity-id values like `a:age:55_64` are cleaned first).

    Returns:
        `{"facet:label": weight}`. Empty if no recognized questions are present.

    Example:
        ```python
        survey_affinity({
            "q:age": "55_64",
            "q:gender": "female",
            "q:nationality": "france",              # not a core country
            "q:personalization_theme": "forced labour",
        })
        # {
        #     'person_who.age_group:age 55-64': 0.5,
        #     'person_who.gender_and_age:female': 0.3,
        #     'person_who.city_village_country:From: France': 0.4,
        #     'person_who.city_village_country:International': 0.3,
        #     'theme_what:Forced Labor': 1.0,
        # }
        ```
    """
    out: dict[str, float] = {}

    for v in _vals(answers, *_AGE_QIDS):
        if v in _AGE:
            out[f"person_who.age_group:{_AGE[v]}"] = 0.5
    for v in _vals(answers, *_GENDER_QIDS):
        if v in _GENDER:
            out[f"person_who.gender_and_age:{_GENDER[v]}"] = 0.3
    for v in _vals(answers, *_NAT_QIDS):
        raw = str(v).strip().casefold().replace(" ", "_")
        # answers arrive as demonyms ("dutch"); the core-country set and the
        # content's From: tags use country nouns - map before comparing, or a
        # Dutch visitor gets the International rollup (and misses NL content)
        country = _DEMONYM_COUNTRY.get(raw, str(v).replace("_", " ").title())
        out[f"person_who.city_village_country:From: {country}"] = 0.4
        if country.strip().casefold() not in CORE_COUNTRIES:
            out["person_who.city_village_country:International"] = 0.3
    # visitor's NL province -> person_who.province_netherlands, the facet the content
    # province tags use per the authoritative taxonomy (tags.json). score_tag boosts
    # same-province stories. Keep the label as-is (hyphens matter: "Zuid-Holland").
    for v in _vals(answers, *_PROVINCE_QIDS):
        if v:
            out[f"person_who.province_netherlands:{str(v).strip()}"] = 0.5

    # explicit preference questions: value IS the taxonomy label -> strong weight.
    # canonicalize the value so slug/underscore/spelling variants still match content.
    for qid, facet in _PERSONALIZATION.items():
        if facet == "place_where.camp_areas":
            continue    # AR areas are a proximity FILTER, not a taste signal:
                        # every recommendation is already within the site
        for v in _vals(answers, qid):
            if not v:
                continue
            if str(v).strip().isdigit():
                continue    # bare content ids (interest picks) are not taxonomy labels
            out[f"{facet}:{_canonical_label(v)}"] = 1.0

    return out

Ranking

scorers

scorers

Pure scorers. CONTRACT: every scorer returns a value in [0, 1].

That contract is what makes the weighted sum in fusion valid without rescaling.

cosine 

cosine(a: Optional[Vector], b: Optional[Vector]) -> float

Cosine similarity in [-1, 1]; 0 if either side is missing/zero.

Source code in ai-engine/src/ai_engine/recsys/ranking/scorers.py 
13
14
15
16
17
18
19
20
21
22
def cosine(a: Optional[Vector], b: Optional[Vector]) -> float:
    """Cosine similarity in [-1, 1]; 0 if either side is missing/zero."""
    if not a or not b or len(a) != len(b):
        return 0.0
    dot = sum(x * y for x, y in zip(a, b))
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(y * y for y in b))
    if na == 0 or nb == 0:
        return 0.0
    return dot / (na * nb)

score_semantic 

score_semantic(signals: UserSignals, candidate_vector: Optional[Vector]) -> float

How close the candidate is to the user's taste vector. -> [0, 1].

Source code in ai-engine/src/ai_engine/recsys/ranking/scorers.py 
25
26
27
28
29
def score_semantic(signals: UserSignals, candidate_vector: Optional[Vector]) -> float:
    """How close the candidate is to the user's taste vector. -> [0, 1]."""
    if signals.taste_vector is None or candidate_vector is None:
        return 0.0
    return (cosine(signals.taste_vector, candidate_vector) + 1.0) / 2.0

haversine_m 

haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float

Great-circle distance between two lat/lon points, in metres.

Source code in ai-engine/src/ai_engine/recsys/ranking/scorers.py 
32
33
34
35
36
37
38
39
def haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    """Great-circle distance between two lat/lon points, in metres."""
    r = 6371000.0
    p1, p2 = math.radians(lat1), math.radians(lat2)
    dphi = math.radians(lat2 - lat1)
    dlambda = math.radians(lon2 - lon1)
    a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlambda / 2) ** 2
    return 2 * r * math.asin(min(1.0, math.sqrt(a)))

score_geo 

score_geo(content: Optional[Content], ref: Optional[tuple], scale_m: float) -> float

Proximity of the candidate to a reference point (the user's CURRENT location, a per-request signal, NOT part of the stored user model). exp(-distance/scale). -> [0,1]; 0 if either side lacks coordinates. Independent of the tag system.

Source code in ai-engine/src/ai_engine/recsys/ranking/scorers.py 
42
43
44
45
46
47
48
49
50
51
def score_geo(content: Optional[Content], ref: Optional[tuple], scale_m: float) -> float:
    """Proximity of the candidate to a reference point (the user's CURRENT location,
    a per-request signal, NOT part of the stored user model). exp(-distance/scale).
    -> [0,1]; 0 if either side lacks coordinates. Independent of the tag system."""
    if content is None or ref is None or scale_m <= 0:
        return 0.0
    if content.lat is None or content.lon is None:
        return 0.0
    d = haversine_m(ref[0], ref[1], content.lat, content.lon)
    return math.exp(-d / scale_m)

score_recency 

score_recency(signals: UserSignals, candidate_vector: Optional[Vector]) -> float

Sequence awareness: closeness to the user's MOST-RECENT view (vs the whole-history taste vector). Boosts 'more like what you just read'. -> [0, 1].

Source code in ai-engine/src/ai_engine/recsys/ranking/scorers.py 
54
55
56
57
58
59
def score_recency(signals: UserSignals, candidate_vector: Optional[Vector]) -> float:
    """Sequence awareness: closeness to the user's MOST-RECENT view (vs the whole-history
    taste vector). Boosts 'more like what you just read'. -> [0, 1]."""
    if signals.recency_vector is None or candidate_vector is None:
        return 0.0
    return (cosine(signals.recency_vector, candidate_vector) + 1.0) / 2.0

score_tag 

score_tag(signals: UserSignals, content: Optional[Content], topk: int = 6) -> float

How well a candidate item matches the visitor's tag interests, in [0, 1].

The visitor's persona is a set of weighted tags (signals.tag_affinity, each tag to a weight in [0, 1]); the content carries its own weighted tags. For every tag the visitor has that the content also has, this adds visitor_weight * content_weight. That sum is then divided by the total weight of the visitor's topk strongest interests, giving a score where roughly 1.0 means the content covers the visitor's top interests and 0.0 means it shares none of their tags.

score = sum over shared tags of (visitor_weight * content_weight)
        / (sum of the visitor's topk largest weights)

Dividing by only the strongest topk interests (not every tag the visitor has) stops a long tail of faint browsing tags from shrinking the score: a visitor with one dominant survey interest plus many weak ones still scores well on content that hits the dominant one, instead of having it averaged away.

Parameters:

Name Type Description Default
signals UserSignals

The visitor model; only tag_affinity is used.

 required
content Optional[Content]

The candidate item (its tags). Returns 0.0 if it is None, or if the visitor has no tag affinity yet.

 required
topk int

How many of the visitor's strongest interests to normalize by.

 6

Returns:

Type Description
float

A match score in [0, 1].
Example 
# Visitor persona (content tag weights are all 1.0 here):
#   theme_what:forced labor                    1.0
#   theme_how.type_of_stores:personal stories  0.7
#   person_who.age_group:age 55-64             0.5
# topk = 6, so the denominator is 1.0 + 0.7 + 0.5 = 2.2

# Story A, tagged {forced labor, personal stories}:
#   matched = 1.0*1.0 + 0.7*1.0 = 1.7   ->  score = 1.7 / 2.2 = 0.77

# Story C, tagged {daily life} (none of the visitor's interests):
#   matched = 0                          ->  score = 0.0
Source code in ai-engine/src/ai_engine/recsys/ranking/scorers.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def score_tag(signals: UserSignals, content: Optional[Content], topk: int = 6) -> float:
    """How well a candidate item matches the visitor's tag interests, in [0, 1].

    The visitor's persona is a set of weighted tags (`signals.tag_affinity`, each tag to a
    weight in [0, 1]); the content carries its own weighted tags. For every tag the visitor
    has that the content also has, this adds `visitor_weight * content_weight`. That sum is
    then divided by the total weight of the visitor's `topk` strongest interests, giving a
    score where roughly 1.0 means the content covers the visitor's top interests and 0.0
    means it shares none of their tags.

        score = sum over shared tags of (visitor_weight * content_weight)
                / (sum of the visitor's topk largest weights)

    Dividing by only the strongest `topk` interests (not every tag the visitor has) stops a
    long tail of faint browsing tags from shrinking the score: a visitor with one dominant
    survey interest plus many weak ones still scores well on content that hits the dominant
    one, instead of having it averaged away.

    Args:
        signals: The visitor model; only `tag_affinity` is used.
        content: The candidate item (its `tags`). Returns 0.0 if it is None, or if the
            visitor has no tag affinity yet.
        topk: How many of the visitor's strongest interests to normalize by.

    Returns:
        A match score in [0, 1].

    Example:
        ```python
        # Visitor persona (content tag weights are all 1.0 here):
        #   theme_what:forced labor                    1.0
        #   theme_how.type_of_stores:personal stories  0.7
        #   person_who.age_group:age 55-64             0.5
        # topk = 6, so the denominator is 1.0 + 0.7 + 0.5 = 2.2

        # Story A, tagged {forced labor, personal stories}:
        #   matched = 1.0*1.0 + 0.7*1.0 = 1.7   ->  score = 1.7 / 2.2 = 0.77

        # Story C, tagged {daily life} (none of the visitor's interests):
        #   matched = 0                          ->  score = 0.0
        ```
    """
    if content is None or not signals.tag_affinity:
        return 0.0
    # match on canonical form: survey-derived keys and content keys must agree
    # despite casing/whitespace/accents/separators/typos. normalize_key is the
    # single source of truth, applied symmetrically to both sides.
    cand_weights = {normalize_key(t.key): t.weight for t in content.tags}
    top = sorted(signals.tag_affinity.values(), reverse=True)[:max(topk, 1)]
    total = sum(top)
    if total <= 0:
        return 0.0
    matched = sum(
        aff * cand_weights.get(normalize_key(key), 0.0)
        for key, aff in signals.tag_affinity.items()
    )
    return max(0.0, min(matched / total, 1.0))

score_aversion 

score_aversion(signals: UserSignals, content: Optional[Content]) -> float

Overlap between the candidate's tags and themes the user DISLIKED. -> [0, 1]. Mirrors score_tag over tag_aversion; fused with a NEGATIVE weight so a candidate sharing themes with abandoned content is pushed down.

Source code in ai-engine/src/ai_engine/recsys/ranking/scorers.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def score_aversion(signals: UserSignals, content: Optional[Content]) -> float:
    """Overlap between the candidate's tags and themes the user DISLIKED. -> [0, 1].
    Mirrors score_tag over tag_aversion; fused with a NEGATIVE weight so a candidate
    sharing themes with abandoned content is pushed down."""
    if content is None or not signals.tag_aversion:
        return 0.0
    cand_weights = {normalize_key(t.key): t.weight for t in content.tags}
    total = sum(signals.tag_aversion.values())
    if total <= 0:
        return 0.0
    matched = sum(
        av * cand_weights.get(normalize_key(key), 0.0)
        for key, av in signals.tag_aversion.items()
    )
    return max(0.0, min(matched / total, 1.0))

fusion

fusion

Pure fusion + diversity. No IO.

weighted_fuse: combine per-scorer [0,1] scores into one fused score + keep the breakdown (explainability). mmr_rerank: greedy Maximal Marginal Relevance to avoid returning 10 near-identical stories.

weighted_fuse 

weighted_fuse(per_scorer: dict[str, float], weights: FusionWeights) -> tuple[float, dict[str, float]]

Return (fused_score, breakdown). breakdown[s] = weight[s] * score[s].

Source code in ai-engine/src/ai_engine/recsys/ranking/fusion.py 
15
16
17
18
19
20
21
22
23
24
25
def weighted_fuse(per_scorer: dict[str, float], weights: FusionWeights) -> tuple[float, dict[str, float]]:
    """Return (fused_score, breakdown). breakdown[s] = weight[s] * score[s]."""
    wmap = {
        "semantic": weights.semantic,
        "tag": weights.tag,
        "recency": weights.recency,
        "aversion": weights.aversion,
        "geo": weights.geo,
    }
    breakdown = {name: wmap.get(name, 0.0) * val for name, val in per_scorer.items()}
    return sum(breakdown.values()), breakdown

mmr_rerank 

mmr_rerank(candidates: list[ScoredCandidate], vectors: dict[str, Optional[Vector]], *, lambda_: float, limit: int) -> list[ScoredCandidate]

Greedy MMR. Relevance = candidate.final_score; diversity = cosine between candidate vectors. lambda_=1 pure relevance, lambda_=0 pure diversity.

Returns up to limit items. Stable: the first pick is always the top-relevance candidate (no selected set to penalize against yet).

Source code in ai-engine/src/ai_engine/recsys/ranking/fusion.py 
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
53
54
55
56
57
def mmr_rerank(
    candidates: list[ScoredCandidate],
    vectors: dict[str, Optional[Vector]],
    *,
    lambda_: float,
    limit: int,
) -> list[ScoredCandidate]:
    """Greedy MMR. Relevance = candidate.final_score; diversity = cosine between
    candidate vectors. lambda_=1 pure relevance, lambda_=0 pure diversity.

    Returns up to `limit` items. Stable: the first pick is always the top-relevance
    candidate (no selected set to penalize against yet).
    """
    pool = sorted(candidates, key=lambda c: c.final_score, reverse=True)
    selected: list[ScoredCandidate] = []
    while pool and len(selected) < limit:
        best_idx, best_val = 0, float("-inf")
        for i, cand in enumerate(pool):
            if not selected:
                mmr = cand.final_score
            else:
                max_sim = max(
                    cosine(vectors.get(cand.content_id), vectors.get(s.content_id))
                    for s in selected
                )
                mmr = lambda_ * cand.final_score - (1.0 - lambda_) * max_sim
            if mmr > best_val:
                best_idx, best_val = i, mmr
        selected.append(pool.pop(best_idx))
    return selected

bandit

bandit

Contextual bandit ranking policy (linear, LinUCB-style). PURE: no IO, no numpy.

Replaces the STATIC weighted fusion with a LEARNED linear reward model over the SAME per-scorer features. The hand-set FusionWeights become the bandit's PRIOR (θ0 = weights via a ridge prior), so enabling the bandit starts at EXACTLY the current behavior and learns away from it as (features -> reward) data arrives.

context  x = [semantic, tag, recency, aversion, geo]   (per candidate)
action   = recommend a candidate
reward   r = realized engagement strength of the view it produced
model    E[r | x] = θ·x      with UCB exploration bonus  α·√(xᵀ A⁻¹ x)

Online update (LinUCB): A += x xᵀ ; b += r x ; θ = A⁻¹ b. The reward is delayed (a view ends after we serve), so updates are applied by the OFFLINE trainer that joins the served-log (features) to the event-log (reward). d is tiny (~6), so a hand-rolled matrix inverse keeps this dependency-free.

LinearBandit 

LinearBandit(A: list[list[float]], b: list[float], *, alpha: float = 0.3, feature_order: Sequence[str] = FEATURE_ORDER, ridge: float = 1.0, n_updates: int = 0)
Source code in ai-engine/src/ai_engine/recsys/ranking/bandit.py 
62
63
64
65
66
67
68
69
70
71
def __init__(self, A: list[list[float]], b: list[float], *,
             alpha: float = 0.3, feature_order: Sequence[str] = FEATURE_ORDER,
             ridge: float = 1.0, n_updates: int = 0):
    self.A = A
    self.b = b
    self.alpha = alpha
    self.feature_order = tuple(feature_order)
    self.d = len(feature_order)
    self.ridge = ridge          # prior strength (A0 = ridge*I); lets health() report data gain
    self.n_updates = n_updates  # rewarded impressions folded in so far

with_prior classmethod 

with_prior(weights: dict, *, ridge: float = 1.0, alpha: float = 0.3, feature_order: Sequence[str] = FEATURE_ORDER) -> 'LinearBandit'

Ridge prior centred on the static fusion weights: A0 = ridgeI, b0 = ridgew => theta0 = A0^-1 b0 = w. Starts identical to weighted fusion, then learns.

Source code in ai-engine/src/ai_engine/recsys/ranking/bandit.py 
73
74
75
76
77
78
79
80
81
82
@classmethod
def with_prior(cls, weights: dict, *, ridge: float = 1.0, alpha: float = 0.3,
               feature_order: Sequence[str] = FEATURE_ORDER) -> "LinearBandit":
    """Ridge prior centred on the static fusion weights: A0 = ridge*I, b0 = ridge*w
    => theta0 = A0^-1 b0 = w. Starts identical to weighted fusion, then learns."""
    order = tuple(feature_order)
    d = len(order)
    A = [[ridge if i == j else 0.0 for j in range(d)] for i in range(d)]
    b = [ridge * float(weights.get(name, 0.0)) for name in order]
    return cls(A, b, alpha=alpha, feature_order=order, ridge=ridge)

update 

update(x: list[float], reward: float, *, weight: float = 1.0) -> None

Online LinUCB update: A += wx xT ; b += wreward*x. weight < 1 down-weights a sample (e.g. bootstrap data from another study, so live data dominates later).

Source code in ai-engine/src/ai_engine/recsys/ranking/bandit.py 
87
88
89
90
91
92
93
94
95
def update(self, x: list[float], reward: float, *, weight: float = 1.0) -> None:
    """Online LinUCB update: A += w*x xT ; b += w*reward*x. `weight` < 1 down-weights
    a sample (e.g. bootstrap data from another study, so live data dominates later)."""
    for i in range(self.d):
        self.b[i] += weight * reward * x[i]
        xi, row = weight * x[i], self.A[i]
        for j in range(self.d):
            row[j] += xi * x[j]
    self.n_updates += 1

health 

health() -> dict

Training diagnostics: how much data each weight has seen and how confident it is. std[i] = posterior std of theta_i (sqrt of A^-1 diagonal) -> shrinks with data. data[i] = A_ii - ridge = total x_i^2 mass observed -> 0 means that feature never fired.

Source code in ai-engine/src/ai_engine/recsys/ranking/bandit.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
def health(self) -> dict:
    """Training diagnostics: how much data each weight has seen and how confident it is.
    std[i] = posterior std of theta_i (sqrt of A^-1 diagonal) -> shrinks with data.
    data[i] = A_ii - ridge = total x_i^2 mass observed -> 0 means that feature never fired."""
    A_inv = _mat_inverse(self.A)
    return {
        "n_updates": self.n_updates,
        "ridge": self.ridge,
        "feature_order": list(self.feature_order),
        "std": [max(A_inv[i][i], 0.0) ** 0.5 for i in range(self.d)],
        "data": [self.A[i][i] - self.ridge for i in range(self.d)],
    }

rank_scores 

rank_scores(feats: dict[str, list[float]], *, explore: bool = True) -> dict[str, float]

Score each candidate by θ·x (+ UCB bonus). Inverts A once for the batch.

Source code in ai-engine/src/ai_engine/recsys/ranking/bandit.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
def rank_scores(self, feats: dict[str, list[float]], *, explore: bool = True) -> dict[str, float]:
    """Score each candidate by θ·x (+ UCB bonus). Inverts A once for the batch."""
    A_inv = _mat_inverse(self.A)
    th = _matvec(A_inv, self.b)
    out: dict[str, float] = {}
    for cid, x in feats.items():
        mean = _dot(th, x)
        if explore and self.alpha > 0:
            var = max(_dot(x, _matvec(A_inv, x)), 0.0)
            out[cid] = mean + self.alpha * math.sqrt(var)
        else:
            out[cid] = mean
    return out

feature_vector 

feature_vector(per: dict, order: Sequence[str] = FEATURE_ORDER) -> list[float]

Ordered context vector from a per-scorer dict (missing scorer -> 0.0).

Source code in ai-engine/src/ai_engine/recsys/ranking/bandit.py 
26
27
28
def feature_vector(per: dict, order: Sequence[str] = FEATURE_ORDER) -> list[float]:
    """Ordered context vector from a per-scorer dict (missing scorer -> 0.0)."""
    return [float(per.get(name, 0.0)) for name in order]

Adapters

rudderstack

rudderstack

Pure RudderStack -> InteractionEvent normalizer.

No IO, no infra: takes raw RudderStack track payloads (the same shape RudderStack delivers to a webhook or writes to its warehouse) and maps them to canonical events. This is the single place source-specific shape is handled, and it is fully testable with plain dicts. A PostHog normalizer would live beside this and emit the same type.

normalize_content_id 

normalize_content_id(raw_id: Optional[str]) -> Optional[str]

'content_1234' -> '1234'; '841' -> '841'; None -> None.

Bridges the event schema's string ids to the Qdrant integer point ids.

Source code in ai-engine/src/ai_engine/recsys/adapters/rudderstack.py 
19
20
21
22
23
24
25
26
27
def normalize_content_id(raw_id: Optional[str]) -> Optional[str]:
    """'content_1234' -> '1234'; '841' -> '841'; None -> None.

    Bridges the event schema's string ids to the Qdrant integer point ids.
    """
    if raw_id is None:
        return None
    m = _DIGITS.search(str(raw_id))
    return m.group(0) if m else str(raw_id)

normalize_event 

normalize_event(raw: dict) -> Optional[InteractionEvent]

Map one RudderStack track/identify payload to an InteractionEvent (or None).

Source code in ai-engine/src/ai_engine/recsys/adapters/rudderstack.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def normalize_event(raw: dict) -> Optional[InteractionEvent]:
    """Map one RudderStack track/identify payload to an InteractionEvent (or None)."""
    if not isinstance(raw, dict):
        return None
    # identify call (app sends it AFTER the survey, traits = persona/demographics).
    # No `event` field -> route traits through the survey/demographics fold.
    if raw.get("type") == "identify" or (raw.get("traits") and not raw.get("event")):
        user_id = raw.get("userId") or raw.get("anonymousId")
        if not user_id:
            return None
        return InteractionEvent(
            user_id=str(user_id),
            event="IDENTIFY",
            ts=_parse_ts(raw.get("timestamp") or raw.get("sentAt")),
            survey_answers={k: v for k, v in (raw.get("traits") or {}).items() if v is not None},
            raw=raw,
        )

    event = raw.get("event")
    user_id = raw.get("userId") or raw.get("anonymousId")
    if not event or not user_id:
        return None

    props = raw.get("properties") or {}
    content = props.get("content") or {}
    details = props.get("details") or {}
    context = props.get("context") or {}

    # `content` / candidates may arrive as a dict ({content_id: ...}) OR as the bare
    # content_id string, RudderStack/clients send either. Accept both.
    def _cid(x):
        return x.get("content_id") if isinstance(x, dict) else x

    content_id = normalize_content_id(_cid(content))

    impressions = [
        normalize_content_id(_cid(c))
        for c in (context.get("candidates") or [])
        if _cid(c)
    ]
    impressions = [i for i in impressions if i]

    survey_answers: dict = {}
    for ans in (props.get("answers") or []):
        qid, val = ans.get("question_id"), ans.get("answer_value")
        if qid is not None and val is not None:
            if qid in survey_answers:           # multi-select -> collect into a list
                ex = survey_answers[qid]
                survey_answers[qid] = (ex if isinstance(ex, list) else [ex]) + [val]
            else:
                survey_answers[qid] = val
        if ans.get("question_type") == "rating" and val is not None:
            try:
                survey_answers["rating"] = float(val)
            except (TypeError, ValueError):
                pass

    # request_id: the app echoes the rec response's id on the resulting view, so the
    # bandit trainer can join this reward to the exact impression (its feature vector).
    request_id = (details.get("request_id") or context.get("request_id")
                  or props.get("request_id"))

    return InteractionEvent(
        user_id=str(user_id),
        event=str(event),
        ts=_parse_ts(raw.get("timestamp") or raw.get("sentAt")),
        session_id=context.get("session_id") or raw.get("sessionId"),
        request_id=request_id,
        content_id=content_id,
        dwell_seconds=details.get("dwell_seconds"),
        end_reason=_end_reason(details.get("reason")),
        query_text=details.get("query_text"),
        clicked_id=normalize_content_id(details.get("clicked_id")),
        impressions=impressions,
        survey_answers=survey_answers,
        raw=raw,
    )

normalize_events 

normalize_events(raws: Iterable[dict]) -> list[InteractionEvent]

Normalize a batch of raw RudderStack payloads into InteractionEvents.

Applies normalize_event to each, drops any that don't map to a known event (returns None), and returns the survivors sorted by timestamp (oldest first). This is the single entry the POST /api/ingest webhook calls on every request.

Note on dwell time: a CONTENT_VIEW_ENDED does NOT carry the dwell duration. The event catalog's ended-event only sends the end reason; dwell is computed later, a posteriori, by pairing the CONTENT_VIEW_STARTED and CONTENT_VIEW_ENDED timestamps (ai_engine.recsys.signals.signal_builder.aggregate_views, span = end_ts - start_ts). An explicit details.dwell_seconds is read only as a fallback if some source happens to provide it. So dwell_seconds below is None on the ended event itself.

Examples:

A track view event. Note content_5567 is stripped to 5567, the candidates become impressions (soft negatives), and request_id is lifted out so a later reward can be joined to the impression that produced it:

raw = {
    "type": "track",
    "event": "CONTENT_VIEW_ENDED",
    "userId": "u-42",
    "timestamp": "2026-06-29T12:00:00Z",
    "properties": {
        "content": {"content_id": "content_5567"},
        "details": {"reason": "next_button", "request_id": "req-abc"},
        "context": {"session_id": "s-9",
                    "candidates": ["content_5567", "content_5568"]},
    },
}

normalize_events([raw])
# [InteractionEvent(user_id="u-42", event="CONTENT_VIEW_ENDED",
#     content_id="5567", dwell_seconds=None,   # derived later from the START/END pair
#     end_reason=EndReason.next_button, request_id="req-abc", session_id="s-9",
#     impressions=["5567", "5568"], survey_answers={}, ts=datetime(...))]

An identify call carries survey/demographic traits instead of a view (no event field → routed to IDENTIFY, traits land in survey_answers):

raw = {"type": "identify", "userId": "u-42",
       "traits": {"q:age": "55_64", "q:personalization_theme": "forced labour"}}

normalize_events([raw])
# [InteractionEvent(user_id="u-42", event="IDENTIFY",
#     survey_answers={"q:age": "55_64",
#                     "q:personalization_theme": "forced labour"}, ...)]
Source code in ai-engine/src/ai_engine/recsys/adapters/rudderstack.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def normalize_events(raws: Iterable[dict]) -> list[InteractionEvent]:
    """Normalize a batch of raw RudderStack payloads into `InteractionEvent`s.

    Applies `normalize_event` to each, drops any that don't map to a known event
    (returns `None`), and returns the survivors sorted by timestamp (oldest first).
    This is the single entry the `POST /api/ingest` webhook calls on every request.

    Note on dwell time: a `CONTENT_VIEW_ENDED` does NOT carry the dwell duration. The
    event catalog's ended-event only sends the end `reason`; dwell is computed later, a
    posteriori, by pairing the `CONTENT_VIEW_STARTED` and `CONTENT_VIEW_ENDED` timestamps
    (`ai_engine.recsys.signals.signal_builder.aggregate_views`, `span = end_ts - start_ts`).
    An explicit `details.dwell_seconds` is read only as a fallback if some source happens
    to provide it. So `dwell_seconds` below is `None` on the ended event itself.

    Examples:
        A `track` view event. Note `content_5567` is stripped to `5567`, the
        `candidates` become `impressions` (soft negatives), and `request_id` is lifted
        out so a later reward can be joined to the impression that produced it:

        ```python
        raw = {
            "type": "track",
            "event": "CONTENT_VIEW_ENDED",
            "userId": "u-42",
            "timestamp": "2026-06-29T12:00:00Z",
            "properties": {
                "content": {"content_id": "content_5567"},
                "details": {"reason": "next_button", "request_id": "req-abc"},
                "context": {"session_id": "s-9",
                            "candidates": ["content_5567", "content_5568"]},
            },
        }

        normalize_events([raw])
        # [InteractionEvent(user_id="u-42", event="CONTENT_VIEW_ENDED",
        #     content_id="5567", dwell_seconds=None,   # derived later from the START/END pair
        #     end_reason=EndReason.next_button, request_id="req-abc", session_id="s-9",
        #     impressions=["5567", "5568"], survey_answers={}, ts=datetime(...))]
        ```

        An `identify` call carries survey/demographic traits instead of a view (no
        `event` field → routed to `IDENTIFY`, traits land in `survey_answers`):

        ```python
        raw = {"type": "identify", "userId": "u-42",
               "traits": {"q:age": "55_64", "q:personalization_theme": "forced labour"}}

        normalize_events([raw])
        # [InteractionEvent(user_id="u-42", event="IDENTIFY",
        #     survey_answers={"q:age": "55_64",
        #                     "q:personalization_theme": "forced labour"}, ...)]
        ```
    """
    out = [normalize_event(r) for r in raws]
    out = [e for e in out if e is not None]
    out.sort(key=lambda e: e.ts)
    return out

qdrant_store

qdrant_store

Qdrant-backed ContentStore. Requires qdrant-client (not needed for tests).

Tags live in the point payload (decided design): a tags list of {facet,label,weight} plus a flat tag_labels ("facet:label") KEYWORD-indexed field used for tag recall.

QdrantContentStore 

QdrantContentStore(client: QdrantClient, collection_name: str)
Source code in ai-engine/src/ai_engine/recsys/adapters/qdrant_store.py 
87
88
89
def __init__(self, client: QdrantClient, collection_name: str):
    self.client = client
    self.collection_name = collection_name

raw_payloads 

raw_payloads(ids: Sequence[str]) -> dict[str, dict]

Raw Qdrant payloads keyed by id (image_url / public_url / creator / time_metadata …) , the full content the recsys Content model drops. Used to open item detail cards.

Source code in ai-engine/src/ai_engine/recsys/adapters/qdrant_store.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
def raw_payloads(self, ids: Sequence[str]) -> dict[str, dict]:
    """Raw Qdrant payloads keyed by id (image_url / public_url / creator / time_metadata …)
   , the full content the recsys `Content` model drops. Used to open item detail cards."""
    if not ids:
        return {}
    try:
        res = self.client.retrieve(collection_name=self.collection_name,
                                   ids=[self._pid(i) for i in ids],
                                   with_payload=True, with_vectors=False)
    except Exception as exc:
        logger.warning("raw_payloads retrieve failed: %s", exc)
        return {}
    return {str(p.id): (p.payload or {}) for p in res}

vocab 

vocab(*, sample: int = 4000) -> dict

Distinct tag vocabulary (facet -> labels, "facet:label" -> count) for the collection. Scrolls payloads once; used by the evaluation tool to build / validate synthetic personas.

Source code in ai-engine/src/ai_engine/recsys/adapters/qdrant_store.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def vocab(self, *, sample: int = 4000) -> dict:
    """Distinct tag vocabulary (facet -> labels, "facet:label" -> count) for the collection.
    Scrolls payloads once; used by the evaluation tool to build / validate synthetic personas."""
    from collections import Counter
    counts: Counter = Counter()
    facets: dict[str, set] = {}
    try:
        points, off = [], None
        fetched = 0
        while fetched < sample:
            page, off = self.client.scroll(
                collection_name=self.collection_name, offset=off,
                limit=min(512, sample - fetched), with_payload=["tags"], with_vectors=False,
            )
            if not page:
                break
            points += page
            fetched += len(page)
            if off is None:
                break
    except Exception as exc:
        logger.warning("vocab scroll failed: %s", exc)
        return {"facets": {}, "tags": [], "counts": {}}
    for p in points:
        for t in ((p.payload or {}).get("tags") or []):
            facet, label = t.get("facet", "unknown"), t.get("label", "")
            key = f"{facet}:{label}"
            counts[key] += 1
            facets.setdefault(facet, set()).add(label)
    return {"facets": {f: sorted(ls) for f, ls in facets.items()},
            "tags": [k for k, _ in counts.most_common()],
            "counts": dict(counts)}

redis_store

redis_store

Redis-backed online stores. Requires redis (not needed for tests).

  • RedisEventBuffer: hot per-user event buffer (sorted set by ts, time-windowed). The ingestion webhook calls append; the updater reads via fetch_events.
  • RedisUserModelStore: materialized UserSignals cache (one JSON value per user).

RedisImpressionStore 

RedisImpressionStore(client: 'redis.Redis', *, ttl_seconds: int = 24 * 3600, key_prefix: str = 'imp')

request_id -> {content_id: feature_vector} (JSON), TTL'd. Feeds online bandit updates.

Source code in ai-engine/src/ai_engine/recsys/adapters/redis_store.py 
40
41
42
43
def __init__(self, client: "redis.Redis", *, ttl_seconds: int = 24 * 3600, key_prefix: str = "imp"):
    self.client = client
    self.ttl = ttl_seconds
    self.prefix = key_prefix

consume 

consume(request_id: str, content_id: str) -> None

Drop a served item's context after it has produced one reward, so a redelivered reward (retry / at-least-once webhook) can't double-count the update.

Source code in ai-engine/src/ai_engine/recsys/adapters/redis_store.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def consume(self, request_id: str, content_id: str) -> None:
    """Drop a served item's context after it has produced one reward, so a redelivered
    reward (retry / at-least-once webhook) can't double-count the update."""
    key = f"{self.prefix}:{request_id}"
    raw = self.client.get(key)
    if not raw:
        return
    d = json.loads(raw)
    if content_id in d:
        del d[content_id]
        if d:
            self.client.set(key, json.dumps(d), ex=self.ttl)
        else:
            self.client.delete(key)

RedisUserModelStore 

RedisUserModelStore(client: 'redis.Redis', *, ttl_seconds: int = 7 * 24 * 3600, key_prefix: str = 'umodel')
Source code in ai-engine/src/ai_engine/recsys/adapters/redis_store.py 
71
72
73
74
def __init__(self, client: "redis.Redis", *, ttl_seconds: int = 7 * 24 * 3600, key_prefix: str = "umodel"):
    self.client = client
    self.ttl = ttl_seconds
    self.prefix = key_prefix

iter_signals 

iter_signals() -> list[UserSignals]

All materialized user models (for cohort-wide content statistics).

Source code in ai-engine/src/ai_engine/recsys/adapters/redis_store.py 
86
87
88
89
90
91
92
93
94
95
96
def iter_signals(self) -> list[UserSignals]:
    """All materialized user models (for cohort-wide content statistics)."""
    out: list[UserSignals] = []
    for key in self.client.scan_iter(match=f"{self.prefix}:*"):
        raw = self.client.get(key)
        if raw:
            try:
                out.append(UserSignals.model_validate_json(raw))
            except Exception:
                pass
    return out

fastembed_model

fastembed_model

fastembed-backed EmbeddingModel. Requires fastembed (not needed for tests).

Orchestration

recommender

recommender

Serving side: read the user model, match it against content structure, rank.

Reads the materialized UserSignals from the UserModelStore, so a request is a fast read + candidate scoring, not a rebuild.

Recommender 

Recommender(content_store: ContentStore, model_store: UserModelStore, cfg: RecConfig, policy: Optional[LinearBandit] = None)

Read-time orchestrator: user model + content stores → a ranked Recommendation.

Generates candidates (semantic / tag / filter / geo), scores each with the pure scorers, fuses them (or overrides with the learned bandit when cfg.ranking_mode == "bandit"), diversifies with MMR, and injects a distractor. Constructor-injected stores and config keep the core testable without IO.

Source code in ai-engine/src/ai_engine/recsys/recommender.py 
50
51
52
53
54
55
56
def __init__(self, content_store: ContentStore, model_store: UserModelStore, cfg: RecConfig,
             policy: Optional[LinearBandit] = None):
    self.content_store = content_store
    self.model_store = model_store
    self.cfg = cfg
    # learned ranking policy; used only when cfg.ranking_mode == "bandit" and set.
    self.policy = policy

recommend 

recommend(user_id: str, *, filter: Optional[str] = None, near: Optional[tuple] = None, geo_radius_m: Optional[float] = None) -> Recommendation

Recommend for a stored user. Loads their UserSignals (empty persona if none yet: cold-start diverse fallback, never empty-handed), then delegates to recommend_for_signals. filter restricts to a tag (e.g. an AR location); near/geo_radius_m add an independent geo restriction.

Source code in ai-engine/src/ai_engine/recsys/recommender.py 
58
59
60
61
62
63
64
65
66
def recommend(self, user_id: str, *, filter: Optional[str] = None,
              near: Optional[tuple] = None, geo_radius_m: Optional[float] = None) -> Recommendation:
    """Recommend for a stored user. Loads their `UserSignals` (empty persona if none
    yet: cold-start diverse fallback, never empty-handed), then delegates to
    `recommend_for_signals`. `filter` restricts to a tag (e.g. an AR location);
    `near`/`geo_radius_m` add an independent geo restriction."""
    # no stored model -> empty persona -> cold-start diverse fallback (never empty-handed)
    signals = self.model_store.get_signals(user_id) or UserSignals(user_id=user_id)
    return self.recommend_for_signals(signals, filter=filter, near=near, geo_radius_m=geo_radius_m)

score_features 

score_features(signals: UserSignals, content, vec, cfg: RecConfig, near: Optional[tuple] = None) -> dict

The per-scorer feature dict for one candidate: the SINGLE source of truth for the context vector, shared by serving (Recommender) and offline replay/bootstrap.

Source code in ai-engine/src/ai_engine/recsys/recommender.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
def score_features(signals: UserSignals, content, vec, cfg: RecConfig,
                   near: Optional[tuple] = None) -> dict:
    """The per-scorer feature dict for one candidate: the SINGLE source of truth for
    the context vector, shared by serving (Recommender) and offline replay/bootstrap."""
    per = {
        "semantic": score_semantic(signals, vec),
        "tag": score_tag(signals, content, cfg.tag_match_topk),
        "recency": score_recency(signals, vec),
        "aversion": score_aversion(signals, content),
    }
    if near is not None:
        per["geo"] = score_geo(content, near, cfg.geo_scale_m)
    return per

updater

updater

Ingestion side: events -> user model -> store.

The webhook appends each RudderStack event to the EventSource buffer, then calls refresh to rebuild the materialized UserSignals and save it. Rebuild-from-buffer (rather than fragile true-incremental decay math) keeps build_user_signals as the single source of truth, while staying fast (hot buffer read + pure fold).

UserModelUpdater 

UserModelUpdater(content_store: ContentStore, model_store: UserModelStore, cfg: RecConfig)
Source code in ai-engine/src/ai_engine/recsys/updater.py 
19
20
21
22
def __init__(self, content_store: ContentStore, model_store: UserModelStore, cfg: RecConfig):
    self.content_store = content_store
    self.model_store = model_store
    self.cfg = cfg

build 

build(user_id: str, events: Sequence[InteractionEvent], *, now: datetime, demographics: Optional[dict] = None) -> UserSignals

Fold events into the user model (fetching only the content they touched).

Source code in ai-engine/src/ai_engine/recsys/updater.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def build(
    self,
    user_id: str,
    events: Sequence[InteractionEvent],
    *,
    now: datetime,
    demographics: Optional[dict] = None,
) -> UserSignals:
    """Fold events into the user model (fetching only the content they touched)."""
    engaged_ids = list(aggregate_views(events).keys())
    contents = self.content_store.get(engaged_ids) if engaged_ids else {}
    vectors = self.content_store.get_vectors(engaged_ids) if engaged_ids else {}
    return build_user_signals(
        user_id=user_id,
        events=events,
        contents=contents,
        vectors=vectors,
        now=now,
        cfg=self.cfg,
        demographics=demographics,
    )

refresh 

refresh(user_id: str, source: EventSource, *, now: datetime, demographics: Optional[dict] = None) -> UserSignals

Rebuild one visitor's user model from their buffered events and persist it.

Called once per visitor touched by an ingest batch (see ai_engine.recsys.api.ingest), right after the new events are appended to the buffer. It pulls that visitor's recent events back out of the hot buffer, rebuilds the whole UserSignals from scratch via build (rebuild-from-buffer, not incremental, so build_user_signals stays the single source of truth), saves it to the UserModelStore, and returns it. The next /api/recommend reads this freshly saved model, so recommendations reflect the just-ingested events.

Parameters:

Name Type Description Default
user_id str

The visitor whose model to rebuild.

 required
source EventSource

The event buffer to read this visitor's recent events from.

 required
now datetime

Current time, passed through for recency decay (kept as an argument so the rebuild is deterministic and testable).

 required
demographics Optional[dict]

Optional cold-start demographics (age/gender/nationality/ province) merged into the survey side of the tag affinity.

 None

Returns:

Type Description
UserSignals

The rebuilt, already-persisted UserSignals.
Example 
updater = UserModelUpdater(content_store, model_store, cfg)
buffer.append(view_event)          # a fresh CONTENT_VIEW_ENDED for u-42
signals = updater.refresh(
    "u-42", buffer, now=datetime.now(timezone.utc),
    demographics={"age": 60, "province": "Drenthe"},
)
signals.is_cold          # False once the visitor has a positive view
# False
signals.tag_affinity     # rebuilt from ALL of u-42's buffered events
# {'theme_what:forced labor': 1.0, 'person_who.province_netherlands:drenthe': 0.7}
Source code in ai-engine/src/ai_engine/recsys/updater.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def refresh(
    self,
    user_id: str,
    source: EventSource,
    *,
    now: datetime,
    demographics: Optional[dict] = None,
) -> UserSignals:
    """Rebuild one visitor's user model from their buffered events and persist it.

    Called once per visitor touched by an ingest batch (see
    `ai_engine.recsys.api.ingest`), right after the new events are appended to the
    buffer. It pulls that visitor's recent events back out of the hot buffer,
    rebuilds the whole `UserSignals` from scratch via `build` (rebuild-from-buffer,
    not incremental, so `build_user_signals` stays the single source of truth), saves
    it to the `UserModelStore`, and returns it. The next `/api/recommend` reads this
    freshly saved model, so recommendations reflect the just-ingested events.

    Args:
        user_id: The visitor whose model to rebuild.
        source: The event buffer to read this visitor's recent events from.
        now: Current time, passed through for recency decay (kept as an argument so
            the rebuild is deterministic and testable).
        demographics: Optional cold-start demographics (age/gender/nationality/
            province) merged into the survey side of the tag affinity.

    Returns:
        The rebuilt, already-persisted `UserSignals`.

    Example:
        ```python
        updater = UserModelUpdater(content_store, model_store, cfg)
        buffer.append(view_event)          # a fresh CONTENT_VIEW_ENDED for u-42
        signals = updater.refresh(
            "u-42", buffer, now=datetime.now(timezone.utc),
            demographics={"age": 60, "province": "Drenthe"},
        )
        signals.is_cold          # False once the visitor has a positive view
        # False
        signals.tag_affinity     # rebuilt from ALL of u-42's buffered events
        # {'theme_what:forced labor': 1.0, 'person_who.province_netherlands:drenthe': 0.7}
        ```
    """
    events = source.fetch_events(user_id)
    signals = self.build(user_id, events, now=now, demographics=demographics)
    self.model_store.save_signals(signals)
    return signals

composition

composition

Composition root: assemble the recsys components from environment.

If REDIS_URL / QDRANT_API_URL are set, use the real adapters; otherwise fall back to in-memory fakes (with dev fixtures) so the service runs locally with no infra. This is the ONE place IO backends are chosen, everything else takes ports.

ComponentManager 

ComponentManager(registry=None)

Builds + caches one Components per tenant, over shared Redis & Qdrant clients.

Source code in ai-engine/src/ai_engine/recsys/composition.py 
249
250
251
252
253
254
255
256
257
258
def __init__(self, registry=None):
    from .tenancy import TenantRegistry
    self.registry = registry or TenantRegistry.from_env()
    self._cache: dict = {}
    self._redis = None
    self._redis_init = False
    self._qdrant = None
    self._qdrant_init = False
    self._tenant_store = None
    self._ts_init = False

tenant_store property 

tenant_store

Runtime tenant registry, merged over the TENANTS_PATH baseline. Backing preference: durable file on the PVC (TENANT_STORE_PATH) > Redis > in-memory. File store keeps /admin-added tenants permanent (no redeploy, survives a Redis wipe).

list_tenants 

list_tenants() -> list[dict]

All tenants: config baseline + runtime-created, runtime wins. For the admin UI. API keys are NEVER returned raw, only a count, so the admin view can't leak secrets.

Source code in ai-engine/src/ai_engine/recsys/composition.py 
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
def list_tenants(self) -> list[dict]:
    """All tenants: config baseline + runtime-created, runtime wins. For the admin UI.
    API keys are NEVER returned raw, only a count, so the admin view can't leak secrets."""
    out: dict[str, dict] = {}
    for tid in self.registry.ids():
        s = self.registry.get(tid)
        out[tid] = {"tenant_id": tid, "collection": s.collection, "source": "config",
                    "bandit_state_path": s.bandit_state_path,
                    "cluster_model_path": s.cluster_model_path,
                    "api_keys_count": len(s.api_keys or [])}
    for d in self.tenant_store.all():
        row = {k: v for k, v in d.items() if k not in ("api_keys", "api_key_hashes")}
        row["api_keys_count"] = len(d.get("api_key_hashes") or []) + len(d.get("api_keys") or [])
        out[d["tenant_id"]] = {**row, "source": "runtime"}
    return sorted(out.values(), key=lambda t: t["tenant_id"])

tenant_for_key 

tenant_for_key(key: Optional[str]) -> Optional[str]

Resolve a per-tenant API key to its tenant_id (runtime store first, then baseline). Matches the key's sha256 against stored hashes (constant time); also tolerates legacy plaintext keys. Returns None for the global key / unknown keys.

Source code in ai-engine/src/ai_engine/recsys/composition.py 
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
def tenant_for_key(self, key: Optional[str]) -> Optional[str]:
    """Resolve a per-tenant API key to its tenant_id (runtime store first, then baseline).
    Matches the key's sha256 against stored hashes (constant time); also tolerates legacy
    plaintext keys. Returns None for the global key / unknown keys."""
    if not key:
        return None
    import hmac
    from .tenancy import hash_key
    h = hash_key(key)

    def _match(spec_get_hashes, spec_get_plain) -> bool:
        for kh in (spec_get_hashes or []):
            if hmac.compare_digest(h, kh):
                return True
        for k in (spec_get_plain or []):                 # legacy plaintext fallback
            if hmac.compare_digest(key, k):
                return True
        return False

    seen: set[str] = set()
    for d in self.tenant_store.all():
        seen.add(d["tenant_id"])
        if _match(d.get("api_key_hashes"), d.get("api_keys")):
            return d["tenant_id"]
    for tid in self.registry.ids():
        if tid in seen:
            continue
        s = self.registry.get(tid)
        if _match(s.api_key_hashes, s.api_keys):
            return tid
    return None

build_components_for 

build_components_for(spec, mgr) -> Components

Build a tenant-scoped Components: its own Qdrant collection + Redis key-prefix + config / bandit / cluster / event-log, sharing the manager's Redis & Qdrant clients.

Source code in ai-engine/src/ai_engine/recsys/composition.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def build_components_for(spec, mgr) -> Components:
    """Build a tenant-scoped Components: its own Qdrant collection + Redis key-prefix +
    config / bandit / cluster / event-log, sharing the manager's Redis & Qdrant clients."""
    from .adapters.config_store import deep_merge
    cfg = _build_config()
    if spec.config_overrides:
        try:
            cfg = RecConfig.model_validate(deep_merge(cfg.model_dump(), spec.config_overrides))
        except Exception:
            pass

    qc = mgr.qdrant_client
    if qc is not None:
        from .adapters.qdrant_store import QdrantContentStore
        content_store = QdrantContentStore(qc, spec.collection or os.getenv("COLLECTION_NAME", "omeka-items"))
    else:
        from .testing.fakes import FakeContentStore
        from .testing.fixtures import make_contents_and_vectors, make_payloads
        contents, vectors = make_contents_and_vectors()
        content_store = FakeContentStore(contents, vectors, payloads=make_payloads())

    rc = mgr.redis_client
    p = spec.prefix
    if rc is not None:
        from .adapters.redis_store import RedisEventBuffer, RedisUserModelStore, RedisImpressionStore
        from .adapters.config_store import RedisConfigStore
        # dashboards + serving both read these; MODEL_TTL_DAYS controls how long a
        # visitor stays known without new events (models are re-derivable via replay,
        # so a long TTL costs only a little redis memory - 7d made the cohort forget
        # everyone between museum visits)
        window_days = int(os.getenv("EVENT_WINDOW_DAYS", "30"))
        model_ttl = int(float(os.getenv("MODEL_TTL_DAYS", "90")) * 86400)
        event_buffer = RedisEventBuffer(rc, key_prefix=f"{p}:evt", window_days=window_days)
        model_store = RedisUserModelStore(rc, key_prefix=f"{p}:umodel", ttl_seconds=model_ttl)
        impressions = RedisImpressionStore(rc, key_prefix=f"{p}:imp")
        # config override: durable file on the PVC when the log volume exists
        # (survives deploys + redis wipes; operator-set values must stick),
        # redis otherwise
        log_base = os.getenv("EVENT_LOG_DIR")
        if log_base:
            from .adapters.config_store import FileConfigStore
            config_store = FileConfigStore(os.path.join(log_base, "registry", f"config-{spec.tenant_id}.json"))
            if config_store.get() is None:      # one-time migration of a live redis override
                try:
                    legacy = RedisConfigStore(rc, key=f"{p}:recsys:config").get()
                    if legacy:
                        config_store.set(legacy)
                except Exception:
                    pass
        else:
            config_store = RedisConfigStore(rc, key=f"{p}:recsys:config")
    else:
        from .testing.fakes import FakeEventSource, InMemoryUserModelStore, InMemoryImpressionStore
        event_buffer, model_store, impressions = FakeEventSource(), InMemoryUserModelStore(), InMemoryImpressionStore()
        log_base = os.getenv("EVENT_LOG_DIR")
        if log_base:                            # dev without redis: same durable file
            from .adapters.config_store import FileConfigStore
            config_store = FileConfigStore(os.path.join(log_base, "registry", f"config-{spec.tenant_id}.json"))
        else:
            from .adapters.config_store import InMemoryConfigStore
            config_store = InMemoryConfigStore()

    override = config_store.get()
    if override:
        try:
            cfg = RecConfig.model_validate(deep_merge(cfg.model_dump(), override))
        except Exception:
            pass

    base = os.getenv("EVENT_LOG_DIR")
    if base:
        from .adapters.event_log import ParquetEventLog
        event_log = ParquetEventLog(os.path.join(base, spec.tenant_id))   # per-tenant partition
    else:
        from .adapters.event_log import NullEventLog
        event_log = NullEventLog()

    return Components(
        cfg=cfg, content_store=content_store, event_buffer=event_buffer, model_store=model_store,
        updater=UserModelUpdater(content_store, model_store, cfg),
        recommender=Recommender(content_store, model_store, cfg,
                                policy=_build_policy_for(cfg, spec.bandit_state_path)),
        demographics=_build_demographics(), event_log=event_log, impressions=impressions,
        config_store=config_store, cluster_model_path=spec.cluster_model_path,
        bandit_state_path=spec.bandit_state_path,
    )

api

api

FastAPI surface for the recommendation engine.

  • POST /api/ingest : the ingest WEBHOOK. RudderStack POSTs user events here (single object or list). Normalize -> buffer -> rebuild the user model.
  • GET /api/recommend: serve recommendations for a user (reads the user model).
  • GET /api/usermodel: debug, inspect the current user model.

Mount router into the main service, or run app standalone. With no REDIS_URL / QDRANT_API_URL set it runs fully in-memory on dev fixtures.

PreviewSpec pydantic-model

Bases: BaseModel

A hand-authored user model for testing recs without going through events.

Fields:

  • tag_affinity (dict[str, float])
  • like_items (list[str])
  • demographics (dict)
  • limit (Optional[int])

EvalRun pydantic-model

Bases: BaseModel

Run a synthetic persona across scenarios (module-level so FastAPI reads it as a body).

Fields:

  • spec (PreviewSpec)
  • user_id (Optional[str])
  • scenarios (Optional[list[dict]])
  • cold (bool)

make_tenant_admin_router 

make_tenant_admin_router(manager) -> APIRouter

Control-plane: runtime tenant management (list / create / delete) WITHOUT a redeploy. A separate router so it stays out of the app-facing serving surface. INGEST_API_KEY-guarded. Note: this registers the tenant SLICE; its content must still be ingested into the tenant's Qdrant collection (content-engine) separately.

Source code in ai-engine/src/ai_engine/recsys/api.py 
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
def make_tenant_admin_router(manager) -> APIRouter:
    """Control-plane: runtime tenant management (list / create / delete) WITHOUT a redeploy.
    A separate router so it stays out of the app-facing serving surface. INGEST_API_KEY-guarded.
    Note: this registers the tenant SLICE; its content must still be ingested into the tenant's
    Qdrant collection (content-engine) separately."""
    router = APIRouter(prefix="/api/tenants", tags=["tenants"], dependencies=[Depends(_require_api_key)])

    @router.get("")
    def list_tenants() -> dict:
        items = manager.list_tenants()
        qc = getattr(manager, "qdrant_client", None)
        if qc is not None:
            for t in items:                                  # best-effort: show content count
                col = t.get("collection")
                try:
                    t["content_count"] = qc.count(col).count if col else None
                except Exception:
                    t["content_count"] = None
        return {"result": items}

    @router.post("")
    def upsert_tenant(t: TenantIn) -> dict:
        import secrets
        spec = t.model_dump(exclude={"generate_api_key"})
        generated = None
        if t.generate_api_key:
            generated = secrets.token_urlsafe(32)
            spec.setdefault("api_keys", []).append(generated)
        manager.upsert_tenant(spec)                           # hashes keys; never persists plaintext
        # never echo stored keys back; surface a freshly minted key ONCE
        safe = {k: v for k, v in spec.items() if k not in ("api_keys", "api_key_hashes")}
        resp = {"result": safe, "status": "saved"}
        if generated:
            resp["api_key"] = generated
            resp["note"] = "store this key now: only its hash is kept, it cannot be retrieved later"
        return resp

    @router.delete("/{tenant_id}")
    def delete_tenant(tenant_id: str) -> dict:
        manager.delete_tenant(tenant_id)
        return {"status": "deleted", "tenant_id": tenant_id}

    return router

create_app 

create_app(components: Optional[Components] = None) -> FastAPI

Build the FastAPI app. Pass a fixed Components for single-tenant/tests; omit it for the multi-tenant server (a ComponentManager + tenant middleware resolve a per-tenant Components from the X-Tenant-Id header). Mounts the /api router, the search router, health, and the inspector dashboard.

Source code in ai-engine/src/ai_engine/recsys/api.py 
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
def create_app(components: Optional[Components] = None) -> FastAPI:
    """Build the FastAPI app. Pass a fixed `Components` for single-tenant/tests;
    omit it for the multi-tenant server (a `ComponentManager` + tenant middleware
    resolve a per-tenant `Components` from the `X-Tenant-Id` header). Mounts the
    `/api` router, the search router, health, and the inspector dashboard."""
    from fastapi.middleware.cors import CORSMiddleware

    # Swagger docs left on in prod (operator request). Note: /docs + /openapi.json
    # expose the full API map publicly until the ingress is fronted with auth.
    app = FastAPI(title="AI-Engine Recsys")
    # browser test UIs (ui4testing) call /api/* directly; allow cross-origin in dev
    app.add_middleware(
        CORSMiddleware,
        allow_origins=os.getenv("AI_ENGINE_CORS", "*").split(","),
        allow_methods=["*"],
        allow_headers=["*"],
    )
    # request timing for the NFR-01/02 budgets. Added last of the always-on middleware,
    # so it wraps everything below it and measures what the caller experiences
    # (auth + tenant resolution + handler).
    from .latency import RECORDER, LatencyASGIMiddleware
    app.add_middleware(LatencyASGIMiddleware, recorder=RECORDER)

    # Multi-tenancy: a fixed Components (tests) serves a single tenant; otherwise build a
    # ComponentManager and resolve the tenant per request from the X-Tenant-Id header.
    from .tenancy import TenantProxy, TenantASGIMiddleware
    manager = None
    if components is not None:
        c_or_proxy = components                          # tests / single fixed tenant
    else:
        from .composition import ComponentManager
        manager = ComponentManager()
        c_or_proxy = TenantProxy(manager)
        # key_resolver -> derive tenant from a per-tenant key (trust boundary, not the raw header)
        app.add_middleware(TenantASGIMiddleware, key_resolver=manager.tenant_for_key)

    app.include_router(make_router(c_or_proxy))
    if manager is not None:
        app.include_router(make_tenant_admin_router(manager))

    # Qdrant search surface (ported from legacy ai-engine-api service.py).
    # Guarded so the recsys API still boots if the search stack (embedding model
    # download / Qdrant connectivity) fails at startup.
    try:
        from ai_engine.search.api import make_search_router
        app.include_router(make_search_router())
        logger.info("Search router mounted (/api/search*)")
    except Exception as e:  # noqa: BLE001
        logger.warning(f"Search router NOT mounted: {e}")

    @app.get("/health", tags=["Health"])
    def health() -> dict:
        # APP_VERSION is set from the deployed image tag via the Flux imagepolicy
        # setter (see deployment.yaml), so this reports the running release.
        # `latency` is the live NFR-01/NFR-02 verdict for THIS pod: p50/p95/p99 over the
        # recent request window against the 1s budget (no_data until traffic arrives).
        from .latency import RECORDER
        return {"status": "ok", "version": os.getenv("APP_VERSION", "unknown"),
                "latency": RECORDER.nfr_health()}

    @app.get("/dashboard", tags=["Control panel"])
    def dashboard():
        # the holistic control panel (inspect + admin: tenants, config). Served at /dashboard.
        # Inert HTML shell, no secrets: the operator pastes the API key in the page; its JS
        # sends it as X-API-Key on the guarded calls, so no data is reachable without the key.
        from fastapi.responses import HTMLResponse
        path = os.path.join(os.path.dirname(__file__), "static", "dashboard.html")
        if not os.path.exists(path):
            return HTMLResponse("<h1>dashboard.html missing</h1>", status_code=404)
        with open(path, encoding="utf-8") as fh:
            return HTMLResponse(fh.read())

    # static assets for the dashboard (favicons / icons). Inert public files.
    try:
        from fastapi.staticfiles import StaticFiles
        static_dir = os.path.join(os.path.dirname(__file__), "static")
        app.mount("/static", StaticFiles(directory=static_dir), name="static")

        @app.get("/favicon.ico", include_in_schema=False)
        def favicon():
            from fastapi.responses import FileResponse, Response
            ico = os.path.join(static_dir, "favicon", "favicon.ico")
            return FileResponse(ico) if os.path.exists(ico) else Response(status_code=404)
    except Exception as e:  # noqa: BLE001
        logger.warning(f"static mount failed: {e}")

    return app

Testing helpers

fakes

fakes

In-memory implementations of the three ports + the user-model store.

These let the whole pipeline run offline, deterministic, no Qdrant / Redis / network. Each satisfies the corresponding Protocol in contracts.ports.

FakeContentStore 

FakeContentStore(contents: dict[str, Content], vectors: dict[str, Vector], payloads: Optional[dict[str, dict]] = None)

ContentStore backed by dicts. search_vector = brute-force cosine; search_tags = tag-key overlap.

Source code in ai-engine/src/ai_engine/recsys/testing/fakes.py 
19
20
21
22
23
def __init__(self, contents: dict[str, Content], vectors: dict[str, Vector],
             payloads: Optional[dict[str, dict]] = None):
    self._contents = contents
    self._vectors = vectors
    self._payloads = payloads or {}

FakeEventSource 

FakeEventSource(events_by_user: Optional[dict[str, list[InteractionEvent]]] = None)

EventSource backed by an in-memory per-user buffer (mimics the Redis hot buffer fed by the ingestion webhook).

Source code in ai-engine/src/ai_engine/recsys/testing/fakes.py 
108
109
def __init__(self, events_by_user: Optional[dict[str, list[InteractionEvent]]] = None):
    self._events: dict[str, list[InteractionEvent]] = events_by_user or {}

InMemoryImpressionStore 

InMemoryImpressionStore()

ImpressionStore backed by a dict (no TTL needed for tests).

Source code in ai-engine/src/ai_engine/recsys/testing/fakes.py 
120
121
def __init__(self) -> None:
    self._d: dict[str, dict] = {}

InMemoryUserModelStore 

InMemoryUserModelStore()

UserModelStore backed by a dict (recompute-backed equivalent of Redis).

Source code in ai-engine/src/ai_engine/recsys/testing/fakes.py 
141
142
def __init__(self) -> None:
    self._signals: dict[str, UserSignals] = {}

InMemoryEmbeddingModel 

InMemoryEmbeddingModel(dim: int = 8)

Deterministic text -> vector (hash buckets). For cold-start / profile paths in tests; no model download.

Source code in ai-engine/src/ai_engine/recsys/testing/fakes.py 
158
159
def __init__(self, dim: int = 8):
    self._dim = dim

fixtures

fixtures

A tiny, hand-built Bergen-Belsen content world with KNOWN structure, so synthetic scenarios have a predictable right answer.

Three semantic clusters on orthogonal axes (6-dim vectors), each with matching expert tags from the real taxonomy (theme_what facet):

A = Forced Labor      axis 0
B = Family / Children axis 1
C = Liberation        axis 2

make_payloads 

make_payloads()

Raw payload dicts (location, time_metadata, …) matching the fixture world.

Source code in ai-engine/src/ai_engine/recsys/testing/fixtures.py 
73
74
75
def make_payloads():
    """Raw payload dicts (location, time_metadata, …) matching the fixture world."""
    return {cid: cv[2] for cid, cv in _WORLD.items()}

view_events 

view_events(user_id: str, content_id: str, *, dwell: float, reason: str, base_ts: datetime, visits: int = 1) -> list[InteractionEvent]

Emit a START + END pair (separate events) for one content.

Source code in ai-engine/src/ai_engine/recsys/testing/fixtures.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def view_events(
    user_id: str,
    content_id: str,
    *,
    dwell: float,
    reason: str,
    base_ts: datetime,
    visits: int = 1,
) -> list[InteractionEvent]:
    """Emit a START + END pair (separate events) for one content."""
    out: list[InteractionEvent] = []
    for k in range(visits):
        t0 = base_ts + timedelta(minutes=k)
        out.append(InteractionEvent(
            user_id=user_id, event="CONTENT_VIEW_STARTED",
            content_id=content_id, session_id="s1", ts=t0,
        ))
        out.append(InteractionEvent(
            user_id=user_id, event="CONTENT_VIEW_ENDED",
            content_id=content_id, session_id="s1",
            ts=t0 + timedelta(seconds=dwell),
            dwell_seconds=dwell, end_reason=reason,
        ))
    return out