一次召回事故：离职流程被答成了请假

上个月给公司内部知识库做RAG问答，用户问“离职流程”，系统返回的是“请假审批规范”。查了半天，根因是Chunk把一条完整制度拆碎，向量检索只命中了“审批”两个字。

这件事让我意识到，RAG系统的天花板不在模型，在检索。今天把整套方案和坑都写出来，代码直接抄。

问题拆解：RAG召回质量差的三个根源

知识库文档不是纯文本，有表格、条款、嵌套列表。我遇到的三个具体问题：

• Chunk边界切断了语义：固定512字符把“离职流程”的申请条件和审批节点拆到两个块里，检索时只能召回一半信息
• 向量检索吃掉关键词：Embedding模型对“离职”、“审批”、“流程”这类高频业务词区分度不够，ES向量检索返回的前20条里9条无关
• Top-K固定导致精度不可控：K=5时遗漏了关键条款，K=20时灌入大量噪音

针对这三件事，我做了三组对比实验，变量控制在同一份测试集上。

方案对比：三种路线，差别比想象中大

测试集是公司HR制度文档120份，共38万字符。用50个真实业务问题做召回评估，指标是Recall@5（前5条命中正确答案的比例）。

方案A是大部分教程的默认配置，直接PASS。方案B见效最猛，加BM25混合检索就提升了12个百分点。方案C是把精度推到91%的关键，代价是多花40ms。

方案C完整实现：语义Chunk + 混合检索 + 重排序

1. 语义Chunk：用LLM切分 + 规则兜底

纯LLM切分成本高，我用两层策略：先按Markdown标题和表格结构切分，再用LLM合并、重写块标题。代码在Python 3.11.8下运行，依赖langchain 0.3.7。

from langchain.text_splitter import MarkdownHeaderTextSplitter
import re

def semantic_chunk(text: str) -> list[dict]:
    """先按标题结构切，再合并过短块"""
    headers_to_split_on = [
        ("#", "H1"),
        ("##", "H2"),
        ("###", "H3"),
    ]
    splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=headers_to_split_on, strip_headers=False
    )
    chunks = splitter.split_text(text)

    merged = []
    buffer = ""
    for chunk in chunks:
        content = chunk.page_content
        if len(content) < 200:
            buffer += content + "\n"
            continue
        if buffer:
            merged.append({"title": chunk.metadata.get("H1", ""), "content": buffer + content})
            buffer = ""
        else:
            merged.append({"title": chunk.metadata.get("H1", ""), "content": content})
    if buffer:
        merged.append({"title": "未分类", "content": buffer})
    return merged

2. 混合检索：ES 8.12的kNN + BM25并行查询

ES 8.12开始kNN检索从插件转正，可以原生和BM25做RRF融合。我用的是BGE-M3模型，向量维度1024，ES索引配置如下。

# 创建索引，dense_vector类型必须指定dims和similarity
curl -X PUT "localhost:9200/kb_index" -H 'Content-Type: application/json' -d '{
  "mappings": {
    "properties": {
      "title": {"type": "text", "analyzer": "ik_max_word"},
      "content": {"type": "text", "analyzer": "ik_max_word"},
      "content_vector": {
        "type": "dense_vector",
        "dims": 1024,
        "index": true,
        "similarity": "cosine"
      }
    }
  }
}'

from elasticsearch import Elasticsearch
from langchain_community.embeddings import HuggingFaceBGEEmbeddings

es = Elasticsearch("http://localhost:9200")
embedding_model = HuggingFaceBGEEmbeddings(
    model_name="BAAI/bge-m3",
    model_kwargs={"device": "cuda"},
    encode_kwargs={"normalize_embeddings": True},
)

def hybrid_search(query: str, top_k: int = 10) -> list[dict]:
    """向量检索与BM25并行，RRF融合结果"""
    query_vector = embedding_model.embed_query(query)

    # ES 8.12支持一条查询里同时跑kNN和BM25
    body = {
        "knn": {
            "field": "content_vector",
            "query_vector": query_vector,
            "k": 30,
            "num_candidates": 100
        },
        "query": {
            "multi_match": {
                "query": query,
                "fields": ["title^2", "content"]
            }
        },
        "rank": {
            "rrf": {
                "window_size": 50,
                "rank_constant": 60
            }
        },
        "size": top_k
    }
    resp = es.search(index="kb_index", body=body)
    return [hit["_source"] for hit in resp["hits"]["hits"]]

