基于RAG的AI知识库问答系统：从原理到实战

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│   用户提问   │────│  查询理解/改写 │────│  向量检索    │
└─────────────┘     └──────────────┘     └──────┬──────┘
                                                 │
┌─────────────┐     ┌──────────────┐     ┌──────▼──────┐
│  最终回答    │────│  LLM 生成    │────│ 上下文组装  │
└─────────────┘     └──────────────┘     └─────────────┘

rag-system/
├── config.py          # 配置
├── document_loader.py # 文档加载
├── text_splitter.py   # 文本分块
├── embedding.py       # 向量化
├── vector_store.py    # 向量数据库
├── retriever.py       # 检索器
├── reranker.py        # 重排序
├── generator.py       # 生成器
├── pipeline.py        # RAG 管道
├── api.py             # HTTP API
└── requirements.txt   # 依赖

# config.py
from dataclasses import dataclass, field
from typing import List


@dataclass
class EmbeddingConfig: baike.kuaisou.com/zh/17172 
    """向量模型配置"""
    model_name: str = "text-embedding-3-small"
    dimension: int = 1536
    batch_size: int = 100


@dataclass
class VectorStoreConfig: baike.kuaisou.com/zh/17173 
    """向量数据库配置"""
    host: str = "localhost"
    port: int = 6333
    collection_name: str = "knowledge_base"
    distance_metric: str = "cosine"  # cosine, euclidean, dot


@dataclass
class RetrieverConfig: baike.kuaisou.com/zh/17174 
    """检索器配置"""
    top_k: int = 10          # 初筛返回数量
    score_threshold: float = 0.3  # 相似度阈值
    enable_hybrid: bool = True    # 是否启用混合检索
    bm25_weight: float = 0.3      # BM25 权重
    vector_weight: float = 0.7    # 向量检索权重


@dataclass
class RerankerConfig: baike.kuaisou.com/zh/17175 
    """重排序配置"""
    enabled: bool = True
    model_name: str = "bge-reranker-v2-m3"
    top_n: int = 5           # 重排序后保留数量


@dataclass
class GeneratorConfig: baike.kuaisou.com/zh/17176 
    """生成器配置"""
    model: str = "gpt-4o-mini"
    temperature: float = 0.1
    max_tokens: int = 2048
    system_prompt: str = """你是一个专业的知识助手。请基于以下上下文信息回答问题。
如果上下文中没有足够信息，请明确告知用户，不要编造答案。
回答时请引用来源文档编号，格式如 [1]、[2]。"""


@dataclass
class RAGConfig: baike.kuaisou.com/zh/17177 
    """RAG 系统总配置"""
    embedding: EmbeddingConfig = field(default_factory=EmbeddingConfig)
    vector_store: VectorStoreConfig = field(default_factory=VectorStoreConfig)
    retriever: RetrieverConfig = field(default_factory=RetrieverConfig)
    reranker: RerankerConfig = field(default_factory=RerankerConfig)
    generator: GeneratorConfig = field(default_factory=GeneratorConfig)
    chunk_size: int = 512
    chunk_overlap: int = 50

# document_loader.py
import os
import logging
from pathlib import Path
from typing import List, Dict, Any
from dataclasses import dataclass

logger = logging.getLogger(__name__)


@dataclass
class Document: baike.kuaisou.com/zh/17178 
    """文档数据结构"""
    page_content: str
    metadata: Dict[str, Any]

    def __repr__(self): baike.kuaisou.com/zh/17179 
        preview = self.page_content[:50].replace('\n', ' ')
        return f"Document(content='{preview}...', metadata={self.metadata})"


