RAG检索增强生成系统构建完全教程

一、RAG概述与原理

1.1 什么是RAG

RAG（Retrieval-Augmented Generation，检索增强生成）是一种将信息检索与大语言模型（LLM）相结合的技术架构。它的核心思想是：在LLM生成回答之前，先从外部知识库中检索相关信息，然后将这些信息作为上下文提供给LLM，从而生成更准确、更有依据的回答。

RAG解决了一个根本性问题：LLM的知识是静态的，它只能基于训练数据中的信息回答问题。当面对企业内部文档、最新资讯、专业领域知识等训练数据之外的内容时，LLM会产生"幻觉"——编造看似合理但实际错误的答案。RAG通过引入外部知识源，让LLM能够基于真实数据回答问题。

1.2 RAG的核心原理

RAG的工作流程可以概括为三个阶段：

索引阶段（Indexing）：离线处理。将原始文档加载、分块、向量化后存入向量数据库。这是RAG系统的基础准备工作。

检索阶段（Retrieval）：在线处理。当用户提出问题时，将问题向量化，在向量数据库中搜索最相关的文档片段。

生成阶段（Generation）：将检索到的相关文档片段与用户问题组合成Prompt，发送给LLM生成最终回答。

用户问题
    │
    ▼
┌─────────────┐    ┌──────────────┐    ┌─────────────┐
│  文本向量化   │───▶│  向量检索     │───▶│  相关文档     │
│  (Embedding) │    │ (Similarity) │    │  (Top-K)     │
└─────────────┘    └──────────────┘    └──────┬──────┘
                                              │
                                              ▼
                                       ┌──────────────┐
                                       │  Prompt组装    │
                                       │  (Template)    │
                                       └──────┬───────┘
                                              │
                                              ▼
                                       ┌──────────────┐
                                       │  LLM生成回答   │
                                       │  (Generation)  │
                                       └──────────────┘

1.3 RAG vs 微调 vs 长上下文

二、文档加载与预处理

2.1 文档加载器设计

RAG系统需要处理多种格式的文档。设计一个统一的文档加载接口：

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
import hashlib

@dataclass
class Document:
    """统一文档格式"""
    content: str           # 文本内容
    metadata: dict         # 元数据（来源、页码、时间等）
    doc_id: Optional[str] = None  # 文档唯一标识

    def __post_init__(self):
        if not self.doc_id:
            self.doc_id = hashlib.md5(
                self.content.encode()
            ).hexdigest()

class BaseLoader(ABC):
    """文档加载器基类"""

    @abstractmethod
    def load(self, source: str) -> list[Document]:
        """加载文档，返回Document列表"""
        pass

    @abstractmethod
    def supported_extensions(self) -> list[str]:
        """返回支持的文件扩展名"""
        pass

2.2 PDF文档加载

import fitz  # PyMuPDF

class PDFLoader(BaseLoader):
    def __init__(self, extract_images: bool = False):
        self.extract_images = extract_images

    def supported_extensions(self):
        return [".pdf"]

    def load(self, source: str) -> list[Document]:
        documents = []
        pdf = fitz.open(source)

        for page_num in range(len(pdf)):
            page = pdf[page_num]
            text = page.get_text("text")

            # 基本清理
            text = self._clean_text(text)

            if text.strip():
                documents.append(Document(
                    content=text,
                    metadata={
                        "source": source,
                        "page": page_num + 1,
                        "total_pages": len(pdf),
                        "format": "pdf"
                    }
                ))

            # 提取图片描述（如果启用）
            if self.extract_images:
                images = page.get_images()
                for img_idx, img in enumerate(images):
                    img_text = self._extract_image_text(page, img)
                    if img_text:
                        documents.append(Document(
                            content=img_text,
                            metadata={
                                "source": source,
                                "page": page_num + 1,
                                "type": "image_description"
                            }
                        ))

        pdf.close()
        return documents

    def _clean_text(self, text: str) -> str:
        """清理文本：去除多余空白、修复断行"""
        import re
        # 合并多个空格
        text = re.sub(r'\s+', ' ', text)
        # 修复被断开的单词
        text = re.sub(r'(\w)-\s+(\w)', r'\1\2', text)
        return text.strip()

2.3 Word文档加载

from docx import Document as DocxDocument

class WordLoader(BaseLoader):
    def supported_extensions(self):
        return [".docx"]

    def load(self, source: str) -> list[Document]:
        documents = []
        doc = DocxDocument(source)

        current_section = ""
        for para in doc.paragraphs:
            # 检测标题层级
            if para.style.name.startswith("Heading"):
                level = para.style.name.replace("Heading ", "")
                current_section = para.text

            if para.text.strip():
                documents.append(Document(
                    content=para.text,
                    metadata={
                        "source": source,
                        "section": current_section,
                        "style": para.style.name,
                        "format": "docx"
                    }
                ))

        # 处理表格
        for table_idx, table in enumerate(doc.tables):
            table_text = self._table_to_text(table)
            documents.append(Document(
                content=table_text,
                metadata={
                    "source": source,
                    "type": "table",
                    "table_index": table_idx,
                    "format": "docx"
                }
            ))

        return documents

    def _table_to_text(self, table) -> str:
        """将表格转换为结构化文本"""
        rows = []
        headers = [cell.text.strip() for cell in table.rows[0].cells]
        rows.append(" | ".join(headers))
        rows.append("-" * len(rows[0]))

        for row in table.rows[1:]:
            cells = [cell.text.strip() for cell in row.cells]
            rows.append(" | ".join(cells))

        return "\n".join(rows)

2.4 网页内容加载

import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse

