混合检索 RAG：BM25 与向量融合

score(D, Q) = Σ  IDF(qi) × (tf(qi, D) × (k1 + 1))
                     ─────────────────────────────
                     tf(qi, D) + k1 × (1 - b + b × |D| / avgdl)

IDF(qi) = ln( (N - n(qi) + 0.5) / (n(qi) + 0.5) + 1 )

# bm25_score.py — BM25 分数计算（示意）
import math

def bm25_score(tf, dl, avgdl, n_docs, df, k1=1.5, b=0.75) -> float:
    """单词项 BM25 分数：词频饱和 + 长度归一化。"""
    idf = math.log((n_docs - df + 0.5) / (df + 0.5) + 1)
    tf_norm = tf * (k1 + 1) / (tf + k1 * (1 - b + b * dl / avgdl))
    return idf * tf_norm

# vector_search.py — 向量检索
def vector_search(query_emb, index, top_k=10, filters=None) -> list[dict]:
    """ANN 检索 + 可选元数据过滤。"""
    hits = index.search(query_emb, top_k=top_k * 3)   # 多召回再过滤
    if filters:
        hits = [h for h in hits if match_filters(h["metadata"], filters)]
    return hits[:top_k]

query
  ├──▶ BM25 检索（倒排索引） ──▶ 稀疏结果列表
  │                             │
  └──▶ 向量检索（ANN 索引） ──▶ 稠密结果列表
                               │
                               ▼
                    融合（RRF / 加权）
                               │
                               ▼
                    重排器（Cross-Encoder）
                               │
                               ▼
                    Top-K 送入 LLM 上下文

# hybrid_search.py — 并行混合检索骨架
class HybridRetriever:
    def __init__(self, bm25_index, vector_index, embed_fn):
        self.bm25 = bm25_index
        self.vector = vector_index
        self.embed = embed_fn

    def retrieve(self, query: str, top_k: int = 10) -> list[dict]:
        # 1. 两条通道并行召回
        sparse_hits = self.bm25.search(query, top_k=top_k * 2)
        dense_hits = self.vector.search(self.embed(query), top_k=top_k * 2)

        # 2. 融合成一个列表（RRF / 加权）
        fused = fuse(sparse_hits, dense_hits, top_k=top_k)

        # 3. 可选：重排器精排
        if self.reranker:
            fused = self.reranker.rerank(query, fused, top_k=top_k)
        return fused

def build_hybrid_index(documents: list[dict]):
    """同一批文档构建 BM25 倒排与向量索引。"""
    # BM25 索引（ES / OpenSearch / 自建）
    for doc in documents:
        bm25_index.add_document(doc["id"], doc["text"], doc["metadata"])

    # 向量索引（FAISS / Qdrant / pgvector）
    embeddings = embed_batch([d["text"] for d in documents])
    vector_index.add_vectors(
        ids=[d["id"] for d in documents],
        vectors=embeddings,
        metadatas=[d["metadata"] for d in documents],
    )

RRF_score(D) = Σ  1 / (k + rank_i(D))

k 为平滑常数（经验值 60）
rank_i(D) 为文档 D 在第 i 个列表中的排名

# rrf.py — 倒数排名融合
def rrf_fuse(lists: list[list[dict]], k=60, top_k=20) -> list[dict]:
    """对多个召回列表做 RRF 融合。"""
    scores = {}
    for ranked in lists:
        for rank, doc in enumerate(ranked, start=1):
            doc_id = doc["id"]
            scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank)
            # 保留分数与元数据引用
            meta[doc_id] = doc
    # 按融合分降序
    ordered = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    return [meta[doc_id] | {"rrf_score": score}
            for doc_id, score in ordered[:top_k]]

def weighted_rrf(lists, weights, k=60, top_k=20) -> list[dict]:
    """加权 RRF：按通道重要性给权重。"""
    scores = {}
    for weight, ranked in zip(weights, lists):
        for rank, doc in enumerate(ranked, start=1):
            doc_id = doc["id"]
            scores[doc_id] = scores.get(doc_id, 0) + weight / (k + rank)
    ordered = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    return [doc_id for doc_id, _ in ordered[:top_k]]

