基于RAG架构的DeepSeek大模型本地知识库构建实战（一站式企业级解决方案）

# 安装依赖
pip install vllm transformers torch

# 下载模型（提前使用 modelscope 或 huggingface）
git lfs install
git clone https://modelscope.cn/deepseek-ai/DeepSeek-R1-Distill-Qwen-14B.git /models/deepseek-14b

# 启动 vLLM 服务（生产环境关键参数）
python -m vllm.entrypoints.openai.api_server \
  --model /models/deepseek-14b \
  --tensor-parallel-size 1 \          # 单卡设为1，多卡使用2/4
  --gpu-memory-utilization 0.85 \     # 预留15%显存给KV Cache
  --max-model-len 8192 \              # 最大上下文长度
  --max-num-seqs 32 \                 # 最大并发序列数
  --swap-space 16 \                   # CPU 交换空间
  --dtype bfloat16 \
  --port 8000 \
  --served-model-name deepseek-14b

curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "deepseek-14b",
    "messages": [{"role": "user", "content": "请介绍RAG架构的核心组件"}],
    "temperature": 0.1,
    "max_tokens": 1024
  }'

# 下载并导入 GGUF 量化模型
ollama pull deepseek-r1:14b-q4_K_M
ollama list

# 启动服务（默认 11434 端口）
ollama serve

┌─────────────┐    ┌─────────────┐    ┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  Query      │    │  Query      │    │  Hybrid     │    │  Rerank &   │    │  LLM        │
│  Rewrite    │ -> │  Expansion  │ -> │  Retrieval  │ -> │  Context    │ -> │  Generation │
│  (HyDE)     │    │  (Multi-    │    │  (Dense+    │    │  Compress   │    │  (with      │
│             │    │   Query)    │    │   Sparse)   │    │  (LLMLingua)│    │   Citations)│
└─────────────┘    └─────────────┘    └─────────────┘    └─────────────┘    └─────────────┘
       │                   │                   │                   │                   │
       └───────────────────┴───────────────────┴───────────────────┴───────────────────┘
                                         │
                                     ┌────┴────┐
                                     │ Milvus  │
                                     │ (LVQ+   │
                                     │  IVF)   │
                                     └─────────┘

# docker-compose-milvus-cluster.yml
version: '3.5'
services:
  etcd:
    image: quay.io/coreos/etcd:v3.5.5
    environment:
      - ETCD_AUTO_COMPACTION_MODE=revision
      - ETCD_AUTO_COMPACTION_RETENTION=1000
    volumes:
      - ./volumes/etcd:/etcd
    command: etcd -advertise-client-urls=http://0.0.0.0:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd

  minio:
    image: minio/minio:RELEASE.2023-03-20T20-16-18Z
    environment:
      MINIO_ACCESS_KEY: minioadmin
      MINIO_SECRET_KEY: minioadmin
    volumes:
      - ./volumes/minio:/minio_data
    command: minio server /minio_data
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 30s
      timeout: 20s
      retries: 3

  standalone:
    image: milvusdb/milvus:v2.4.0
    command: ["milvus", "run", "standalone"]
    environment:
      ETCD_ENDPOINTS: etcd:2379
      MINIO_ADDRESS: minio:9000
      MINIO_ACCESS_KEY_ID: minioadmin
      MINIO_SECRET_ACCESS_KEY: minioadmin
    ports:
      - "19530:19530"
      - "9091:9091"
    depends_on:
      - etcd
      - minio
    volumes:
      - ./volumes/milvus:/var/lib/milvus

# connect_milvus.py
from pymilvus import connections, Collection, CollectionSchema, FieldSchema, DataType, utility

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

# 定义 Schema - 支持多租户 (tenant_id 作为分区键)
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024),  # BGE-M3 为 1024 维
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="doc_name", dtype=DataType.VARCHAR, max_length=255),
    FieldSchema(name="chunk_index", dtype=DataType.INT64),
    FieldSchema(name="tenant_id", dtype=DataType.VARCHAR, max_length=64, is_partition_key=True),  # 自动分区
    FieldSchema(name="metadata", dtype=DataType.JSON),  # 存储标签、时间等
]
schema = CollectionSchema(fields, description="Enterprise RAG Knowledge Base")
collection = Collection("rag_knowledge", schema, consistency_level="Strong")