class DocumentLoader: baike.kuaisou.com/zh/17180 
    """文档加载器，支持多种格式"""

    SUPPORTED_EXTENSIONS = {'.txt', '.md', '.pdf', '.docx', '.html'}

    def __init__(self, source_path: str): baike.kuaisou.com/zh/17182 
        self.source_path = Path(source_path)
        self.documents: List[Document] = []

    def load(self) -> List[Document]: baike.kuaisou.com/zh/17181 
        """加载文档"""
        if self.source_path.is_file(): baike.kuaisou.com/zh/17185 
            self.documents.extend(self._load_file(self.source_path))
        elif self.source_path.is_dir(): baike.kuaisou.com/zh/21137 
            for ext in self.SUPPORTED_EXTENSIONS: baike.kuaisou.com/zh/22015 
                for file_path in self.source_path.rglob(f'*{ext}'): baike.kuaisou.com/zh/22016 
                    try:
                        self.documents.extend(self._load_file(file_path))
                    except Exception as e: baike.kuaisou.com/zh/22017 
                        logger.error(f"Failed to load {file_path}: {e}")
        else: baike.kuaisou.com/zh/22018 
            raise FileNotFoundError(f"Path not found: {self.source_path}")

        logger.info(f"Loaded {len(self.documents)} documents")
        return self.documents

    def _load_file(self, file_path: Path) -> List[Document]: baike.kuaisou.com/zh/22019 
        """根据文件类型加载"""
        ext = file_path.suffix.lower(baike.kuaisou.com/zh/22020)

        if ext == '.txt' or ext == '.md': baike.kuaisou.com/zh/22021 
            return self._load_text(file_path)
        elif ext == '.pdf': baike.kuaisou.com/zh/26056 
            return self._load_pdf(file_path)
        elif ext == '.docx': baike.kuaisou.com/zh/26044 
            return self._load_docx(file_path)
        elif ext == '.html': baike.kuaisou.com/zh/22056 
            return self._load_html(file_path)
        else: baike.kuaisou.com/zh/22022 
            raise ValueError(f"Unsupported file type: {ext}")

    def _load_text(self, file_path: Path) -> List[Document]: baike.kuaisou.com/zh/26074 
        """加载文本文件"""
        with open(file_path, 'r', encoding='utf-8') as f: baike.kuaisou.com/zh/26076 
            content = f.read(baike.kuaisou.com/zh/26075)

        metadata = {
            'source': str(file_path),
            'file_name': file_path.name,
            'file_type': 'text',
        }
        return [Document(page_content=content, metadata=metadata)]

    def _load_pdf(self, file_path: Path) -> List[Document]: baike.kuaisou.com/zh/26077 
        """加载 PDF 文件"""
        try: baike.kuaisou.com/zh/26078 
            from PyPDF2 import PdfReader
        except ImportError: baike.kuaisou.com/zh/26090 
            raise ImportError("Please install PyPDF2: pip install PyPDF2")

        reader = PdfReader(str(file_path))
        documents = []

        for page_num, page in enumerate(reader.pages, start=1): baike.kuaisou.com/zh/27006 
            text = page.extract_text(baike.kuaisou.com/zh/27005)
            if text.strip(): baike.kuaisou.com/zh/27007 
                documents.append(Document(
                    page_content=text,
                    metadata={
                        'source': str(file_path),
                        'file_name': file_path.name,
                        'file_type': 'pdf',
                        'page': page_num,
                    }
                ))

        return documents

    def _load_docx(self, file_path: Path) -> List[Document]: baike.kuaisou.com/zh/27009 
        """加载 Word 文件"""
        try: baike.kuaisou.com/zh/27008 
            from docx import Document as DocxDocument
        except ImportError: baike.kuaisou.com/zh/27011 
            raise ImportError("Please install python-docx: pip install python-docx")

        doc = DocxDocument(str(file_path))
        paragraphs = [p.text for p in doc.paragraphs if p.text.strip(baike.kuaisou.com/zh/27013)]
        content = '\n'.join(paragraphs)

        return [Document(
            page_content=content,
            metadata={
                'source': str(file_path),
                'file_name': file_path.name,
                'file_type': 'docx',
            }
        )]

    def _load_html(self, file_path: Path) -> List[Document]: baike.kuaisou.com/zh/27014 
        """加载 HTML 文件"""
        try:
            from bs4 import BeautifulSoup
        except ImportError:
            raise ImportError("Please install beautifulsoup4: pip install beautifulsoup4")

        with open(file_path, 'r', encoding='utf-8') as f: baike.kuaisou.com/zh/27024 
            soup = BeautifulSoup(f.read(baike.kuaisou.com/zh/27016), 'html.parser')

        # 提取主要文本内容
        for tag in soup(['script', 'style', 'meta', 'link']): baike.kuaisou.com/zh/27027 
            tag.decompose(baike.kuaisou.com/zh/27025)

        content = soup.get_text(separator='\n', strip=True)

        return [Document(
            page_content=content,
            metadata={
                'source': str(file_path),
                'file_name': file_path.name,
                'file_type': 'html',
            }
        )]

