生产级RAG系统构建实践：多路召回、融合重排与自我修正架构详解

客户端请求 (HTTP/WebSocket)
    ↓
FastAPI 接入层（限流、鉴权、日志）
    ↓
路由调度器（基于意图识别，选择检索策略）
    ├─ 常规问答 → 三路召回
    ├─ 代码/公式 → 增强符号检索
    ├─ 表格查询 → 结构化检索（SQL/Elasticsearch）
    └─ 多跳推理 → 图数据库路径检索
    ↓
多路召回并行执行（Dense + Sparse + Graph）
    ↓
融合与重排序（RRF + Cross-Encoder）
    ↓
上下文构建 + 大模型生成（vLLM推理）
    ↓
自我修正模块（幻觉检测与修正）
    ↓
流式/非流式返回

# retrieval/base.py
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
from dataclasses import dataclass

@dataclass
class Chunk:
    id: str
    text: str
    metadata: Dict[str, Any]
    score: float

class BaseRetriever(ABC):
    @abstractmethod
    async def retrieve(self, query: str, top_k: int = 10) -> List[Chunk]:
        pass

    @abstractmethod
    async def batch_retrieve(self, queries: List[str], top_k: int = 10) -> List[List[Chunk]]:
        pass

# retrieval/dense_retriever.py
import numpy as np
from vllm import LLM, SamplingParams
from vllm.embeddings import EmbeddingPoolingMode
import faiss
from pathlib import Path

class BGEM3DenseRetriever(BaseRetriever):
    def __init__(self, model_name: str = "BAAI/bge-m3", index_path: str = "./data/faiss_index"):
        self.llm = LLM(
            model=model_name,
            task="embedding",
            pooling_mode=EmbeddingPoolingMode.CLS,
            max_model_len=8192,
            tensor_parallel_size=1,
            dtype="float16",
            enable_prefix_caching=True,
        )
        self.index = faiss.read_index(str(Path(index_path) / "hnsw_pq.index"))
        self.chunk_ids = np.load(Path(index_path) / "chunk_ids.npy")
        self.chunk_metadata = np.load(Path(index_path) / "metadata.npy", allow_pickle=True)

    async def retrieve(self, query: str, top_k: int = 10) -> List[Chunk]:
        outputs = self.llm.encode([query])
        emb = np.array(outputs[0].outputs.embedding, dtype=np.float32).reshape(1, -1)
        faiss.normalize_L2(emb)
        distances, indices = self.index.search(emb, top_k)
        chunks = []
        for idx, dist in zip(indices[0], distances[0]):
            sim = 1.0 / (1.0 + dist)   # 距离转相似度
            meta = self.chunk_metadata[idx]
            chunks.append(Chunk(
                id=str(self.chunk_ids[idx]),
                text=meta.get("text", ""),
                metadata=meta,
                score=sim
            ))
        return chunks

# retrieval/bm25_retriever.py
from elasticsearch import AsyncElasticsearch

class BM25Retriever(BaseRetriever):
    def __init__(self, hosts: str = "localhost:9200", index_name: str = "doc_chunks"):
        self.es = AsyncElasticsearch(hosts=hosts)
        self.index_name = index_name

    async def retrieve(self, query: str, top_k: int = 10) -> List[Chunk]:
        resp = await self.es.search(
            index=self.index_name,
            body={
                "query": {"match": {"text": {"query": query, "operator": "and"}}},
                "size": top_k,
                "_source": ["text", "metadata"]
            }
        )
        chunks = []
        for hit in resp["hits"]["hits"]:
            src = hit["_source"]
            chunks.append(Chunk(
                id=hit["_id"],
                text=src["text"],
                metadata=src.get("metadata", {}),
                score=hit["_score"]
            ))
        return chunks

# retrieval/graph_retriever.py
from neo4j import AsyncGraphDatabase