# 创建索引（IVF_SQ8 平衡性能与精度，生产用 HNSW）
index_params = {
    "metric_type": "IP",  # 内积，配合 BGE 使用
    "index_type": "IVF_SQ8",
    "params": {"nlist": 4096}
}
collection.create_index(field_name="embedding", index_params=index_params)
collection.load()
print("Collection ready.")

# parsers/document_parser.py
import fitz  # PyMuPDF
from unstructured.partition.docx import partition_docx
import pytesseract
from PIL import Image

def parse_pdf_with_ocr(file_path: str) -> list[str]:
    doc = fitz.open(file_path)
    pages_text = []
    for page_num in range(len(doc)):
        page = doc[page_num]
        text = page.get_text()
        if text.strip():
            pages_text.append(text)
        else:
            # 若无可提取文本，触发 OCR
            pix = page.get_pixmap(dpi=150)
            img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
            ocr_text = pytesseract.image_to_string(img, lang='chi_sim+eng')
            pages_text.append(ocr_text)
    return pages_text

# splitters/semantic_chunker.py
from langchain.text_splitter import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer
import numpy as np

embedder = SentenceTransformer('BAAI/bge-m3', device='cuda')

def semantic_split(text: str, max_chunk_size: int = 512, threshold: float = 0.3):
    sentences = RecursiveCharacterTextSplitter(
        separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],
        chunk_size=50,
        chunk_overlap=10,
        length_function=len
    ).split_text(text)
    
    if len(sentences) <= 1:
        return [text]
    
    # 计算每个句子的向量
    sentence_embeddings = embedder.encode(sentences, normalize_embeddings=True)
    
    chunks = []
    current_chunk = [sentences[0]]
    current_embedding = sentence_embeddings[0]
    
    for i in range(1, len(sentences)):
        # 计算当前句子与现有块平均向量的余弦相似度
        sim = np.dot(current_embedding, sentence_embeddings[i])
        if sim < threshold or len("".join(current_chunk)) > max_chunk_size:
            chunks.append("".join(current_chunk))
            current_chunk = [sentences[i]]
            current_embedding = sentence_embeddings[i]
        else:
            current_chunk.append(sentences[i])
            # 滑动平均更新块向量
            current_embedding = (current_embedding + sentence_embeddings[i]) / 2
            current_embedding = current_embedding / np.linalg.norm(current_embedding)
    
    if current_chunk:
        chunks.append("".join(current_chunk))
    return chunks

# retrievers/hybrid_retriever.py
from elasticsearch import Elasticsearch
from pymilvus import Collection
from sentence_transformers import SentenceTransformer
import numpy as np

es_client = Elasticsearch("http://localhost:9200")
milvus_collection = Collection("rag_knowledge")
embedder = SentenceTransformer('BAAI/bge-m3')

