RAG 检索增强生成实战：从零搭建企业级知识库问答系统

RAG（Retrieval-Augmented Generation，检索增强生成）是当前企业落地大模型最主流的方案。它通过"先检索、再生成"的方式，让大模型基于你的私有数据回答问题，有效解决幻觉问题。本文从零开始，手把手搭建一个完整的企业级 RAG 知识库问答系统，包含文档处理、向量化、检索优化、流式输出和前端界面。

一、RAG 系统核心架构

一个生产级 RAG 系统的关键组件：

二、环境准备与依赖安装

# 创建虚拟环境
python -m venv rag_env
source rag_env/bin/activate

# 安装核心依赖
pip install langchain langchain-community langchain-openai
pip install qdrant-client fastapi uvicorn
pip install unstructured[all-docs] python-multipart
pip install pydantic-settings

# 安装 Embedding 模型（本地部署用）
pip install sentence-transformers

三、文档处理与向量化模块

3.1 文档加载与智能切分

"""document_processor.py - 文档处理与智能切分"""

import re
from pathlib import Path
from typing import List
from dataclasses import dataclass, field


@dataclass
class Document:
    """文档块数据结构"""
    content: str
    metadata: dict = field(default_factory=dict)
    chunk_id: str = ""


class DocumentProcessor:
    """文档处理器：加载、清洗、切分"""
    
    def __init__(self, chunk_size: int = 512, chunk_overlap: int = 64):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
    
    def load_text_file(self, filepath: str) -> str:
        """加载文本文件"""
        path = Path(filepath)
        if not path.exists():
            raise FileNotFoundError(f"文件不存在: {filepath}")
        
        content = path.read_text(encoding="utf-8", errors="ignore")
        return self._clean_text(content)
    
    def load_markdown(self, filepath: str) -> List[Document]:
        """加载 Markdown 文件并按标题层级切分"""
        content = self.load_text_file(filepath)
        return self._split_by_headings(content, filepath)
    
    def load_directory(self, dirpath: str, extensions: List[str] = None) -> List[Document]:
        """批量加载目录下的文件"""
        if extensions is None:
            extensions = [".md", ".txt", ".py", ".js", ".ts", ".json"]
        
        docs = []
        dir_path = Path(dirpath)
        
        for filepath in dir_path.rglob("*"):
            if not filepath.is_file():
                continue
            if filepath.suffix not in extensions:
                continue
            
            if filepath.suffix == ".md":
                docs.extend(self.load_markdown(str(filepath)))
            else:
                content = self.load_text_file(str(filepath))
                chunks = self._recursive_split(content, str(filepath))
                docs.extend(chunks)
        
        print(f"从 {dirpath} 加载了 {len(docs)} 个文档块")
        return docs
    
    def _clean_text(self, text: str) -> str:
        """清洗文本"""
        # 移除多余空行
        text = re.sub(r'\n{3,}', '\n\n', text)
        # 移除行首行尾空白
        lines = [line.strip() for line in text.splitlines()]
        return '\n'.join(lines)
    
    def _split_by_headings(self, md_content: str, source: str) -> List[Document]:
        """按 Markdown 标题层级切分"""
        sections = []
        current_section = {"title": "开头", "level": 0, "content": []}
        
        for line in md_content.splitlines():
            heading_match = re.match(r'^(#{1,6})\s+(.+)$', line)
            if heading_match:
                # 保存上一个 section
                if current_section["content"]:
                    content = '\n'.join(current_section["content"]).strip()
                    if content:
                        sections.append({
                            "title": current_section["title"],
                            "level": current_section["level"],
                            "content": content
                        })
                current_section = {
                    "title": heading_match.group(2).strip(),
                    "level": len(heading_match.group(1)),
                    "content": [line]
                }
            else:
                current_section["content"].append(line)
        
        # 最后一个 section
        if current_section["content"]:
            content = '\n'.join(current_section["content"]).strip()
            if content:
                sections.append({
                    "title": current_section["title"],
                    "level": current_section["level"],
                    "content": content
                })
        
        # 合并小 section，确保每个块 >= chunk_size
        docs = []
        buffer_content = []
        buffer_title = ""
        
        for section in sections:
            buffer_content.append(f"## {section['title']}\n{section['content']}")
            buffer_title = section['title']
            
            total_len = sum(len(c) for c in buffer_content)
            if total_len >= self.chunk_size or section == sections[-1]:
                combined = '\n\n'.join(buffer_content)
                # 如果合并后太长，递归切分
                if len(combined) > self.chunk_size * 2:
                    chunks = self._recursive_split(combined, source, buffer_title)
                    docs.extend(chunks)
                else:
                    docs.append(Document(
                        content=combined,
                        metadata={"source": source, "section": buffer_title},
                        chunk_id=f"{Path(source).stem}_{len(docs)}"
                    ))
                buffer_content = []
        
        return docs
    
    def _recursive_split(self, text: str, source: str, section: str = "") -> List[Document]:
        """递归字符切分"""
        if len(text) <= self.chunk_size:
            return [Document(
                content=text,
                metadata={"source": source, "section": section},
                chunk_id=f"{Path(source).stem}_chunk"
            )]
        
        chunks = []
        start = 0
        chunk_idx = 0
        
        while start < len(text):
            end = start + self.chunk_size
            
            # 尝试在段落边界处切分
            if end < len(text):
                # 找最后一个换行符
                newline_pos = text.rfind('\n', start, end)
                if newline_pos > start + self.chunk_size // 2:
                    end = newline_pos + 1
                else:
                    # 找最后一个句号
                    period_pos = text.rfind('。', start, end)
                    if period_pos > start + self.chunk_size // 2:
                        end = period_pos + 1
            
            chunk = text[start:end].strip()
            if chunk:
                chunks.append(Document(
                    content=chunk,
                    metadata={"source": source, "section": section},
                    chunk_id=f"{Path(source).stem}_{chunk_idx}"
                ))
                chunk_idx += 1
            
            start = end - self.chunk_overlap if end < len(text) else end
        
        return chunks


