大模型RAG全鏈路實戰：從零構建生產級檢索增強生成系統

摘要

RAG（檢索增強生成）是大模型落地生產的核心架構，解決幻覺、知識過時、領域缺失三大痛點
生產級RAG的5大關鍵環節：文件解析→分塊策略→Embedding→向量檢索→重排序，每環節都有優化空間
智慧分塊策略（語意分塊+滑動視窗）比固定長度分塊的檢索召回率提升20-30%
混合檢索（向量+關鍵詞+知識圖譜）比純向量檢索的召回率提升15-25%
重排序模型（BGE-Reranker/Cohere Rerank）將最終答案準確率從75%提升到90%+
本文提供從文件處理到生產部署的完整RAG方案，含Python實現與效能基準測試

RAG為什麼是大模型落地的核心架構

大模型有三大固有缺陷：幻覺（生成不存在的事實）、知識過時（訓練資料截止後無法取得新知識）、領域缺失（缺乏垂直領域專業知識）。RAG透過在生成前檢索外部知識庫，將相關上下文注入Prompt，從根源上解決這三大問題。

┌──────────────────────────────────────────────────────────────────┐
│              RAG系統全鏈路架構                                     │
│                                                                    │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐  │
│  │ 1.文件解析│──→│ 2.智慧分塊│──→│ 3.Embed  │──→│ 4.向量儲存│  │
│  │ PDF/DOCX │    │ 語意分塊  │    │ 管線     │    │ Milvus   │  │
│  │ HTML/MD  │    │ 滑動視窗  │    │ 批量嵌入  │    │ HNSW索引 │  │
│  └──────────┘    └──────────┘    └──────────┘    └──────────┘  │
│                                                       │          │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐         │          │
│  │ 8.答案生成│←──│ 7.Prompt │←──│ 6.重排序  │←────────┤          │
│  │ LLM生成  │    │ 工程組裝  │    │ BGE-Rerank│  5.混合檢索│      │
│  │ 引用溯源  │    │ 上下文視窗│    │ Top-K過濾 │  向量+BM25│      │
│  └──────────┘    └──────────┘    └──────────┘    └──────────┘  │
└──────────────────────────────────────────────────────────────────┘

RAG vs 純LLM關鍵指標對比

維度	純LLM	RAG增強LLM
事實準確率	60-70%	90-95%
幻覺率	15-30%	3-5%
領域知識	通用	可定製
知識更新	需重新訓練	增量更新知識庫
可解釋性	低	高（引用溯源）
成本	高（大模型推理）	中（檢索+小模型推理）

文件解析與智慧分塊

智慧分塊策略

class SemanticChunker:
    def __init__(self, embedding_client, max_chunk_tokens: int = 512, overlap_tokens: int = 64, similarity_threshold: float = 0.75):
        self.embedding_client = embedding_client
        self.max_chunk_tokens = max_chunk_tokens
        self.overlap_tokens = overlap_tokens
        self.similarity_threshold = similarity_threshold

    async def chunk_document(self, doc: ParsedDocument) -> list[Chunk]:
        sentences = self._split_sentences(doc.content)
        embeddings = await self._batch_embed(sentences)
        chunks = []
        current_chunk = [sentences[0]]
        current_tokens = self._count_tokens(sentences[0])

        for i in range(1, len(sentences)):
            similarity = self._cosine_similarity(embeddings[i - 1], embeddings[i])
            if similarity < self.similarity_threshold or current_tokens + self._count_tokens(sentences[i]) > self.max_chunk_tokens:
                chunk_content = ' '.join(current_chunk)
                chunks.append(Chunk(chunk_id=f"{doc.doc_id}_c{len(chunks)}", content=chunk_content, metadata={**doc.metadata, "chunk_index": len(chunks)}, token_count=current_tokens))
                overlap_start = max(0, len(current_chunk) - self._sentences_for_tokens(self.overlap_tokens))
                current_chunk = current_chunk[overlap_start:] + [sentences[i]]
                current_tokens = sum(self._count_tokens(s) for s in current_chunk)
            else:
                current_chunk.append(sentences[i])
                current_tokens += self._count_tokens(sentences[i])

        if current_chunk:
            chunks.append(Chunk(chunk_id=f"{doc.doc_id}_c{len(chunks)}", content=' '.join(current_chunk), metadata={**doc.metadata, "chunk_index": len(chunks)}, token_count=current_tokens))
        return chunks