class GraphRetriever(BaseRetriever):
    def __init__(self, uri: str, user: str, password: str):
        self.driver = AsyncGraphDatabase.driver(uri, auth=(user, password))

    async def retrieve(self, query: str, top_k: int = 10) -> List[Chunk]:
        # 简单实体识别（示例），实际使用NER模型提取实体
        entities = self._extract_entities(query)  
        cypher = """
        MATCH (e:Entity)-[r:RELATED_TO]->(n)
        WHERE e.name IN $entities
        RETURN n.chunk_id as cid, n.text as text, n.metadata as meta
        LIMIT $limit
        """
        async with self.driver.session() as session:
            result = await session.run(cypher, entities=entities, limit=top_k)
            records = await result.data()
        chunks = []
        for rec in records:
            chunks.append(Chunk(
                id=rec["cid"],
                text=rec["text"],
                metadata=rec["meta"],
                score=1.0   # 图召回不设距离分数，后续由融合层调整
            ))
        return chunks

# retrieval/fusion.py
import torch
import torch.nn as nn
from typing import List

class LearnableRRFFusion:
    def __init__(self, num_retrievers: int, k: int = 60):
        self.k = k
        self.weights = nn.Parameter(torch.ones(num_retrievers, dtype=torch.float32))

    def fuse(self, retriever_results: List[List[Chunk]], top_k: int) -> List[Chunk]:
        w = torch.softmax(self.weights, dim=0).detach().numpy()
        rank_dict = {}
        for ridx, chunks in enumerate(retriever_results):
            for rank, chunk in enumerate(chunks):
                if chunk.id not in rank_dict:
                    rank_dict[chunk.id] = {}
                rank_dict[chunk.id][ridx] = rank

        fusion_scores = {}
        for cid, rid_rank in rank_dict.items():
            score = 0.0
            for ridx, rank in rid_rank.items():
                score += w[ridx] / (self.k + rank + 1)
            fusion_scores[cid] = score

        sorted_ids = sorted(fusion_scores.keys(), key=lambda x: fusion_scores[x], reverse=True)[:top_k]
        # 合并各检索器返回的Chunk元数据（优先取第一个）
        id_to_chunk = {}
        for chunks in retriever_results:
            for ch in chunks:
                if ch.id not in id_to_chunk:
                    id_to_chunk[ch.id] = ch
        return [id_to_chunk[cid] for cid in sorted_ids]

# rerank/cross_encoder.py
from transformers import AutoModelForSequenceClassification, AutoTokenizer
import torch

class CrossEncoderReranker:
    def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3", device: str = "cuda:0"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name).to(device)
        self.model.eval()
        self.device = device

    @torch.no_grad()
    def rerank(self, query: str, candidates: List[Chunk], top_k: int = 5) -> List[Chunk]:
        pairs = [(query, chunk.text) for chunk in candidates]
        inputs = self.tokenizer(
            pairs,
            padding=True,
            truncation=True,
            max_length=512,
            return_tensors="pt"
        ).to(self.device)
        scores = self.model(**inputs).logits.squeeze(-1).sigmoid().cpu().numpy()
        for ch, score in zip(candidates, scores):
            ch.score = float(score)
        candidates.sort(key=lambda x: x.score, reverse=True)
        return candidates[:top_k]

# correction/hallucination_detector.py
from transformers import pipeline

class HallucinationCorrector:
    def __init__(self, model_name: str = "Qwen/Qwen2.5-7B-Instruct"):
        self.detector = pipeline(
            "text-generation",
            model=model_name,
            device_map="auto",
            torch_dtype=torch.bfloat16
        )
        self.prompt_template = """
你是一个事实核查专家。给定检索上下文(Context)和生成答案(Answer)，判断每条陈述是否被支持。
输出格式：
[支持] 陈述
[不支持] 陈述 -> 修正建议: xxx
[部分支持] 陈述 -> 补充: xxx

Context: {context}

Answer: {answer}

核查结果：
"""

    def correct(self, answer: str, context_chunks: List[Chunk]) -> str:
        context = "\n".join([ch.text for ch in context_chunks[:3]])
        prompt = self.prompt_template.format(context=context, answer=answer)
        outputs = self.detector(prompt, max_new_tokens=512, do_sample=False)
        feedback = outputs[0]["generated_text"][len(prompt):]
        if "[不支持]" in feedback or "[部分支持]" in feedback:
            # 提取所有支持的部分
            supported_sentences = []
            for line in feedback.split("\n"):
                if "[支持]" in line:
                    supported_sentences.append(line.replace("[支持]", "").strip())
            # 重新生成答案（仅保留支持句）
            if supported_sentences:
                new_answer = ". ".join(supported_sentences)
                return new_answer
            else:
                # 若无支持句，返回原始答案并标记警告
                return answer + " [警告：内容可能缺乏事实依据]"
        return answer