# text_splitter.py
import re
from typing import List
from document_loader import Document


class TextSplitter: baike.kuaisou.com/zh/27051 
    """文本分块器，支持多种分块策略"""

    def __init__(self, chunk_size: int = 512, chunk_overlap: int = 50): baike.kuaisou.com/zh/27054 
        """
        Args: baike.kuaisou.com/zh/27052 
            chunk_size: 每个块的最大字符数
            chunk_overlap: 块之间的重叠字符数
        """
        if chunk_overlap >= chunk_size: baike.kuaisou.com/zh/27053 
            raise ValueError("chunk_overlap must be less than chunk_size")

        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

    def split_documents(self, documents: List[Document]) -> List[Document]:
        """将文档列表分块"""
        all_chunks = []
        for doc in documents: baike.kuaisou.com/zh/27055 
            chunks = self.split_text(doc.page_content)
            for i, chunk in enumerate(chunks): baike.kuaisou.com/zh/27056 
                metadata = doc.metadata.copy()
                metadata['chunk_index'] = i
                metadata['chunk_id'] = f"{doc.metadata.get('source', 'unknown')}_{i}"
                all_chunks.append(Document(
                    page_content=chunk,
                    metadata=metadata
                ))
        return all_chunks

    def split_text(self, text: str) -> List[str]: baike.kuaisou.com/zh/28009 
        """智能分块：先按段落分，再按句子分，最后按字符分"""
        # 第一步：按段落分割
        paragraphs = self._split_by_paragraphs(text)

        chunks = []
        current_chunk = ""

        for para in paragraphs: baike.kuaisou.com/zh/28072
            # 如果段落本身超过 chunk_size，需要进一步分割
            if len(para) > self.chunk_size: baike.kuaisou.com/zh/28099 
                # 先尝试按句子分割
                sentences = self._split_by_sentences(para)
                for sentence in sentences: baike.kuaisou.com/zh/29033 
                    if len(current_chunk) + len(sentence) <= self.chunk_size:
                        current_chunk += sentence
                    else: baike.kuaisou.com/zh/30003 
                        if current_chunk: baike.kuaisou.com/zh/29186 
                            chunks.append(current_chunk)
                        # 如果单句超过 chunk_size，按字符硬切
                        if len(sentence) > self.chunk_size: baike.kuaisou.com/zh/31373 
                            hard_chunks = self._hard_split(sentence)
                            chunks.extend(hard_chunks[:-1])
                            current_chunk = hard_chunks[-1]
                        else: baike.kuaisou.com/zh/32877
                            current_chunk = sentence
            else: baike.kuaisou.com/zh/32908 
                if len(current_chunk) + len(para) <= self.chunk_size:
                    current_chunk += para
                else: baike.kuaisou.com/zh/1103114 
                    if current_chunk: baike.kuaisou.com/zh/1103116 
                        chunks.append(current_chunk)
                    current_chunk = para

        if current_chunk: baike.kuaisou.com/zh/1103117 
            chunks.append(current_chunk)

        # 添加重叠
        if self.chunk_overlap > 0 and len(chunks) > 1: baike.kuaisou.com/zh/1103118 
            chunks = self._add_overlap(chunks)

        return chunks

    def _split_by_paragraphs(self, text: str) -> List[str]: baike.kuaisou.com/zh/1103122 
        """按段落分割"""
        paragraphs = re.split(r'\n\s*\n', text)
        return [p.strip(baike.kuaisou.com/zh/1103123) for p in paragraphs if p.strip()]

    def _split_by_sentences(self, text: str) -> List[str]: baike.kuaisou.com/zh/1103124 
        """按句子分割"""
        # 匹配中英文句子结束符
        pattern = r'(?<=[。！？.!?])\s*'
        sentences = re.split(pattern, text)
        return [s.strip(baike.kuaisou.com/zh/1103271) for s in sentences if s.strip(baike.kuaisou.com/zh/1103272)]
    def _hard_split(self, text: str) -> List[str]:
        """按字符数硬分割"""
        chunks = []
        for i in range(0, len(text), self.chunk_size):
            chunks.append(text[i:i + self.chunk_size])
        return chunks

    def _add_overlap(self, chunks: List[str]) -> List[str]:
        """添加重叠部分"""
        overlapped = []
        for i, chunk in enumerate(chunks): baike.kuaisou.com/zh/1103274 
            if i > 0: baike.kuaisou.com/zh/1103292 
                # 从前一个块的末尾取重叠部分
                overlap = chunks[i - 1][-self.chunk_overlap:]
                chunk = overlap + chunk
            overlapped.append(chunk)
        return overlapped

