Python 搜推技术栈：从 FastAPI 到 LambdaMART

本文是 搜推算法笔记 第 6 篇（工程篇）。索引见搜推算法 · 知识索引。
上一篇：AB 实验与线上迭代 · 下一篇：Elasticsearch 检索基础

搜推系统中，Python 常用于算法子服务（召回、精排、向量化）和离线训练流水线。主链路可能是 Java/Go/C++，Python 负责算法密集部分。本文介绍常用工具链与工程模式。

0. 常见分层架构

┌──────────── 主服务（Java/Go）────────────┐
│  Query → ES/倒排召回 → 返回 id 列表      │
└──────────────────┬──────────────────────┘
                   │ HTTP / gRPC
                   ▼
┌──────────── 算法子服务（Python）─────────┐
│  Query 理解 → 多路召回(async) → LTR 精排  │
│  向量库 │ Redis 开关 │ 模型热加载           │
└──────────────────────────────────────────┘
                   │
                   ▼
            离线：Pandas/Spark → 训练 → 模型仓库

也可采用全 Python 栈（FastAPI 统一网关），取决于团队规模和 QPS 要求。

1. 为什么搜推常用 Python

Python 在搜推里的定位：算法子服务 + 离线训练，不是替代 Java 主链路。

优势	说明
算法库成熟	NumPy、Pandas、scikit-learn、LightGBM、PyTorch、onnxruntime
迭代快	改特征、改训练脚本成本低
服务化简单	FastAPI / Flask 很快能挂 HTTP
胶水强	轻松连 DB、Redis、对象存储、消息队列

局限	常见解法
极致 QPS / 超低延迟	核心路径用 Java/Go/C++；Python 做算法子服务
GIL 限制 CPU 并行	多进程、或把重计算丢到 C 扩展 / ONNX
部署体积	训练用 PyTorch，在线导出 ONNX

很多公司的典型分工：

1 2	Java/Go：检索网关、交易、埋点 Python：向量化、LTR 精排、离线训练

2. FastAPI：模型服务化

最小服务

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(title="rank-demo")

class SearchReq(BaseModel):
    keyword: str = Field(..., min_length=1, max_length=64)
    page: int = Field(1, ge=1)
    page_size: int = Field(20, ge=1, le=50)

class SearchResp(BaseModel):
    list: list[dict]
    total: int

@app.post("/search/recommend", response_model=SearchResp)
async def recommend(req: SearchReq):
    # 伪代码：理解 → 召回 → 精排 → 分页
    items = [{"item_id": 1, "score": 0.9}]
    start = (req.page - 1) * req.page_size
    end = start + req.page_size
    return SearchResp(list=items[start:end], total=len(items))

为什么搜推爱 FastAPI

类型校验（Pydantic）减少脏输入
原生 async，适合「大量下游 I/O」
自动 OpenAPI 文档，联调方便

async 里能不能跑重计算？

原则：

网络 I/O（HTTP、DB、Redis）→ await 异步客户端
CPU 重（大批 embedding、树模型大批预测）→ 丢线程池/进程池，避免堵事件循环

import asyncio
from functools import partial

def heavy_predict(X):
    return model.predict(X)

async def rank(X):
    loop = asyncio.get_running_loop()
    scores = await loop.run_in_executor(None, partial(heavy_predict, X))
    return scores

Python 3.9+ 也可用 asyncio.to_thread(heavy_predict, X)。

3. asyncio：并发与超时

多路召回必须并发——三路耗时 40ms、50ms、30ms，串行 120ms，并发约 50ms。

gather + 单路失败隔离

import asyncio
from typing import Any

async def recall_a(q: str) -> list[tuple[str, float]]:
    await asyncio.sleep(0.04)
    return [("g1", 0.91), ("g2", 0.88)]

async def recall_b(q: str) -> list[tuple[str, float]]:
    await asyncio.sleep(0.05)
    return [("g2", 0.80), ("g3", 0.77)]

