| """HybridRecommender: v2b for new users, gradually blended toward a warm DeepFM as the user's history grows,
|
| plus a post-age decay and a cold-start push for fresh posts.
|
|
|
| score = (1 - w) * rank(v2b) + w * rank(DeepFM) w = min(w_max, engaged / (engaged + n0)) (w = 0 for a brand-new user)
|
| * (floor + (1 - floor) * 0.5 ** (age_hours / half_life)) (older posts slowly sink)
|
| + push_beta * [age_hours <= push_max_age] / sqrt(1 + impressions) (fresh, rarely shown posts get a lift)
|
| v2b = calibrated cold-model probability + tag_bonus * rank(tag closeness)
|
| """
|
| import json, math, os
|
| import numpy as np
|
| import onnxruntime as ort
|
| from .features import FeatureBuilder, content_vector, rank01
|
|
|
| ART = os.path.join(os.path.dirname(os.path.abspath(__file__)), "artifacts")
|
|
|
|
|
| class HybridRecommender:
|
| def __init__(self, artifacts_dir=ART, **overrides):
|
| self.meta = json.load(open(os.path.join(artifacts_dir, "meta.json"), encoding="utf8"))
|
| self.cfg = {**self.meta["defaults"], **overrides}
|
| self.fb = FeatureBuilder(self.meta)
|
| so = ort.SessionOptions(); so.intra_op_num_threads = 1
|
| self.cold = ort.InferenceSession(os.path.join(artifacts_dir, self.meta["cold"]["onnx"]), so, providers=["CPUExecutionProvider"])
|
| self.warm = ort.InferenceSession(os.path.join(artifacts_dir, self.meta["warm"]["onnx"]), so, providers=["CPUExecutionProvider"])
|
|
|
|
|
| def _tokens(self, which, toks):
|
| spec = self.meta[which]
|
| return [spec["vocab"][f].get(toks[f], 0) for f in spec["token_inputs"]]
|
|
|
| def _floats(self, which, raw):
|
| return [raw[f] for f in self.meta[which]["float_inputs"]]
|
|
|
| @staticmethod
|
| def _prep_history(history, fb):
|
| hs = []
|
| for h in sorted(history or [], key=lambda x: x["ts_ms"]):
|
| e = content_vector(h.get("embedding"), fb.emb_dim)
|
| hs.append({**h, "embedding": e, "pca16": None if e is None else fb.pca16(e)})
|
| return hs
|
|
|
| def warm_weight(self, engaged_count):
|
| n0, wmax = self.cfg["n0_engaged_posts"], self.cfg["w_max"]
|
| return min(wmax, engaged_count / (engaged_count + n0)) if engaged_count > 0 else 0.0
|
|
|
|
|
| def score(self, user, candidates, now_ms):
|
| """user: {"interests": [tag slugs], "history": [interaction dicts]}; candidates: [post dicts]; now_ms: request time (or a list, one per candidate, for replay).
|
| Returns one dict per candidate with the final score and every component (for logging / explanations)."""
|
| fb, cfg = self.fb, self.cfg
|
| if not candidates:
|
| return []
|
| nows = list(now_ms) if isinstance(now_ms, (list, tuple, np.ndarray)) else [now_ms] * len(candidates)
|
| q, r, ms = fb.interest_profile(user.get("interests"))
|
| hist = self._prep_history(user.get("history"), fb)
|
| Tc, Fc, Tw, Fw, s1, eng, ages, imps, topic, content = [], [], [], [], [], [], [], [], [], []
|
| for post, now in zip(candidates, nows):
|
| X = content_vector(post.get("embedding"), fb.emb_dim); post = {**post, "embedding": X}; content.append(X is not None)
|
| base = fb.post_floats(post); toks = fb.post_tokens(post)
|
| tm = fb.tag_match(fb.tag_vector(post.get("tags")), q, r, ms); s1.append(fb.tag_closeness(tm))
|
| topic.append(bool(tm[0] > 0 or tm[1] > 0))
|
| zb = {k: fb.z(k, v) for k, v in base.items()}
|
| zt = dict(zip(self.meta["stage1"]["features"], (tm - fb.s1_mu) / fb.s1_sd))
|
| Tc.append(self._tokens("cold", toks)); Fc.append(self._floats("cold", {**zb, **zt}))
|
| hf, n_eng = fb.history_floats(hist, post, now, X); eng.append(n_eng)
|
| Tw.append(self._tokens("warm", toks)); Fw.append(self._floats("warm", {**zb, **{k: fb.z(k, v) for k, v in hf.items()}}))
|
| ages.append(fb.age_median if post.get("age_hours") is None else float(post["age_hours"])); imps.append(float(post.get("impressions") or 0))
|
| p_cold = self.cold.run(None, {"token_ids": np.array(Tc, np.int64), "float_features": np.array(Fc, np.float32)})[0]
|
| a, b = self.meta["cold"]["platt"]["coef"], self.meta["cold"]["platt"]["intercept"]
|
| lgt = np.log(np.clip(p_cold, 1e-6, 1 - 1e-6) / (1 - np.clip(p_cold, 1e-6, 1 - 1e-6)))
|
| p_cal = 1 / (1 + np.exp(-(a * lgt + b)))
|
| v2b = p_cal + cfg["tag_bonus"] * rank01(s1)
|
| w = np.array([self.warm_weight(e) for e in eng])
|
| if (w > 0).any():
|
| p_warm = self.warm.run(None, {"token_ids": np.array(Tw, np.int64), "float_features": np.array(Fw, np.float32)})[0]
|
| else:
|
| p_warm = np.full(len(candidates), np.nan)
|
| blend = (1 - w) * rank01(v2b) + w * (rank01(np.nan_to_num(p_warm)) if (w > 0).any() else 0)
|
| ages = np.array(ages, dtype=float); imps = np.array(imps)
|
| fresh = cfg["decay_floor"] + (1 - cfg["decay_floor"]) * 0.5 ** (np.maximum(ages, 0) / cfg["decay_half_life_hours"])
|
| push = cfg["push_beta"] * (ages <= cfg["push_max_age_hours"]) / np.sqrt(1 + imps)
|
| final = blend * fresh + push
|
| return [dict(post_id=c.get("post_id"), score=float(final[i]), topic_match=topic[i], cold_probability=float(p_cal[i]), tag_closeness=float(s1[i]), v2b=float(v2b[i]),
|
| warm_probability=None if np.isnan(p_warm[i]) else float(p_warm[i]), warm_weight=float(w[i]), engaged_history=int(eng[i]),
|
| freshness_multiplier=float(fresh[i]), push=float(push[i]), has_content_embedding=content[i]) for i, c in enumerate(candidates)]
|
|
|
| def order(self, scored):
|
| """Topic-first ordering: in every block of `topic_block` positions, the first `topic_first_slots` go to topic-matching posts (best score first);
|
| the remaining positions take the best remaining posts of any kind. If too few topic matches exist, the slots fall back to the best remaining posts."""
|
| slots, block = int(self.cfg.get("topic_first_slots", 0)), int(self.cfg.get("topic_block", 10))
|
| rest = sorted(scored, key=lambda d: -d["score"])
|
| if slots <= 0:
|
| return rest
|
| out = []
|
| while rest:
|
| topical = [d for d in rest if d["topic_match"]][:slots]
|
| for d in topical: rest.remove(d)
|
| out += topical
|
| fill = block - len(topical)
|
| out += rest[:fill]; rest = rest[fill:]
|
| for pos, d in enumerate(out): d["position"] = pos + 1
|
| return out
|
|
|
| def rank(self, user, candidates, now_ms):
|
| """All candidates in final feed order (scores + topic-first rule)."""
|
| return self.order(self.score(user, candidates, now_ms))
|
|
|
| def recommend(self, user, candidates, now_ms, k=10):
|
| """Top-k candidates in final feed order."""
|
| return self.rank(user, candidates, now_ms)[:k]
|
|
|
