摘要本文手把手带你用 Python、Milvus 向量数据库与 DeepSeek V4 API从零搭建一套可落地的企业知识库问答系统。涵盖文档解析、Embedding、向量检索、混合重排、Prompt 设计全链路附完整可运行代码。关键词RAG、Milvus、DeepSeek、向量数据库、知识库问答、Python AI开发。一、RAG 核心原理检索增强为什么能减少幻觉大模型的最大软肋之一是知识截止与幻觉问题——它只知道训练数据里的内容遇到企业私有文档、最新政策、专业手册就容易胡说八道。RAGRetrieval-Augmented Generation检索增强生成的思路非常直接用户提问 ↓ 从知识库里检索最相关的文档片段 ↓ 把检索到的内容 用户问题一起塞给大模型 ↓ 大模型基于真实资料作答不是凭空想象为什么能减少幻觉因为模型的回答有了锚点——它只需要在给定材料里找答案而不是凭记忆生成。就像开卷考试 vs 闭卷考试开卷的准确率天然更高。RAG 与微调的对比对比维度RAG微调Fine-tuning知识更新实时改文档即生效需要重新训练数据量要求无要求通常需要几百条以上成本低只需向量检索高GPU 训练费用适合场景知识查询、文档问答风格迁移、特定任务结论企业知识库问答RAG 是首选方案。二、整体架构设计本系统分为离线索引和在线检索两个阶段【离线阶段】 原始文档PDF/Word/网页 → 文本解析与分块Chunking → Embedding 向量化text-embedding-3-small → 存入 Milvus 向量库 【在线阶段】 用户提问 → 问题向量化 → Milvus 相似度检索Top-K → Reranker 重排序 → 拼接 Prompt → DeepSeek 生成答案 → 返回答案 来源引用技术栈选型向量数据库Milvus 2.4开源、性能强、支持混合检索Embedding 模型BAAI/bge-m3中英文双语本地运行免费大模型DeepSeek V4 API性价比极高1元/百万 Token重排序BAAI/bge-reranker-v2-m3文档解析pypdf、python-docx、unstructured三、环境搭建3.1 安装依赖pipinstallpymilvus2.4.0\sentence-transformers\openai\pypdf\python-docx\unstructured\FlagEmbedding\tqdm\python-dotenv3.2 用 Docker 启动 Milvus# 下载 docker-compose 配置curl-sfLhttps://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh-ostandalone_embed.sh# 启动 Milvus standalonebashstandalone_embed.sh start# 验证启动dockerps|grepmilvusMilvus 默认端口19530gRPC、9091HTTP。3.3 配置环境变量# .env 文件DEEPSEEK_API_KEYsk-your-key-hereMILVUS_HOSTlocalhostMILVUS_PORT19530四、文档处理实战解析与分块策略4.1 文档解析# document_parser.pyimportosfrompathlibimportPathfrompypdfimportPdfReaderfromdocximportDocumentdefparse_pdf(file_path:str)-str:解析 PDF提取纯文本readerPdfReader(file_path)textforpageinreader.pages:textpage.extract_text()orreturntextdefparse_docx(file_path:str)-str:解析 Word 文档docDocument(file_path)return\n.join([para.textforparaindoc.paragraphsifpara.text.strip()])defparse_document(file_path:str)-str:根据扩展名自动选择解析器extPath(file_path).suffix.lower()ifext.pdf:returnparse_pdf(file_path)elifextin(.docx,.doc):returnparse_docx(file_path)elifextin(.txt,.md):withopen(file_path,r,encodingutf-8)asf:returnf.read()else:raiseValueError(f不支持的文件类型:{ext})4.2 文本分块Chunking策略分块是 RAG 质量的关键。块太大检索噪声多块太小上下文丢失。# chunker.pyfromtypingimportListdefchunk_by_sliding_window(text:str,chunk_size:int512,overlap:int64)-List[dict]: 滑动窗口分块 - chunk_size: 每块字符数中文约 256-512 字 - overlap: 相邻块重叠字符数保留上下文连续性 chunks[]start0chunk_id0whilestartlen(text):endstartchunk_size chunk_texttext[start:end]# 避免在词语中间截断向前找最近的句号/换行ifendlen(text):forsepin[。,\n,.,,]:last_sepchunk_text.rfind(sep)iflast_sepchunk_size*0.6:# 截断点在后 40% 区域chunk_textchunk_text[:last_sep1]breakchunks.append({chunk_id:chunk_id,text:chunk_text.strip(),start:start,end:startlen(chunk_text)})startlen(chunk_text)-overlap chunk_id1return[cforcinchunksiflen(c[text])20]# 过滤极短片段分块策略对比建议场景chunk_sizeoverlap说明技术文档/手册51264段落完整信息密度高法律/合同文本25648条款短需精准匹配新闻/报告76896篇幅长保留更多上下文五、Milvus 向量库搭建与索引选型5.1 创建 Collection# milvus_store.pyfrompymilvusimport(connections,Collection,CollectionSchema,FieldSchema,DataType,utility)COLLECTION_NAMEenterprise_kbEMBEDDING_DIM1024# bge-m3 输出维度defconnect_milvus(host:strlocalhost,port:str19530):connections.connect(default,hosthost,portport)print(f✅ Milvus 连接成功:{host}:{port})defcreate_collection()-Collection:创建知识库 Collectionifutility.has_collection(COLLECTION_NAME):print(fCollection{COLLECTION_NAME}已存在直接加载)returnCollection(COLLECTION_NAME)# 定义字段fields[FieldSchema(nameid,dtypeDataType.INT64,is_primaryTrue,auto_idTrue),FieldSchema(namedoc_name,dtypeDataType.VARCHAR,max_length256),FieldSchema(namechunk_id,dtypeDataType.INT64),FieldSchema(nametext,dtypeDataType.VARCHAR,max_length4096),FieldSchema(nameembedding,dtypeDataType.FLOAT_VECTOR,dimEMBEDDING_DIM),]schemaCollectionSchema(fields,description企业知识库)collectionCollection(COLLECTION_NAME,schema)# 创建 HNSW 索引检索速度快适合中小规模index_params{metric_type:COSINE,index_type:HNSW,params:{M:16,efConstruction:200}}collection.create_index(embedding,index_params)print(f✅ Collection{COLLECTION_NAME}创建完成)returncollection索引选型说明索引类型适用规模精度速度内存占用HNSW 100万条高快较高IVF_FLAT100万-1000万高中中IVF_SQ8 1000万中快低企业知识库通常在 10 万条以内推荐 HNSW。5.2 Embedding 向量化与入库# embedder.pyfromFlagEmbeddingimportBGEM3FlagModelfrompymilvusimportCollectionfromtqdmimporttqdmimportnumpyasnp# 加载 BGE-M3 模型首次会自动下载embed_modelBGEM3FlagModel(BAAI/bge-m3,use_fp16True)defembed_texts(texts:list[str])-np.ndarray:批量 Embedding返回归一化向量resultembed_model.encode(texts,batch_size32,max_length512,return_denseTrue,)returnresult[dense_vecs]definsert_chunks(collection:Collection,doc_name:str,chunks:list[dict]):将文档块批量入库batch_size64totallen(chunks)foriintqdm(range(0,total,batch_size),descf入库{doc_name}):batchchunks[i:ibatch_size]texts[c[text]forcinbatch]embeddingsembed_texts(texts)data[[doc_name]*len(batch),# doc_name[c[chunk_id]forcinbatch],# chunk_idtexts,# textembeddings.tolist(),# embedding]collection.insert(data)collection.flush()print(f✅{doc_name}入库完成共{total}个 chunk)六、DeepSeek API 接入与 Prompt 设计6.1 检索函数# retriever.pyfrompymilvusimportCollectionfromembedderimportembed_textsdefsearch_similar(collection:Collection,query:str,top_k:int10)-list[dict]:向量相似度检索collection.load()query_vecembed_texts([query])[0].tolist()resultscollection.search(data[query_vec],anns_fieldembedding,param{metric_type:COSINE,params:{ef:128}},limittop_k,output_fields[doc_name,chunk_id,text])hits[]forhitinresults[0]:hits.append({text:hit.entity.get(text),doc_name:hit.entity.get(doc_name),score:hit.score,})returnhits6.2 Reranker 重排序提升召回精度# reranker.pyfromFlagEmbeddingimportFlagReranker rerankerFlagReranker(BAAI/bge-reranker-v2-m3,use_fp16True)defrerank(query:str,hits:list[dict],top_k:int5)-list[dict]:对检索结果重排序过滤噪声pairs[[query,h[text]]forhinhits]scoresreranker.compute_score(pairs,normalizeTrue)forhit,scoreinzip(hits,scores):hit[rerank_score]score rankedsorted(hits,keylambdax:x[rerank_score],reverseTrue)returnranked[:top_k]6.3 DeepSeek 生成答案# generator.pyimportosfromopenaiimportOpenAI# DeepSeek 兼容 OpenAI SDKfromdotenvimportload_dotenv load_dotenv()clientOpenAI(api_keyos.getenv(DEEPSEEK_API_KEY),base_urlhttps://api.deepseek.com)SYSTEM_PROMPT你是一个企业知识库助手。 请严格基于下面提供的【参考资料】回答问题。 - 如果资料中没有相关内容请明确说根据现有资料暂无相关信息不要凭空编造 - 回答要简洁、准确关键信息可以用列表呈现 - 在回答末尾注明信息来源文档名称defgenerate_answer(query:str,context_chunks:list[dict])-str:拼接上下文调用 DeepSeek 生成答案context\n\n.join([f【来源{c[doc_name]}】\n{c[text]}forcincontext_chunks])user_messagef【参考资料】{context}【用户问题】{query}responseclient.chat.completions.create(modeldeepseek-chat,messages[{role:system,content:SYSTEM_PROMPT},{role:user,content:user_message}],temperature0.1,# 降低随机性确保答案稳定max_tokens1024,streamFalse)returnresponse.choices[0].message.content七、整合完整问答流程# main.pyimportosfrompymilvusimportconnections,Collectionfromchunkerimportchunk_by_sliding_windowfromdocument_parserimportparse_documentfrommilvus_storeimportconnect_milvus,create_collectionfromembedderimportinsert_chunksfromretrieverimportsearch_similarfromrerankerimportrerankfromgeneratorimportgenerate_answerdefbuild_index(doc_paths:list[str]):离线索引构建connect_milvus()collectioncreate_collection()forpathindoc_paths:print(f\n 处理文档:{path})textparse_document(path)chunkschunk_by_sliding_window(text,chunk_size512,overlap64)doc_nameos.path.basename(path)insert_chunks(collection,doc_name,chunks)print(\n✅ 所有文档索引构建完成)defask(query:str)-str:在线问答connect_milvus()collectionCollection(enterprise_kb)# 1. 向量检索 Top-10hitssearch_similar(collection,query,top_k10)# 2. Reranker 重排取 Top-5rankedrerank(query,hits,top_k5)# 3. DeepSeek 生成答案answergenerate_answer(query,ranked)returnanswerif__name____main__:# 构建索引首次运行# build_index([docs/产品手册.pdf, docs/常见问题.docx])# 问答测试q产品的退款政策是什么print(f\n❓ 问题{q})print(f\n 回答\n{ask(q)})八、召回率优化混合检索 重排序单纯向量检索对专业术语、型号编码等精确词效果差。推荐混合检索稀疏检索BM25 稠密检索向量两路结果融合后再 Rerank。# Milvus 2.4 支持混合检索frompymilvusimportAnnSearchRequest,RRFRanker,WeightedRankerdefhybrid_search(collection,query,query_vec,top_k10):稠密 稀疏混合检索dense_reqAnnSearchRequest(data[query_vec],anns_fieldembedding,param{metric_type:COSINE,params:{ef:128}},limittop_k)# RRF 融合策略倒数排名融合resultscollection.hybrid_search([dense_req],rerankRRFRanker(k60),limittop_k,output_fields[doc_name,text])returnresults九、效果评测方法与上线注意事项9.1 快速评测指标# 评测脚本示例test_cases[{question:退款流程是什么,expected_keywords:[7天,原路退回]},{question:如何申请发票,expected_keywords:[税务,开具]},]hit_count0forcaseintest_cases:answerask(case[question])ifall(kwinanswerforkwincase[expected_keywords]):hit_count1print(f关键词命中率:{hit_count/len(test_cases):.1%})9.2 上线 Checklist文档分块后人工抽查 10% 的块确认内容完整限制 Milvus 并发查询数防止内存 OOMDeepSeek API 调用加超时timeout30s和重试最多3次敏感信息文档在入库前做脱敏处理建立用户反馈机制踩/赞持续迭代 Prompt十、总结本文完整实现了一套企业 RAG 知识库问答系统核心要点回顾分块策略直接影响召回质量推荐 512 字 64 字重叠的滑动窗口BGE-M3是目前中英文 Embedding 的最优免费选择Reranker 重排序可将准确率提升 15-30%强烈建议加入DeepSeek的 System Prompt 要明确约束不得编造是减少幻觉的最后一道防线生产环境务必加入监控、限流、脱敏三件套如果本文对你有帮助点赞收藏是对我最大的支持有问题欢迎在评论区交流。