class WebLoader(BaseLoader):
    def __init__(self, timeout: int = 30):
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (RAG-Bot/1.0)"
        })

    def supported_extensions(self):
        return [".html", ".htm"]

    def load(self, url: str) -> list[Document]:
        response = self.session.get(url, timeout=self.timeout)
        response.raise_for_status()

        soup = BeautifulSoup(response.text, "html.parser")

        # 移除脚本和样式
        for tag in soup(["script", "style", "nav", "footer", "header"]):
            tag.decompose()

        # 提取主要内容
        main_content = (
            soup.find("main") or
            soup.find("article") or
            soup.find("div", class_="content") or
            soup.find("body")
        )

        if not main_content:
            return []

        # 分段提取
        documents = []
        current_heading = ""

        for element in main_content.find_all(["h1", "h2", "h3", "p", "li"]):
            if element.name.startswith("h"):
                current_heading = element.get_text(strip=True)
            elif element.get_text(strip=True):
                documents.append(Document(
                    content=element.get_text(strip=True),
                    metadata={
                        "source": url,
                        "heading": current_heading,
                        "tag": element.name,
                        "format": "html"
                    }
                ))

        return documents

    def load_sitemap(self, sitemap_url: str) -> list[Document]:
        """加载站点地图中的所有页面"""
        response = self.session.get(sitemap_url)
        soup = BeautifulSoup(response.text, "xml")

        all_docs = []
        for loc in soup.find_all("loc"):
            url = loc.text.strip()
            try:
                docs = self.load(url)
                all_docs.extend(docs)
            except Exception as e:
                print(f"加载失败 {url}: {e}")

        return all_docs

2.5 代码文件加载

import ast
import re

class CodeLoader(BaseLoader):
    def __init__(self):
        self.language_map = {
            ".py": "python", ".js": "javascript",
            ".ts": "typescript", ".java": "java",
            ".go": "go", ".rs": "rust",
        }

    def supported_extensions(self):
        return list(self.language_map.keys())

    def load(self, source: str) -> list[Document]:
        import os
        ext = os.path.splitext(source)[1]
        language = self.language_map.get(ext, "unknown")

        with open(source, "r", encoding="utf-8") as f:
            code = f.read()

        documents = []

        # 提取模块级文档
        if ext == ".py":
            module_doc = self._extract_python_docstring(code)
            if module_doc:
                documents.append(Document(
                    content=module_doc,
                    metadata={
                        "source": source,
                        "type": "module_doc",
                        "language": language
                    }
                ))

        # 按函数/类分块
        chunks = self._split_by_function(code, language)
        for chunk in chunks:
            documents.append(Document(
                content=chunk["code"],
                metadata={
                    "source": source,
                    "type": "function",
                    "name": chunk["name"],
                    "language": language
                }
            ))

        return documents

    def _extract_python_docstring(self, code: str) -> str:
        try:
            tree = ast.parse(code)
            return ast.get_docstring(tree) or ""
        except SyntaxError:
            return ""

    def _split_by_function(self, code: str, language: str) -> list[dict]:
        if language == "python":
            return self._split_python_functions(code)
        # 其他语言使用正则分块
        return self._split_by_regex(code)

    def _split_python_functions(self, code: str) -> list[dict]:
        chunks = []
        try:
            tree = ast.parse(code)
            for node in ast.walk(tree):
                if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
                    start = node.lineno - 1
                    end = node.end_lineno or start + 1
                    lines = code.splitlines()[start:end]
                    chunks.append({
                        "name": node.name,
                        "code": "\n".join(lines)
                    })
        except SyntaxError:
            pass
        return chunks

三、文本分块策略

3.1 为什么分块很重要

分块（Chunking）是RAG系统中最关键的预处理步骤之一。分块质量直接影响检索精度：

• 块太大：包含过多无关信息，稀释了关键内容的相关性
• 块太小：丢失上下文，语义不完整
• 分块位置不当：在句子中间断开，破坏语义连贯性

3.2 固定大小分块

最简单的分块方式，按固定字符数或token数切分：

class FixedSizeChunker:
    def __init__(self, chunk_size: int = 500, overlap: int = 50):
        self.chunk_size = chunk_size
        self.overlap = overlap

    def chunk(self, text: str) -> list[str]:
        chunks = []
        start = 0

        while start < len(text):
            end = start + self.chunk_size

            # 尝试在句子边界切分
            if end < len(text):
                # 找最近的句号、问号、感叹号
                for punct in ['。', '！', '？', '. ', '! ', '? ']:
                    last_punct = text[start:end].rfind(punct)
                    if last_punct > self.chunk_size * 0.5:
                        end = start + last_punct + len(punct)
                        break

            chunk = text[start:end].strip()
            if chunk:
                chunks.append(chunk)

            # 下一块从overlap位置开始
            start = end - self.overlap

        return chunks

3.3 递归字符分块

更智能的分块方式，按层级分隔符递归切分：

