检索增强生成与向量检索实战：Embedding、HNSW 与重排

离线索引：文档 → 分块 → embedding → 向量库
在线查询：问题 → embedding → 检索 Top-K → 重排 → 拼上下文 → LLM 生成

import numpy as np
def normalize(x):
    return x / np.linalg.norm(x, axis=-1, keepdims=True)

# 归一化后，内积 == 余弦相似度，可以用更快的点积索引
emb = normalize(emb)

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-large-zh-v1.5")
docs = ["退换货政策：签收后 7 天内可无理由退货。",
        "发票申请：订单完成后可在个人中心申请电子发票。"]
doc_vecs = model.encode(docs, normalize_embeddings=True)
q_vec = model.encode(["怎么开发票"], normalize_embeddings=True)
print(q_vec @ doc_vecs.T)     # 哪个文档更相关，一眼可见

M              每个节点的邻居数，越大越准也越占内存（常用 16~64）
ef_construction 建图时的候选池大小，越大图质量越好（常用 100~500）
ef_search      查询时的候选池大小，越大召回越高、延迟越大（常用 64~256）

import faiss
d, nlist = 1024, 256
quantizer = faiss.IndexFlatIP(d)
index = faiss.IndexIVFFlat(quantizer, d, nlist, faiss.METRIC_INNER_PRODUCT)
index.train(doc_vecs)          # 必须先训练聚类中心
index.add(doc_vecs)
index.nprobe = 16              # 只搜最近的 16 个桶
D, I = index.search(q_vec, k=5)

index = faiss.IndexHNSWFlat(d, 32)          # M=32
index.hnsw.efConstruction = 200
index.add(doc_vecs)
index.hnsw.efSearch = 128                   # 查询时调
D, I = index.search(q_vec, k=5)

数据 < 10 万且要精确     → Flat
内存充足、要低延迟       → HNSW
数据上亿、内存吃紧       → IVF-PQ
需要按条件过滤           → 带 filter 的向量库（Milvus/Qdrant 等）

def chunk_by_chars(text, size=500, overlap=80):
    chunks, i = [], 0
    while i < len(text):
        chunks.append(text[i:i + size])
        i += size - overlap          # 重叠避免语义被切断
    return chunks

def rrf_fuse(dense_ids, sparse_ids, k=60):
    """Reciprocal Rank Fusion：按排名倒数加权融合"""
    score = {}
    for ids in (dense_ids, sparse_ids):
        for rank, doc_id in enumerate(ids):
            score[doc_id] = score.get(doc_id, 0) + 1 / (k + rank + 1)
    return [d for d, _ in sorted(score.items(), key=lambda x: -x[1])]

向量检索 Top-100  →  Reranker 精排  →  取 Top-5 送 LLM
   （快、粗）            （慢、准）

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-large")
pairs = [(q, docs[i]) for i in candidate_ids]
scores = reranker.predict(pairs)
top5 = [candidate_ids[i] for i in np.argsort(-scores)[:5]]

import faiss, numpy as np
from sentence_transformers import SentenceTransformer
embedder = SentenceTransformer("BAAI/bge-large-zh-v1.5")

def build_index(texts):
    chunks = [c for t in texts for c in chunk_by_chars(t)]
    vecs = embedder.encode(chunks, normalize_embeddings=True,
                           batch_size=64, show_progress_bar=True)
    vecs = np.asarray(vecs, dtype="float32")
    index = faiss.IndexHNSWFlat(vecs.shape[1], 32)
    index.hnsw.efConstruction = 200
    index.add(vecs)
    return index, chunks

def retrieve(index, chunks, query, top_k=100, final_k=5, reranker=None):
    qv = np.asarray(embedder.encode([query], normalize_embeddings=True),
                    dtype="float32")
    _, ids = index.search(qv, top_k)
    cands = [chunks[i] for i in ids[0] if i != -1]
    if reranker is None:
        return cands[:final_k]
    scores = reranker.predict([(query, c) for c in cands])
    order = np.argsort(-scores)[:final_k]
    return [cands[i] for i in order]

PROMPT = """请仅依据以下资料回答问题；资料中没有的信息，请回答「资料未提及」。
资料：
{context}
问题：{question}
"""

def answer(llm, question, contexts):
    ctx = "\n\n".join(f"[{i+1}] {c}" for i, c in enumerate(contexts))
    return llm(PROMPT.format(context=ctx, question=question))

def recall_at_k(retrieved_ids, gold_ids, k):
    hit = set(retrieved_ids[:k]) & set(gold_ids)
    return len(hit) / len(gold_ids)

def mrr(retrieved_ids, gold_ids):
    for rank, doc_id in enumerate(retrieved_ids, 1):
        if doc_id in gold_ids:
            return 1.0 / rank
    return 0.0

JUDGE = """判断回答是否完全由资料支撑，只输出 支持 或 不支持。
资料：{context}
回答：{answer}
"""

def faithfulness(llm, answer, contexts):
    verdict = llm(JUDGE.format(context="\n".join(contexts), answer=answer))
    return 1.0 if "支持" in verdict and "不支持" not in verdict else 0.0

def mmr(query_vec, doc_vecs, doc_ids, lam=0.7, k=5):
    selected, cands = [], list(range(len(doc_ids)))
    while cands and len(selected) < k:
        def score(i):
            rel = float(query_vec @ doc_vecs[i])
            div = max([float(doc_vecs[i] @ doc_vecs[j]) for j in selected],
                      default=0.0)
            return lam * rel - (1 - lam) * div
        best = max(cands, key=score)
        selected.append(best)
        cands.remove(best)
    return [doc_ids[i] for i in selected]

1. 分块策略      ← 收益最大、成本最低
2. 混合召回      ← 解决专有名词与编号
3. 加重排        ← 精度提升 10~25 个点
4. 查询改写      ← 解决短问题与多跳
5. 换更大模型    ← 最后再考虑