RAG 架构实战：从零搭建 Embedding、检索与生成的知识问答系统

系统性讲解 RAG（检索增强生成）完整架构：Embedding 模型选型（text-embedding-3/GTE/E5）、文档分块策略（固定/语义/递归重叠）、混合检索（dense+BM25）、重排序（Reranker）、LangChain vs LlamaIndex 实现对比。附带端到端 FastAPI 服务代码，基于 Qdrant + sentence-transformers。

spcent 6 分钟阅读 2998 字

目录

  1. RAG 架构全景
  2. Embedding 模型选型
  3. 文档分块策略
  4. 向量检索与混合搜索
  5. 重排序（Reranker）
  6. LangChain RAG 实现
  7. LlamaIndex RAG 实现
  8. 端到端 FastAPI RAG 服务
  9. RAG 评估与优化

1. RAG 架构全景

RAG（Retrieval-Augmented Generation）通过将外部知识检索与 LLM 生成结合，解决 hallucination（幻觉）问题。

1.1 标准 RAG 流程

用户查询 → Query Embedding → 向量检索 → Top-K 文档 → Rerank → 上下文拼接 → LLM 生成

from dataclasses import dataclass
from typing import List

@dataclass
class RAGContext:
    query: str
    retrieved_chunks: List[str]
    scores: List[float]
    context_window: str  # 拼接后的上下文

class RAGPipeline:
    """标准 RAG 流水线。"""
    def __init__(self, embedder, retriever, llm_client):
        self.embedder = embedder
        self.retriever = retriever
        self.llm = llm_client

    async def answer(self, query: str, top_k: int = 5) -> str:
        # 1. Query → Embedding
        query_embedding = self.embedder.encode(query)

        # 2. 检索 Top-K
        chunks, scores = self.retriever.search(query_embedding, top_k)

        # 3. 构建上下文
        context = "\n\n---\n\n".join(
            f"[Document {i+1}]\n{chunk}" for i, chunk in enumerate(chunks)
        )

        # 4. LLM 生成
        prompt = f"""Answer the question based on the provided context.
If the answer is not in the context, say "I don't know".

Context:
{context}

Question: {query}

Answer:"""

        return await self.llm.chat_completion([{"role": "user", "content": prompt}])

2. Embedding 模型选型

模型维度最大长度语言性能（MTEB）许可适用
text-embedding-3-small15368192多语言62.3%商业通用，低成本
text-embedding-3-large30728192多语言64.6%商业高质量需求
BGE-M310248192多语言72.3%MIT开源首选
GTE-large1024512多语言69.3%MIT平衡速度与效果
E5-mistral-7b-instruct409632768多语言74.1%MIT长文档，高质量
multilingual-e5-large1024512多语言68.3%MIT中文场景

2.1 本地 Embedding 推理

from sentence_transformers import SentenceTransformer
import numpy as np

class LocalEmbedder:
    """基于 sentence-transformers 的本地嵌入模型。"""
    def __init__(self, model_name: str = "BAAI/bge-m3"):
        self.model = SentenceTransformer(model_name)
        self.dim = self.model.get_sentence_embedding_dimension()

    def encode(self, texts: list[str] | str, normalize: bool = True) -> np.ndarray:
        if isinstance(texts, str):
            texts = [texts]
        # BGE 模型需要添加指令前缀
        instruction = "Represent this sentence for searching relevant passages: "
        texts = [instruction + t for t in texts]
        embeddings = self.model.encode(texts, convert_to_numpy=True)
        if normalize:
            embeddings = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
        return embeddings

    def encode_queries(self, queries: list[str]) -> np.ndarray:
        """Query 使用不同指令。"""
        instruction = "Represent this query for retrieving relevant documents: "
        texts = [instruction + q for q in queries]
        embeddings = self.model.encode(texts, convert_to_numpy=True)
        return embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)

2.2 多模态 Embedding（图像+文本）

from sentence_transformers import SentenceTransformer

class MultimodalEmbedder:
    """CLIP 风格的多模态嵌入。"""
    def __init__(self):
        self.model = SentenceTransformer("clip-ViT-B-32")

    def encode_image(self, image_path: str) -> np.ndarray:
        from PIL import Image
        img = Image.open(image_path)
        return self.model.encode(img)

    def encode_text(self, text: str) -> np.ndarray:
        return self.model.encode(text)

3. 文档分块策略

分块质量直接影响 RAG 效果——太大丢失细节，太小丢失上下文。

策略方法适用缺点
固定大小每 N 个 token/字符简单场景句子截断
递归文本按段落→句子→词递归切分通用文档实现复杂
语义分块按语义边界（句号/主题变化）高质量需求计算开销
重叠窗口固定大小 + M token 重叠避免边界信息丢失冗余增加
代理分块LLM 判断最佳切分点最高质量Token 成本高