class RecursiveChunker:
    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 chunk(self, text: str) -> list[str]:
        return self._recursive_split(text, self.separators)

    def _recursive_split(
        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 ""
        remaining_separators = separators[1:] if len(separators) > 1 else [""]

        # 按分隔符切分
        if separator:
            splits = text.split(separator)
        else:
            # 无分隔符时按字符切分
            splits = [text[i:i+self.chunk_size]
                      for i in range(0, len(text), self.chunk_size)]

        # 合并小块
        chunks = []
        current = ""

        for split in splits:
            candidate = (
                current + separator + split if current else split
            )

            if len(candidate) <= self.chunk_size:
                current = candidate
            else:
                if current:
                    chunks.append(current)
                # 如果单个split仍然太大，递归处理
                if len(split) > self.chunk_size:
                    sub_chunks = self._recursive_split(
                        split, remaining_separators
                    )
                    chunks.extend(sub_chunks[:-1])
                    current = sub_chunks[-1] if sub_chunks else ""
                else:
                    current = split

        if current:
            chunks.append(current)

        # 添加重叠
        if self.chunk_overlap > 0:
            chunks = self._add_overlap(chunks)

        return chunks

    def _add_overlap(self, chunks: list[str]) -> list[str]:
        overlapped = [chunks[0]]
        for i in range(1, len(chunks)):
            # 从前一块的末尾取overlap个字符
            overlap_text = chunks[i-1][-self.chunk_overlap:]
            # 找到完整的词边界
            space_idx = overlap_text.find(" ")
            if space_idx != -1:
                overlap_text = overlap_text[space_idx+1:]
            overlapped.append(overlap_text + chunks[i])
        return overlapped

3.4 语义分块

基于语义相似度的分块方式，在语义变化点切分：

import numpy as np
from sentence_transformers import SentenceTransformer

class SemanticChunker:
    def __init__(
        self,
        model_name: str = "all-MiniLM-L6-v2",
        threshold: float = 0.5,
        min_chunk_size: int = 100
    ):
        self.model = SentenceTransformer(model_name)
        self.threshold = threshold
        self.min_chunk_size = min_chunk_size

    def chunk(self, text: str) -> list[str]:
        # 先按句子切分
        sentences = self._split_sentences(text)
        if len(sentences) <= 1:
            return [text]

        # 计算句子嵌入
        embeddings = self.model.encode(sentences)

        # 计算相邻句子的相似度
        similarities = []
        for i in range(len(embeddings) - 1):
            sim = np.dot(embeddings[i], embeddings[i+1]) / (
                np.linalg.norm(embeddings[i]) *
                np.linalg.norm(embeddings[i+1])
            )
            similarities.append(sim)

        # 在相似度低于阈值的位置切分
        chunks = []
        current_chunk = [sentences[0]]

        for i, sim in enumerate(similarities):
            if sim < self.threshold:
                chunk_text = " ".join(current_chunk)
                if len(chunk_text) >= self.min_chunk_size:
                    chunks.append(chunk_text)
                    current_chunk = [sentences[i+1]]
                else:
                    current_chunk.append(sentences[i+1])
            else:
                current_chunk.append(sentences[i+1])

        if current_chunk:
            chunks.append(" ".join(current_chunk))

        return chunks

    def _split_sentences(self, text: str) -> list[str]:
        import re
        # 中英文混合句子切分
        sentences = re.split(r'(?<=[。！？.!?])\s*', text)
        return [s.strip() for s in sentences if s.strip()]

3.5 分块策略对比

四、Embedding模型选择与对比

4.1 Embedding模型概述

Embedding模型将文本转换为固定维度的向量表示，是RAG系统的核心组件。模型选择直接影响检索质量。

4.2 主流模型对比

4.3 使用Embedding模型

from sentence_transformers import SentenceTransformer
import numpy as np

class EmbeddingService:
    def __init__(self, model_name: str = "BAAI/bge-m3"):
        self.model = SentenceTransformer(model_name)
        self.dimension = self.model.get_sentence_embedding_dimension()

    def encode(
        self,
        texts: list[str],
        batch_size: int = 32,
        show_progress: bool = False
    ) -> np.ndarray:
        """批量编码文本为向量"""
        return self.model.encode(
            texts,
            batch_size=batch_size,
            show_progress=show_progress,
            normalize_embeddings=True  # L2归一化
        )

    def encode_query(self, query: str) -> np.ndarray:
        """编码查询（可能需要添加前缀）"""
        # BGE模型建议查询添加前缀
        query = "为这个句子生成表示以用于检索相关文章：" + query
        return self.model.encode(
            [query],
            normalize_embeddings=True
        )[0]

    def similarity(self, vec1: np.ndarray, vec2: np.ndarray) -> float:
        """计算两个向量的余弦相似度"""
        return float(np.dot(vec1, vec2))

4.4 使用OpenAI Embedding API

from openai import OpenAI
import numpy as np

class OpenAIEmbedding:
    def __init__(
        self,
        model: str = "text-embedding-3-small",
        api_key: str = None
    ):
        self.client = OpenAI(api_key=api_key)
        self.model = model

    def encode(self, texts: list[str]) -> np.ndarray:
        """批量编码"""
        response = self.client.embeddings.create(
            model=self.model,
            input=texts
        )
        return np.array([
            item.embedding for item in response.data
        ])

    def encode_with_cache(
        self,
        texts: list[str],
        cache: dict
    ) -> np.ndarray:
        """带缓存的编码"""
        uncached = [t for t in texts if t not in cache]
        if uncached:
            new_embeddings = self.encode(uncached)
            for text, emb in zip(uncached, new_embeddings):
                cache[text] = emb

        return np.array([cache[t] for t in texts])

五、向量数据库选型

5.1 向量数据库概述

向量数据库是RAG系统的存储层，负责高效存储和检索高维向量。选择合适的向量数据库需要考虑：数据规模、查询性能、运维成本、功能特性。

5.2 Chroma - 轻量级首选

import chromadb
from chromadb.config import Settings

class ChromaVectorStore:
    def __init__(
        self,
        collection_name: str = "default",
        persist_directory: str = "./chroma_db"
    ):
        self.client = chromadb.PersistentClient(
            path=persist_directory,
            settings=Settings(anonymized_telemetry=False)
        )
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}
        )

    def add_documents(
        self,
        documents: list[str],
        embeddings: list[list[float]],
        metadatas: list[dict] = None,
        ids: list[str] = None
    ):
        """添加文档"""
        if ids is None:
            import hashlib
            ids = [hashlib.md5(d.encode()).hexdigest() for d in documents]

        self.collection.add(
            documents=documents,
            embeddings=embeddings,
            metadatas=metadatas,
            ids=ids
        )

    def query(
        self,
        query_embedding: list[float],
        top_k: int = 5,
        filter_dict: dict = None
    ) -> dict:
        """查询相似文档"""
        kwargs = {
            "query_embeddings": [query_embedding],
            "n_results": top_k,
            "include": ["documents", "metadatas", "distances"]
        }
        if filter_dict:
            kwargs["where"] = filter_dict

        return self.collection.query(**kwargs)

    def delete(self, ids: list[str]):
        """删除文档"""
        self.collection.delete(ids=ids)

    def count(self) -> int:
        """返回文档数量"""
        return self.collection.count()