async def multi_recall(q: str) -> list[list[tuple[str, float]]]:
    results: list[Any] = await asyncio.gather(
        recall_a(q),
        recall_b(q),
        return_exceptions=True,  # 关键：一路异常不取消其它路
    )
    return [r for r in results if not isinstance(r, Exception)]

超时降级（生产必会）

async def with_timeout(coro, timeout_ms: int, fallback=None):
    if fallback is None:
        fallback = []
    try:
        return await asyncio.wait_for(coro, timeout=timeout_ms / 1000)
    except asyncio.TimeoutError:
        return fallback

原则：单路超时返回空列表，整请求仍可用其它路；全部空再走兜底或返回空并告警。

这在 Query 理解与多路召回的多路架构里会反复用到。

4. NumPy：向量计算基础

L2 归一化后，点积 = 余弦相似度——向量检索的数学基础：

import numpy as np

def l2_normalize(x: np.ndarray, axis: int = -1, eps: float = 1e-12) -> np.ndarray:
    norm = np.linalg.norm(x, axis=axis, keepdims=True)
    return x / (norm + eps)

def cosine_matrix(q: np.ndarray, docs: np.ndarray) -> np.ndarray:
    q = l2_normalize(np.atleast_2d(q))
    docs = l2_normalize(docs)
    return (q @ docs.T).ravel()

TopK 用 argpartition 而不是全排序——当 (n) 很大、(k) 很小时更省：

def topk_indices(scores: np.ndarray, k: int) -> np.ndarray:
    k = min(k, scores.size)
    idx = np.argpartition(-scores, kth=k - 1)[:k]
    return idx[np.argsort(-scores[idx])]

在线向量常用 float32，省内存、够用。

5. Pandas：样本与特征表

Listwise 样本每一行是「某个列表里的一个文档」：

session_id	item_id	label	scroll_depth	price
s1	101	1	1	19.9
s1	102	0	2	29.9
s1	103	0	3	9.9

同一 session_id 构成一个 group。LightGBM 的 group 参数要求样本按 session 连续排列：

1
2
3

df = df.sort_values(["session_id", "scroll_depth"]).reset_index(drop=True)
group = df.groupby("session_id", sort=False).size().tolist()
# 例：[3, 2] 表示先 3 行属于 s1，再 2 行属于 s2

禁止按行随机切分 train/valid——同一列表的文档进两边，NDCG 虚高。必须按 session 切：

sessions = df["session_id"].unique()
rng = np.random.default_rng(42)
rng.shuffle(sessions)
cut = int(len(sessions) * 0.8)
train_s, valid_s = set(sessions[:cut]), set(sessions[cut:])

位置去偏权重的一种常见写法：

1	df["sample_weight"] = 1.0 / np.log2(df["scroll_depth"] + 1.0)

6. LightGBM：排序模型

最小训练脚本：

import lightgbm as lgb

feature_cols = ["log_price", "sales", "cosine_vec", "term_overlap"]
X_train = train_df[feature_cols].to_numpy()
y_train = train_df["label"].to_numpy()
w_train = train_df["sample_weight"].to_numpy()
group_train = train_df.groupby("session_id", sort=False).size().tolist()

train_set = lgb.Dataset(X_train, label=y_train, group=group_train, weight=w_train)
valid_set = lgb.Dataset(X_valid, label=y_valid, group=group_valid, reference=train_set)

params = {
    "objective": "lambdarank",
    "metric": "ndcg",
    "eval_at": [5, 10],
    "learning_rate": 0.05,
    "num_leaves": 63,
    "min_data_in_leaf": 50,
    "feature_fraction": 0.8,
    "verbosity": -1,
}

model = lgb.train(
    params, train_set, num_boost_round=500,
    valid_sets=[valid_set],
    callbacks=[lgb.early_stopping(50), lgb.log_evaluation(50)],
)
model.save_model("lambdamart.txt")