# api/rag_endpoint.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio, time, uuid

app = FastAPI()

class QueryRequest(BaseModel):
    query: str
    top_k_retrieve: int = 20
    top_k_rerank: int = 5
    enable_correction: bool = True
    stream: bool = False

# 初始化全局组件（单例）
dense = BGEM3DenseRetriever()
bm25 = BM25Retriever()
graph = GraphRetriever(uri="bolt://localhost:7687", user="neo4j", password="pass")
fusion = LearnableRRFFusion(num_retrievers=3)
reranker = CrossEncoderReranker()
corrector = HallucinationCorrector()
# vLLM生成器封装（省略）
llm = vLLMGenerator(model="Qwen/Qwen2.5-72B-Instruct", quantization="awq")

@app.post("/v1/rag/query")
async def rag_query(request: QueryRequest):
    start = time.perf_counter()
    qid = str(uuid.uuid4())

    # 并行召回
    tasks = [
        dense.retrieve(request.query, request.top_k_retrieve),
        bm25.retrieve(request.query, request.top_k_retrieve),
        graph.retrieve(request.query, request.top_k_retrieve)
    ]
    results = await asyncio.gather(*tasks)

    # 融合（返回top_k_retrieve个）
    fused = fusion.fuse(results, request.top_k_retrieve)

    # 重排序（取top_k_rerank个）
    reranked = reranker.rerank(request.query, fused, request.top_k_rerank)

    # 构建上下文
    context = "\n\n".join([f"[{i+1}] {ch.text}" for i, ch in enumerate(reranked)])

    # 生成
    if request.stream:
        # 实现流式生成（使用FastAPI StreamingResponse）
        pass
    else:
        answer = llm.generate(
            system="你是一个专业助手，基于给定上下文回答问题。",
            user=f"上下文：\n{context}\n\n问题：{request.query}",
            max_tokens=1024
        )

    # 修正
    if request.enable_correction:
        answer = corrector.correct(answer, reranked)

    elapsed = (time.perf_counter() - start) * 1000
    return {
        "query_id": qid,
        "answer": answer,
        "chunks_used": [ch.__dict__ for ch in reranked],
        "latency_ms": elapsed
    }

# vllm_config.yaml
model: Qwen/Qwen2.5-72B-Instruct
tensor_parallel_size: 4
gpu_memory_utilization: 0.92
max_num_seqs: 128
max_model_len: 8192
enable_prefix_caching: true
quantization: awq
swap_space: 16  # GB

# 索引构建示例
dim = 1024
quantizer = faiss.IndexHNSWFlat(dim, 32)
index = faiss.IndexIVFPQ(quantizer, dim, 4096, 32, 8)
index.train(training_vectors)   # 需至少 4096*32 条数据
index.add(vectors)
index.nprobe = 64

# 每日更新任务
from bayes_opt import BayesianOptimization

def update_weights(history_data):
    def objective(w1, w2, w3):
        # 临时设置权重并计算在历史数据上的MRR
        return compute_mrr_with_weights([w1, w2, w3], history_data)
    
    optimizer = BayesianOptimization(
        f=objective,
        pbounds={'w1': (0.1, 2.0), 'w2': (0.1, 2.0), 'w3': (0.1, 2.0)}
    )
    optimizer.maximize(init_points=10, n_iter=20)
    best = optimizer.max['params']
    fusion.weights.data = torch.tensor([best['w1'], best['w2'], best['w3']])
    # 保存权重到持久化存储