3.1 递归文本分块（推荐）

from typing import List
import re

class RecursiveTextSplitter:
    """LangChain 风格递归文本分块。"""
    def __init__(
        self,
        chunk_size: int = 500,
        chunk_overlap: int = 50,
        separators: List[str] = None,
    ):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.separators = separators or ["\n\n", "\n", ". ", "? ", "! ", " ", ""]

    def split_text(self, text: str) -> List[str]:
        return self._split_recursive(text, self.separators)

    def _split_recursive(self, text: str, separators: List[str]) -> List[str]:
        if len(text) <= self.chunk_size:
            return [text] if text.strip() else []

        separator = separators[0] if separators else ""
        if separator:
            parts = text.split(separator)
        else:
            parts = list(text)

        chunks = []
        current = ""
        for part in parts:
            candidate = current + (separator if current else "") + part
            if len(candidate) <= self.chunk_size:
                current = candidate
            else:
                if current:
                    chunks.append(current)
                    # 保留重叠部分
                    if self.chunk_overlap > 0:
                        current = current[-self.chunk_overlap:] + separator + part
                    else:
                        current = part
                else:
                    # 单个部分超过 chunk_size，递归拆分
                    sub = self._split_recursive(part, separators[1:])
                    chunks.extend(sub)
                    current = ""

        if current:
            chunks.append(current)

        return chunks

# 按 Markdown 标题智能分块
class MarkdownSplitter:
    def __init__(self, chunk_size: int = 1000):
        self.chunk_size = chunk_size

    def split(self, markdown: str) -> List[dict]:
        """按标题层级分块，保留标题元数据。"""
        chunks = []
        current_lines = []
        current_headers = []

        for line in markdown.split("\n"):
            if line.startswith("#"):
                if current_lines:
                    chunks.append({
                        "content": "\n".join(current_lines),
                        "headers": current_headers.copy(),
                    })
                level = len(line.split()[0])
                current_headers = current_headers[:level-1] + [line.strip()]
                current_lines = []
            else:
                current_lines.append(line)

        if current_lines:
            chunks.append({
                "content": "\n".join(current_lines),
                "headers": current_headers,
            })

        return chunks

3.2 按 Token 计数精确分块

import tiktoken

class TokenAwareSplitter:
    def __init__(self, chunk_size: int = 512, chunk_overlap: int = 50, model: str = "gpt-4o"):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.encoding = tiktoken.encoding_for_model(model)

    def split(self, text: str) -> List[str]:
        tokens = self.encoding.encode(text)
        chunks = []
        start = 0
        while start < len(tokens):
            end = min(start + self.chunk_size, len(tokens))
            chunk_tokens = tokens[start:end]
            chunks.append(self.encoding.decode(chunk_tokens))
            start += self.chunk_size - self.chunk_overlap
        return chunks

4. 向量检索与混合搜索

4.1 Dense 向量检索（Qdrant）

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

class QdrantStore:
    """Qdrant 向量存储封装。"""
    def __init__(self, host: str = "localhost", port: int = 6333, collection: str = "documents"):
        self.client = QdrantClient(host=host, port=port)
        self.collection = collection

    def create_collection(self, dim: int):
        self.client.recreate_collection(
            collection_name=self.collection,
            vectors_config=VectorParams(size=dim, distance=Distance.COSINE),
        )

    def upsert(self, chunks: list[str], embeddings: np.ndarray, metadata: list[dict] = None):
        points = [
            PointStruct(
                id=i,
                vector=emb.tolist(),
                payload={"text": chunk, **(metadata[i] if metadata else {})},
            )
            for i, (emb, chunk) in enumerate(zip(embeddings, chunks))
        ]
        self.client.upsert(collection_name=self.collection, points=points)

    def search(self, query_embedding: np.ndarray, top_k: int = 5) -> list[dict]:
        results = self.client.search(
            collection_name=self.collection,
            query_vector=query_embedding.tolist(),
            limit=top_k,
        )
        return [
            {"text": r.payload["text"], "score": r.score, "metadata": r.payload}
            for r in results
        ]

4.2 Hybrid Search：Dense + Sparse（BM25）

from qdrant_client.models import SparseVector

