DeepSeek RAG 知识库搭建教程

Retrieval（检索）	将用户问题转换为向量，在向量数据库中找到最相关的文档片段。这是 RAG 的"查找"环节。
Augmented（增强）	将检索到的文档片段作为上下文，拼接到 Prompt 中。这是 RAG 的"注入知识"环节。
Generation（生成）	DeepSeek 模型基于增强后的 Prompt 生成答案。这是 RAG 的"输出"环节。

离线阶段（知识库构建）：

• 文档加载：读取 PDF、TXT、Markdown、网页等各类文档
• 文本分割：将长文档切分为适当大小的文本块（Chunk）
• 向量嵌入：将文本块转换为向量表示
• 向量存储：将向量存入 ChromaDB 等向量数据库

在线阶段（问答）：

• 用户提问：接收用户输入的问题
• 向量检索：将问题转为向量，在数据库中查找最相似的文本块
• 上下文拼接：将检索结果作为上下文，构建增强 Prompt
• 模型生成：DeepSeek 基于增强 Prompt 生成答案，并附上来源引用

对比维度	RAG	微调（Fine-tuning）
知识更新	实时更新，增删文档即时生效	需要重新训练，周期长
成本	低，只需向量数据库	高，需要 GPU 训练资源
可解释性	可溯源到具体文档	黑盒，难以追溯知识来源
适用场景	知识库问答、文档检索、客服系统	风格模仿、特定领域术语、指令遵循

# 创建虚拟环境（推荐） python -m venv rag-env # Windows 激活 rag-env\Scripts\activate # macOS/Linux 激活 # source rag-env/bin/activate # 安装核心依赖 pip install langchain langchain-community chromadb sentence-transformers openai # 安装文档加载器（按需选择） pip install pypdf # PDF 加载 pip install unstructured # 通用文档加载（PDF/Word/PPT/HTML） pip install beautifulsoup4 # 网页解析 pip install lxml # XML/HTML 解析加速 # 验证安装 python -c "import langchain; import chromadb; print('RAG 环境就绪')"

# Windows PowerShell $env:DEEPSEEK_API_KEY = "sk-your-api-key-here" # macOS/Linux # export DEEPSEEK_API_KEY="sk-your-api-key-here" # 或在 Python 代码中直接配置 import os os.environ["DEEPSEEK_API_KEY"] = "sk-your-api-key-here"

提示

如果你使用 Ollama 本地部署的 DeepSeek 模型，无需 API Key，只需将 API Base URL 指向 http://localhost:11434/v1。详见 DeepSeek 本地部署教程。

from langchain_community.document_loaders import PyPDFLoader # 加载单个 PDF loader = PyPDFLoader("docs/企业年报2025.pdf") pages = loader.load() # 查看加载结果 print(f"加载了 {len(pages)} 页") print(f"第一页前 200 字: {pages[0].page_content[:200]}") # 批量加载 PDF 文件夹 from langchain_community.document_loaders import DirectoryLoader loader = DirectoryLoader( "docs/", glob="**/*.pdf", loader_cls=PyPDFLoader, ) documents = loader.load() print(f"共加载 {len(documents)} 个文档")

from langchain_community.document_loaders import TextLoader, UnstructuredMarkdownLoader # 加载纯文本文件 txt_loader = TextLoader("docs/产品手册.txt", encoding="utf-8") txt_docs = txt_loader.load() # 加载 Markdown 文件 md_loader = UnstructuredMarkdownLoader("docs/技术文档.md") md_docs = md_loader.load()

from langchain_community.document_loaders import WebBaseLoader # 加载单个网页 web_loader = WebBaseLoader("https://platform.deepseek.com/api-docs") web_docs = web_loader.load() # 批量加载多个网页 urls = [ "https://example.com/doc1", "https://example.com/doc2", ] web_loader = WebBaseLoader(urls) web_docs = web_loader.load()

from langchain_community.document_loaders import UnstructuredFileLoader # 自动识别文件类型并解析 loader = UnstructuredFileLoader("docs/产品方案.docx") docs = loader.load() # 批量加载多种格式 loader = DirectoryLoader( "docs/", glob="**/*.*", loader_cls=UnstructuredFileLoader, ) all_docs = loader.load() print(f"加载了 {len(all_docs)} 个文档片段")

参数	说明	推荐值
chunk_size	每个文本块的最大字符数	500-1000
chunk_overlap	相邻文本块之间的重叠字符数	50-200
separators	分割优先级：先按段落，再按句子，最后按字符	默认即可

