基于RAG的DeepSeek本地知识库构建：从分块策略到混合检索的工程实践

+----------------------------------------------------------+
|                   应用层（Gradio / API Server）             |
+----------------------------------------------------------+
|                   编排层（LangChain / LlamaIndex）          |
|          (Query Rewriting -> Multi-Recall -> Rerank)       |
+----------------------------------------------------------+
|           检索层（Milvus 2.4）                              |
|   [稠密索引: HNSW] + [稀疏索引: Sparse-BM25]               |
+----------------------------------------------------------+
|           嵌入层 & 生成层（本地GPU推理）                    |
|   BGE-M3 (Embedding) + DeepSeek-R1:14B (Generation)       |
+----------------------------------------------------------+
|                   数据层（本地NAS / MinIO）                 |
|      源文件(PDF/Word/Markdown) -> 解析 -> 分块             |
+----------------------------------------------------------+

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import PyPDFLoader, UnstructuredMarkdownLoader
import numpy as np

class SemanticChunker:
    def __init__(self, embedding_model, threshold=0.65):
        self.embedder = embedding_model
        self.threshold = threshold  # 余弦相似度阈值
        self.base_splitter = RecursiveCharacterTextSplitter(
            chunk_size=512,
            chunk_overlap=128,
            separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
        )

    def split(self, text):
        base_chunks = self.base_splitter.split_text(text)
        if len(base_chunks) <= 1:
            return base_chunks
        
        # 计算相邻块的embedding相似度
        final_chunks = []
        current_chunk = base_chunks[0]
        for i in range(1, len(base_chunks)):
            emb_curr = self.embedder.embed_query(base_chunks[i-1])
            emb_next = self.embedder.embed_query(base_chunks[i])
            sim = np.dot(emb_curr, emb_next) / (np.linalg.norm(emb_curr) * np.linalg.norm(emb_next))
            
            if sim < self.threshold:  # 语义突变，断开
                final_chunks.append(current_chunk)
                current_chunk = base_chunks[i]
            else:
                current_chunk += base_chunks[i]
        if current_chunk:
            final_chunks.append(current_chunk)
        return final_chunks

from FlagEmbedding import BGEM3FlagModel

class BGEM3Embedder:
    def __init__(self, model_path="/models/bge-m3"):
        self.model = BGEM3FlagModel(model_path, use_fp16=True, device="cuda")
    
    def encode_documents(self, texts):
        # 返回 Dense 向量 和 Sparse 词权重
        outputs = self.model.encode(texts, return_dense=True, return_sparse=True, return_colbert_vecs=False)
        dense_vecs = outputs['dense_vecs']
        sparse_vecs = outputs['lexical_weights']  # List[Dict[int, float]]
        return dense_vecs, sparse_vecs
    
    def encode_query(self, query):
        # 查询需额外增加指令："Represent this sentence for searching relevant passages: "
        formatted_query = f"Represent this sentence for searching relevant passages: {query}"
        outputs = self.model.encode(formatted_query, return_dense=True, return_sparse=True)
        return outputs['dense_vecs'][0], outputs['lexical_weights'][0]

from pymilvus import CollectionSchema, FieldSchema, DataType, Collection, connections

connections.connect(host="localhost", port="19530")

dense_field = FieldSchema(name="dense_vector", dtype=DataType.FLOAT_VECTOR, dim=1024)
sparse_field = FieldSchema(name="sparse_vector", dtype=DataType.SPARSE_FLOAT_VECTOR)
id_field = FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True)
text_field = FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535)
meta_field = FieldSchema(name="metadata", dtype=DataType.JSON)

schema = CollectionSchema(fields=[id_field, dense_field, sparse_field, text_field, meta_field])
collection = Collection("knowledge_base", schema)