class HybridRetriever:
    """混合检索：Dense 语义 + Sparse 词汇匹配。"""
    def __init__(self, qdrant_store, sparse_model):
        self.dense_store = qdrant_store
        self.sparse_model = sparse_model  # 如 splade或 bm25

    def search(self, query: str, query_emb: np.ndarray, top_k: int = 10) -> list[dict]:
        # Dense 检索
        dense_results = self.dense_store.search(query_emb, top_k=top_k)

        # Sparse 检索（BM25）
        sparse_results = self.sparse_model.search(query, top_k=top_k)

        # Reciprocal Rank Fusion (RRF) 融合
        return self._rrf_fusion(dense_results, sparse_results, k=60)

    def _rrf_fusion(self, dense: list, sparse: list, k: int = 60) -> list[dict]:
        """RRF：基于排名的融合算法。"""
        scores = {}

        for rank, r in enumerate(dense):
            doc_id = r["metadata"].get("doc_id", r["text"][:50])
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
            if doc_id not in scores:
                scores[doc_id] = {"text": r["text"], "metadata": r["metadata"]}

        for rank, r in enumerate(sparse):
            doc_id = r["metadata"].get("doc_id", r["text"][:50])
            if doc_id in scores:
                if isinstance(scores[doc_id], dict) and "text" in scores[doc_id]:
                    scores[doc_id] = 1 / (k + rank + 1)
                else:
                    scores[doc_id] += 1 / (k + rank + 1)
            else:
                scores[doc_id] = 1 / (k + rank + 1)
                scores[doc_id] = {"text": r["text"], "metadata": r["metadata"], "score": scores[doc_id]}

        # 排序返回
        sorted_docs = sorted(
            [(k, v) for k, v in scores.items() if isinstance(v, (int, float)) or "score" in v],
            key=lambda x: x[1] if isinstance(x[1], (int, float)) else x[1]["score"],
            reverse=True,
        )
        return [doc for _, doc in sorted_docs[:10]]

5. 重排序（Reranker）

检索的 Top-K 结果顺序可能不够精确，使用轻量级 Reranker 模型对 Top-K 重新排序。

from sentence_transformers import CrossEncoder

class Reranker:
    """交叉编码器重排序。"""
    def __init__(self, model_name: str = "BAAI/bge-reranker-large"):
        self.model = CrossEncoder(model_name)

    def rerank(self, query: str, documents: list[str], top_k: int = 5) -> list[tuple[str, float]]:
        """对文档列表重排序，返回 (doc, score)。"""
        pairs = [(query, doc) for doc in documents]
        scores = self.model.predict(pairs)

        ranked = sorted(
            zip(documents, scores),
            key=lambda x: x[1],
            reverse=True,
        )
        return ranked[:top_k]

典型 RAG 检索+重排序流程：

async def advanced_retrieve(query: str, top_k: int = 5) -> list[str]:
    embedder = LocalEmbedder()
    retriever = QdrantStore()
    reranker = Reranker()

    # 1. Query embedding
    query_emb = embedder.encode_queries([query])

    # 2. 粗排：向量检索 Top-20
    candidates = retriever.search(query_emb[0], top_k=20)
    candidate_texts = [c["text"] for c in candidates]

    # 3. 精排：Reranker Top-5
    ranked = reranker.rerank(query, candidate_texts, top_k=top_k)
    return [doc for doc, _ in ranked]

6. LangChain RAG 实现

from langchain_community.vectorstores import Qdrant
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

class LangChainRAG:
    def __init__(self, qdrant_url: str):
        embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-m3")
        self.vectorstore = Qdrant.from_existing_collection(
            embedding=embeddings,
            collection_name="documents",
            url=qdrant_url,
        )

        llm = ChatOpenAI(model="gpt-4o", temperature=0)

        prompt = PromptTemplate.from_template("""
You are a helpful assistant. Use the following context to answer the question.
If you don't know, say "I don't know".

Context:
{context}

Question: {question}

Answer:""")

        self.qa_chain = RetrievalQA.from_chain_type(
            llm=llm,
            chain_type="stuff",
            retriever=self.vectorstore.as_retriever(search_kwargs={"k": 5}),
            chain_type_kwargs={"prompt": prompt},
            return_source_documents=True,
        )

    async def ask(self, question: str) -> dict:
        result = await self.qa_chain.ainvoke({"query": question})
        return {
            "answer": result["result"],
            "sources": [d.page_content for d in result["source_documents"]],
        }

7. LlamaIndex RAG 实现

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.llms.openai import OpenAI
from llama_index.core.postprocessor import SentenceTransformerRerank

class LlamaIndexRAG:
    def __init__(self, data_dir: str):
        Settings.embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-m3")
        Settings.llm = OpenAI(model="gpt-4o")

        documents = SimpleDirectoryReader(data_dir).load_data()
        self.index = VectorStoreIndex.from_documents(documents)

        # 添加 Reranker
        self.rerank = SentenceTransformerRerank(
            model="BAAI/bge-reranker-large",
            top_n=5,
        )

    def query(self, question: str) -> str:
        query_engine = self.index.as_query_engine(
            similarity_top_k=20,
            node_postprocessors=[self.rerank],
        )
        response = query_engine.query(question)
        return str(response)

8. 端到端 FastAPI RAG 服务