# 使用示例
if __name__ == "__main__":
    processor = DocumentProcessor(chunk_size=512, chunk_overlap=64)
    docs = processor.load_directory("./knowledge_base", [".md", ".txt"])
    for doc in docs[:3]:
        print(f"[{doc.chunk_id}] {doc.metadata.get('section', '')} ({len(doc.content)} 字符)")
        print(f"  {doc.content[:100]}...")

3.2 向量化与存储

"""vector_store.py - 向量化与 Qdrant 存储"""

import json
from typing import List, Optional
from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue
)
from document_processor import Document


class RAGVectorStore:
    """RAG 向量存储管理"""
    
    def __init__(
        self,
        collection_name: str = "knowledge_base",
        qdrant_url: str = "localhost:6333",
        embedding_dim: int = 1024,
        api_key: Optional[str] = None
    ):
        self.collection_name = collection_name
        self.client = QdrantClient(url=qdrant_url, api_key=api_key)
        self.embedding_dim = embedding_dim
        self._ensure_collection()
    
    def _ensure_collection(self):
        """确保集合存在"""
        collections = [c.name for c in self.client.get_collections().collections]
        if self.collection_name not in collections:
            self.client.create_collection(
                collection_name=self.collection_name,
                vectors_config=VectorParams(
                    size=self.embedding_dim,
                    distance=Distance.COSINE
                )
            )
            print(f"创建集合: {self.collection_name}")
    
    def embed_texts(self, texts: List[str]) -> List[List[float]]:
        """文本向量化（使用 OpenAI Embedding API）"""
        from openai import OpenAI
        client = OpenAI()
        
        # 批量处理，每批 100 条
        all_embeddings = []
        batch_size = 100
        
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            response = client.embeddings.create(
                model="text-embedding-3-small",
                input=batch,
                dimensions=self.embedding_dim
            )
            batch_embeddings = [item.embedding for item in response.data]
            all_embeddings.extend(batch_embeddings)
        
        return all_embeddings
    
    def index_documents(self, documents: List[Document], batch_size: int = 100):
        """将文档向量化并存储"""
        texts = [doc.content for doc in documents]
        
        print(f"正在向量化 {len(texts)} 个文档块...")
        embeddings = self.embed_texts(texts)
        
        # 构建点数据
        points = []
        for idx, (doc, embedding) in enumerate(zip(documents, embeddings)):
            point = PointStruct(
                id=idx,
                vector=embedding,
                payload={
                    "content": doc.content,
                    "source": doc.metadata.get("source", ""),
                    "section": doc.metadata.get("section", ""),
                    "chunk_id": doc.chunk_id
                }
            )
            points.append(point)
        
        # 批量上传
        for i in range(0, len(points), batch_size):
            batch = points[i:i + batch_size]
            self.client.upsert(
                collection_name=self.collection_name,
                points=batch
            )
            print(f"  已上传 {min(i + batch_size, len(points))}/{len(points)}")
        
        print(f"索引完成，共 {len(documents)} 个文档块")
    
    def search(
        self,
        query: str,
        top_k: int = 5,
        source_filter: Optional[str] = None
    ) -> List[dict]:
        """语义检索"""
        query_embedding = self.embed_texts([query])[0]
        
        search_filter = None
        if source_filter:
            search_filter = Filter(
                must=[FieldCondition(key="source", match=MatchValue(value=source_filter))]
            )
        
        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_embedding,
            limit=top_k,
            query_filter=search_filter,
            with_payload=True
        )
        
        return [
            {
                "content": hit.payload["content"],
                "source": hit.payload["source"],
                "section": hit.payload.get("section", ""),
                "score": hit.score
            }
            for hit in results
        ]
    
    def get_stats(self) -> dict:
        """获取集合统计信息"""
        info = self.client.get_collection(self.collection_name)
        return {
            "collection": self.collection_name,
            "vector_count": info.points_count,
            "vector_dim": self.embedding_dim
        }