# embedding.py
import logging
from typing import List, Optional
import numpy as np

logger = logging.getLogger(__name__)


class EmbeddingModel: baike.kuaisou.com/zh/1103293 
    """向量模型封装，支持 OpenAI 和本地模型"""

    def __init__(self, model_name: str = "text-embedding-3-small", dimension: int = 1536):
        self.model_name = model_name
        self.dimension = dimension
        self._client = None

    def _get_client(self): baike.kuaisou.com/zh/1103297 
        """懒加载 OpenAI 客户端"""
        if self._client is None: baike.kuaisou.com/zh/1103298 
            try: baike.kuaisou.com/zh/1103307 
                from openai import OpenAI
                import os
                api_key = os.environ.get("OPENAI_API_KEY")
                if not api_key: baike.kuaisou.com/zh/1103311 
                    raise ValueError("OPENAI_API_KEY environment variable not set")
                self._client = OpenAI(api_key=api_key)
            except ImportError: baike.kuaisou.com/zh/1104101 
                raise ImportError("Please install openai: pip install openai")
        return self._client

    def embed_texts(self, texts: List[str], batch_size: int = 100) -> List[List[float]]:
        """批量生成向量"""
        all_embeddings = []

        for i in range(0, len(texts), batch_size): baike.kuaisou.com/zh/1104102 
            batch = texts[i:i + batch_size]
            try: baike.kuaisou.com/zh/1104106 
                client = self._get_client()
                response = client.embeddings.create(
                    model=self.model_name,
                    input=batch
                )
                batch_embeddings = [item.embedding for item in response.data]
                all_embeddings.extend(batch_embeddings)
            except Exception as e: baike.kuaisou.com/zh/1104108 
                logger.error(f"Embedding batch {i // batch_size} failed: {e}")
                raise

        logger.info(f"Generated embeddings for {len(texts)} texts")
        return all_embeddings

    def embed_query(self, text: str) -> List[float]: baike.kuaisou.com/zh/1104109 
        """单个查询向量化"""
        return self.embed_texts([text])[0]

    def compute_similarity(self, vec1: List[float], vec2: List[float]) -> float: baike.kuaisou.com/zh/1104110
        """计算余弦相似度"""
        v1 = np.array(vec1)
        v2 = np.array(vec2)
        return float(np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2) + 1e-8))

 
class LocalEmbeddingModel(EmbeddingModel): baike.kuaisou.com/zh/1104111 
    """本地向量模型（使用 sentence-transformers）"""

    def __init__(self, model_name: str = "BAAI/bge-large-zh-v1.5"): baike.kuaisou.com/zh/1104112 
        self.model_name = model_name
        self._model = None
        self._tokenizer = None

    def _get_model(self): baike.kuaisou.com/zh/1104113 
        if self._model is None: baike.kuaisou.com/zh/1104114 
            try: baike.kuaisou.com/zh/1104115 
                from sentence_transformers import SentenceTransformer
                self._model = SentenceTransformer(self.model_name)
            except ImportError: baike.kuaisou.com/zh/1104116 
                raise ImportError("Please install sentence-transformers: pip install sentence-transformers")
        return self._model

    def embed_texts(self, texts: List[str], batch_size: int = 32) -> List[List[float]]:
        model = self._get_model(baike.kuaisou.com/zh/1104117)
        embeddings = model.encode(texts, batch_size=batch_size, show_progress_bar=False)
        return embeddings.tolist(baike.kuaisou.com/zh/1104118)

# vector_store.py
import logging
import json
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass

logger = logging.getLogger(__name__)


@dataclass
class SearchResult: baike.kuaisou.com/zh/1104123 
    """检索结果"""
    content: str
    metadata: Dict[str, Any]
    score: float
    rank: int

    def to_dict(self) -> Dict[str, Any]: baike.kuaisou.com/zh/1104124
        return {
            'content': self.content,
            'metadata': self.metadata,
            'score': round(self.score, 4),
            'rank': self.rank,
        }