from fastapi import FastAPI, UploadFile, File, HTTPException
from pydantic import BaseModel
import tempfile
import shutil
from pathlib import Path

app = FastAPI(title="RAG Knowledge Base API")

class QueryRequest(BaseModel):
    question: str
    top_k: int = 5
    rerank: bool = True

class QueryResponse(BaseModel):
    answer: str
    sources: list[dict]
    latency_ms: float

class IngestRequest(BaseModel):
    text: str
    metadata: dict = {}

class DocumentIngestor:
    def __init__(self):
        self.embedder = LocalEmbedder()
        self.splitter = RecursiveTextSplitter(chunk_size=500, chunk_overlap=50)
        self.store = QdrantStore()
        self.store.create_collection(self.embedder.dim)
        self.reranker = Reranker()

    async def ingest_text(self, text: str, metadata: dict = None):
        chunks = self.splitter.split_text(text)
        embeddings = self.embedder.encode(chunks)
        metas = [metadata or {} for _ in chunks]
        self.store.upsert(chunks, embeddings, metas)
        return len(chunks)

    async def ingest_file(self, file: UploadFile):
        """支持 txt, md, pdf 文件上传。"""
        suffix = Path(file.filename).suffix
        with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
            shutil.copyfileobj(file.file, tmp)
            tmp_path = tmp.name

        try:
            if suffix == ".pdf":
                text = self._extract_pdf(tmp_path)
            else:
                with open(tmp_path, "r", encoding="utf-8") as f:
                    text = f.read()
            return await self.ingest_text(text, {"source": file.filename})
        finally:
            Path(tmp_path).unlink()

    def _extract_pdf(self, path: str) -> str:
        from pypdf import PdfReader
        reader = PdfReader(path)
        return "\n".join(page.extract_text() or "" for page in reader.pages)

ingestor = DocumentIngestor()

@app.post("/ingest/text")
async def ingest_text(req: IngestRequest):
    count = await ingestor.ingest_text(req.text, req.metadata)
    return {"chunks_ingested": count}

@app.post("/ingest/file")
async def ingest_file(file: UploadFile = File(...)):
    count = await ingestor.ingest_file(file)
    return {"chunks_ingested": count, "filename": file.filename}

@app.post("/query", response_model=QueryResponse)
async def query(req: QueryRequest):
    import time
    start = time.perf_counter()

    embedder = LocalEmbedder()
    store = QdrantStore()

    # 检索
    query_emb = embedder.encode_queries([req.question])
    results = store.search(query_emb[0], top_k=20 if req.rerank else req.top_k)
    texts = [r["text"] for r in results]
    sources = [{"text": r["text"], "score": r["score"]} for r in results]

    # 重排序
    if req.rerank:
        reranked = ingestor.reranker.rerank(req.question, texts, top_k=req.top_k)
        texts = [doc for doc, _ in reranked]
        sources = [{"text": doc, "score": float(score)} for doc, score in reranked]

    # 生成
    context = "\n\n---\n\n".join(f"[{i+1}] {t}" for i, t in enumerate(texts))
    from openai import AsyncOpenAI
    oai = AsyncOpenAI()

    response = await oai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Answer based on the provided context only."},
            {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {req.question}"},
        ],
        temperature=0.0,
    )

    latency = (time.perf_counter() - start) * 1000
    return QueryResponse(
        answer=response.choices[0].message.content,
        sources=sources,
        latency_ms=round(latency, 2),
    )

9. RAG 评估与优化

9.1 评估指标

指标含义计算方法
Context Precision检索结果中相关文档比例相关文档数 / Top-K
Context Recall所有相关文档中被检索到的比例检索到的相关 / 总相关
Faithfulness生成内容是否忠实于上下文LLM-as-Judge
Answer Relevance回答与问题的相关度Embedding 相似度
Answer Correctness回答的正确性与标准答案比对

9.2 Ragas 自动评估

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision
from datasets import Dataset

def evaluate_rag(questions: list[str], answers: list[str], contexts: list[list[str]]):
    """使用 Ragas 框架评估 RAG 质量。"""
    data = Dataset.from_dict({
        "question": questions,
        "answer": answers,
        "contexts": contexts,
    })

    result = evaluate(
        dataset=data,
        metrics=[faithfulness, answer_relevancy, context_precision],
    )
    return result.to_pandas()

9.3 RAG 优化检查清单

  • Embedding 模型与数据领域匹配（通用/代码/医疗）
  • 分块大小通过实验确定（512-1024 token 是常见起点）
  • 混合检索（Dense + BM25）覆盖语义和关键词
  • Reranker 提升 Top-K 排序质量
  • Query 重写/扩展（如 HyDE：生成假答案再检索）
  • 元数据过滤（如按日期、作者过滤）
  • 上下文窗口不超过 LLM 限制（留出生成空间）

交叉链接：

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页