5.3 Milvus - 大规模生产级

from pymilvus import (
    connections, Collection, FieldSchema,
    CollectionSchema, DataType, utility
)

class MilvusVectorStore:
    def __init__(
        self,
        collection_name: str = "default",
        host: str = "localhost",
        port: int = 19530,
        dimension: int = 1024
    ):
        connections.connect(host=host, port=port)
        self.collection_name = collection_name
        self.dimension = dimension
        self._create_collection()

    def _create_collection(self):
        if utility.has_collection(self.collection_name):
            self.collection = Collection(self.collection_name)
            return

        fields = [
            FieldSchema("id", DataType.VARCHAR, is_primary=True, max_length=64),
            FieldSchema("embedding", DataType.FLOAT_VECTOR, dim=self.dimension),
            FieldSchema("content", DataType.VARCHAR, max_length=65535),
            FieldSchema("metadata", DataType.JSON),
        ]
        schema = CollectionSchema(fields)
        self.collection = Collection(self.collection_name, schema)

        # 创建索引
        self.collection.create_index(
            field_name="embedding",
            index_params={
                "metric_type": "COSINE",
                "index_type": "HNSW",
                "params": {"M": 16, "efConstruction": 256}
            }
        )

    def add_documents(
        self,
        ids: list[str],
        embeddings: list[list[float]],
        documents: list[str],
        metadatas: list[dict] = None
    ):
        data = [ids, embeddings, documents, metadatas or [{}] * len(ids)]
        self.collection.insert(data)
        self.collection.flush()

    def query(
        self,
        query_embedding: list[float],
        top_k: int = 5,
        filter_expr: str = None
    ) -> list[dict]:
        self.collection.load()

        search_params = {"metric_type": "COSINE", "params": {"ef": 128}}
        results = self.collection.search(
            data=[query_embedding],
            anns_field="embedding",
            param=search_params,
            limit=top_k,
            output_fields=["content", "metadata"],
            expr=filter_expr
        )

        return [
            {
                "id": hit.id,
                "score": hit.score,
                "content": hit.entity.get("content"),
                "metadata": hit.entity.get("metadata")
            }
            for hit in results[0]
        ]

5.4 Qdrant - 高性能选择

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue
)

class QdrantVectorStore:
    def __init__(
        self,
        collection_name: str = "default",
        host: str = "localhost",
        port: int = 6333,
        dimension: int = 1024
    ):
        self.client = QdrantClient(host=host, port=port)
        self.collection_name = collection_name
        self.dimension = dimension
        self._create_collection()

    def _create_collection(self):
        collections = self.client.get_collections().collections
        names = [c.name for c in collections]

        if self.collection_name not in names:
            self.client.create_collection(
                collection_name=self.collection_name,
                vectors_config=VectorParams(
                    size=self.dimension,
                    distance=Distance.COSINE
                )
            )

    def add_documents(
        self,
        ids: list[str],
        embeddings: list[list[float]],
        documents: list[str],
        metadatas: list[dict] = None
    ):
        points = []
        for i, (id_, emb, doc) in enumerate(zip(ids, embeddings, documents)):
            payload = {"content": doc}
            if metadatas and i < len(metadatas):
                payload.update(metadatas[i])

            points.append(PointStruct(
                id=id_,
                vector=emb,
                payload=payload
            ))

        self.client.upsert(
            collection_name=self.collection_name,
            points=points
        )

    def query(
        self,
        query_embedding: list[float],
        top_k: int = 5,
        filter_dict: dict = None
    ) -> list[dict]:
        query_filter = None
        if filter_dict:
            conditions = [
                FieldCondition(
                    key=k,
                    match=MatchValue(value=v)
                )
                for k, v in filter_dict.items()
            ]
            query_filter = Filter(must=conditions)

        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_embedding,
            limit=top_k,
            query_filter=query_filter
        )

        return [
            {
                "id": r.id,
                "score": r.score,
                "content": r.payload.get("content"),
                "metadata": {k: v for k, v in r.payload.items() if k != "content"}
            }
            for r in results
        ]

5.5 选型建议

六、检索策略

6.1 基础相似度检索

最简单的检索方式，直接按向量相似度返回Top-K结果：

class BasicRetriever:
    def __init__(self, vector_store, embedding_service):
        self.vector_store = vector_store
        self.embedding_service = embedding_service

    def retrieve(self, query: str, top_k: int = 5) -> list[dict]:
        query_embedding = self.embedding_service.encode_query(query)
        results = self.vector_store.query(
            query_embedding=query_embedding.tolist(),
            top_k=top_k
        )
        return results

6.2 混合检索

结合向量检索和关键词检索（BM25），综合利用语义匹配和精确匹配的优势：

from rank_bm25 import BM25Okapi
import jieba
import numpy as np

class HybridRetriever:
    def __init__(
        self,
        vector_store,
        embedding_service,
        alpha: float = 0.7  # 向量检索权重
    ):
        self.vector_store = vector_store
        self.embedding_service = embedding_service
        self.alpha = alpha
        self.bm25 = None
        self.corpus = []

    def build_bm25_index(self, documents: list[str]):
        """构建BM25索引"""
        self.corpus = documents
        tokenized = [list(jieba.cut(doc)) for doc in documents]
        self.bm25 = BM25Okapi(tokenized)

    def retrieve(self, query: str, top_k: int = 5) -> list[dict]:
        # 向量检索
        query_embedding = self.embedding_service.encode_query(query)
        vector_results = self.vector_store.query(
            query_embedding=query_embedding.tolist(),
            top_k=top_k * 2
        )

        # BM25检索
        tokenized_query = list(jieba.cut(query))
        bm25_scores = self.bm25.get_scores(tokenized_query)
        bm25_top_indices = np.argsort(bm25_scores)[::-1][:top_k * 2]

        # 分数融合（RRF - Reciprocal Rank Fusion）
        scores = {}
        k = 60  # RRF常数

        for rank, result in enumerate(vector_results):
            doc_id = result["id"]
            scores[doc_id] = scores.get(doc_id, 0) + self.alpha / (k + rank + 1)

        for rank, idx in enumerate(bm25_top_indices):
            doc_id = str(idx)
            scores[doc_id] = scores.get(doc_id, 0) + (1 - self.alpha) / (k + rank + 1)

        # 按融合分数排序
        sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)

        # 返回结果
        id_to_result = {r["id"]: r for r in vector_results}
        results = []
        for doc_id in sorted_ids[:top_k]:
            if doc_id in id_to_result:
                result = id_to_result[doc_id]
                result["hybrid_score"] = scores[doc_id]
                results.append(result)

        return results