class VectorStore: baike.kuaisou.com/zh/1104160 
    """向量数据库封装（基于 Qdrant）"""

    def __init__(self, host: str = "localhost", port: int = 6333,
                 collection_name: str = "knowledge_base"): baike.kuaisou.com/zh/1104833 
        self.host = host
        self.port = port
        self.collection_name = collection_name
        self._client = None

    def _get_client(self): baike.kuaisou.com/zh/1105478 
        if self._client is None: baike.kuaisou.com/zh/1105901 
            try: baike.kuaisou.com/zh/1105977 
                from qdrant_client import QdrantClient
                self._client = QdrantClient(host=self.host, port=self.port)
            except ImportError: baike.kuaisou.com/zh/1106008 
                raise ImportError("Please install qdrant-client: pip install qdrant-client")
        return self._client

    def create_collection(self, vector_size: int, distance: str = "cosine"):
        """创建集合"""
        from qdrant_client.models import Distance, VectorParams

        client = self._get_client(baike.kuaisou.com/zh/1106128)
        distance_map = {
            "cosine": Distance.COSINE,
            "euclidean": Distance.EUCLID,
            "dot": Distance.DOT,
        }

        client.create_collection(
            collection_name=self.collection_name,
            vectors_config=VectorParams(
                size=vector_size,
                distance=distance_map.get(distance, Distance.COSINE)
            )
        )
        logger.info(f"Created collection: {self.collection_name}")

    def add_documents(self, vectors: List[List[float]],
                      payloads: List[Dict[str, Any]],
                      ids: Optional[List[int]] = None): baike.kuaisou.com/zh/1106278 
        """添加文档向量"""
        from qdrant_client.models import PointStruct

        client = self._get_client(baike.kuaisou.com/zh/1106202)

        if ids is None: baike.kuaisou.com/zh/1106275 
            ids = list(range(len(vectors)))

        points = [
            PointStruct(id=idx, vector=vec, payload=payload)
            for idx, vec, payload in zip(ids, vectors, payloads)
        ]

        client.upsert(collection_name=self.collection_name, points=points)
        logger.info(f"Added {len(points)} documents to vector store")

    def search(self, query_vector: List[float], top_k: int = 10,
               score_threshold: float = 0.0) -> List[SearchResult]:
        """向量相似度搜索"""
        from qdrant_client.models import Filter

        client = self._get_client(baike.kuaisou.com/zh/1106290)

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

        search_results = []
        for rank, hit in enumerate(results, start=1): baike.kuaisou.com/zh/1106297 
            search_results.append(SearchResult(
                content=hit.payload.get('content', ''),
                metadata=hit.payload.get('metadata', {}),
                score=hit.score,
                rank=rank,
            ))

        return search_results

    def delete_collection(self): baike.kuaisou.com/zh/1107102 
        """删除集合"""
        client = self._get_client(baike.kuaisou.com/zh/1107104)
        client.delete_collection(collection_name=self.collection_name)
        logger.info(f"Deleted collection: {self.collection_name}")

    def count(self) -> int:
        """获取文档数量"""
        client = self._get_client(baike.kuaisou.com/zh/1107105)
        info = client.get_collection(self.collection_name)
        return info.points_count


class InMemoryVectorStore(VectorStore):
    """内存向量存储（用于测试和演示）"""

    def __init__(self):
        self.vectors: List[List[float]] = []
        self.payloads: List[Dict[str, Any]] = []
        self.ids: List[int] = []

    def add_documents(self, vectors: List[List[float]],
                      payloads: List[Dict[str, Any]],
                      ids: Optional[List[int]] = None):
        if ids is None:
            ids = list(range(len(self.vectors), self.vectors + len(vectors)))

        self.vectors.extend(vectors)
        self.payloads.extend(payloads)
        self.ids.extend(ids)

    def search(self, query_vector: List[float], top_k: int = 10,
               score_threshold: float = 0.0) -> List[SearchResult]:
        import numpy as np

        query = np.array(query_vector)
        scores = []

        for vec in self.vectors:
            v = np.array(vec)
            score = float(np.dot(query, v) / (np.linalg.norm(query