基于 RAG 架构的 DeepSeek 大模型本地知识库构建

用户Query → Embedding → 向量检索 → 召回TopK → Prompt组装 → DeepSeek生成 → 返回答案

# 创建虚拟环境
conda create -n rag_deepseek python=3.10
conda activate rag_deepseek

# 核心库
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118
pip install langchain langchain-community chromadb sentence-transformers
pip install fastapi uvicorn pypdf python-docx markdown
pip install ollama  # 用于调用DeepSeek

# 安装 Ollama（Linux/macOS/WSL2）
curl -fsSL https://ollama.com/install.sh | sh

# 拉取 DeepSeek-R1 蒸馏版（14B 量化）
ollama pull deepseek-r1:14b

# 测试推理
ollama run deepseek-r1:14b "你好，请介绍一下RAG"

from langchain.document_loaders import PyPDFLoader, Docx2txtLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

def load_documents(file_paths):
    docs = []
    for path in file_paths:
        if path.endswith('.pdf'):
            loader = PyPDFLoader(path)
        elif path.endswith('.docx'):
            loader = Docx2txtLoader(path)
        else:
            loader = TextLoader(path, encoding='utf-8')
        docs.extend(loader.load())
    return docs

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,          # 每个块最大字符数
    chunk_overlap=100,       # 重叠字符
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],
    length_function=len,
)
chunks = text_splitter.split_documents(docs)

from sentence_transformers import SentenceTransformer

embedding_model = SentenceTransformer('BAAI/bge-large-zh-v1.5')
# 也可用 'shibing624/text2vec-large-chinese'

def get_embedding(text):
    return embedding_model.encode(text, normalize_embeddings=True).tolist()

import chromadb
from chromadb.config import Settings

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection(
    name="knowledge_base",
    metadata={"hnsw:space": "cosine"}  # 使用余弦距离
)

# 批量写入
ids = [f"chunk_{i}" for i in range(len(chunks))]
metadatas = [{"source": doc.metadata.get("source", "")} for doc in chunks]
documents = [chunk.page_content for chunk in chunks]
embeddings = [get_embedding(text) for text in documents]

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

def retrieve(query, top_k=5):
    query_embedding = get_embedding(query)
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k,
        include=["documents", "distances", "metadatas"]
    )
    # 返回文档列表及分数
    return results['documents'][0], results['distances'][0]

system_prompt = """你是一个专业的知识助手。请严格基于以下提供的参考信息回答用户问题。
如果参考信息不足以回答问题，请明确说明“根据已有知识无法回答”，不要编造。
参考信息：
{context}
"""

def build_prompt(query, context_docs):
    context = "\n\n".join([f"[文档{i+1}] {doc}" for i, doc in enumerate(context_docs)])
    return system_prompt.format(context=context) + f"\n用户问题：{query}\n回答："

import requests
import json

def generate_deepseek(prompt, model="deepseek-r1:14b", temperature=0.3):
    url = "http://localhost:11434/api/generate"
    payload = {
        "model": model,
        "prompt": prompt,
        "stream": False,
        "options": {"temperature": temperature, "top_p": 0.9}
    }
    response = requests.post(url, json=payload)
    if response.status_code == 200:
        return response.json()["response"]
    else:
        raise Exception(f"Ollama error: {response.text}")

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="DeepSeek RAG Knowledge Base")

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

class QueryResponse(BaseModel):
    answer: str
    retrieved_docs: list
    scores: list

@app.post("/ask", response_model=QueryResponse)
async def ask(request: QueryRequest):
    try:
        docs, scores = retrieve(request.question, request.top_k)
        prompt = build_prompt(request.question, docs)
        answer = generate_deepseek(prompt)
        return QueryResponse(answer=answer, retrieved_docs=docs, scores=scores)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# 启动：uvicorn main:app --host 0.0.0.0 --port 8000

rag_deepseek/
├── index.py         # 索引构建脚本
├── retrieval.py     # 检索模块
├── generation.py    # DeepSeek 调用
├── app.py           # FastAPI 服务
├── config.py        # 配置项
├── documents/       # 原始文档存放
└── chroma_db/       # 向量库持久化

EMBEDDING_MODEL = "BAAI/bge-large-zh-v1.5"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 100
COLLECTION_NAME = "knowledge_base"
OLLAMA_MODEL = "deepseek-r1:14b"
TOP_K = 5

curl -X POST "http://localhost:8000/ask" \
     -H "Content-Type: application/json" \
     -d '{"question": "什么是RAG？", "top_k": 3}'