6.3 重排序（Reranking）

使用交叉编码器对检索结果进行二次排序，提高精度：

from sentence_transformers import CrossEncoder

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

    def rerank(
        self,
        query: str,
        documents: list[dict],
        top_k: int = 5
    ) -> list[dict]:
        if not documents:
            return []

        # 构造query-document对
        pairs = [(query, doc["content"]) for doc in documents]

        # 计算相关性分数
        scores = self.model.predict(pairs)

        # 按分数排序
        scored_docs = list(zip(scores, documents))
        scored_docs.sort(key=lambda x: x[0], reverse=True)

        results = []
        for score, doc in scored_docs[:top_k]:
            doc["rerank_score"] = float(score)
            results.append(doc)

        return results

6.4 多查询检索

通过生成多个查询变体来提高召回率：

class MultiQueryRetriever:
    def __init__(self, retriever, llm_client):
        self.retriever = retriever
        self.llm = llm_client

    def retrieve(self, query: str, top_k: int = 5) -> list[dict]:
        # 生成查询变体
        variations = self._generate_variations(query)

        # 收集所有结果
        all_results = {}
        for q in [query] + variations:
            results = self.retriever.retrieve(q, top_k=top_k)
            for r in results:
                doc_id = r["id"]
                if doc_id not in all_results or r["score"] > all_results[doc_id]["score"]:
                    all_results[doc_id] = r

        # 按分数排序返回
        sorted_results = sorted(
            all_results.values(),
            key=lambda x: x["score"],
            reverse=True
        )
        return sorted_results[:top_k]

    def _generate_variations(self, query: str) -> list[str]:
        prompt = f"""请为以下问题生成3个不同角度的查询变体，
每个变体单独一行，不要编号：

原始问题：{query}

查询变体："""
        response = self.llm.generate(prompt)
        return [q.strip() for q in response.strip().split("\n") if q.strip()]

七、Prompt模板设计

7.1 基础RAG Prompt

RAG_PROMPT_TEMPLATE = """你是一个专业的知识助手。请基于以下参考文档回答用户的问题。

要求：
1. 只使用参考文档中的信息回答问题
2. 如果参考文档中没有相关信息，请明确说明"根据现有资料无法回答该问题"
3. 回答要准确、完整、有条理
4. 适当引用文档来源

参考文档：
{context}

用户问题：{question}

回答："""

7.2 带来源引用的Prompt

RAG_CITATION_PROMPT = """你是一个严谨的知识助手。请基于参考文档回答问题，并在回答中标注信息来源。

规则：
1. 每个关键信息点都需要标注来源，格式为 [来源X]
2. 只使用参考文档中的信息
3. 如果信息不足，明确指出
4. 保持客观中立

参考文档：
{context}

其中每个文档的来源信息：
{sources}

用户问题：{question}

请提供带来源标注的回答："""

7.3 对话式RAG Prompt

CONVERSATIONAL_RAG_PROMPT = """你是一个友好的知识助手，正在与用户进行对话。

对话历史：
{chat_history}

参考文档：
{context}

当前用户问题：{question}

要求：
1. 结合对话历史理解用户问题的上下文
2. 基于参考文档提供准确回答
3. 如果需要更多信息才能回答，请向用户提问
4. 保持对话的自然流畅

回答："""

7.4 Prompt构建器

class PromptBuilder:
    def __init__(self, template: str = None):
        self.template = template or RAG_PROMPT_TEMPLATE

    def build(
        self,
        question: str,
        documents: list[dict],
        chat_history: list[dict] = None
    ) -> str:
        # 构建上下文
        context_parts = []
        for i, doc in enumerate(documents, 1):
            source = doc.get("metadata", {}).get("source", "未知来源")
            context_parts.append(
                f"[文档{i}] (来源: {source})\n{doc['content']}"
            )
        context = "\n\n".join(context_parts)

        # 构建对话历史
        history = ""
        if chat_history:
            history = "\n".join([
                f"{msg['role']}: {msg['content']}"
                for msg in chat_history[-5:]  # 只保留最近5轮
            ])

        # 填充模板
        return self.template.format(
            context=context,
            question=question,
            chat_history=history,
            sources="\n".join([
                f"[文档{i+1}]: {doc.get('metadata', {}).get('source', '未知')}"
                for i, doc in enumerate(documents)
            ])
        )

八、RAG系统评估指标

8.1 评估维度

RAG系统的评估需要从多个维度进行：

检索质量：检索到的文档是否与问题相关 生成质量：生成的回答是否准确、完整 端到端质量：最终回答是否满足用户需求

8.2 检索评估指标

import numpy as np