from langchain.text_splitter import RecursiveCharacterTextSplitter # 创建文本分割器 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个 chunk 最多 500 字符 chunk_overlap=100, # 相邻 chunk 重叠 100 字符 length_function=len, separators=["\n\n", "\n", "。", "！", "？", ".", " ", ""], ) # 分割文档 chunks = text_splitter.split_documents(documents) print(f"原始文档数: {len(documents)}") print(f"分割后 chunk 数: {len(chunks)}") # 查看分割效果 for i, chunk in enumerate(chunks[:3]): print(f"--- Chunk {i+1} ({len(chunk.page_content)} 字符) ---") print(chunk.page_content[:200]) print()

• chunk_size 按内容类型调整：技术文档用 500-800，长篇文章用 800-1200，FAQ 类用 300-500
• chunk_overlap 不能省略：重叠可以避免关键信息被切分在边界处，建议设为 chunk_size 的 10%-20%
• 保留元数据：分割时保留文档来源、页码等元数据，方便后续溯源
• 中文分割：在 separators 中加入中文标点（。！？），确保按语义边界分割
• 先加载再分割：先加载完整文档，再统一分割，避免重复加载和分割

模型名称	维度	中文支持	模型大小	推荐场景
BAAI/bge-large-zh-v1.5	1024	极佳	1.3GB	中文文档首选
BAAI/bge-small-zh-v1.5	512	极佳	96MB	轻量级中文
sentence-transformers/all-MiniLM-L6-v2	384	一般	80MB	英文为主
shibing624/text2vec-base-chinese	768	极佳	400MB	中文语义匹配

from langchain_community.embeddings import HuggingFaceEmbeddings # 使用 BGE 中文嵌入模型（推荐） embedding_model = HuggingFaceEmbeddings( model_name="BAAI/bge-small-zh-v1.5", model_kwargs={"device": "cpu"}, # 有 GPU 改为 "cuda" encode_kwargs={"normalize_embeddings": True}, # 归一化，提高检索精度 ) # 测试嵌入效果 test_text = "DeepSeek 是一个强大的开源大语言模型" embedding = embedding_model.embed_query(test_text) print(f"向量维度: {len(embedding)}") print(f"向量前 5 个值: {embedding[:5]}") # 也可以使用 DeepSeek API 进行嵌入（如果官方支持） # 或使用 OpenAI 兼容的嵌入 API from langchain_openai import OpenAIEmbeddings api_embedding = OpenAIEmbeddings( model="text-embedding-3-small", # 或使用其他兼容 API base_url="https://api.deepseek.com/v1", api_key="sk-your-api-key-here", )

嵌入模型选择建议

中文文档强烈推荐 BGE 系列模型，在中文语义理解上明显优于通用英文模型。本地部署用 HuggingFaceEmbeddings 加载，无需 API 费用。如果追求极致性能和稳定性，可以使用云端嵌入 API。

from langchain_community.vectorstores import Chroma # 将分割后的文档存入 ChromaDB vectorstore = Chroma.from_documents( documents=chunks, # 分割后的文本块 embedding=embedding_model, # 嵌入模型 persist_directory="./chroma_db", # 持久化存储路径 collection_name="deepseek_knowledge", # 集合名称 ) # 数据已自动持久化，下次使用时直接加载 print(f"向量数据库中已有 {vectorstore._collection.count()} 条记录")

# 从磁盘加载已有向量数据库 vectorstore = Chroma( persist_directory="./chroma_db", embedding_function=embedding_model, collection_name="deepseek_knowledge", ) print(f"加载了已有数据库，共 {vectorstore._collection.count()} 条记录")

# 基础相似度搜索 query = "DeepSeek 模型的 API 如何调用？" results = vectorstore.similarity_search(query, k=4) for i, doc in enumerate(results): print(f"--- 结果 {i+1} (来源: {doc.metadata.get('source', '未知')}) ---") print(doc.page_content[:300]) print() # 带相似度分数的搜索 results_with_scores = vectorstore.similarity_search_with_score(query, k=4) for doc, score in results_with_scores: print(f"相似度分数: {score:.4f} | 内容: {doc.page_content[:100]}...") # MMR 搜索（最大边际相关性）— 兼顾相关性和多样性 mmr_results = vectorstore.max_marginal_relevance_search( query, k=4, fetch_k=10, lambda_mult=0.7 )

