一、为什么需要 RAG在开发 Agent 应用时我发现一个核心问题大模型虽然强大但它不知道我公司的私有数据。比如问它“我们公司的 API 鉴权流程是什么”它只能瞎编。RAGRetrieval-Augmented Generation就是解决方案先检索相关知识再让 LLM 基于这些知识回答。本文记录我从零构建 RAG 系统的完整过程。二、整体架构设计text┌─────────────────────────────────────────────────────────────────┐ │ 建库阶段离线 │ ├─────────────────────────────────────────────────────────────────┤ │ PDF 文档 → Markdown 转换 → 智能切分 → 向量化 → Qdrant 存储 │ │ ↓ │ │ BM25 关键词索引 │ └─────────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────────┐ │ 检索阶段在线 │ ├─────────────────────────────────────────────────────────────────┤ │ 用户查询 → 向量召回 BM25 召回 → RRF 融合 → Rerank → LLM 回答 │ └─────────────────────────────────────────────────────────────────┘三、核心代码实现3.1 配置文件config.py所有配置集中管理方便维护和切换。python# 路径配置 RAW_DIR Path(data/raw) # 原始 PDF 存放位置 MARKDOWN_DIR Path(data/markdown) # 转换后的 Markdown IMAGES_DIR Path(data/images) # 提取的图片 CHUNKS_DIR Path(data/chunks) # 切分结果 # Qdrant 配置 QDRANT_HOST localhost QDRANT_PORT 6333 COLLECTION_NAME company_knowledge # Embedding 配置 EMBEDDING_MODEL text-embedding-v3 EMBEDDING_DIMENSION 1024 # v3 支持 64-1024 # 切分配置 CHUNK_SIZE 1000 # 目标字符数 MIN_CHUNK_SIZE 100 # 最小 chunk 大小设计要点所有路径用Path对象跨平台兼容配置与代码分离修改配置不需要改代码敏感信息API Key放在.env文件3.2 PDF 转 Markdownconvert_pdf_to_md.pypythondef convert_pdf_to_markdown(pdf_path: Path, output_path: Path) - str: md_text pymupdf4llm.to_markdown( str(pdf_path), write_imagesTrue, # 提取图片 image_pathstr(IMAGES_DIR), # 图片保存目录 image_formatpng, # 图片格式 dpi150, # 图片分辨率 ) output_path.write_text(md_text, encodingutf-8) return md_text踩坑记录page_chunksTrue时返回的是列表不是字符串需要合并中文 PDF 需要确保编码正确图片提取后需要在 Markdown 中保留引用3.3 智能切分chunk_markdown.pypythondef chunk_by_headers(md_text: str) - List[Dict]: 按标题层级切分 Markdown headers_to_split_on [ (#, title), (##, section), (###, subsection), ] splitter MarkdownHeaderTextSplitter(headers_to_split_on) splits splitter.split_text(md_text) chunks [] for i, split in enumerate(splits): # 构建面包屑路径保留层级结构 breadcrumbs [] if split.metadata.get(title): breadcrumbs.append(split.metadata[title]) # ... 处理 section 和 subsection chunk { id: fchunk_{i:04d}, content: split.page_content.strip(), metadata: { breadcrumbs: breadcrumbs, source: , chunk_type: section, length: len(split.page_content) } } chunks.append(chunk) return chunks为什么用 MarkdownHeaderTextSplitter按标题边界切分保证语义完整性自动提取标题层级作为元数据支持面包屑路径检索时能知道来源章节3.4 向量存储store_to_qdrant.pypythondef get_embedding(text: str) - List[float]: 使用 DashScope 生成向量 resp dashscope.TextEmbedding.call( modelEMBEDDING_MODEL, inputtext, text_typedocument, dimensionsEMBEDDING_DIMENSION, ) return resp.output[embeddings][0][embedding] def store_chunks_to_qdrant(chunks: List[Dict]): client QdrantClient(hostQDRANT_HOST, portQDRANT_PORT) # 创建 Collection client.create_collection( collection_nameCOLLECTION_NAME, vectors_configVectorParams( sizeEMBEDDING_DIMENSION, distanceDistance.COSINE ) ) # 批量存储 for i, chunk in enumerate(chunks): vector get_embedding(chunk[content]) point PointStruct( idi, vectorvector, payload{ content: chunk[content], **chunk[metadata] } ) client.upsert(collection_nameCOLLECTION_NAME, points[point])Embedding 模型选型选择 DashScope text-embedding-v3维度 1024中文优化好成本低。3.5 BM25 关键词索引bm25_index.pypythonclass BM25Index: def __init__(self, db_path: str bm25_index.db): self.db_path db_path self.k1 1.5 # 词频饱和控制 self.b 0.75 # 文档长度归一化 self._init_db() def _init_db(self): 初始化 SQLite 表结构 # documents: 文档主表 # chunks: 切分块表 # keywords: 关键词表 # inverted_index: 倒排索引表 def _extract_keywords(self, text: str, top_k: int 10) - List[str]: 提取关键词分词 频率统计 停用词过滤 def search(self, query: str, top_n: int 5) - List[Dict]: BM25 搜索 # 1. 提取查询关键词 # 2. 计算 IDF # 3. 计算每个 chunk 的 BM25 分数 # 4. 返回 Top-N为什么用 SQLite 存 BM25轻量、零配置、Python 自带支持复杂查询和过滤数据持久化重启不丢失3.6 多路召回与融合fusion.pypythondef multi_channel_recall(query: str, top_n: int 20) - List[Dict]: 多路召回 results [] # 通道1向量召回 vector_results vector_recall(query, top_ntop_n) for r in vector_results: r[recall_method] vector results.extend(vector_results) # 通道2BM25 召回 bm25_results bm25_index.search(query, top_ntop_n) for r in bm25_results: r[recall_method] bm25 results.extend(bm25_results) # RRF 融合去重 return rrf_fusion(results, top_ntop_n) def rrf_fusion(results: List[Dict], k: int 60) - List[Dict]: RRFReciprocal Rank Fusion融合算法 scores {} for item in results: doc_id item.get(id) or hash(item[content]) scores[doc_id] scores.get(doc_id, 0) 1 / (k item.get(rank, 1)) # 按融合分数排序返回RRF 公式score Σ 1/(k rank)k 通常取 60。3.7 重排与生成rerank.pypythondef bge_rerank(query: str, passages: List[str]) - List[float]: 使用 Qwen3-Rerank 进行重排 response TextReRank.call( modelqwen3-rerank, queryquery, documentspassages, top_klen(passages) ) # 返回每个 passage 的相关性分数 def generate_answer(query: str, retrieved_docs: List[Dict]) - str: 基于检索结果生成回答 context \n\n.join([doc[content] for doc in retrieved_docs]) prompt f基于以下文档回答用户问题 文档内容{context} 用户问题{query} # 调用 DeepSeek API 生成回答 return call_llm(prompt)为什么需要 Rerank向量召回和 BM25 召回都是粗排可能有噪音Cross-Encoder 模型能更精确地判断相关性用 5% 的召回率损失换 100 倍的速度提升四、完整流程演示4.1 建库pythonrag RAG() result rag.build_knowledge_base() print(result) # 输出: 转换了 10 个 PDF生成 245 个 chunks存入 Qdrant构建 BM25 索引4.2 检索pythonresults rag.retrieve(什么是统一力位控制, top_k5) for r in results: print(f来源: {r[source]}) print(f内容: {r[content][:100]}...)4.3 完整 RAGpythonresult rag.rag_pipeline(什么是统一力位控制, top_k5) print(result[answer]) # 输出: 基于文档内容的回答五、踩坑与优化问题解决方案PDF 图片丢失设置write_imagesTrue双栏论文顺序错乱用marker包替代中文分词不准添加自定义词到 jieba向量召回精度低切换到 DashScope v3/v4BM25 内存不足从 Pickle 切换到 SQLiteRerank 延迟高缩小候选集到 30-50 条六、性能指标指标数据建库速度~2 秒/PDF检索延迟 500ms召回率0.89精确率0.83七、后续计划增量更新检测文件变化混合检索向量 BM25 RRF缓存热门查询接入 Agent 作为工具八、完整代码项目已开源 https://github.com/aminga123321/RAG/tree/main/rag代码结构rag/├── data/ # 数据目录│ ├── raw/ # 原始 PDF 文件│ ├── markdown/ # 转换后的 Markdown 文件│ └── chunks/ # 切分后的文本块├── scripts/ # 核心脚本│ ├── config.py # 配置文件│ ├── convert_pdf_to_md.py # PDF 转 Markdown│ ├── chunk_markdown.py # Markdown 切分│ ├── store_to_qdrant.py # 向量存储│ ├── retrieve.py # 向量召回│ ├── fusion.py # 多路召回融合│ ├── rerank.py # 重排│ ├── bm25_index.py # BM25 索引│ └── rag.py # RAG 系统类├── test_rag_class.py # RAG 类测试├── test_bm25_simple.py # BM25 索引测试├── requirements.txt # 依赖文件├── .env.example # 环境变量示例└── .gitignore # Git 忽略文件