def hybrid_search(query: str, tenant_id: str, top_k: int = 20):
    # 1. Dense 向量检索
    query_emb = embedder.encode(query, normalize_embeddings=True).tolist()
    
    milvus_res = milvus_collection.search(
        data=[query_emb],
        anns_field="embedding",
        param={"metric_type": "IP", "params": {"nprobe": 16}},
        limit=top_k,
        expr=f'tenant_id == "{tenant_id}"',
        output_fields=["id", "text", "doc_name", "metadata"]
    )
    dense_hits = {hit.id: (hit.score, hit.entity.get('text')) for hit in milvus_res[0]}
    
    # 2. Sparse 检索 (BM25 via ES)
    es_query = {
        "bool": {
            "must": {"match": {"text": query}},
            "filter": {"term": {"tenant_id": tenant_id}}
        }
    }
    es_res = es_client.search(index="rag_index", query=es_query, size=top_k)
    sparse_hits = {hit['_id']: (hit['_score'], hit['_source']['text']) for hit in es_res['hits']['hits']}
    
    # 3. RRF 融合 (Reciprocal Rank Fusion)
    all_ids = set(dense_hits.keys()) | set(sparse_hits.keys())
    rrf_scores = {}
    rank_constant = 60
    
    for doc_id in all_ids:
        score = 0
        if doc_id in dense_hits:
            rank = list(dense_hits.keys()).index(doc_id) + 1
            score += 1 / (rank + rank_constant)
        if doc_id in sparse_hits:
            rank = list(sparse_hits.keys()).index(doc_id) + 1
            score += 1 / (rank + rank_constant)
        rrf_scores[doc_id] = score
    
    sorted_ids = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)[:top_k]
    results = []
    for doc_id, score in sorted_ids:
        text = dense_hits.get(doc_id, [None, None])[1] or sparse_hits.get(doc_id, [None, None])[1]
        results.append({"id": doc_id, "text": text, "rrf_score": score})
    return results

# rerankers/bge_reranker.py
from FlagEmbedding import BGEM3FlagModel
model = BGEM3FlagModel('BAAI/bge-m3', use_fp16=True)

def rerank(query: str, passages: list[str], top_n: int = 5):
    # 计算交叉编码器分数（比双编码器更准）
    scores = model.compute_lexical_matching_scores(query, passages)  # 或 compute_similarity 直接计算
    # 注意：bge-m3 的 compute_similarity 返回 [query_len, passage_len]
    # 这里简化使用 sentence_transformers 的 CrossEncoder
    from sentence_transformers import CrossEncoder
    ce_model = CrossEncoder('BAAI/bge-reranker-v2-m3', max_length=512)
    
    pairs = [[query, p] for p in passages]
    ce_scores = ce_model.predict(pairs)
    
    sorted_idx = np.argsort(ce_scores)[::-1][:top_n]
    return [passages[i] for i in sorted_idx], [ce_scores[i] for i in sorted_idx]

# compressors/llm_lingua.py
from llmlingua import PromptCompressor
compressor = PromptCompressor()

def compress_context(context: str, ratio: float = 0.3):
    compressed = compressor.compress_prompt(
        context,
        rate=ratio,
        force_tokens=['\n', '。', '！', '？']  # 强制保留标点，确保语义完整
    )
    return compressed['compressed_prompt']

# tools/rag_tools.py
from langchain_core.tools import tool

@tool
def retrieve_knowledge(query: str) -> str:
    """检索内部知识库文档，返回最相关的文本片段"""
    results = hybrid_search(query, tenant_id="default", top_k=5)
    return "\n\n".join([r['text'] for r in results])

@tool
def calculate_expression(expression: str) -> float:
    """计算数学表达式，如 'SUM(100, 200)' 或 'AVG([1,2,3])'"""
    try:
        return eval(expression)  # 生产需使用 ast.literal_eval 或沙箱
    except:
        return 0.0

# graph/rag_agent.py
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

# 使用本地 vLLM 接口
llm = ChatOpenAI(
    base_url="http://localhost:8000/v1",
    api_key="EMPTY",
    model="deepseek-14b",
    temperature=0.1
)

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    context: str
    iterations: int

def query_rewriter(state: AgentState):
    """节点1：改写用户问题，生成3个假设性问句（HyDE）"""
    messages = state['messages']
    last_question = messages[-1].content
    prompt = f"请将以下问题改写为3个不同视角的相似问句，用于检索：{last_question}"
    response = llm.invoke(prompt)
    return {"messages": [response], "iterations": state.get('iterations', 0) + 1}

def retriever_node(state: AgentState):
    """节点2：执行检索并注入上下文"""
    query = state['messages'][-1].content
    docs = hybrid_search(query, "default", top_k=5)
    context = "\n\n---\n\n".join([d['text'] for d in docs])
    return {"context": context}