• FAISS：Meta 开源的向量索引库，纯内存运算，检索速度极快，适合百万级数据
• Milvus：企业级分布式向量数据库，支持十亿级向量，适合大规模生产环境
• Weaviate：开源向量数据库，内置向量化和混合搜索，支持 GraphQL API
• Pinecone：商业向量数据库服务，免运维，适合不想管理基础设施的团队
• Qdrant：高性能 Rust 编写的向量数据库，支持过滤和分组检索

from langchain_openai import ChatOpenAI from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate # 初始化 DeepSeek 模型（通过 OpenAI 兼容接口） llm = ChatOpenAI( model="deepseek-chat", base_url="https://api.deepseek.com/v1", api_key="sk-your-api-key-here", temperature=0.3, # RAG 场景建议低温，保证准确性 max_tokens=2048, ) # 自定义 Prompt 模板 prompt_template = """你是一个基于知识库的专业问答助手。请根据以下文档内容回答用户问题。 要求： 1. 仅根据提供的文档内容回答，不要编造信息 2. 如果文档内容不足以回答问题，请明确说明"根据现有文档，无法回答该问题" 3. 回答时引用具体的文档来源 4. 使用中文回答，保持专业、清晰 文档内容： {context} 用户问题：{question} 回答：""" PROMPT = PromptTemplate( template=prompt_template, input_variables=["context", "question"], ) # 构建 RetrievalQA 链 qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", # 将所有检索结果填入 Prompt retriever=vectorstore.as_retriever( search_kwargs={"k": 4} # 检索 4 个最相关文档 ), chain_type_kwargs={"prompt": PROMPT}, return_source_documents=True, # 返回源文档 )

# 提问并获取答案 question = "DeepSeek API 的收费标准是什么？" result = qa_chain.invoke({"query": question}) print("问题：", question) print("答案：", result["result"]) print("\n参考来源：") for i, doc in enumerate(result["source_documents"]): source = doc.metadata.get("source", "未知") print(f" {i+1}. {source}")

类型	说明	适用场景
stuff	将所有检索结果一次性放入 Prompt	文档较少（k <= 4），最常用
map_reduce	先对每个文档单独总结，再合并总结	文档多且长，需要全局理解
refine	逐个文档迭代优化答案	需要高质量答案，不介意延迟
map_rerank	对每个文档打分，选最高分答案	需要精确匹配特定文档

def format_qa_response(result): """格式化 RAG 问答结果，包含源引用""" answer = result["result"] sources = result["source_documents"] # 提取唯一来源 unique_sources = [] seen = set() for doc in sources: source = doc.metadata.get("source", "未知来源") page = doc.metadata.get("page", None) source_key = f"{source}_{page}" if source_key not in seen: seen.add(source_key) unique_sources.append({ "source": source, "page": page, "preview": doc.page_content[:150] + "..." }) return { "answer": answer, "sources": unique_sources, "source_count": len(unique_sources), } # 使用示例 result = qa_chain.invoke({"query": "DeepSeek R1 模型有什么特点？"}) formatted = format_qa_response(result) print("=" * 60) print(formatted["answer"]) print("\n--- 参考来源 (", formatted["source_count"], "个) ---") for i, s in enumerate(formatted["sources"]): page_info = f"第{s['page']}页" if s['page'] else "" print(f"{i+1}. {s['source']} {page_info}") print(f" {s['preview']}")

# 带编号引用的 Prompt 模板 prompt_with_citation = """你是一个基于知识库的专业问答助手。请根据以下文档内容回答用户问题。 要求： 1. 仅根据提供的文档内容回答，不要编造信息 2. 在回答中引用具体来源，格式为 [来源X] 3. 如果文档内容不足以回答问题，请明确说明 4. 使用中文回答，保持专业、清晰 文档内容： {context} 用户问题：{question} 请回答（引用的来源用 [来源1]、[来源2] 等标记）："""

# pip install fastapi uvicorn from fastapi import FastAPI from pydantic import BaseModel app = FastAPI(title="DeepSeek RAG API") class Question(BaseModel): text: str class Answer(BaseModel): question: str answer: str sources: list @app.post("/ask", response_model=Answer) async def ask_question(q: Question): result = qa_chain.invoke({"query": q.text}) formatted = format_qa_response(result) return Answer( question=q.text, answer=formatted["answer"], sources=[s["source"] for s in formatted["sources"]], ) # 启动服务: uvicorn main:app --reload --port 8000