class RetrievalEvaluator:
    @staticmethod
    def precision_at_k(retrieved: list, relevant: set, k: int) -> float:
        """P@K: 前K个结果中相关文档的比例"""
        top_k = retrieved[:k]
        hits = sum(1 for doc in top_k if doc in relevant)
        return hits / k

    @staticmethod
    def recall_at_k(retrieved: list, relevant: set, k: int) -> float:
        """R@K: 前K个结果覆盖了多少相关文档"""
        top_k = retrieved[:k]
        hits = sum(1 for doc in top_k if doc in relevant)
        return hits / len(relevant) if relevant else 0

    @staticmethod
    def mrr(retrieved: list, relevant: set) -> float:
        """MRR: 第一个相关文档的排名倒数"""
        for i, doc in enumerate(retrieved):
            if doc in relevant:
                return 1.0 / (i + 1)
        return 0.0

    @staticmethod
    def ndcg_at_k(retrieved: list, relevance: dict, k: int) -> float:
        """NDCG@K: 归一化折损累积增益"""
        dcg = sum(
            relevance.get(doc, 0) / np.log2(i + 2)
            for i, doc in enumerate(retrieved[:k])
        )
        ideal = sorted(relevance.values(), reverse=True)[:k]
        idcg = sum(
            rel / np.log2(i + 2)
            for i, rel in enumerate(ideal)
        )
        return dcg / idcg if idcg > 0 else 0.0

8.3 生成评估指标

from collections import Counter

class GenerationEvaluator:
    @staticmethod
    def faithfulness(answer: str, context: str) -> float:
        """忠实度: 回答是否基于给定上下文"""
        # 简单实现：检查回答中的关键信息是否在上下文中出现
        answer_words = set(answer.split())
        context_words = set(context.split())
        overlap = answer_words & context_words
        return len(overlap) / len(answer_words) if answer_words else 0

    @staticmethod
    def answer_relevancy(answer: str, question: str) -> float:
        """答案相关性: 回答是否与问题相关"""
        # 使用简单的词重叠作为近似
        q_words = set(question.split())
        a_words = set(answer.split())
        overlap = q_words & a_words
        return len(overlap) / len(q_words) if q_words else 0

    @staticmethod
    def completeness(answer: str, reference: str) -> float:
        """完整性: 回答是否涵盖了参考答案的要点"""
        ref_words = set(reference.split())
        ans_words = set(answer.split())
        overlap = ref_words & ans_words
        return len(overlap) / len(ref_words) if ref_words else 0

8.4 端到端评估框架

class RAGEvaluator:
    def __init__(self, rag_system):
        self.rag = rag_system
        self.retrieval_eval = RetrievalEvaluator()
        self.generation_eval = GenerationEvaluator()

    def evaluate(self, test_cases: list[dict]) -> dict:
        """
        test_cases: [{
            "question": "...",
            "reference_answer": "...",
            "relevant_doc_ids": [...]
        }]
        """
        results = {
            "precision@5": [],
            "recall@5": [],
            "mrr": [],
            "faithfulness": [],
            "relevancy": []
        }

        for case in test_cases:
            # 执行RAG查询
            rag_result = self.rag.query(case["question"])

            # 检索评估
            retrieved_ids = [r["id"] for r in rag_result["documents"]]
            relevant = set(case.get("relevant_doc_ids", []))

            results["precision@5"].append(
                self.retrieval_eval.precision_at_k(retrieved_ids, relevant, 5)
            )
            results["recall@5"].append(
                self.retrieval_eval.recall_at_k(retrieved_ids, relevant, 5)
            )
            results["mrr"].append(
                self.retrieval_eval.mrr(retrieved_ids, relevant)
            )

            # 生成评估
            context = " ".join([d["content"] for d in rag_result["documents"]])
            results["faithfulness"].append(
                self.generation_eval.faithfulness(rag_result["answer"], context)
            )
            results["relevancy"].append(
                self.generation_eval.answer_relevancy(
                    rag_result["answer"], case["question"]
                )
            )

        # 计算平均值
        return {k: np.mean(v) for k, v in results.items()}

九、高级RAG技术

9.1 HyDE（假设文档嵌入）

HyDE的核心思想：先让LLM生成一个假设性答案，然后用这个假设答案去检索，而不是直接用问题检索。

class HyDERetriever:
    def __init__(self, retriever, llm_client):
        self.retriever = retriever
        self.llm = llm_client

    def retrieve(self, query: str, top_k: int = 5) -> list[dict]:
        # 生成假设性文档
        hyde_prompt = f"""请针对以下问题，写一段可能包含答案的文档内容。
不需要准确，只需要看起来合理即可。

问题：{query}

假设文档："""
        hypothetical_doc = self.llm.generate(hyde_prompt)

        # 用假设文档进行检索
        return self.retriever.retrieve(hypothetical_doc, top_k=top_k)

9.2 Self-RATE（自适应检索增强）

Self-RATE让LLM先判断是否需要检索，避免对简单问题进行不必要的检索：

class SelfRATERetriever:
    def __init__(self, retriever, llm_client):
        self.retriever = retriever
        self.llm = llm_client

    def query(self, question: str) -> dict:
        # 第一步：判断是否需要检索
        decision_prompt = f"""判断以下问题是否需要外部知识来回答。

问题：{question}

如果这个问题可以基于常识回答，回复 "NO_RETRIEVAL"
如果需要查询特定信息才能准确回答，回复 "NEED_RETRIEVAL"

判断："""
        decision = self.llm.generate(decision_prompt).strip()

        if "NO_RETRIEVAL" in decision:
            # 直接回答
            answer = self.llm.generate(f"请回答以下问题：{question}")
            return {"answer": answer, "documents": [], "retrieved": False}
        else:
            # 执行检索增强回答
            results = self.retriever.retrieve(question)
            context = "\n".join([r["content"] for r in results])
            prompt = f"基于以下信息回答问题：\n{context}\n\n问题：{question}"
            answer = self.llm.generate(prompt)
            return {"answer": answer, "documents": results, "retrieved": True}

9.3 CRAG（纠正性RAG）

CRAG在检索后增加一个评估和纠正步骤，过滤掉不相关的检索结果：