# 使用示例
if __name__ == "__main__":
    from document_processor import DocumentProcessor
    
    # 1. 加载文档
    processor = DocumentProcessor(chunk_size=512, chunk_overlap=64)
    docs = processor.load_directory("./knowledge_base")
    
    # 2. 向量化存储
    store = RAGVectorStore(
        collection_name="my_knowledge",
        qdrant_url="localhost:6333",
        embedding_dim=1024
    )
    store.index_documents(docs)
    
    # 3. 检索测试
    results = store.search("如何配置 Nginx 反向代理？", top_k=3)
    for r in results:
        print(f"[{r['score']:.3f}] {r['section']}")
        print(f"  {r['content'][:100]}...")

四、RAG 问答引擎

"""rag_engine.py - RAG 问答引擎：检索 + 生成"""

import json
from typing import List, Optional, AsyncGenerator
from openai import OpenAI
from vector_store import RAGVectorStore


RAG_SYSTEM_PROMPT = """你是一个专业的企业知识库助手。请根据提供的参考资料回答用户问题。

规则：
1. 只基于参考资料回答，如果资料中没有相关信息，明确告知用户
2. 引用来源时标注文档名称
3. 回答要结构化，使用列表和标题组织内容
4. 技术问题要提供具体的配置示例或代码
5. 如果参考资料信息不足，可以结合通用知识补充，但要明确区分"""

RAG_USER_TEMPLATE = """参考资料：
{context}

用户问题：{question}

请基于以上参考资料回答。"""


class RAGEngine:
    """RAG 问答引擎"""
    
    def __init__(
        self,
        vector_store: RAGVectorStore,
        model: str = "gpt-4o-mini",
        top_k: int = 5,
        max_context_chars: int = 6000
    ):
        self.vector_store = vector_store
        self.client = OpenAI()
        self.model = model
        self.top_k = top_k
        self.max_context_chars = max_context_chars
    
    def _build_context(self, search_results: List[dict]) -> str:
        """构建上下文文本"""
        context_parts = []
        total_chars = 0
        
        for result in search_results:
            source = result.get("source", "未知来源")
            section = result.get("section", "")
            content = result["content"]
            
            # 估算加上后的总长度
            part = f"【{source} - {section}】\n{content}\n"
            if total_chars + len(part) > self.max_context_chars:
                # 截断当前块以适应限制
                remaining = self.max_context_chars - total_chars
                if remaining > 200:
                    part = f"【{source} - {section}】\n{content[:remaining]}...\n"
                    context_parts.append(part)
                break
            
            context_parts.append(part)
            total_chars += len(part)
        
        return '\n'.join(context_parts)
    
    def ask(self, question: str) -> dict:
        """同步问答"""
        # 1. 检索
        search_results = self.vector_store.search(question, top_k=self.top_k)
        
        # 2. 构建上下文
        context = self._build_context(search_results)
        
        # 3. 生成回答
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": RAG_SYSTEM_PROMPT},
                {"role": "user", "content": RAG_USER_TEMPLATE.format(
                    context=context,
                    question=question
                )}
            ],
            temperature=0.3,
            max_tokens=2000
        )
        
        return {
            "question": question,
            "answer": response.choices[0].message.content,
            "sources": [
                {"source": r["source"], "section": r["section"], "score": r["score"]}
                for r in search_results
            ],
            "model": self.model,
            "tokens_used": response.usage.total_tokens if response.usage else 0
        }
    
    def ask_stream(self, question: str):
        """流式问答"""
        search_results = self.vector_store.search(question, top_k=self.top_k)
        context = self._build_context(search_results)
        
        stream = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": RAG_SYSTEM_PROMPT},
                {"role": "user", "content": RAG_USER_TEMPLATE.format(
                    context=context,
                    question=question
                )}
            ],
            temperature=0.3,
            max_tokens=2000,
            stream=True
        )
        
        sources = [
            {"source": r["source"], "section": r["section"], "score": r["score"]}
            for r in search_results
        ]
        
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield {
                    "type": "token",
                    "content": chunk.choices[0].delta.content
                }
        
        yield {"type": "done", "sources": sources}