from langchain.retrievers.multi_query import MultiQueryRetriever # 创建多查询检索器 multi_query_retriever = MultiQueryRetriever.from_llm( retriever=vectorstore.as_retriever(), llm=llm, ) # 用户问题: "模型怎么部署？" # 自动生成多个查询: # 1. "DeepSeek 模型本地部署方法" # 2. "Ollama 部署 DeepSeek 教程" # 3. "DeepSeek 模型服务器部署步骤" unique_docs = multi_query_retriever.invoke("模型怎么部署？") print(f"多查询检索到 {len(unique_docs)} 个相关文档")

from langchain.retrievers import ParentDocumentRetriever from langchain.storage import InMemoryStore # 创建两个分割器：子文档用小块，父文档用大块 child_splitter = RecursiveCharacterTextSplitter(chunk_size=200) parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000) # 父文档检索器 parent_retriever = ParentDocumentRetriever( vectorstore=Chroma( collection_name="parent_docs", embedding_function=embedding_model, ), docstore=InMemoryStore(), child_splitter=child_splitter, parent_splitter=parent_splitter, ) parent_retriever.add_documents(documents) # 检索时用子文档匹配，但返回完整的父文档 results = parent_retriever.invoke("DeepSeek API 定价")

from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor # 创建压缩器：用 LLM 提取文档中与问题相关的部分 compressor = LLMChainExtractor.from_llm(llm) compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=vectorstore.as_retriever(search_kwargs={"k": 6}), ) # 检索并压缩文档 compressed_docs = compression_retriever.invoke( "DeepSeek V3 和 R1 的区别是什么？" ) for doc in compressed_docs: print(f"压缩后内容长度: {len(doc.page_content)} 字符")

# pip install sentence-transformers from sentence_transformers import CrossEncoder # 加载重排序模型 reranker = CrossEncoder("BAAI/bge-reranker-large") # 第一步：向量检索获取候选文档（多取一些） candidate_docs = vectorstore.similarity_search(query, k=20) # 第二步：用重排序模型重新打分 pairs = [[query, doc.page_content] for doc in candidate_docs] scores = reranker.predict(pairs) # 第三步：按分数排序，取前 K 个 sorted_docs = sorted( zip(candidate_docs, scores), key=lambda x: x[1], reverse=True ) top_docs = [doc for doc, score in sorted_docs[:4]] print(f"重排序后保留了 {len(top_docs)} 个最相关文档")

技巧	解决的问题	成本
多查询检索	用户问题表述不精确导致检索失败	额外 LLM 调用
父文档检索	小 chunk 丢失上下文	存储翻倍
上下文压缩	检索结果包含大量无关内容	额外 LLM 调用
重排序	向量检索不够精准	额外模型推理