分塊策略對比

策略	召回率	上下文完整度	實作複雜度
固定長度(512 tokens)	65%	低（截斷語意）	簡單
段落分塊	72%	中	簡單
語意分塊	85%	高	中等
語意分塊+滑動視窗	92%	高	中等

Embedding管線與向量索引構建

批量Embedding管線

class EmbeddingPipeline:
    def __init__(self, model: str = "BAAI/bge-large-zh-v1.5", batch_size: int = 64):
        self.client = AsyncOpenAI(base_url="http://localhost:8000/v1")
        self.model = model
        self.batch_size = batch_size

    async def embed_chunks(self, chunks: list[Chunk]) -> list[dict]:
        all_embeddings = []
        for i in range(0, len(chunks), self.batch_size):
            batch = chunks[i:i + self.batch_size]
            texts = [f"檢索查詢：{c.content}" for c in batch]
            response = await self.client.embeddings.create(model=self.model, input=texts)
            for j, item in enumerate(response.data):
                all_embeddings.append({"chunk_id": batch[j].chunk_id, "content": batch[j].content, "metadata": batch[j].metadata, "embedding": item.embedding, "token_count": batch[j].token_count})
        return all_embeddings

向量索引構建

class RAGVectorStore:
    def __init__(self, milvus_uri: str = "http://localhost:19530", collection_name: str = "rag_chunks"):
        self.client = MilvusClient(uri=milvus_uri)
        self.collection_name = collection_name
        self._ensure_collection()

    def _ensure_collection(self):
        if self.client.has_collection(self.collection_name):
            return
        schema = self.client.create_schema(auto_id=True, enable_dynamic_field=True)
        schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
        schema.add_field(field_name="chunk_id", datatype=DataType.VARCHAR, max_length=256)
        schema.add_field(field_name="content", datatype=DataType.VARCHAR, max_length=65535)
        schema.add_field(field_name="embedding", datatype=DataType.FLOAT_VECTOR, dim=1024)
        schema.add_field(field_name="source", datatype=DataType.VARCHAR, max_length=512)
        index_params = self.client.prepare_index_params()
        index_params.add_index(field_name="embedding", index_type="HNSW", metric_type="COSINE", params={"M": 16, "efConstruction": 200})
        self.client.create_collection(collection_name=self.collection_name, schema=schema, index_params=index_params)

    def search(self, query_embedding: list[float], top_k: int = 10, ef: int = 100) -> list[dict]:
        results = self.client.search(self.collection_name, data=[query_embedding], limit=top_k, output_fields=["chunk_id", "content", "source"], search_params={"metric_type": "COSINE", "params": {"ef": ef}})
        return [{"chunk_id": r["entity"]["chunk_id"], "content": r["entity"]["content"], "source": r["entity"]["source"], "score": r["distance"]} for r in results[0]]

混合檢索與重排序

混合檢索：向量+BM25

class HybridRetriever:
    def __init__(self, vector_store: RAGVectorStore, bm25: BM25Retriever, vector_weight: float = 0.7, bm25_weight: float = 0.3):
        self.vector_store = vector_store
        self.bm25 = bm25
        self.vector_weight = vector_weight
        self.bm25_weight = bm25_weight

    async def search(self, query: str, query_embedding: list[float], top_k: int = 10) -> list[dict]:
        vector_results = self.vector_store.search(query_embedding, top_k=top_k * 2)
        bm25_results = self.bm25.search(query, top_k=top_k * 2)
        merged: dict[str, dict] = {}
        for r in vector_results:
            merged[r["chunk_id"]] = {**r, "vector_score": r["score"], "bm25_score": 0.0}
        for r in bm25_results:
            if r["chunk_id"] in merged:
                merged[r["chunk_id"]]["bm25_score"] = r["score"]
            else:
                merged[r["chunk_id"]] = {**r, "vector_score": 0.0, "bm25_score": r["score"]}
        max_vector = max((m["vector_score"] for m in merged.values()), default=1.0) or 1.0
        max_bm25 = max((m["bm25_score"] for m in merged.values()), default=1.0) or 1.0
        for m in merged.values():
            m["combined_score"] = self.vector_weight * m["vector_score"] / max_vector + self.bm25_weight * m["bm25_score"] / max_bm25
        results = sorted(merged.values(), key=lambda x: x["combined_score"], reverse=True)
        return results[:top_k]