class CRAGRetriever:
    def __init__(self, retriever, llm_client):
        self.retriever = retriever
        self.llm = llm_client

    def retrieve(self, query: str, top_k: int = 5) -> list[dict]:
        # 初始检索
        initial_results = self.retriever.retrieve(query, top_k=top_k * 2)

        # 评估每个文档的相关性
        relevant_docs = []
        for doc in initial_results:
            eval_prompt = f"""判断以下文档是否与问题相关。

问题：{query}

文档内容：
{doc['content'][:500]}

只回答 "相关" 或 "不相关"："""
            evaluation = self.llm.generate(eval_prompt).strip()

            if "相关" in evaluation and "不相关" not in evaluation:
                relevant_docs.append(doc)

        # 如果没有相关文档，使用网络搜索补充
        if not relevant_docs:
            web_results = self._web_search(query)
            return web_results[:top_k]

        return relevant_docs[:top_k]

    def _web_search(self, query: str) -> list[dict]:
        """网络搜索作为后备方案"""
        # 这里可以集成搜索引擎API
        # 返回格式与向量检索一致
        return []

9.4 上下文压缩

在生成前压缩检索到的文档，去除无关内容：

class ContextCompressor:
    def __init__(self, llm_client):
        self.llm = llm_client

    def compress(self, query: str, documents: list[dict]) -> list[dict]:
        compressed = []
        for doc in documents:
            compress_prompt = f"""从以下文档中提取与问题最相关的关键信息。
去除无关内容，保留核心信息。

问题：{query}

文档：
{doc['content']}

关键信息："""
            key_info = self.llm.generate(compress_prompt)
            compressed.append({
                **doc,
                "content": key_info,
                "original_length": len(doc["content"]),
                "compressed_length": len(key_info)
            })
        return compressed

十、实战案例：企业知识库问答系统

10.1 系统架构

class EnterpriseRAGSystem:
    """企业级RAG问答系统"""

    def __init__(self, config: dict):
        self.config = config

        # 初始化组件
        self.embedding_service = EmbeddingService(
            model_name=config.get("embedding_model", "BAAI/bge-m3")
        )

        self.vector_store = ChromaVectorStore(
            collection_name=config.get("collection", "enterprise_kb"),
            persist_directory=config.get("persist_dir", "./data/chroma")
        )

        self.retriever = HybridRetriever(
            vector_store=self.vector_store,
            embedding_service=self.embedding_service,
            alpha=config.get("hybrid_alpha", 0.7)
        )

        self.reranker = Reranker()
        self.prompt_builder = PromptBuilder()
        self.llm = LLMClient(model=config.get("llm_model", "gpt-4"))

    def ingest_documents(self, source_dir: str):
        """批量导入文档"""
        loaders = {
            ".pdf": PDFLoader(),
            ".docx": WordLoader(),
            ".html": WebLoader(),
            ".py": CodeLoader(),
        }

        chunker = RecursiveChunker(
            chunk_size=self.config.get("chunk_size", 500),
            chunk_overlap=self.config.get("chunk_overlap", 50)
        )

        all_chunks = []
        all_metadata = []

        import glob
        for filepath in glob.glob(f"{source_dir}/**/*", recursive=True):
            import os
            ext = os.path.splitext(filepath)[1].lower()
            loader = loaders.get(ext)

            if loader:
                print(f"处理文件: {filepath}")
                documents = loader.load(filepath)

                for doc in documents:
                    chunks = chunker.chunk(doc.content)
                    for chunk in chunks:
                        all_chunks.append(chunk)
                        all_metadata.append(doc.metadata)

        # 批量向量化
        print(f"正在向量化 {len(all_chunks)} 个文本块...")
        embeddings = self.embedding_service.encode(
            all_chunks, batch_size=64, show_progress=True
        )

        # 存入向量数据库
        import hashlib
        ids = [hashlib.md5(c.encode()).hexdigest() for c in all_chunks]

        self.vector_store.add_documents(
            ids=ids,
            embeddings=embeddings.tolist(),
            documents=all_chunks,
            metadatas=all_metadata
        )

        # 构建BM25索引
        self.retriever.build_bm25_index(all_chunks)

        print(f"导入完成，共 {len(all_chunks)} 个文本块")

    def query(
        self,
        question: str,
        top_k: int = 5,
        use_reranker: bool = True,
        chat_history: list[dict] = None
    ) -> dict:
        """执行RAG查询"""
        # 第一步：混合检索
        retrieved = self.retriever.retrieve(question, top_k=top_k * 2)

        # 第二步：重排序（可选）
        if use_reranker and retrieved:
            retrieved = self.reranker.rerank(question, retrieved, top_k=top_k)

        # 第三步：构建Prompt
        prompt = self.prompt_builder.build(
            question=question,
            documents=retrieved,
            chat_history=chat_history
        )

        # 第四步：生成回答
        answer = self.llm.generate(prompt)

        return {
            "answer": answer,
            "documents": retrieved,
            "prompt": prompt
        }

10.2 Web API接口

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="企业知识库问答系统")
rag_system = None

class QueryRequest(BaseModel):
    question: str
    top_k: int = 5
    use_reranker: bool = True
    chat_history: list[dict] = None

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

@app.on_event("startup")
async def startup():
    global rag_system
    config = {
        "embedding_model": "BAAI/bge-m3",
        "collection": "enterprise_kb",
        "persist_dir": "./data/chroma",
        "hybrid_alpha": 0.7,
        "chunk_size": 500,
        "chunk_overlap": 50,
        "llm_model": "gpt-4"
    }
    rag_system = EnterpriseRAGSystem(config)