"""DeepSeek RAG 知识库问答系统 — 完整实现""" import os from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Chroma from langchain_openai import ChatOpenAI from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate # ========== 配置 ========== DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "sk-your-api-key") DOCS_DIR = "./documents" CHROMA_DIR = "./chroma_db" EMBEDDING_MODEL = "BAAI/bge-small-zh-v1.5" CHUNK_SIZE = 500 CHUNK_OVERLAP = 100 RETRIEVAL_K = 4 # ========== 1. 加载文档 ========== def load_documents(docs_dir): """加载 docs_dir 下所有 PDF 和 TXT 文件""" loaders = { "**/*.pdf": PyPDFLoader, "**/*.txt": TextLoader, } all_docs = [] for pattern, loader_cls in loaders.items(): loader = DirectoryLoader(docs_dir, glob=pattern, loader_cls=loader_cls) try: docs = loader.load() all_docs.extend(docs) print(f" [OK] {pattern}: {len(docs)} 个文档") except Exception as e: print(f" [SKIP] {pattern}: {e}") return all_docs # ========== 2. 分割文档 ========== def split_documents(documents): splitter = RecursiveCharacterTextSplitter( chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP, separators=["\n\n", "\n", "。", "！", "？", ".", " ", ""], ) chunks = splitter.split_documents(documents) print(f" 原始文档: {len(documents)}, 分割后: {len(chunks)} 个 chunk") return chunks # ========== 3. 创建向量数据库 ========== def create_vectorstore(chunks, force_rebuild=False): embedding = HuggingFaceEmbeddings( model_name=EMBEDDING_MODEL, model_kwargs={"device": "cpu"}, encode_kwargs={"normalize_embeddings": True}, ) if os.path.exists(CHROMA_DIR) and not force_rebuild: print(" 加载已有向量数据库...") vectorstore = Chroma( persist_directory=CHROMA_DIR, embedding_function=embedding, ) else: print(" 创建新的向量数据库...") vectorstore = Chroma.from_documents( documents=chunks, embedding=embedding, persist_directory=CHROMA_DIR, ) print(f" 向量数据库: {vectorstore._collection.count()} 条记录") return vectorstore # ========== 4. 构建问答链 ========== def create_qa_chain(vectorstore): llm = ChatOpenAI( model="deepseek-chat", base_url="https://api.deepseek.com/v1", api_key=DEEPSEEK_API_KEY, temperature=0.3, max_tokens=2048, ) prompt = PromptTemplate( template="""你是一个基于知识库的专业问答助手。请根据以下文档内容回答用户问题。 要求： 1. 仅根据提供的文档内容回答，不要编造信息 2. 如果文档内容不足以回答问题，请明确说明 3. 回答时引用文档来源，格式为 [来源X] 4. 使用中文回答，保持专业、清晰 文档内容： {context} 用户问题：{question} 回答：""", input_variables=["context", "question"], ) qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=vectorstore.as_retriever( search_kwargs={"k": RETRIEVAL_K} ), chain_type_kwargs={"prompt": prompt}, return_source_documents=True, ) return qa_chain # ========== 5. 交互式问答 ========== def interactive_qa(qa_chain): print("\n" + "=" * 60) print(" DeepSeek RAG 知识库问答系统") print(" 输入 'quit' 退出, 'sources' 显示上次来源") print("=" * 60) last_result = None while True: try: question = input("\n你的问题: ").strip() except (EOFError, KeyboardInterrupt): break if not question: continue if question.lower() == "quit": break if question.lower() == "sources" and last_result: print("\n--- 上次回答的参考来源 ---") for i, doc in enumerate(last_result["source_documents"]): source = doc.metadata.get("source", "未知") print(f" [来源{i+1}] {source}") continue print(" 思考中...", end="\r") result = qa_chain.invoke({"query": question}) last_result = result print("\n" + "-" * 60) print(result["result"]) print("-" * 60) # 显示来源摘要 sources = set() for doc in result["source_documents"]: sources.add(doc.metadata.get("source", "未知")) print(f"参考来源 ({len(sources)} 个): {', '.join(list(sources)[:3])}") print("\n 再见！") # ========== 主程序 ========== if __name__ == "__main__": print("[1/4] 加载文档...") documents = load_documents(DOCS_DIR) if not documents: print(" 错误: 未找到任何文档，请在 ./documents 目录下放置 PDF/TXT 文件") exit(1) print("\n[2/4] 分割文档...") chunks = split_documents(documents) print("\n[3/4] 创建向量数据库...") vectorstore = create_vectorstore(chunks) print("\n[4/4] 构建问答链...") qa_chain = create_qa_chain(vectorstore) interactive_qa(qa_chain)

# 1. 创建文档目录 mkdir documents # 2. 放入你的 PDF/TXT 文件 # 将企业文档、技术手册等放入 documents/ 目录 # 3. 设置 API Key # Windows: # $env:DEEPSEEK_API_KEY = "sk-your-api-key" # macOS/Linux: # export DEEPSEEK_API_KEY="sk-your-api-key" # 4. 运行 python deepseek_rag.py

RAG 环节	代码实现	Dify 对应功能
文档加载	PyPDFLoader / TextLoader	知识库 → 上传文档（支持 PDF/Word/TXT/网页）
文本分割	RecursiveCharacterTextSplitter	知识库 → 分段设置（自定义分段长度和重叠）
向量嵌入	HuggingFaceEmbeddings	模型供应商 → 嵌入模型配置
向量存储	ChromaDB	Dify 内置向量数据库（Qdrant/Weaviate/ChromaDB）
检索	vectorstore.similarity_search()	知识库 → 召回设置（TopK、分数阈值）
生成	ChatOpenAI + RetrievalQA	应用编排 → 添加 DeepSeek 模型 + 知识库节点

Dify 的优势

• 可视化操作，无需编写代码
• 内置向量数据库，自动管理索引
• 支持混合检索（向量检索 + 关键词检索）
• 内置引用溯源，自动标注答案来源
• 对话日志和分析，持续优化问答效果
• 一键发布为 API，方便集成到业务系统