# rerank.py — Cross-Encoder 重排
def rerank(query: str, candidates: list[dict], top_k=5) -> list[dict]:
    """用 Cross-Encoder 对候选精排，只对少量候选执行。"""
    scores = cross_encoder.score(
        [(query, c["text"]) for c in candidates])
    ranked = sorted(zip(candidates, scores),
                    key=lambda x: x[1], reverse=True)
    return [c for c, s in ranked[:top_k]]

def two_stage_retrieve(query, retriever, reranker,
                       recall_k=50, final_k=5) -> list[dict]:
    """两阶段：粗召回 50 → 重排精排 5。"""
    recall = retriever.retrieve(query, top_k=recall_k)
    return reranker.rerank(query, recall, top_k=final_k)

# query_rewrite.py — 多路查询改写
def rewrite_queries(query: str, n_variants=3, llm=None) -> list[str]:
    """LLM 生成多个改写版本。"""
    prompt = (
        f"把用户问题改写为 {n_variants} 个更利于检索的表述，"
        f"保留原意，覆盖不同用词。原问题：{query}"
    )
    return llm(prompt)   # ["退订会员如何操作", "取消自动续费入口", "会员扣费停止办法"]

def multiway_retrieve(query: str, hybrid, variants_fn, top_k=10):
    """多路改写 + 每路混合检索 + 汇总融合。"""
    queries = [query] + variants_fn(query)
    all_lists = [hybrid.retrieve(q, top_k=top_k * 2) for q in queries]
    return rrf_fuse(all_lists, top_k=top_k)

# retrieval_eval.py — 检索指标计算
def recall_at_k(retrieved_ids, relevant_ids, k) -> float:
    """Recall@K：Top-K 中相关文档占比。"""
    hit = set(retrieved_ids[:k]) & set(relevant_ids)
    return len(hit) / len(relevant_ids)

def mrr(retrieved_ids, relevant_ids) -> float:
    """MRR：第一个相关结果的排名倒数。"""
    for rank, doc_id in enumerate(retrieved_ids, start=1):
        if doc_id in relevant_ids:
            return 1.0 / rank
    return 0.0

def rag_end_to_end_eval(queries, pipeline, judge_fn, golden):
    """端到端：检索层指标 + 生成层指标一起看。"""
    return {
        "retrieval": evaluate_retrieval(queries, pipeline.retriever, golden),
        "generation": {
            "faithfulness": judge_fn.avg_faithfulness(queries, pipeline),
            "answer_relevancy": judge_fn.avg_relevancy(queries, pipeline),
        },
    }

改造前：纯向量检索，命中率低、术语文档常漏
  症状：用户问"续费失败"，检索不到写"扣款异常"的文档
改造后：BM25 + 向量 + RRF 融合 + 重排
  效果：字面与语义两条路都覆盖，召回显著提升

# kb_hybrid.py — 知识库混合检索完整实现
class KBHybridSearch:
    def __init__(self, bm25, vector, reranker, embed_fn, rewrite_fn=None):
        self.bm25 = bm25
        self.vector = vector
        self.reranker = reranker
        self.embed = embed_fn
        self.rewrite = rewrite_fn or (lambda q: [q])

    def search(self, query: str, top_k: int = 5) -> list[dict]:
        queries = self.rewrite(query)            # 多路改写
        recalls = []
        for q in queries:
            sparse = self.bm25.search(q, 50)
            dense = self.vector.search(self.embed(q), 50)
            recalls.append(rrf_fuse([sparse, dense], top_k=30))
        fused = rrf_fuse(recalls, top_k=20)      # 汇总各路
        return self.reranker.rerank(query, fused, top_k=top_k)

上线前必须回答：
- Recall@10 是否优于任一单通道？
- 重排器是否带来 NDCG 提升（且能覆盖成本）？
- 查询改写是否值得（Delta MAP > 5%）？
- 端到端 Faithfulness 是否未下降？
- 延迟 P95 是否在预算内？