# 创建索引
dense_index = {"index_type": "HNSW", "metric_type": "IP", "params": {"M": 16, "efConstruction": 200}}
sparse_index = {"index_type": "SPARSE_INVERTED_INDEX", "metric_type": "IP"}  # IP 即内积

collection.create_index("dense_vector", dense_index)
collection.create_index("sparse_vector", sparse_index)
collection.load()

def ingest_documents(chunks, embedder, collection):
    texts = [chunk.page_content for chunk in chunks]
    dense_list, sparse_list = embedder.encode_documents(texts)
    
    entities = [
        dense_list.tolist(),  # 注意转list
        sparse_list,
        texts,
        [chunk.metadata for chunk in chunks]
    ]
    collection.insert([dense_vector, sparse_vector, text, metadata])
    collection.flush()

def hybrid_search(query, collection, embedder, top_k=10, alpha=0.7):
    dense_q, sparse_q = embedder.encode_query(query)
    
    # Milvus 2.4 Hybrid Search 语法
    search_params = {
        "dense": {"metric_type": "IP", "params": {"ef": 64}},
        "sparse": {"metric_type": "IP"}
    }
    
    result = collection.hybrid_search(
        reqs=[
            {"vector": dense_q, "anns_field": "dense_vector", "param": search_params["dense"], "limit": top_k * 2},
            {"vector": sparse_q, "anns_field": "sparse_vector", "param": search_params["sparse"], "limit": top_k * 2}
        ],
        rerank={"strategy": "weighted", "weights": [alpha, 1-alpha]},
        limit=top_k,
        output_fields=["text", "metadata"]
    )
    return result

python -m vllm.entrypoints.openai.api_server \
  --model /models/deepseek-r1-14b \
  --served-model-name deepseek-r1 \
  --tensor-parallel-size 1 \
  --max-model-len 8192 \
  --enforce-eager \
  --kv-cache-dtype fp8_e4m3  # 启用 FP8 缓存，显存占用降低 40%

SYSTEM_PROMPT = """
你是一个严谨的知识库助手。你必须严格基于以下【上下文】内容回答用户问题。
规则：
1. 如果上下文中没有提及任何相关信息，请直接回复“根据当前知识库，无法找到相关信息”，严禁编造。
2. 如果上下文中存在信息，请提取最相关的部分进行简洁、条理清晰的回答，并标注信息来源（引用文件名）。
3. 回答中禁止出现“根据我的理解”、“我认为”等主观表述。
"""

def build_prompt(query, context_chunks):
    context_text = "\n\n---\n\n".join([
        f"【来源：{chunk['metadata']['source']}】{chunk['text']}" 
        for chunk in context_chunks
    ])
    return f"{SYSTEM_PROMPT}\n\n【上下文】\n{context_text}\n\n【用户问题】\n{query}\n\n【回答】"

import openai
client = openai.OpenAI(base_url="http://localhost:8000/v1", api_key="EMPTY")

def generate_response(query, retrieved_chunks):
    prompt = build_prompt(query, retrieved_chunks)
    response = client.chat.completions.create(
        model="deepseek-r1",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,  # 低温度减少发散
        top_p=0.9,
        max_tokens=2048
    )
    return response.choices[0].message.content

from math_utils import shannon_entropy
def filter_noise(text, min_entropy=4.5):
    # 中文汉字 Unicode 范围，计算信息熵
    chars = [c for c in text if '\u4e00' <= c <= '\u9fff']
    if len(chars) < 10:
        return False
    entropy = shannon_entropy(''.join(chars))
    return entropy > min_entropy

def evaluate(retriever, qa_pairs):
    hits = 0
    for q, a, relevant_id in qa_pairs:
        res = retriever.search(q, top_k=5)
        if relevant_id in [r.id for r in res]:
            hits += 1
    return hits / len(qa_pairs)

torch==2.1.2
vllm==0.5.4
pymilvus==2.4.5
FlagEmbedding==1.2.11
langchain==0.3.7
pypdfium2==4.30.0