前言在前两篇文章中我们完成了基础设施的搭建和 Embedding 模型的封装。现在系统有了向量数据库具备了将文本转换为高维向量的能力。接下来要做的就是把知识真正“喂”给系统。数据是 RAG 系统的“底座”后续所有的检索和问答质量都直接依赖于此。本文将完整记录 Milvus Collection 设计、文档分块模块开发、数据灌入脚本编写、AI 爬虫实战以及知识库从 0 到近千条向量的规模化演进过程。但本文不止于此——我们还将深入实现混合检索、集成 Reranker 精排并通过量化对比验证优化效果。一、设计 Milvus Collection Schema在灌入数据之前我们需要先设计好 Collection 的结构。Collection 相当于关系型数据库中的“表”我们需要预先定义好字段类型和向量索引。1.1 Schema 设计字段名类型说明约束idVARCHAR文档块的唯一标识主键最大长度 100textVARCHAR原始文本内容最大长度 65535vectorFLOAT_VECTOREmbedding 向量维度 384需与 BGE-small 模型一致sourceVARCHAR来源文件路径最大长度 5001.2 索引配置索引类型IVF_FLAT后续可换 HNSW 或 IVF_SQ8 做对比实验度量类型COSINE因为 Embedding 已归一化余弦相似度最合适nlist128聚类中心数平衡精度与速度1.3 代码实现创建app/core/retrieval/milvus_client.py封装数据库连接与管理逻辑Pythonfrom pymilvus import connections, Collection, CollectionSchema, FieldSchema, DataType, utility COLLECTION_NAME lite_rag_docs VECTOR_DIM 384 # bge-small-en-v1.5 的维度 def connect_milvus(host: str localhost, port: str 19530): 建立与 Milvus 的连接 connections.connect(aliasdefault, hosthost, portport) print(f✅ 已连接到 Milvus ({host}:{port})) def create_collection() - Collection: 创建 Collection如果已存在则先清空重建 if utility.has_collection(COLLECTION_NAME): utility.drop_collection(COLLECTION_NAME) print(f⚠️ 已删除旧 Collection: {COLLECTION_NAME}) fields [ FieldSchema(nameid, dtypeDataType.VARCHAR, is_primaryTrue, max_length100), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length65535), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dimVECTOR_DIM), FieldSchema(namesource, dtypeDataType.VARCHAR, max_length500), ] schema CollectionSchema(fields, descriptionLiteRAG 知识库) collection Collection(COLLECTION_NAME, schema) index_params { metric_type: COSINE, index_type: IVF_FLAT, params: {nlist: 128} } collection.create_index(vector, index_params) print(f✅ Collection {COLLECTION_NAME} 创建成功) return collection def get_collection() - Collection: 获取当前可用的 Collection if not utility.has_collection(COLLECTION_NAME): raise RuntimeError(fCollection {COLLECTION_NAME} 不存在) return Collection(COLLECTION_NAME) if __name__ __main__: connect_milvus() create_collection()运行测试bashpython -m app.core.retrieval.milvus_client预期输出text✅ 已连接到 Milvus (localhost:19530) ⚠️ 已删除旧 Collection: lite_rag_docs ✅ Collection lite_rag_docs 创建成功二、文档分块Chunking模块开发在大模型 RAG 系统中受限于 LLM 的上下文窗口长文档必须切分成小块Chunks才能进行有效的检索。我们需要一个支持多格式、可配置的分块模块。2.1 安装依赖bashpip install pypdf langchain langchain-community langchain-text-splitters2.2 代码实现创建app/core/splitter.pypythonimport os from typing import List, Dict, Any from langchain_community.document_loaders import PyPDFLoader, TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from loguru import logger class DocumentSplitter: 文档加载与分块服务 def __init__(self, chunk_size: int 512, chunk_overlap: int 50): self.chunk_size chunk_size self.chunk_overlap chunk_overlap # 使用递归字符分割器按自然段落和标点进行切分 self.text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , , , ], length_functionlen, ) def load_and_split(self, file_path: str) - List[Dict[str, Any]]: if not os.path.exists(file_path): logger.error(f文件不存在: {file_path}) return [] ext os.path.splitext(file_path)[1].lower() try: if ext .pdf: loader PyPDFLoader(file_path) elif ext in [.md, .markdown, .txt, .log, .json, .yaml, .yml]: loader TextLoader(file_path, encodingutf-8) else: logger.warning(f不支持的文件类型: {ext}) return [] documents loader.load() except Exception as e: logger.error(f加载文件失败 {file_path}: {e}) return [] chunks self.text_splitter.split_documents(documents) result [] for i, chunk in enumerate(chunks): result.append({ text: chunk.page_content, metadata: { source: file_path, chunk_index: i, **chunk.metadata } }) logger.info(f文件 {file_path} 切分为 {len(result)} 个文本块) return result def load_directory(self, directory: str) - List[Dict[str, Any]]: all_chunks [] supported_exts {.pdf, .md, .markdown, .txt} for root, _, files in os.walk(directory): for file in files: ext os.path.splitext(file)[1].lower() if ext in supported_exts: file_path os.path.join(root, file) chunks self.load_and_split(file_path) all_chunks.extend(chunks) logger.info(f目录 {directory} 共生成 {len(all_chunks)} 个文本块) return all_chunks设计决策我们刻意避开了UnstructuredMarkdownLoader因为它依赖庞大的unstructured库安装时常因系统 C 依赖缺失而失败。改用TextLoader统一处理所有文本类文件更轻量、更稳定。三、数据灌入脚本万事俱备我们需要一个核心脚本将全流程串联起来遍历目录 → 文档分块 → 向量化编码 → 插入 Milvus。创建scripts/ingest.pypython#!/usr/bin/env python import hashlib import sys from pathlib import Path # 确保能导入 app 模块 sys.path.insert(0, str(Path(__file__).parent.parent)) from loguru import logger from app.core.splitter import DocumentSplitter from app.core.embedding import get_embedding_service from app.core.retrieval.milvus_client import connect_milvus, get_collection, COLLECTION_NAME def generate_doc_id(source: str, chunk_index: int) - str: 根据文件路径和块索引生成唯一的短 ID raw f{source}_{chunk_index} return hashlib.md5(raw.encode()).hexdigest()[:16] def ingest_docs(directory: str docs): # 1. 连接 Milvus connect_milvus() collection get_collection() logger.info(f 已连接 Collection: {COLLECTION_NAME}) # 2. 加载并分块 splitter DocumentSplitter(chunk_size512, chunk_overlap50) chunks splitter.load_directory(directory) logger.info(f 共加载 {len(chunks)} 个文本块) if not chunks: logger.warning(没有找到任何文档块退出) return # 3. 向量化 embedder get_embedding_service() texts [chunk[text] for chunk in chunks] sources [chunk[metadata][source] for chunk in chunks] chunk_indices [chunk[metadata][chunk_index] for chunk in chunks] logger.info( 正在调用模型生成向量...) vectors embedder.encode(texts) # 4. 插入数据库 ids [generate_doc_id(src, idx) for src, idx in zip(sources, chunk_indices)] data [ids, texts, vectors.tolist(), sources] logger.info(f 正在插入 {len(ids)} 条数据到 Milvus...) collection.insert(data) collection.flush() logger.success(f✅ 成功当前 Collection 共有 {collection.num_entities} 条向量) if __name__ __main__: ingest_docs()四、知识库规模化从玩具数据到生产级语料仅有几条测试数据的知识库无法验证系统的真实性能。为了让项目具备面试展示价值我们需要灌入大规模、高质量的 AI 领域专业语料。4.1 AI 爬虫环境搭建Crawl4AI Playwright传统的requests难以应对现代动态渲染网页。我们选择Crawl4AI——一款专为大模型时代设计的开源爬虫它能直接解析网页并输出极度干净的 Markdown。 踩坑与修复lxml 依赖冲突Python 3.13 环境下解析失败。解决办法升级 pip 并强制使用清华源拉取预编译包。浏览器内核下载超时默认从国外 CDN 下载 Chromium 频繁中断。解决办法降级 Playwright 至 1.48.0 版本并手动通过apt补齐系统底层依赖。4.2 批量抓取 AI 技术文章编写带重试机制的爬虫脚本定点抓取涵盖 RAG、Agent、向量数据库等前沿方向的高质量文章。抓取结果统计来源平台成功数典型内容arXiv5 篇ACL26 RAG 前沿研究、Agent 综述论文阿里云开发者社区4 篇RAG 企业级实战、AI 应用开发指南Machine Learning Mastery1 篇Graph-RAG图检索深度教程经济观察网1 篇兆瓦级算力系统新闻知乎 / 百度0 篇反爬严重直接放弃经验总结对于强反爬平台如知乎、百度开发者社区即使使用 Playwright 模拟浏览器也难以稳定抓取。实践中应优先选择提供 RSS、API 或反爬较弱的平台如 arXiv、各大技术社区。4.3 规模化向量入库清理掉抓取失败的空文件后执行一键入库脚本python scripts/ingest.py。最终剔除掉早期测试文件后知识库内沉淀了14 篇高质量长文共计切分生成了718 条高维向量注意初始入库时包含测试文件test.txt后续已彻底清理最终知识库规模为718 条向量。五、混合检索模块让召回更精准仅有向量检索是不够的。向量擅长捕捉语义相似性但对精确术语如HNSW、IVF_FLAT的匹配能力较弱。为此我们实现了混合检索——融合向量检索和 BM25 关键词检索取长补短。5.1 核心设计组件实现方式说明向量检索Milvussearch() BGE-small384 维 COSINE 相似度BM25 关键词检索rank_bm25库全量文档构建索引中文分词使用jieba融合算法RRFReciprocal Rank Fusionk60排名加权融合索引缓存pickle序列化到bm25_index.pkl首次构建后秒级加载5.2 中文分词适配BM25 默认的空格分词对中文完全失效。我们引入jieba并实现智能分词pythondef tokenize(text: str) - List[str]: if any(\u4e00 char \u9fff for char in text): return list(jieba.cut(text)) # 中文用 jieba else: return text.lower().split() # 英文用空格5.3 踩坑与修复问题解决方案BM25 对中文无效引入jieba智能分词Milvus 连接未建立导致get_collection()失败在__init__中显式调用connect_milvus()向量格式错误required argument is not a float显式转换为float32并flatten().tolist()collection.is_loaded属性不存在直接调用幂等的collection.load()六、Reranker 精排画龙点睛之笔混合检索已能返回不错的结果但在高级 RAG 系统中Reranker重排序是提升精度的关键组件。它对召回的候选文档块进行二次精排筛选出最相关的几条作为 LLM 的上下文。概念澄清这里精排的“候选文档”指的是文本块Chunks而非原始文章。我们的知识库包含 14 篇文章但被切分成了718 个文本块。检索时是从这 718 个块中召回 Top-20 个块Reranker 再从中精选 Top-3。6.1 模型选型与显存管理选用BAAI/bge-reranker-base约 1.1GB 核心权重。针对 3060 6GB 显存的限制我们设计了动态加载/释放机制pythonclass RerankerService: def load_model(self): # 仅在需要时加载到 GPU if self.model is None: self.model AutoModelForSequenceClassification.from_pretrained( model_name, torch_dtypetorch.float16 ).to(self.device) def unload_model(self): # 用完后立即释放显存 del self.model torch.cuda.empty_cache()6.2 模型本地化下载使用hf download预先下载模型支持完全离线运行bashexport HF_ENDPOINThttps://hf-mirror.com hf download BAAI/bge-reranker-base --local-dir ./app/models/bge-reranker-base6.3 集成到检索流程修改检索服务的search()方法添加use_rerank参数pythondef search(self, query: str, top_k: int 5, use_rerank: bool False): # 1. 召回阶段混合检索返回 Top-20 个文本块 candidates self.reciprocal_rank_fusion(vector_hits, bm25_hits) # 2. 精排阶段可选 if use_rerank: reranker get_reranker_service() passages [doc[text] for doc in candidates] reranked reranker.rerank(query, passages, top_ktop_k) return reranked return candidates[:top_k]七、效果量化对比Reranker 带来了什么我们设计了对比实验对同一查询分别测试有/无 Reranker 的 Top-1 结果查询无 Reranker Top-1有 Reranker Top-1关键变化什么是RAGarticle_1709964.mdarticle_1709964.md来源一致精排优化了片段选择How does RAG work?...3-tiered-graph-rag-system.mdarticle_1709964.md⭐来源改变Reranker 纠正了 BM25 偏差RAG系统中如何优化检索召回率article_1709964.mdarticle_1709964.md来源一致精排优化了片段选择核心发现Reranker 具备语义纠偏能力对于英文查询能将更通用、更适合回答基础概念的文章排到首位。分数尺度变化RRF 0.03 → Reranker 5.05是正常现象两者算法不同我们关注的是相对排序的改善。面试话术参考“我通过对比实验验证了 Reranker 的有效性。对于英文查询 How does RAG work?原始混合检索由于 BM25 权重较高返回了一篇深度 Graph-RAG 文章而 Reranker 精准地将其纠正为更通用、更适合回答基础概念的中文 RAG 文章。”八、工程规范化一个优秀的项目不仅代码能跑还需要具备可复现性和清晰的模块划分。8.1 导出依赖文件bashpip freeze requirements.txt8.2 配置.gitignoretextvenv/ __pycache__/ app/models/ volumes/ .env *.pyc .DS_Store8.3 项目目录结构textLiteRAG/ ├── app/ │ ├── core/ │ │ ├── embedding.py # Embedding 服务 │ │ ├── splitter.py # 文档分块 │ │ ├── reranker.py # Reranker 精排 │ │ └── retrieval/ │ │ ├── milvus_client.py # Milvus 连接与 Collection 管理 │ │ └── hybrid_search.py # 混合检索 RRF │ └── models/ # 本地模型缓存 ├── scripts/ │ ├── ingest.py # 数据灌入脚本 │ ├── fetch_articles.py # 第一批爬虫 │ └── fetch_articles_batch2.py # 第二批爬虫 ├── docs/ # 原始文档 ├── docker-compose.yml # 6 个服务容器编排 ├── requirements.txt └── .gitignore九、踩坑记录完整版问题现象 / 报错根本原因解决方案No module named langchain.text_splitter新版 LangChain 将分块器拆分到了独立包中安装并修改导入from langchain_text_splitters import ...FileNotFoundError: Path ... not foundEmbedding 模块中模型路径使用了未解析的占位符去缓存目录使用ls查看真实哈希文件夹名并更新路径UnstructuredMarkdownLoader各种报错底层unstructured库体积大、系统 C 依赖复杂放弃该方案改用更轻量的TextLoader统一处理文本Crawl4AI 安装时lxml冲突Python 3.13 环境太新部分轮子尚未提供预编译包升级 pip并强制使用清华镜像源安装Playwright 浏览器内核下载失败/超时默认的国外 CDN 节点在国内访问不稳定降级 Playwright 至 1.48.0并使用apt手动安装系统底层依赖BM25 对中文完全失效默认空格分词无法处理中文引入jieba智能分词自动检测中英文Milvus 向量格式错误传入的向量包含numpy.float32而非纯 Python 浮点数astype(np.float32).flatten().tolist()collection.is_loaded不存在pymilvus版本差异直接调用幂等的collection.load()Reranker 测试时测试文件仍为 Top-1Milvus 中残留旧数据彻底删除 Collection 并重新灌库volumes/目录无法纳入版本控制宿主机的 Docker 数据卷目录归属 root权限被拒绝将volumes/加入.gitignore十、当前成果速览核心指标当前数值知识库文档总数14 篇高质量长文已剔除测试文件文本块 (Chunks) 总数718 条纯净知识库向量维度384 (BGE-Small)Milvus 索引结构IVF_FLAT COSINE检索方式向量检索 BM25 RRF 混合融合精排组件BGE-Reranker-base本地化动态显存管理Reranker 纠偏效果英文查询 Top-1 来源被正确纠正工程规范requirements.txt、.gitignore、模块化设计离线可用性所有模型本地化无需联网十一、下一步计划现在我们拥有了一个包含718 条高质量向量、混合检索与 Reranker 精排双引擎驱动的 RAG 检索系统。数据底座和检索链路已完全打通下一篇文章我们将进入 RAG 系统的核心交互环节LLM 集成封装大模型 API如 DeepSeek、Qwen设计 Prompt 模板FastAPI 问答接口实现POST /chat接收用户问题返回流式/非流式答案Redis 缓存加速为高频查询添加缓存层端到端测试用真实 AI 面试题验证系统问答质量本文是【从0到1搭建企业级RAG系统】系列的第三篇如果你在数据处理、混合检索或 Reranker 集成过程中遇到任何问题欢迎在评论区交流