重排序模型

class Reranker:
    def __init__(self, llm_client, model: str = "BAAI/bge-reranker-v2-m3"):
        self.llm = llm_client
        self.model = model

    async def rerank(self, query: str, candidates: list[dict], top_k: int = 5) -> list[dict]:
        pairs = [[query, c["content"][:512]] for c in candidates]
        scores = await self._compute_scores(pairs)
        for i, candidate in enumerate(candidates):
            candidate["rerank_score"] = scores[i]
        candidates.sort(key=lambda x: x["rerank_score"], reverse=True)
        return candidates[:top_k]

RAG全鏈路優化策略

RAG全鏈路效能基準

環節	耗時(P50)	耗時(P99)	說明
文件解析(PDF 10頁)	500ms	1.5s	PyMuPDF
語意分塊(10頁)	2s	5s	含Embedding呼叫
批量Embedding(64 chunks)	800ms	2s	BGE-large
向量檢索(top-10)	5ms	15ms	Milvus HNSW
BM25檢索(top-10)	2ms	5ms	記憶體索引
混合檢索融合	1ms	3ms	分數歸一化
重排序(top-5)	200ms	500ms	BGE-Reranker
LLM生成(7B)	1.5s	3s	Qwen2.5-7B
端到端RAG	3s	6s	完整鏈路

檢索召回率對比

檢索方式	Top-5召回率	Top-10召回率	Top-20召回率
純向量檢索	72%	82%	88%
純BM25	65%	75%	80%
混合檢索	82%	90%	95%
混合+重排序	88%	94%	97%

生產部署與可觀測性

RAG服務K8s部署

apiVersion: apps/v1
kind: Deployment
metadata:
  name: rag-service
  namespace: ai-rag
spec:
  replicas: 2
  selector:
    matchLabels:
      app: rag-service
  template:
    metadata:
      labels:
        app: rag-service
    spec:
      containers:
        - name: rag-api
          image: myregistry/rag-service:v1.0
          ports:
            - containerPort: 8000
          resources:
            requests:
              cpu: "2"
              memory: 4Gi
            limits:
              cpu: "4"
              memory: 8Gi
          env:
            - name: MILVUS_URI
              value: "http://milvus:19530"
            - name: LLM_API_BASE
              value: "http://vllm-qwen2-72b:8000/v1"

總結與引流

RAG是大模型落地生產的核心架構。5大關鍵環節（文件解析→分塊→Embedding→檢索→重排序）每環節都有優化空間。語意分塊+滑動視窗比固定分塊召回率提升20-30%，混合檢索比純向量檢索召回率提升15-25%，重排序將最終準確率從75%提升到90%+。

開發要點回顧：

文件解析：PDF用PyMuPDF，Markdown按標題分節，HTML用BeautifulSoup去噪
智慧分塊：語意分塊+滑動視窗，max_chunk_tokens=512, overlap_tokens=64
Embedding：BGE-large-zh-v1.5，批量64，查詢前綴「檢索查詢：」
混合檢索：向量權重0.7 + BM25權重0.3，分數歸一化後加權融合
重排序：BGE-Reranker-v2-m3，從top-20候選中精選top-5

相關閱讀：

Rust向量資料庫內核架構與效能優化實戰 — RAG檢索後端的向量索引優化
K8s 1.30+大模型推理彈性調度實戰 — RAG推理服務的K8s編排
AI Agent多智能體編排實戰 — Agent系統中的RAG整合

權威參考：