企业级RAG系统2.0：混合检索+重排序+Self-RAG架构设计与实践

为什么需要RAG 2.0

初代RAG系统（2023-2024）的核心问题：
- 召回率不足：纯向量检索对关键词匹配场景效果差
- 幻觉问题：检索到相关文档但LLM仍然编造答案
- 多跳推理弱：跨文档关联推理能力不足
- 上下文利用率低：检索到5段文档，LLM只参考了1段

RAG 2.0通过系统性架构升级，全面解决上述问题。

一、混合检索架构

1.1 稠密+稀疏向量融合

from qdrant_client import QdrantClient
from qdrant_client.models import (
    VectorParams, Distance, SparseVectorParams,
    NamedVector, NamedSparseVector
)

class HybridRetriever:
    """混合检索器：融合稠密向量与稀疏BM25检索"""

    def __init__(self):
        self.client = QdrantClient(host="localhost", port=6333)
        self.dense_encoder = SentenceTransformer("BAAI/bge-m3")
        self.sparse_encoder = BM25Encoder()

        # 创建支持混合检索的集合
        self.client.create_collection(
            collection_name="enterprise_kb",
            vectors_config={
                "dense": VectorParams(
                    size=1024,
                    distance=Distance.COSINE
                )
            },
            sparse_vectors_config={
                "sparse": SparseVectorParams()
            }
        )

    def search(self, query: str, top_k: int = 10) -> list[dict]:
        """执行混合检索，结合RRF融合算法"""

        # 编码查询
        dense_vector = self.dense_encoder.encode(query).tolist()
        sparse_vector = self.sparse_encoder.encode(query)

        # 稠密检索
        dense_results = self.client.search(
            collection_name="enterprise_kb",
            query_vector=NamedVector(name="dense", vector=dense_vector),
            limit=top_k * 2
        )

        # 稀疏检索（关键词匹配）
        sparse_results = self.client.search(
            collection_name="enterprise_kb",
            query_vector=NamedSparseVector(
                name="sparse",
                vector=sparse_vector
            ),
            limit=top_k * 2
        )

        # RRF（倒序排名融合）
        return self._reciprocal_rank_fusion(
            [dense_results, sparse_results],
            k=60  # RRF参数
        )[:top_k]

    def _reciprocal_rank_fusion(self, result_lists, k=60):
        """倒序排名融合算法"""
        scores = {}
        for result_list in result_lists:
            for rank, item in enumerate(result_list):
                doc_id = item.id
                if doc_id not in scores:
                    scores[doc_id] = {"score": 0, "item": item}
                scores[doc_id]["score"] += 1 / (k + rank + 1)

        sorted_results = sorted(
            scores.values(),
            key=lambda x: x["score"],
            reverse=True
        )
        return [r["item"] for r in sorted_results]

1.2 查询扩展与改写

async def expand_query(query: str, llm) -> list[str]:
    """使用LLM扩展查询，提升召回率"""

    prompt = f"""
给定用户查询，生成3个语义等价但表述不同的查询版本。
目的是提高检索系统的召回率。

原始查询：{query}

只输出JSON数组，每个元素是一个查询字符串：
["查询1", "查询2", "查询3"]
"""
    response = await llm.complete(prompt)
    expanded = json.loads(response)
    return [query] + expanded  # 原始查询 + 扩展版本

二、多路重排序

2.1 Cross-Encoder精排

from sentence_transformers import CrossEncoder

class RerankerPipeline:
    """多阶段重排序流水线"""

    def __init__(self):
        # 中文场景推荐 BAAI/bge-reranker-v2-m3
        self.cross_encoder = CrossEncoder("BAAI/bge-reranker-v2-m3")

    def rerank(
        self,
        query: str,
        candidates: list[dict],
        top_k: int = 5
    ) -> list[dict]:
        """精排候选文档"""

        # 构建(query, document)对
        pairs = [(query, doc["text"]) for doc in candidates]

        # Cross-Encoder打分（考虑query-doc交互）
        scores = self.cross_encoder.predict(pairs)

        # 按分数排序
        ranked = sorted(
            zip(candidates, scores),
            key=lambda x: x[1],
            reverse=True
        )

        return [
            {**doc, "rerank_score": float(score)}
            for doc, score in ranked[:top_k]
        ]

    def contextual_compression(
        self,
        query: str,
        documents: list[dict],
        llm
    ) -> list[dict]:
        """上下文压缩：只保留文档中与查询相关的部分"""

        compressed = []
        for doc in documents:
            prompt = f"""
从以下文档中提取与查询直接相关的段落。
只输出相关内容，不添加任何解释。如果整个文档都不相关，输出"NO_RELEVANT"。

查询：{query}
文档：{doc['text']}
"""
            result = llm.complete(prompt)
            if result != "NO_RELEVANT":
                compressed.append({**doc, "text": result})

        return compressed