参数	含义
`objective=lambdarank`	Listwise 排序目标（LambdaMART 族）
`group`	每个 query/session 的文档数，必须与行对齐
`weight`	样本权重（去偏）
`eval_at`	NDCG@K 的 K 列表

在线预测时不必传 group——一次请求一个列表，直接 booster.predict(X) 再 argsort 降序。

7. ONNX Runtime：在线推理

Embedding 模型（BERT/BGE）用 PyTorch 在线推理依赖重、启动慢。导出 ONNX 后运行时更轻，与训练框架解耦：

import onnxruntime as ort
import numpy as np

session = ort.InferenceSession("bge-base.onnx", providers=["CPUExecutionProvider"])

def encode(input_ids: np.ndarray, attention_mask: np.ndarray) -> np.ndarray:
    outs = session.run(None, {"input_ids": input_ids, "attention_mask": attention_mask})
    last_hidden = outs[0]
    mask = attention_mask[..., None].astype(np.float32)
    emb = (last_hidden * mask).sum(axis=1) / mask.sum(axis=1).clip(min=1e-9)
    return emb / (np.linalg.norm(emb, axis=1, keepdims=True) + 1e-12)

工程实践中 tokenization 仍要在 Python 侧做，再喂给 ONNX。

8. 与中间件协作

组件	典型用途	Python 常见库
PostgreSQL / pgvector	存向量、ANN	asyncpg / psycopg
Redis	缓存、开关、会话	redis.asyncio
Kafka	日志	aiokafka
OSS/S3	模型与 Parquet	boto3 / oss2

模式：

1 2	离线：SQL 拉日志 → Pandas/Spark → 训练 → 上传 model.txt 在线：启动加载模型 → Redis 读 kill-switch → 召回 → predict

搜索零结果兜底等场景常用 pgvector 存向量、Redis 做 kill-switch，是上述模式的典型组合。

9. 工程习惯与坑

随机种子：切分 session、训练要固定 seed，保证可复现
空 Query：入口直接返回，不要打下游
监控：每路召回耗时、空结果率、降级次数、模型版本号
不要在 async 里同步阻塞 HTTP（用 httpx.AsyncClient）
特征列顺序：训练与预测列顺序必须一致

10. 技术选型对比

FastAPI vs Flask：搜推服务怎么选？

维度	FastAPI	Flask
异步支持	原生 async/await	需配合 gevent/eventlet
类型校验	Pydantic 内置	需额外库
文档	自动生成 OpenAPI	需插件
性能（I/O 密集）	优秀	一般
生态成熟度	较新但增长快	极成熟

搜推在线服务大量 I/O（多路召回、Redis、DB），FastAPI 是更自然的选择。Flask 更适合简单脚本或 legacy 项目。

LightGBM vs XGBoost：排序模型怎么选？

维度	LightGBM	XGBoost
训练速度	更快（直方图 + leaf-wise）	较慢（level-wise）
内存	更省	较大
排序支持	原生 `lambdarank`	需自定义或 XGBRanker
大规模稀疏特征	优秀	优秀
工业界采用	搜推精排主流	通用 ML 主流

精排场景 LightGBM 几乎是默认答案——原生 Listwise 目标、训练快、线上推理轻。

ONNX vs 原生 PyTorch 推理

维度	ONNX Runtime	PyTorch 原生
依赖体积	小	大（含 torch）
启动时间	快	慢
CPU 推理	优化好	一般
灵活性	导出后固定	随时改模型
适用	Embedding 在线编码	训练、实验迭代

典型路径：PyTorch 训练 → 导出 ONNX → 在线 ORT 推理。tokenization 仍留 Python 侧。

下一步

Query 理解 + 召回流水线 → Query 理解与多路召回
LambdaMART 原理与 NDCG → 精排与评价指标
ES 全文检索工程 → Elasticsearch 检索基础

脑瓜呆呆

记录 Java 后端、中间件、搜推算法、产品与管理类读书笔记。从 2021 年写到现在的个人学习笔记。