def generator_node(state: AgentState):
    """节点3：基于上下文生成最终答案（带引用）"""
    context = state.get('context', '')
    user_question = state['messages'][0].content
    prompt = f"""
    你是一个企业知识库助手。请基于以下上下文回答用户问题。
    如果上下文不足以回答问题，请明确告知用户“知识库中未找到相关信息”。
    请按 [1][2] 格式标注引用来源。

    上下文：
    {context}

    用户问题：{user_question}
    回答：
    """
    response = llm.invoke(prompt)
    return {"messages": [response]}

# 构建图
graph = StateGraph(AgentState)
graph.add_node("rewrite", query_rewriter)
graph.add_node("retrieve", retriever_node)
graph.add_node("generate", generator_node)

graph.set_entry_point("rewrite")
graph.add_edge("rewrite", "retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)

app = graph.compile()

# cdc/update_engine.py
import redis
import json
from pymilvus import Collection

r = redis.Redis(host='localhost', port=6379, decode_responses=True)
milvus_collection = Collection("rag_knowledge")

def index_consumer():
    while True:
        _, message = r.blpop("knowledge_updates", timeout=5)
        if not message:
            continue
        data = json.loads(message)
        if data['action'] == 'insert':
            # 生成 embedding 并插入
            emb = embedder.encode(data['text'], normalize_embeddings=True).tolist()
            milvus_collection.insert([{
                "embedding": emb,
                "text": data['text'],
                "doc_name": data['doc_name'],
                "tenant_id": data['tenant_id'],
                "metadata": data['metadata']
            }])
            milvus_collection.flush()
        elif data['action'] == 'delete':
            # 按 ID 删除
            milvus_collection.delete(f"id in {data['ids']}")

# 动态调整 nprobe (在搜索时)
search_params = {"metric_type": "IP", "params": {"nprobe": 32}}

# docker-compose-full.yml
version: '3.8'
services:
  etcd: ...  # 同前
  minio: ...
  milvus:
    image: milvusdb/milvus:v2.4.0
    # ...
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.10.0
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - "ES_JAVA_OPTS=-Xms4g -Xmx4g"
    ports:
      - "9200:9200"
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
  llm-server:
    build:
      context: .
      dockerfile: Dockerfile.vllm
    ports:
      - "8000:8000"
    volumes:
      - /models/deepseek-14b:/models
    environment:
      - CUDA_VISIBLE_DEVICES=0
    command: python -m vllm.entrypoints.openai.api_server --model /models --port 8000
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
  rag-api:
    build:
      context: ./app
      dockerfile: Dockerfile
    ports:
      - "8080:8080"
    depends_on:
      - milvus
      - elasticsearch
      - redis
      - llm-server
    environment:
      - LLM_BASE_URL=http://llm-server:8000/v1
      - MILVUS_HOST=milvus
      - ES_HOST=elasticsearch
      - REDIS_HOST=redis

# app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from graph.rag_agent import app as rag_app

api = FastAPI()

class QueryRequest(BaseModel):
    question: str
    tenant_id: str
    history: list[dict] = []

@api.post("/v1/rag/chat")
async def chat(req: QueryRequest):
    # 调用 LangGraph 获取结果
    result = rag_app.invoke({
        "messages": [{"role": "user", "content": req.question}],
        "context": "",
        "iterations": 0
    })
    return {"answer": result['messages'][-1].content}

# evaluate.py
from ragas.metrics import faithfulness, answer_relevancy, context_relevancy, context_recall
from ragas.llms import LangchainLLM
from datasets import Dataset
import pandas as pd

# 准备测试集 (question, answer, contexts)
test_data = {
    "question": ["什么是RAG？", "怎么部署DeepSeek？"],
    "answer": ["...", "..."],
    "contexts": [["doc1..."], ["doc2..."]]
}
dataset = Dataset.from_pandas(pd.DataFrame(test_data))

# 使用 DeepSeek 作为评估 Judge（自我纠偏）
evaluator_llm = LangchainLLM(llm=llm)
result = dataset.evaluate(
    metrics=[faithfulness, answer_relevancy, context_relevancy, context_recall],
    llm=evaluator_llm
)

print(result)  # 输出分数 0-1，>0.85 可上线