三、Self-RAG自我反思机制

3.1 Self-RAG的核心思想

Self-RAG通过引入特殊控制token，让LLM在生成过程中主动判断：
1. 是否需要检索（[Retrieve] / [No Retrieve]）
2. 检索结果是否相关（[Relevant] / [Irrelevant]）
3. 生成内容是否有幻觉（[Supported] / [Partially Supported] / [No Support]）

class SelfRAGPipeline:
    """Self-RAG流水线实现"""

    def __init__(self, llm, retriever, reranker):
        self.llm = llm
        self.retriever = retriever
        self.reranker = reranker

    async def generate(self, query: str) -> dict:
        """Self-RAG生成流程"""

        # Step 1: 判断是否需要检索
        need_retrieval = await self._check_retrieval_needed(query)

        if not need_retrieval:
            # 直接生成（事实性强的问题不需要检索）
            answer = await self.llm.complete(query)
            return {"answer": answer, "retrieved": False, "docs": []}

        # Step 2: 检索文档
        raw_docs = self.retriever.search(query, top_k=10)
        reranked_docs = self.reranker.rerank(query, raw_docs, top_k=5)

        # Step 3: 过滤不相关文档
        relevant_docs = await self._filter_relevant(query, reranked_docs)

        if not relevant_docs:
            # 检索无果，提示用户
            return {
                "answer": "根据知识库内容，无法回答此问题。",
                "retrieved": True,
                "docs": []
            }

        # Step 4: 生成答案
        answer = await self._generate_with_context(query, relevant_docs)

        # Step 5: 验证答案忠实度
        faithfulness_score = await self._check_faithfulness(
            query, answer, relevant_docs
        )

        if faithfulness_score < 0.7:
            # 答案可信度低，重新生成
            answer = await self._regenerate_conservative(
                query, relevant_docs
            )

        return {
            "answer": answer,
            "retrieved": True,
            "docs": relevant_docs,
            "faithfulness": faithfulness_score
        }

    async def _check_faithfulness(
        self,
        query: str,
        answer: str,
        docs: list[dict]
    ) -> float:
        """NLI方式验证答案是否有文档支撑"""

        context = "\n\n".join([d["text"] for d in docs])

        prompt = f"""
判断以下"答案"是否完全由"上下文"所支持。
请只输出0到1之间的数字（1=完全支持，0=完全无支持）。

上下文：
{context}

答案：{answer}

忠实度分数：
"""
        score_str = await self.llm.complete(prompt)
        try:
            return float(score_str.strip())
        except:
            return 0.5

四、GraphRAG知识图谱增强

对于需要跨文档推理的复杂问题，GraphRAG通过构建知识图谱大幅提升性能：

from neo4j import GraphDatabase

class GraphRAGEnhancer:
    """GraphRAG知识图谱检索增强"""

    def __init__(self, neo4j_uri, neo4j_auth):
        self.driver = GraphDatabase.driver(neo4j_uri, auth=neo4j_auth)

    def multi_hop_search(self, entities: list[str], hops: int = 2) -> str:
        """多跳图检索，支持跨文档推理"""

        query = """
        MATCH path = (start:Entity)-[*1..{hops}]-(end:Entity)
        WHERE start.name IN $entities
        WITH path, relationships(path) as rels
        RETURN path, 
               [r IN rels | r.evidence] as evidence_list
        LIMIT 20
        """.format(hops=hops)

        with self.driver.session() as session:
            results = session.run(query, entities=entities)

            context_pieces = []
            for record in results:
                for evidence in record["evidence_list"]:
                    if evidence:
                        context_pieces.append(evidence)

            return "\n".join(set(context_pieces))  # 去重

五、RAG 2.0性能基准

在企业内部知识库QA任务上的测试结果（1000个问题）：

从65%到91%的准确率提升，是RAG 2.0对企业知识库应用的核心价值。

六、生产部署建议

1. 分级检索策略：简单问题用基础RAG（快），复杂问题用完整RAG 2.0流水线（准）
2. 缓存机制：相同或相似查询缓存检索结果（可降低70%重复检索开销）
3. 异步重排序：将重排序放到后台，先返回初步结果，再更新精排结果
4. 知识库更新：增量索引 + 定期全量重建，确保知识时效性
5. 评估体系：部署RAGAs等自动评估框架，持续监控系统质量

RAG 2.0不是单一技术，而是一套面向生产的系统工程。掌握混合检索、重排序和Self-RAG的组合拳，是2026年企业AI知识库建设的核心竞争力。