本文基于一个完整的实战项目带你从零构建一个高精度的智能问答机器人。我们会使用Milvus 向量数据库存储知识库结合BM25 关键词检索和向量检索进行混合召回再通过多查询生成和FlashRank 重排序提升答案准确率。所有代码都有详细注释读完你就能理解现代 RAG 系统的核心设计思路。环境准备Python 版本建议使用Python 3.12如果下载速度慢可配置国内 PyPI 镜像。项目背景为什么要这么复杂一个简单的 RAG 流程是用户提问 → 向量数据库检索相似段落 → 拼接上下文 → 大模型生成答案。但在实际落地中你会遇到以下问题用户问法千奇百怪同一个意图“文石哪款 pad 尺寸最大”用户可能问“最大的电纸书是哪台”向量检索可能召回不同结果。向量相似 ≠ 语义相关向量模型可能把“大尺寸”和“大电池”混淆导致无关段落排在前面。纯向量检索会漏掉专有名词、型号等精确关键词比如“Tab13”这类词向量模型可能不认识。因此我们设计了如下流水线多查询生成让大模型根据原始问题生成 3 个不同角度的问法分别去检索扩大召回范围这个我们上一篇也专门聊过。混合召回每条问法既走Milvus 向量检索语义相似也走BM25 关键词检索字面匹配合并候选文档。重排序用专门的轻量模型对所有候选段落按与原始问题的真实相关性重新打分只保留最相关的前 2 条送入大模型这个上一篇也用过。最终大模型只看到少量高质量上下文答案自然更准确、更省钱。完整代码与逐行解析3.1 基础配置与镜像劫持import os import shutil import time os.environ[HF_ENDPOINT] https://hf-mirror.com from dotenv import load_dotenv from langchain_core.documents import Document from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_deepseek import ChatDeepSeek from langchain_huggingface import HuggingFaceEmbeddings from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough, RunnableLambda from langchain_core.output_parsers import StrOutputParser from pymilvus import MilvusClient, DataType from flashrank import Ranker, RerankRequest from langchain_community.retrievers import BM25Retriever import requests作用HF_ENDPOINT设置 HuggingFace 国内镜像避免下载模型超时。从.env文件加载DEEPSEEK_API_KEY。导入了 Milvus 客户端、FlashRank 重排序器、BM25 检索器等关键组件。3.2 拦截 HuggingFace 下载请求防止模型下载失败original_get requests.get def mirror_get(url, *args, **kwargs): if huggingface.co in url: url url.replace(https://huggingface.co, https://hf-mirror.com) return original_get(url, *args, **kwargs) requests.get mirror_get有些库如 FlashRank内部使用requests下载模型文件此段代码强制替换为国内镜像保证下载成功。3.3 清理 Milvus 数据库文件Windows 下常见占用问题DB_PATH milvus_demo.db COLLECTION_NAME docs def clean_milvus_db(db_path: str, retry_times3): # 先尝试关闭所有可能的连接 for _ in range(2): try: tmp_cli MilvusClient(db_path) tmp_cli.close() del tmp_cli except Exception: pass time.sleep(0.6) # 删除 LOCK 文件Windows 下常因进程未释放而残留 lock_path os.path.join(db_path, LOCK) if os.path.exists(lock_path): try: os.unlink(lock_path) except Exception: pass # 循环删除整个数据库目录 for i in range(retry_times): try: if os.path.exists(db_path): shutil.rmtree(db_path) print(f[清理成功] {db_path} 已删除) return except PermissionError as e: print(f[清理重试{i1}] 文件占用: {e}) time.sleep(1) raise Exception(自动清理失败执行这条命令手动清理Get-Process milvus-lite* -EA SilentlyContinue | Stop-Process -Force; Remove-Item milvus_demo.db -Recurse -Force)为什么需要这个函数Milvus Lite 在 Windows 下可能因为异常退出、多进程访问等原因导致数据库目录被占用重新运行时会出现 PermissionError。这个函数尝试优雅关闭连接、删除锁文件实在不行就抛出提示让用户手动清理。3.4 读取文档并切分成小块with open(knowledge.txt, r, encodingutf-8) as f: text f.read() documents [Document(page_contenttext)] text_splitter RecursiveCharacterTextSplitter( chunk_size300, # 每块最多300字符 chunk_overlap50, # 重叠50字符避免语义断裂 separators[\n\n, \n, 。, , , , , ] ) docs text_splitter.split_documents(documents) target_chunk_num len(docs) # 记录切片总数用于后续判断数据库是否需要重建切分策略块大小 300 字符适合大部分段落重叠 50 字符确保上下文连贯。分隔符顺序从大到小优先在段落、换行、句号处切分保证语义完整性。3.5 初始化向量化模型embeddings HuggingFaceEmbeddings( model_nameBAAI/bge-large-zh-v1.5, model_kwargs{device: cpu}, encode_kwargs{normalize_embeddings: True} ) dim len(embeddings.embed_query(test)) # 获取向量维度bge-large-zh-v1.5是中文语义向量模型的优选效果好、速度快。normalize_embeddingsTrue会将向量归一化使得余弦相似度计算更稳定。3.6 Milvus 初始化与数据写入3.6.1 判断是否需要重建数据库need_rebuild True try: check_client MilvusClient(DB_PATH) if check_client.has_collection(COLLECTION_NAME): stats check_client.get_collection_stats(COLLECTION_NAME) db_chunk_num stats.get(row_count, 0) if db_chunk_num target_chunk_num: need_rebuild False print(f✅ 向量库匹配复用现有数据库内切片:{db_chunk_num}) else: print(f 切片数量不一致执行重建库:{db_chunk_num} 本地:{target_chunk_num}) check_client.close() del check_client except Exception as e: print(f 向量库读取异常强制重建: {e}) try: check_client.close() except: pass del check_client time.sleep(0.5)这段代码避免了重复写入。如果本地知识库没有变化切片数量相同就直接复用已有的 Milvus 数据库加快启动速度。3.6.2 创建集合并写入向量client None if need_rebuild: client MilvusClient(DB_PATH) # 定义表结构 schema client.create_schema(auto_idFalse, enable_dynamic_fieldFalse) schema.add_field(field_nameid, datatypeDataType.INT64, is_primaryTrue) schema.add_field(field_namevector, datatypeDataType.FLOAT_VECTOR, dimdim) schema.add_field(field_nametext, datatypeDataType.VARCHAR, max_length2000) # 创建索引FLAT 索引余弦相似度 index_params client.prepare_index_params() index_params.add_index(field_namevector, index_typeFLAT, metric_typeCOSINE) client.create_collection(collection_nameCOLLECTION_NAME, schemaschema, index_paramsindex_params) # 批量写入向量 insert_data [] for idx, doc in enumerate(docs): vec embeddings.embed_query(doc.page_content) insert_data.append({id: idx, vector: vec, text: doc.page_content}) client.insert(collection_nameCOLLECTION_NAME, datainsert_data) # 关键将集合加载到内存否则检索会报错 client.load_collection(COLLECTION_NAME) print(f✅ 向量库构建完成并加载写入 {len(insert_data)} 条切片) else: client MilvusClient(DB_PATH) client.load_collection(COLLECTION_NAME) # 复用库也要加载解释id是主键手动赋值为 0,1,2,… 方便对应。text字段保存切片的原文max_length2000足够容纳 300 字符的文本。FLAT索引是暴力搜索适合小规模数据几万条以内准确率 100%COSINE相似度与归一化向量配合使用。重要Milvus Lite 需要显式调用load_collection()将数据加载到内存后才能执行search否则会报错。3.7 带阈值过滤的 Milvus 检索函数def milvus_retrieve(query, top_k5, score_threshold0.6): query_vec embeddings.embed_query(query) results client.search( collection_nameCOLLECTION_NAME, data[query_vec], limittop_k, output_fields[text] ) doc_list [] for hit in results[0]: if hit[distance] score_threshold: # distance 是余弦相似度值越大越相似 doc_list.append(Document(page_contenthit[entity][text])) return doc_listscore_threshold0.6用来过滤掉相似度太低的片段避免引入噪声。注意由于向量已归一化余弦相似度范围是 [-1,1]0.6 表示中等以上相似度。返回的仍是 LangChain 的Document对象方便后续统一处理。3.8 初始化大模型与重排序器llm ChatDeepSeek(modeldeepseek-v4-pro, temperature0)ranker Ranker()temperature0保证大模型输出确定、严谨。Ranker()默认使用ms-marco-TinyBERT-L-2-v2模型也可通过参数指定其他模型。3.9 多查询生成query_gen_prompt ChatPromptTemplate.from_template( 为以下问题生成3个不同角度的相关中文问题换行分隔。\n原问题{question}\n相关问法 ) query_gen_chain query_gen_prompt | llm | StrOutputParser()大模型会输出类似文石最大屏幕的电子书阅读器是哪款哪款文石平板尺寸最大文石电纸书尺寸对比随后按换行分割成列表。3.10 构建 BM25 检索器bm25_retriever BM25Retriever.from_documents(docs) # docs 是所有切片文档BM25 是经典的关键词检索算法擅长精确匹配专有名词、型号等。我们基于同样的切片构建 BM25 索引实现关键词层面的召回。3.11 多查询 BM25 混合召回def multi_query_retrieve(question): variants query_gen_chain.invoke(question).split(\n) all_docs [] seen set() # 用于去重 q_list [question] [v.strip() for v in variants if v.strip()] # 1. 对每个问法进行 Milvus 向量检索 for q in q_list: chunk_list milvus_retrieve(q, top_k3) for chunk in chunk_list: cnt chunk.page_content if cnt not in seen: seen.add(cnt) all_docs.append(chunk) # 2. BM25 关键词召回 bm25_docs bm25_retriever.invoke(question) # 默认 top_k4 for doc in bm25_docs: if doc.page_content not in seen: seen.add(doc.page_content) all_docs.append(doc) return all_docs[:8] # 限制候选池大小流程说明原问题 3 个生成变体共 4 个查询每个查询用 Milvus 召回 top 3理论上最多 12 条。BM25 再召回 4 条合并后去重。最终限制最多 8 条候选文档进入重排序避免候选太多导致重排过慢。3.12 重排序选出最相关的 2 条def retrieve_multi_and_rerank(question): raw_docs multi_query_retrieve(question) if not raw_docs: return [] passages [{text: doc.page_content} for doc in raw_docs] rerank_res ranker.rerank(RerankRequest(queryquestion, passagespassages)) top_texts [item[text] for item in rerank_res[:2]] return [d for d in raw_docs if d.page_content in top_texts]FlashRank 会对每个候选段落与原始问题计算相关性分数并按分数降序排列。我们只取前 2 条确保送入大模型的上下文最相关。3.13 组装 RAG 链并执行template 根据下面提供的上下文回答问题。如果不知道答案就说不知道。 上下文 {context} 问题{question} 回答 prompt ChatPromptTemplate.from_template(template) def format_docs(docs): return \n\n.join(doc.page_content for doc in docs) rag_chain ( { context: RunnableLambda(retrieve_multi_and_rerank) | RunnableLambda(format_docs), question: RunnablePassthrough() } | prompt | llm | StrOutputParser() ) if __name__ __main__: question 文石哪款的pad尺寸最大 answer rag_chain.invoke(question) print(回答, answer) # 释放资源 client.release_collection(COLLECTION_NAME) client.close()LCEL 链执行顺序用户问题同时传给两个分支上下文分支和问题分支。上下文分支retrieve_multi_and_rerank获得最佳文档列表 →format_docs拼成字符串。将上下文和问题填入 Prompt交给 DeepSeek 生成答案。StrOutputParser提取纯文本输出。运行效果演示假设knowledge.txt内容如下文石 Tab10 拥有一块 10.3 英寸的电子墨水屏适合日常阅读。文石 Tab13 配备 13.3 英寸柔性屏幕是目前文石尺寸最大的电纸书适合阅读 PDF。文石 Nova Air 采用 7.8 英寸屏幕主打便携。文石 Leaf 2 是 7 英寸非常轻巧。运行脚本后控制台会输出类似 向量库读取异常强制重建: ...✅ 向量库构建完成并加载写入 4 条切片回答 文石 Tab13 的 pad 尺寸最大它配备了 13.3 英寸的屏幕。常见问题与解决方案5.1 Windows 下 Milvus 数据库文件被占用症状PermissionError: [WinError 32]或[WinError 5]。解决确保上次运行的 Python 进程已完全退出或手动运行提示中的 PowerShell 命令强制结束 milvus-lite 进程并删除文件夹。5.2 向量检索返回空列表检查score_threshold是否设置过高可尝试降低到 0.3。确认client.load_collection(COLLECTION_NAME)是否已执行否则search会报错或返回空。5.3 重排序模型下载失败请确保requests.get镜像劫持代码在Ranker()初始化之前已执行。也可以手动下载模型并指定cache_dir参数如Ranker(model_namems-marco-TinyBERT-L-2-v2, cache_dir你的路径)。持久化向量库Milvus Lite 让数据落盘重启不丢失也方便接入Milvus 。混合召回向量检索 BM25 关键词检索互补增强。多查询生成自动扩展用户意图提高召回覆盖率。重排序精排FlashRank 轻量高效筛选出真正相关的段落。阈值过滤过滤低相似度噪声提升上下文质量。接下来我们会逐步的封装成可对外使用的api。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
