multi_channel_retrieval_service.py

1"""2对应正文第 4 章 §4.2「多路召回架构」—— 完整的多路召回服务链路。34正文把多路召回划分为三个阶段，本脚本按这个结构逐阶段实现：5  1) 召回前处理：构建触发信号 Trigger（近期点击/点赞/收藏 + 兴趣标签）、6     获取辅助特征（画像/上下文/设备）、访问外部存储（Redis、索引服务、画像服务）；7  2) 多路召回执行：8 路召回源（热门/I2I/U2I/垂类Tag/U2U2I/U2I2I/实时行为/图网络）8     用线程池并发触发，每路有独立超时；超时或异常的召回源走降级兜底；9  3) 召回后处理：按 Item ID 去重（保留召回来源 reason）、轻量级过滤10     （下架/违规/用户屏蔽）、Quota 配额控制、整体截断。1112关键机制与公式：13  - 并发召回总耗时 ≈ max_k(T_k)，而非 Σ_k T_k（正文：异步 RPC / 多线程调度）14  - 单路超时保护：T_k > timeout_k 时丢弃该路结果并计入超时率15  - 兜底：若最终候选数 < min_candidates，用热门池补齐（正文 §4.4 兜底机制）16  - Quota 截断：对召回源 k 保留 min(|R_k|, quota_k)，再整体截断到 max_candidates1718[MOCK] 本文件中的 MockRedis、MockForwardIndex、MockInvertedIndex、MockVectorIndex、19       MockProfileService 全部是教学用的内存假实现，用 time.sleep 模拟 RPC 网络延迟，20       不访问任何真实存储；延迟与失败率由构造参数注入，便于复现超时/降级场景。21"""2223import random24import threading25import time26from collections import defaultdict27from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout28from dataclasses import dataclass, field293031# ============================================================ [MOCK] 存储与索引层32class MockLatency:33    """[MOCK] 统一的延迟/故障注入器：按名字给出一次调用的模拟耗时与是否失败。"""3435    def __init__(self, rng: random.Random):36        self.rng = rng37        self.lock = threading.Lock()38        self.calls: dict[str, int] = defaultdict(int)3940    def hit(self, name: str, base_ms: float, jitter_ms: float, fail_rate: float = 0.0) -> None:41        with self.lock:42            self.calls[name] += 143            ms = base_ms + self.rng.random() * jitter_ms44            failed = self.rng.random() < fail_rate45        time.sleep(ms / 1000.0)46        if failed:47            raise RuntimeError(f"[MOCK] {name} RPC failed")484950class MockRedis:51    """[MOCK] 内存 KV，模拟 Redis：兴趣 Tag -> 物品列表、用户近期行为等。"""5253    def __init__(self, lat: MockLatency):54        self.store: dict[str, object] = {}55        self.lat = lat5657    def set(self, key: str, value: object) -> None:58        self.store[key] = value5960    def get(self, key: str, default=None):61        self.lat.hit("redis.get", base_ms=1.0, jitter_ms=1.0)62        return self.store.get(key, default)636465class MockForwardIndex:66    """[MOCK] 正排索引：Item ID -> 物品完整特征（标题/类目/标签/统计/状态/向量）。"""6768    def __init__(self, items: dict[str, dict], lat: MockLatency):69        self.items = items70        self.lat = lat7172    def mget(self, item_ids: list[str]) -> dict[str, dict]:73        self.lat.hit("forward_index.mget", base_ms=2.0, jitter_ms=2.0)74        return {i: self.items[i] for i in item_ids if i in self.items}757677class MockInvertedIndex:78    """[MOCK] 倒排索引：触发项（tag/类目/地域/item）-> 倒排拉链（关联物品列表）。"""7980    def __init__(self, lists: dict[str, list[str]], lat: MockLatency, fail_rate: float = 0.0):81        self.lists = lists82        self.lat = lat83        self.fail_rate = fail_rate8485    def seek(self, key: str, top_n: int = 200) -> list[str]:86        self.lat.hit("inverted_index.seek", base_ms=3.0, jitter_ms=3.0, fail_rate=self.fail_rate)87        return self.lists.get(key, [])[:top_n]888990class MockVectorIndex:91    """[MOCK] ANN 向量索引服务：用户向量 -> Top-N 近邻物品（这里用暴力内积代替 ANN）。"""9293    def __init__(self, vectors: dict[str, list[float]], lat: MockLatency,94                 base_ms: float = 12.0, fail_rate: float = 0.0):95        self.vectors = vectors96        self.lat = lat97        self.base_ms = base_ms98        self.fail_rate = fail_rate99100    def search(self, query: list[float], top_n: int) -> list[tuple[str, float]]:101        self.lat.hit("vector_index.search", base_ms=self.base_ms, jitter_ms=8.0,102                     fail_rate=self.fail_rate)103        scored = [(iid, sum(a * b for a, b in zip(query, vec)))104                  for iid, vec in self.vectors.items()]105        scored.sort(key=lambda x: -x[1])106        return scored[:top_n]107108109class MockProfileService:110    """[MOCK] 用户画像服务：长期兴趣标签、人口属性、地域。"""111112    def __init__(self, profiles: dict[str, dict], lat: MockLatency):113        self.profiles = profiles114        self.lat = lat115116    def get(self, uid: str) -> dict:117        self.lat.hit("profile.get", base_ms=4.0, jitter_ms=3.0)118        return self.profiles.get(uid, {})119120121# ============================================================ 数据结构122@dataclass123class RetrieveRequest:124    """一次召回请求：用户、上下文、设备信息。"""125    uid: str126    scene: str = "home_feed"127    city: str = "beijing"128    device: str = "ios"129    hour: int = 20130131132@dataclass133class TriggerContext:134    """召回前处理产出：触发信号 + 辅助特征。"""135    uid: str136    recent_clicks: list[str] = field(default_factory=list)137    recent_likes: list[str] = field(default_factory=list)138    realtime_clicks: list[str] = field(default_factory=list)139    interest_tags: list[str] = field(default_factory=list)140    user_vector: list[float] = field(default_factory=list)141    profile: dict = field(default_factory=dict)142    city: str = ""143    pre_ms: float = 0.0144145146@dataclass147class ChannelResult:148    """单路召回结果与可观测指标。"""149    name: str150    items: list[tuple[str, float]]151    cost_ms: float152    status: str  # ok / timeout / error / fallback153    quota: int154155156# ============================================================ 召回前处理157def pre_retrieval(req: RetrieveRequest, redis: MockRedis, profile_svc: MockProfileService,158                  pool: ThreadPoolExecutor) -> TriggerContext:159    """160    §4.2 召回前处理：三件事并发做——取行为序列（Redis）、取画像（画像服务）、取用户向量。161    注意正文强调「不直接使用全部用户行为」，这里对点击序列只截取最近 N 条作为 Trigger。162    """163    t0 = time.perf_counter()164    f_behavior = pool.submit(redis.get, f"behavior:{req.uid}", {})165    f_profile = pool.submit(profile_svc.get, req.uid)166    f_vector = pool.submit(redis.get, f"user_vec:{req.uid}", [])167168    behavior = f_behavior.result() or {}169    profile = f_profile.result() or {}170    user_vec = f_vector.result() or []171172    ctx = TriggerContext(173        uid=req.uid,174        recent_clicks=behavior.get("clicks", [])[:10],      # Trigger: 近期点击175        recent_likes=behavior.get("likes", [])[:5],         # Trigger: 近期点赞176        realtime_clicks=behavior.get("clicks", [])[:2],     # Trigger: 当前会话即时行为177        interest_tags=profile.get("tags", [])[:4],          # Trigger: 兴趣标签178        user_vector=user_vec,179        profile=profile,180        city=req.city,181    )182    ctx.pre_ms = (time.perf_counter() - t0) * 1000183    return ctx184185186# ============================================================ 8 路召回源187def ch_hot(ctx: TriggerContext, res: dict) -> list[tuple[str, float]]:188    """热门召回：全站热度 Top-N，保障基础覆盖与新用户冷启动。"""189    hot = res["redis"].get("hot:global", [])190    return [(iid, float(sc)) for iid, sc in hot]191192193def ch_i2i(ctx: TriggerContext, res: dict) -> list[tuple[str, float]]:194    """I2I 召回：对每个 Trigger 物品查 ItemCF 倒排拉链，分数按 trigger 位置衰减。"""195    out: dict[str, float] = {}196    inv: MockInvertedIndex = res["i2i_index"]197    for pos, trig in enumerate(ctx.recent_clicks[:5]):198        decay = 0.9 ** pos199        for rank, iid in enumerate(inv.seek(f"i2i:{trig}", top_n=30)):200            out[iid] = max(out.get(iid, 0.0), decay / (1 + rank) ** 0.5)201    return sorted(out.items(), key=lambda x: -x[1])202203204def ch_u2i(ctx: TriggerContext, res: dict) -> list[tuple[str, float]]:205    """U2I 召回：双塔用户向量过 ANN 索引（正文：双塔 + ANN 检索）。"""206    if not ctx.user_vector:207        return []208    return res["vector_index"].search(ctx.user_vector, top_n=120)209210211def ch_tag(ctx: TriggerContext, res: dict) -> list[tuple[str, float]]:212    """垂类/Tag 召回：按用户兴趣标签查 Tag 倒排拉链。"""213    out: dict[str, float] = {}214    inv: MockInvertedIndex = res["tag_index"]215    for pos, tag in enumerate(ctx.interest_tags):216        w = 1.0 / (1 + pos)217        for rank, iid in enumerate(inv.seek(f"tag:{tag}", top_n=40)):218            out[iid] = max(out.get(iid, 0.0), w * 0.98 ** rank)219    return sorted(out.items(), key=lambda x: -x[1])220221222def ch_u2u2i(ctx: TriggerContext, res: dict) -> list[tuple[str, float]]:223    """U2U2I 召回：先查相似用户拉链，再聚合相似用户的交互物品（UserCF 思想）。"""224    inv: MockInvertedIndex = res["u2u_index"]225    redis: MockRedis = res["redis"]226    out: dict[str, float] = defaultdict(float)227    for sim_uid in inv.seek(f"u2u:{ctx.uid}", top_n=5):228        peer = redis.get(f"behavior:{sim_uid}", {}) or {}229        for rank, iid in enumerate(peer.get("clicks", [])[:20]):230            out[iid] += 1.0 / (1 + rank)231    return sorted(out.items(), key=lambda x: -x[1])232233234def ch_u2i2i(ctx: TriggerContext, res: dict) -> list[tuple[str, float]]:235    """U2I2I 召回：在历史物品上做两跳扩展（trigger -> 相似物品 -> 再相似物品）。"""236    inv: MockInvertedIndex = res["i2i_index"]237    out: dict[str, float] = defaultdict(float)238    for trig in ctx.recent_likes[:3]:239        hop1 = inv.seek(f"i2i:{trig}", top_n=8)240        for a in hop1:241            for rank, b in enumerate(inv.seek(f"i2i:{a}", top_n=8)):242                out[b] += 0.5 / (1 + rank)243    return sorted(out.items(), key=lambda x: -x[1])244245246def ch_realtime(ctx: TriggerContext, res: dict) -> list[tuple[str, float]]:247    """实时行为召回：只用当前会话刚刚发生的行为触发，权重最高。"""248    inv: MockInvertedIndex = res["i2i_index"]249    out: dict[str, float] = {}250    for trig in ctx.realtime_clicks:251        for rank, iid in enumerate(inv.seek(f"i2i:{trig}", top_n=25)):252            out[iid] = max(out.get(iid, 0.0), 1.5 / (1 + rank) ** 0.5)253    return sorted(out.items(), key=lambda x: -x[1])254255256def ch_graph(ctx: TriggerContext, res: dict) -> list[tuple[str, float]]:257    """图网络召回：GNN 产出的图 Embedding 索引（此处复用一个独立的慢索引服务）。"""258    if not ctx.user_vector:259        return []260    return res["graph_index"].search(ctx.user_vector, top_n=60)261262263# ============================================================ 多路召回执行264def _timed(fn, ctx: TriggerContext, res: dict, box: dict):265    """在工作线程里记录该路召回自身的真实耗时（不含主线程排队等待）。"""266    def run():267        t0 = time.perf_counter()268        try:269            return fn(ctx, res)270        finally:271            box["cost_ms"] = (time.perf_counter() - t0) * 1000272    return run273274275def execute_channels(ctx: TriggerContext, res: dict, channels: list[dict],276                     pool: ThreadPoolExecutor) -> tuple[list[ChannelResult], float]:277    """278    §4.2 多路召回执行：线程池并发提交所有召回源，每路独立 timeout。279280    两个容易写错的细节，这里都按线上正确做法实现：281      1) 超时必须用「绝对 deadline」判定（submit 时刻 + timeout），282         否则按列表顺序逐个 result(timeout=...) 时，前面的等待会把后面的超时吃掉；283      2) 单路耗时要在工作线程内部测量，否则测到的是「排队 + 执行」的混合值。284    超时 / 异常的召回源标记 status 并返回空结果，后续由兜底机制补齐。285    """286    t0 = time.perf_counter()287    futures, deadlines, boxes = {}, {}, {}288    for cfg in channels:289        name = cfg["name"]290        boxes[name] = {"cost_ms": 0.0}291        deadlines[name] = time.perf_counter() + cfg["timeout_ms"] / 1000.0292        futures[name] = pool.submit(_timed(cfg["fn"], ctx, res, boxes[name]))293294    results: list[ChannelResult] = []295    for cfg in channels:296        name = cfg["name"]297        left = max(0.0, deadlines[name] - time.perf_counter())298        try:299            items = futures[name].result(timeout=left)300            status = "ok" if items else "empty"301        except FutureTimeout:302            items, status = [], "timeout"303            futures[name].cancel()304            # 超时时 worker 还在跑，box 里没有耗时；对外记为「至少用掉了超时预算」305            boxes[name]["cost_ms"] = cfg["timeout_ms"]306        except Exception:307            items, status = [], "error"308        results.append(ChannelResult(name, items, boxes[name]["cost_ms"], status, cfg["quota"]))309    return results, (time.perf_counter() - t0) * 1000310311312# ============================================================ 召回后处理313def post_retrieval(results: list[ChannelResult], res: dict, ctx: TriggerContext,314                   max_candidates: int, min_candidates: int) -> dict:315    """316    §4.2 召回后处理 + §4.4 融合：317      去重（保留 reason）-> 轻量级过滤 -> Quota 配额 -> 整体截断 -> 兜底补齐318    reason 采用「召回源优先级」口径（正文给出的两种做法之一，实现最简单）。319    """320    priority = {c.name: idx for idx, c in enumerate(results)}  # 列表顺序即优先级321322    # ---- Quota 配额控制：每路先按自身分数截到 quota323    quota_kept: list[tuple[str, str, float]] = []324    quota_stat = []325    for c in results:326        kept = sorted(c.items, key=lambda x: -x[1])[:c.quota]327        quota_stat.append((c.name, len(c.items), len(kept)))328        quota_kept.extend((iid, c.name, sc) for iid, sc in kept)329330    # ---- 去重：同一 item 命中多路，reason 取优先级最高的那一路331    best: dict[str, tuple[str, float]] = {}332    hit_sources: dict[str, set[str]] = defaultdict(set)333    for iid, src, sc in quota_kept:334        hit_sources[iid].add(src)335        if iid not in best or priority[src] < priority[best[iid][0]]:336            best[iid] = (src, sc)337    dedup_in, dedup_out = len(quota_kept), len(best)338339    # ---- 轻量级过滤：下架 / 违规 / 用户屏蔽作者（正文：简单且计算成本低的规则）340    fwd: MockForwardIndex = res["forward_index"]341    blocked_authors = set(ctx.profile.get("blocked_authors", []))342    feats = fwd.mget(list(best.keys()))343    filtered, drop_reason = {}, defaultdict(int)344    for iid, (src, sc) in best.items():345        f = feats.get(iid)346        if f is None:347            drop_reason["no_feature"] += 1348            continue349        if f["status"] != "online":350            drop_reason[f"status={f['status']}"] += 1351            continue352        if f["author"] in blocked_authors:353            drop_reason["blocked_author"] += 1354            continue355        filtered[iid] = (src, sc)356357    # ---- 整体截断：跨源按「源权重 × 源内分数」排序后截断358    src_weight = {c.name: 1.0 / (1 + priority[c.name] * 0.1) for c in results}359    ordered = sorted(filtered.items(), key=lambda kv: -src_weight[kv[1][0]] * kv[1][1])360    truncated = ordered[:max_candidates]361362    # ---- 兜底：候选不足时用热门池补齐（§4.4 召回源兜底机制）363    backup_added = 0364    if len(truncated) < min_candidates:365        have = {iid for iid, _ in truncated}366        hot = [(iid, sc) for iid, sc in res["redis"].get("hot:global", []) if iid not in have]367        # 一次批量取特征，而不是逐个物品打 RPC（线上同理：兜底也要走 batch 接口）368        hot_feats = fwd.mget([iid for iid, _ in hot if iid not in feats])369        for iid, sc in hot:370            if len(truncated) >= min_candidates:371                break372            f = feats.get(iid) or hot_feats.get(iid)373            if f is None or f["status"] != "online" or f["author"] in blocked_authors:374                continue375            truncated.append((iid, ("backup_hot", float(sc))))376            have.add(iid)377            backup_added += 1378379    return {380        "candidates": [(iid, src, sc) for iid, (src, sc) in truncated],381        "quota_stat": quota_stat,382        "dedup_in": dedup_in,383        "dedup_out": dedup_out,384        "drop_reason": dict(drop_reason),385        "backup_added": backup_added,386        "multi_hit": sum(1 for s in hit_sources.values() if len(s) > 1),387    }388389390# ============================================================ 服务编排391def retrieve_service(req: RetrieveRequest, res: dict, channels: list[dict],392                     pool: ThreadPoolExecutor, max_candidates: int = 3000,393                     min_candidates: int = 60) -> dict:394    """一次完整的召回服务调用：前处理 -> 并发多路召回 -> 后处理。"""395    t0 = time.perf_counter()396    ctx = pre_retrieval(req, res["redis"], res["profile"], pool)397    results, exec_ms = execute_channels(ctx, res, channels, pool)398    post = post_retrieval(results, res, ctx, max_candidates, min_candidates)399    post.update({400        "ctx": ctx,401        "channel_results": results,402        "exec_ms": exec_ms,403        "total_ms": (time.perf_counter() - t0) * 1000,404    })405    return post406407408# ============================================================ 合成数据 [MOCK]409def build_world(rng: random.Random, n_items: int = 600, n_users: int = 40,410                dim: int = 8, u2i_fail: float = 0.0, graph_slow_ms: float = 12.0):411    """[MOCK] 造一个小世界：物品库、正排、倒排拉链、向量索引、用户行为与画像。"""412    lat = MockLatency(rng)413    tags = ["tech", "sports", "food", "travel", "game", "movie"]414    authors = [f"a{i}" for i in range(30)]415416    items, vectors = {}, {}417    for i in range(n_items):418        iid = f"i{i:04d}"419        status = "online"420        r = rng.random()421        if r < 0.05:422            status = "offline"423        elif r < 0.08:424            status = "illegal"425        items[iid] = {426            "item_id": iid,427            "tag": rng.choice(tags),428            "category": rng.choice(["c1", "c2", "c3", "c4"]),429            "author": rng.choice(authors),430            "status": status,431            "ctr": round(rng.betavariate(2, 20), 4),432        }433        vectors[iid] = [rng.gauss(0, 1) for _ in range(dim)]434435    all_ids = list(items.keys())436    # 倒排拉链：i2i 相似物品、tag 拉链、u2u 相似用户437    i2i_lists = {f"i2i:{iid}": rng.sample(all_ids, 40) for iid in all_ids}438    tag_lists = {}439    for t in tags:440        pool_t = [iid for iid, f in items.items() if f["tag"] == t]441        tag_lists[f"tag:{t}"] = sorted(pool_t, key=lambda x: -items[x]["ctr"])[:60]442443    redis = MockRedis(lat)444    uids = [f"u{i:03d}" for i in range(n_users)]445    for uid in uids:446        redis.set(f"behavior:{uid}", {447            "clicks": rng.sample(all_ids, 24),448            "likes": rng.sample(all_ids, 8),449        })450        redis.set(f"user_vec:{uid}", [rng.gauss(0, 1) for _ in range(dim)])451    hot = sorted(items.values(), key=lambda f: -f["ctr"])[:200]452    redis.set("hot:global", [(f["item_id"], f["ctr"]) for f in hot])453454    u2u_lists = {f"u2u:{uid}": rng.sample([x for x in uids if x != uid], 8) for uid in uids}455    profiles = {456        uid: {457            "tags": rng.sample(tags, 4),458            "age": rng.randint(18, 50),459            "blocked_authors": rng.sample(authors, 2),460        }461        for uid in uids462    }463464    res = {465        "lat": lat,466        "redis": redis,467        "forward_index": MockForwardIndex(items, lat),468        "i2i_index": MockInvertedIndex(i2i_lists, lat),469        "tag_index": MockInvertedIndex(tag_lists, lat),470        "u2u_index": MockInvertedIndex(u2u_lists, lat),471        "vector_index": MockVectorIndex(vectors, lat, base_ms=12.0, fail_rate=u2i_fail),472        "graph_index": MockVectorIndex(vectors, lat, base_ms=graph_slow_ms, fail_rate=0.0),473        "profile": MockProfileService(profiles, lat),474        "items": items,475    }476    return res, uids477478479def default_channels() -> list[dict]:480    """8 路召回源配置：顺序即 reason 优先级，quota 对应正文表 4.3 的配额思想。"""481    return [482        {"name": "hot",      "fn": ch_hot,      "timeout_ms": 20.0, "quota": 500},483        {"name": "realtime", "fn": ch_realtime, "timeout_ms": 30.0, "quota": 300},484        {"name": "i2i",      "fn": ch_i2i,      "timeout_ms": 40.0, "quota": 1000},485        {"name": "u2i",      "fn": ch_u2i,      "timeout_ms": 35.0, "quota": 1500},486        {"name": "tag",      "fn": ch_tag,      "timeout_ms": 40.0, "quota": 600},487        {"name": "u2u2i",    "fn": ch_u2u2i,    "timeout_ms": 40.0, "quota": 400},488        {"name": "u2i2i",    "fn": ch_u2i2i,    "timeout_ms": 60.0, "quota": 400},489        {"name": "graph",    "fn": ch_graph,    "timeout_ms": 25.0, "quota": 1000},490    ]491492493def print_report(out: dict) -> None:494    print(f"  前处理耗时 {out['ctx'].pre_ms:6.2f} ms | 多路并发执行 {out['exec_ms']:6.2f} ms "495          f"| 服务总耗时 {out['total_ms']:6.2f} ms")496    print(f"  {'召回源':<10}{'状态':<9}{'耗时(ms)':>10}{'返回数':>8}{'quota':>7}{'保留数':>8}")497    print("  " + "-" * 54)498    kept_by_src = defaultdict(int)499    for iid, src, _ in out["candidates"]:500        kept_by_src[src] += 1501    quota_map = {n: (raw, kept) for n, raw, kept in out["quota_stat"]}502    seq_sum = 0.0503    for c in out["channel_results"]:504        raw, kept = quota_map[c.name]505        seq_sum += c.cost_ms506        print(f"  {c.name:<10}{c.status:<9}{c.cost_ms:>10.2f}{raw:>8}{c.quota:>7}"507              f"{kept_by_src[c.name]:>8}")508    print(f"  串行累加耗时 {seq_sum:.2f} ms  vs  并发耗时 {out['exec_ms']:.2f} ms "509          f"(并发收益 {seq_sum / max(out['exec_ms'], 1e-9):.2f}x)")510    print(f"  去重: {out['dedup_in']} -> {out['dedup_out']} "511          f"(去重率 {1 - out['dedup_out'] / max(out['dedup_in'], 1):.2%}, "512          f"多源命中 {out['multi_hit']} 个)")513    print(f"  轻量级过滤丢弃: {out['drop_reason']}")514    print(f"  兜底补充: {out['backup_added']} 个 | 最终候选集: {len(out['candidates'])}")515516517def main() -> None:518    random.seed(42)519    rng = random.Random(42)520521    print("=" * 78)522    print("场景 1：全部召回源健康")523    print("=" * 78)524    res, uids = build_world(rng)525    with ThreadPoolExecutor(max_workers=12) as pool:526        out = retrieve_service(RetrieveRequest(uid=uids[0]), res, default_channels(), pool)527        print_report(out)528        print("  候选集前 8 条 (item, reason, score):")529        for iid, src, sc in out["candidates"][:8]:530            print(f"    {iid}  reason={src:<9} score={sc:.4f}  tag={res['items'][iid]['tag']}")531532    print()533    print("=" * 78)534    print("场景 2：U2I 向量索引服务 100% 故障 + 图召回变慢（30ms > 25ms 超时阈值）")535    print("=" * 78)536    rng2 = random.Random(42)537    res2, uids2 = build_world(rng2, u2i_fail=1.0, graph_slow_ms=30.0)538    with ThreadPoolExecutor(max_workers=12) as pool:539        out2 = retrieve_service(RetrieveRequest(uid=uids2[0]), res2, default_channels(), pool)540        print_report(out2)541542    print()543    print("=" * 78)544    print("场景 3：极端降级——只剩热门召回可用（min_candidates 兜底生效）")545    print("=" * 78)546    rng3 = random.Random(42)547    res3, uids3 = build_world(rng3)548    only_hot = [c for c in default_channels() if c["name"] == "hot"]549    only_hot[0] = dict(only_hot[0], quota=30)  # 故意把配额压到低于 min_candidates550    with ThreadPoolExecutor(max_workers=4) as pool:551        out3 = retrieve_service(RetrieveRequest(uid=uids3[0]), res3, only_hot, pool,552                                min_candidates=60)553        print_report(out3)554555    print()556    print("=" * 78)557    print("场景 4：吞吐压测——20 个并发请求的 P50/P99 耗时")558    print("=" * 78)559    rng4 = random.Random(7)560    res4, uids4 = build_world(rng4)561    chans = default_channels()562    lat_list = []563    with ThreadPoolExecutor(max_workers=24) as pool:564        for uid in uids4[:20]:565            o = retrieve_service(RetrieveRequest(uid=uid), res4, chans, pool)566            lat_list.append((o["total_ms"], len(o["candidates"])))567    costs = sorted(c for c, _ in lat_list)568    sizes = [n for _, n in lat_list]569    print(f"  请求数 {len(costs)} | 耗时 P50={costs[len(costs) // 2]:.2f}ms "570          f"P90={costs[int(len(costs) * 0.9)]:.2f}ms max={costs[-1]:.2f}ms")571    print(f"  候选集规模 min={min(sizes)} avg={sum(sizes) / len(sizes):.1f} max={max(sizes)}")572    print(f"  [MOCK] 存储调用次数统计: {dict(res4['lat'].calls)}")573574575if __name__ == "__main__":576    main()