# 使用示例
if __name__ == "__main__":
    store = RAGVectorStore(collection_name="my_knowledge")
    engine = RAGEngine(vector_store=store)
    
    result = engine.ask("如何配置 Nginx 负载均衡？")
    print(f"问题：{result['question']}")
    print(f"回答：{result['answer']}")
    print(f"来源：{json.dumps(result['sources'], ensure_ascii=False, indent=2)}")
    print(f"Token 消耗：{result['tokens_used']}")

五、FastAPI 接口服务

"""api_server.py - RAG 问答 API 服务"""

import os
from typing import Optional
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from rag_engine import RAGEngine
from vector_store import RAGVectorStore
from document_processor import DocumentProcessor
import json


app = FastAPI(title="RAG 知识库问答 API", version="1.0.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# 全局引擎实例
engine: Optional[RAGEngine] = None


class QueryRequest(BaseModel):
    question: str
    stream: bool = False
    top_k: int = 5


class IndexRequest(BaseModel):
    directory: str
    extensions: list = [".md", ".txt", ".py", ".json"]


def get_engine() -> RAGEngine:
    """获取或初始化 RAG 引擎"""
    global engine
    if engine is None:
        store = RAGVectorStore(
            collection_name=os.getenv("QDRANT_COLLECTION", "knowledge_base"),
            qdrant_url=os.getenv("QDRANT_URL", "localhost:6333"),
            api_key=os.getenv("QDRANT_API_KEY"),
            embedding_dim=int(os.getenv("EMBEDDING_DIM", "1024"))
        )
        engine = RAGEngine(
            vector_store=store,
            model=os.getenv("LLM_MODEL", "gpt-4o-mini")
        )
    return engine


@app.post("/api/index")
async def index_documents(req: IndexRequest):
    """索引文档目录"""
    try:
        processor = DocumentProcessor(chunk_size=512, chunk_overlap=64)
        docs = processor.load_directory(req.directory, req.extensions)
        
        store = get_engine().vector_store
        store.index_documents(docs)
        
        stats = store.get_stats()
        return {"status": "ok", "indexed": len(docs), "stats": stats}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/ask")
async def ask_question(req: QueryRequest):
    """问答接口"""
    eng = get_engine()
    eng.top_k = req.top_k
    
    if req.stream:
        def generate():
            for chunk in eng.ask_stream(req.question):
                yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
        
        return StreamingResponse(generate(), media_type="text/event-stream")
    
    result = eng.ask(req.question)
    return result


@app.get("/api/stats")
async def get_stats():
    """获取索引统计"""
    store = get_engine().vector_store
    return store.get_stats()


@app.get("/api/health")
async def health():
    return {"status": "ok", "service": "RAG Knowledge Base"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

启动服务：

# 设置环境变量
export QDRANT_URL=localhost:6333
export QDRANT_COLLECTION=knowledge_base
export LLM_MODEL=gpt-4o-mini
export OPENAI_API_KEY=your-api-key

# 启动
python api_server.py

六、检索质量优化：混合检索 + 重排序

"""hybrid_search.py - 混合检索与重排序"""

import re
import math
from typing import List
from vector_store import RAGVectorStore


class HybridSearcher:
    """混合检索：语义检索 + 关键词检索 + 重排序"""
    
    def __init__(self, vector_store: RAGVectorStore):
        self.vector_store = vector_store
    
    def keyword_search(
        self,
        query: str,
        documents: List[dict],
        top_k: int = 20
    ) -> List[dict]:
        """TF-IDF 风格的关键词检索"""
        # 分词
        query_terms = set(re.findall(r'\w+', query.lower()))
        
        scored_docs = []
        for doc in documents:
            content_terms = set(re.findall(r'\w+', doc["content"].lower()))
            
            # BM25 简化版评分
            common_terms = query_terms & content_terms
            if not common_terms:
                continue
            
            # 词频
            tf = len(common_terms) / len(content_terms) if content_terms else 0
            # 逆文档频率（简化）
            idf = math.log(len(documents) / max(sum(1 for d in documents if any(t in d["content"].lower() for t in common_terms)), 1))
            
            score = tf * idf
            scored_docs.append({**doc, "keyword_score": score})
        
        scored_docs.sort(key=lambda x: x["keyword_score"], reverse=True)
        return scored_docs[:top_k]
    
    def reciprocal_rank_fusion(
        self,
        semantic_results: List[dict],
        keyword_results: List[dict],
        k: int = 60
    ) -> List[dict]:
        """RRF 融合排序"""
        scores = {}
        
        for rank, doc in enumerate(semantic_results):
            key = doc["content"][:100]  # 用内容前100字符作为唯一键
            scores[key] = scores.get(key, 0) + 1 / (k + rank + 1)
            if key not in scores or "merged" not in scores:
                scores[f"{key}_data"] = doc
        
        for rank, doc in enumerate(keyword_results):
            key = doc["content"][:100]
            scores[key] = scores.get(key, 0) + 1 / (k + rank + 1)
            if f"{key}_data" not in scores:
                scores[f"{key}_data"] = doc
        
        # 排序
        fused = []
        seen_keys = set()
        for key, score in sorted(scores.items(), key=lambda x: x[1], reverse=True):
            if key.endswith("_data"):
                data = scores[key]
                content_key = key[:-5]
                if content_key not in seen_keys:
                    data["rrf_score"] = scores[content_key]
                    fused.append(data)
                    seen_keys.add(content_key)
        
        return fused
    
    def search(
        self,
        query: str,
        top_k: int = 5
    ) -> List[dict]:
        """执行混合检索"""
        # 1. 语义检索
        semantic_results = self.vector_store.search(query, top_k=20)
        
        # 2. 关键词检索（从语义结果中重排）
        keyword_results = self.keyword_search(query, semantic_results, top_k=20)
        
        # 3. RRF 融合
        fused = self.reciprocal_rank_fusion(semantic_results, keyword_results)
        
        return fused[:top_k]


# 使用示例
if __name__ == "__main__":
    store = RAGVectorStore(collection_name="my_knowledge")
    searcher = HybridSearcher(store)
    
    results = searcher.search("Nginx HTTPS 证书配置", top_k=5)
    for r in results:
        print(f"[RRF: {r.get('rrf_score', 0):.4f}] {r.get('section', '')}")
        print(f"  {r['content'][:120]}...")

七、Docker 一键部署

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

RUN pip install --no-cache-dir \
    fastapi uvicorn qdrant-client \
    langchain langchain-openai \
    unstructured[all-docs] \
    sentence-transformers \
    python-multipart

COPY . .

EXPOSE 8000

CMD ["python", "api_server.py"]

# docker-compose.yml
version: '3.8'
services:
  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
    volumes:
      - qdrant_data:/qdrant/storage

  rag-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - QDRANT_URL=http://qdrant:6333
      - QDRANT_COLLECTION=knowledge_base
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    depends_on:
      - qdrant

volumes:
  qdrant_data:

启动：docker-compose up -d

常见问题 FAQ

Q1: chunk_size 和 chunk_overlap 如何选择？

A: chunk_size 建议在 256-1024 之间，取决于你的文档类型。技术文档用 512 效果好，法律文档可能需要更大的块。chunk_overlap 设为 chunk_size 的 10-20%（如 512 对应 64），确保上下文连贯。

Q2: 向量数据库选 Qdrant 还是 pgvector？

A: 小型项目（< 10 万条）用 pgvector 最方便，不需要额外服务。中大型项目（> 10 万条）推荐 Qdrant 或 Milvus，它们支持更丰富的过滤和更快的检索速度。如果已有 PostgreSQL 基础设施，pgvector 的运维成本最低。

Q3: RAG 的回答质量不好怎么办？

A: 按优先级排查：1) 文档切分是否合理（检查是否有语义被切断）；2) Embedding 模型是否匹配你的语言（中文推荐 BGE-M3）；3) 检索结果是否相关（加 Reranker）；4) Prompt 是否清晰（明确要求基于资料回答）。

Q4: 如何处理 PDF 中的表格和图片？

A: 使用 Unstructured 库可以提取 PDF 中的表格。图片则需要 OCR，推荐使用 Surya 或 PaddleOCR。表格建议转为 Markdown 格式存储，这样大模型能更好地理解。

Q5: 实时数据如何集成到 RAG 中？

A: 对于频繁更新的数据（如数据库记录），可以：1) 定时全量重建索引（适合小数据量）；2) 增量索引 + 变更监听（适合大数据量）；3) 对于实时性要求极高的场景，直接将 SQL 查询结果作为上下文注入 Prompt，跳过向量检索。