RRF（Reciprocal Rank Fusion）不是简单加和，它把两个结果集的排名序号融合。BM25排第1的文档和向量检索排第3的文档，融合分是1/60 + 1/63，而不是1/1 + 1/1。这样的好处是不需要归一化两个分数，直接处理排名。rank_constant=60是ES默认值，实测对长尾召回更稳。

3. 重排序：Cross-Encoder的第二轮精排

混合检索返回Top-10，但前3条可能被标题词带偏。用bge-reranker-base做二次精排，交叉编码器把query和每条文档拼接后逐字比较，精度高，但速度慢，所以只对Top-10重排。

from FlagEmbedding import FlagReranker

reranker = FlagReranker("BAAI/bge-reranker-base", use_fp16=True)

def rerank(query: str, docs: list[dict], top_k: int = 5) -> list[dict]:
    """Cross-Encoder重排，只处理前10条"""
    pairs = [(query, doc["title"] + "\n" + doc["content"][:800]) for doc in docs]
    scores = reranker.compute_score(pairs, normalize=True)
    scored = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, score in scored[:top_k]]

4. 拼接Prompt，调用Qwen2.5-14B

from openai import OpenAI

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")

def rag_answer(question: str) -> str:
    docs = hybrid_search(question)
    final_docs = rerank(question, docs, top_k=5)
    context = "\n\n".join(
        [f"[来源{doc['title']}]\n{doc['content']}" for doc in final_docs]
    )
    prompt = f"""你是知识库助手，只能根据下方资料回答，资料中没有的信息要明确说不知道。
资料：
{context}

问题：{question}
回答："""
    resp = client.chat.completions.create(
        model="qwen2.5:14b",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
        max_tokens=512
    )
    return resp.choices[0].message.content

5. 完整服务：FastAPI + 压测接口

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Query(BaseModel):
    question: str

@app.post("/rag")
def rag(query: Query):
    answer = rag_answer(query.question)
    return {"answer": answer}

@app.get("/health")
def health():
    return {"status": "ok"}

pip install fastapi uvicorn elasticsearch langchain langchain-community FlagEmbedding openai
uvicorn rag_api:app --host 0.0.0.0 --port 8000 --workers 2

效果数据：方案C的真实表现

压测环境：ECS 2核4G + RTX 4090（跑Embedding和Reranker），数据库是MySQL 8.0.35存的元数据，ES 8.12.0单独一台2核4G。

重排序多花40ms，但Recall@5从74%涨到91%。QPS不算高，混合检索单QPS约9次/秒，对内部知识库够用。如果并发要求高，把Reranker换成ONNX Runtime部署，P99能压到220ms。

避坑：这四个坑，我替你踩了

坑1：ES 8.12的kNN和旧版写法不兼容

网上大量教程还在用“_knn_search”接口，ES 8.0以后直接报错。必须在search body里用knn字段，而且dense_vector索引映射里必须有"index": true，否则es.search会抛“vector index is not enabled”。解决方式：删除索引重建，我花了半天才定位到是映射问题。

坑2：BGE-M3的中文Embedding必须归一化

BGE系列在encode时需要设normalize_embeddings=True。如果不设，ES的cosine相似度会算出一堆负数，RRF融合结果直接乱序。我一开始没设，Recall@5掉到58%，比方案A还差。

坑3：RRF的window_size别改成10

ES文档里window_size默认50。我试过调小到10，想让融合更激进，结果BM25排30名开外的文档直接消失，长尾信息召回率掉到66%。这个参数控制融合时考虑的排名范围，不是越大越好，但10太小，建议保持在50。

坑4：语义Chunk的标题信息不能丢

早期版本我只存content，不存title。检索时BM25只用content字段，导致“离职流程”这种标题词完全失去作用。后来给每个Chunk配上父文档标题，BM25查询里title权重设成2倍，Recall@5直接涨了8个百分点。Chunk不能脱离上下文，标题就是最便宜的上下文。

一些结论

RAG项目上线后，人工客服的“简单问题重复答”量降了35%。混合检索是性价比最高的升级路径——不用换模型，不用加GPU，代码改动20行，Recall涨10个百分点以上。重排序则适合对精度要求高的场景，代价是多40ms延迟。

你的业务如果只需要答“XX是什么”，纯向量检索够用。但如果答的是“XX流程怎么做”，建议直接上方案C。

完整代码在内部GitLab（rag_hybrid_search），直接拉下来改ES地址和模型路径就能跑。