@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
    try:
        result = rag_system.query(
            question=request.question,
            top_k=request.top_k,
            use_reranker=request.use_reranker,
            chat_history=request.chat_history
        )
        return QueryResponse(
            answer=result["answer"],
            sources=[
                {
                    "content": doc["content"][:200],
                    "source": doc.get("metadata", {}).get("source", ""),
                    "score": doc.get("rerank_score", doc.get("score", 0))
                }
                for doc in result["documents"]
            ]
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/ingest")
async def ingest(source_dir: str):
    try:
        rag_system.ingest_documents(source_dir)
        return {"status": "success"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

10.3 前端界面

<!DOCTYPE html>
<html>
<head>
    <title>企业知识库</title>
    <style>
        body { font-family: sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }
        .chat-container { border: 1px solid #ddd; border-radius: 8px; padding: 20px; }
        .message { margin: 10px 0; padding: 10px; border-radius: 8px; }
        .user { background: #e3f2fd; text-align: right; }
        .assistant { background: #f5f5f5; }
        .sources { font-size: 0.9em; color: #666; margin-top: 5px; }
        input, button { padding: 10px; margin: 5px 0; }
        input { width: 80%; }
        button { background: #1976d2; color: white; border: none; cursor: pointer; }
    </style>
</head>
<body>
    <h1>📚 企业知识库问答</h1>
    <div class="chat-container" id="chat"></div>
    <div>
        <input type="text" id="question" placeholder="输入您的问题..." />
        <button onclick="ask()">发送</button>
    </div>

    <script>
    async function ask() {
        const input = document.getElementById('question');
        const question = input.value.trim();
        if (!question) return;

        // 显示用户消息
        appendMessage('user', question);
        input.value = '';

        // 调用API
        const response = await fetch('/query', {
            method: 'POST',
            headers: {'Content-Type': 'application/json'},
            body: JSON.stringify({question, top_k: 5})
        });
        const data = await response.json();

        // 显示回答
        let sourcesHtml = '<div class="sources">来源：' +
            data.sources.map(s => s.source).join(', ') + '</div>';
        appendMessage('assistant', data.answer + sourcesHtml);
    }

    function appendMessage(role, content) {
        const chat = document.getElementById('chat');
        const div = document.createElement('div');
        div.className = `message ${role}`;
        div.innerHTML = content;
        chat.appendChild(div);
        chat.scrollTop = chat.scrollHeight;
    }
    </script>
</body>
</html>

十一、最佳实践

11.1 分块策略调优

• 从500字开始：大多数场景下，500字符的块大小是一个好的起点
• 保留50-100字的重叠：确保上下文连贯性
• 按文档类型调整：代码文档可能需要更大的块，FAQ可以按条目分块
• 测试不同策略：用真实查询评估不同分块方案的效果

11.2 检索优化

• 混合检索是标配：向量检索+BM25的效果通常优于单一检索
• 重排序提升明显：使用交叉编码器重排序可以显著提高Top-5精度
• 调整返回数量：检索时多取一些结果（如Top-10），重排序后再精选Top-5

11.3 Prompt设计

• 明确指令：告诉LLM"只使用提供的文档回答"
• 设置兜底：当文档不足以回答时，让LLM明确说明
• 控制长度：根据应用场景设定回答长度要求
• 来源引用：要求LLM标注信息来源，提高可信度

11.4 生产部署

# 关键配置项
PRODUCTION_CONFIG = {
    # 分块
    "chunk_size": 500,
    "chunk_overlap": 50,

    # Embedding
    "embedding_model": "BAAI/bge-m3",
    "embedding_batch_size": 64,

    # 向量数据库
    "vector_db": "milvus",
    "index_type": "HNSW",
    "ef_construction": 256,
    "m": 16,

    # 检索
    "retrieval_top_k": 10,
    "final_top_k": 5,
    "hybrid_alpha": 0.7,
    "use_reranker": True,

    # 生成
    "llm_model": "gpt-4",
    "temperature": 0.1,
    "max_tokens": 2000,

    # 缓存
    "enable_cache": True,
    "cache_ttl": 3600,
}

十二、常见问题

Q1: 检索结果不相关怎么办？

排查步骤：

1. 检查分块是否合理——块是否太大或太小
2. 检查Embedding模型是否适合你的语言和领域
3. 尝试混合检索，结合关键词匹配
4. 添加重排序步骤
5. 优化查询——使用HyDE或多查询技术

Q2: 如何处理多语言文档？

建议：

1. 使用多语言Embedding模型（如BGE-M3、multilingual-e5）
2. 为不同语言建立独立的向量集合
3. 查询时自动检测语言并路由到对应集合

Q3: 知识库如何更新？

实现增量更新：

class IncrementalIndexer:
    def update(self, source_dir: str):
        # 计算文件哈希，检测变更
        current_files = self._scan_files(source_dir)
        indexed_files = self._get_indexed_files()

        # 新增文件
        new_files = current_files - indexed_files
        # 修改文件
        modified = self._detect_modifications(current_files, indexed_files)
        # 删除文件
        deleted = indexed_files - current_files

        # 执行增量更新
        for f in new_files | modified:
            self._reindex_file(f)
        for f in deleted:
            self._remove_file(f)

Q4: 如何控制成本？

• 使用本地Embedding模型替代API调用
• 实施查询缓存，对相似查询返回缓存结果
• 使用较小的LLM生成回答（如GPT-3.5而非GPT-4）
• 限制检索文档数量和上下文长度

Q5: 如何评估RAG系统效果？

1. 构建测试数据集（100+条带标准答案的问答对）
2. 使用自动化评估脚本定期运行
3. 关注指标：检索召回率、回答准确率、忠实度
4. 收集用户反馈，持续优化

十三、总结

RAG系统是当前最实用的AI应用架构之一，它让LLM能够基于最新、最准确的外部知识回答问题。构建高质量的RAG系统需要关注以下关键环节：

1. 数据质量：好的文档预处理和分块策略是基础
2. 检索精度：混合检索+重排序是当前最佳实践
3. 生成质量：精心设计的Prompt模板确保回答准确可靠
4. 持续优化：通过评估指标驱动系统迭代改进

随着技术的发展，RAG系统正在向更智能的方向演进：自适应检索、多模态RAG、图谱增强RAG等新技术不断涌现。掌握RAG核心技术，将帮助你构建更加强大的AI知识应用。

推荐学习资源

• LangChain官方文档 - RAG章节
• LlamaIndex官方文档
• 向量数据库（Chroma/Milvus/Qdrant）官方文档
• "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks" 原始论文
• MTEB Embedding模型排行榜

本教程基于2024年RAG技术生态编写，相关工具和最佳实践持续演进中。