RAG全链路数据清洗、切分、Embedding、向量库、召回、重排、答案生成的设计逻辑离线阶段文档 → 清洗 → 切分 → Embedding → 向量库1文档加载import os import fitz # PyMuPDF from loguru import logger from tqdm import tqdm log_dir log if not os.path.exists(log_dir): os.makedirs(log_dir) logger.add( f{log_dir}/pdf_load.log, rotation200 MB, retention1, levelINFO, format{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}, encodingutf-8)2数据清洗import re #页级清洗过滤版权/图片授权页 NOISE_KEYWORDS [ unsplash license, creative commons, attribution, photo by, image by, reused, license, copyright, all rights reserved, ] def is_noise_page(text): lower_text text.lower() hit_count sum(keyword in lower_text for keyword in NOISE_KEYWORDS) # 如果一页里多个版权/授权关键词同时出现大概率是噪声页 if hit_count 3: return True # URL 很多的页通常是引用/授权页 url_count len(re.findall(rhttps?://|www\., lower_text)) if url_count 5: return True return False #过滤空格等特殊字符合并多个换行符 def clean_text(text): text text.replace(\xa0, ) text re.sub(r[ \t], , text) text re.sub(r(?!\n)\n(?!\n), , text) text re.sub(r\n{3,}, \n\n, text) return text.strip() def load_pdf(pdf_path): try: pages [] with fitz.open(pdf_path) as doc: logger.info(f开始加载PDF: {pdf_path}, 总页数: {doc.page_count}) for page_index, page in tqdm(enumerate(doc), totaldoc.page_count): text clean_text(page.get_text()) if not text: logger.info(f第{page_index}页为空跳过) continue if is_noise_page(text): logger.info(f第{page_index}页为噪声页跳过) continue pages.append({ page: page_index, text: text }) logger.info(fPDF加载完成有效页数: {len(pages)}) return pages except Exception as error: logger.exception(f加载PDF失败: {pdf_path}, error: {error}) raise pages load_pdf(rag/simple-local-rag/human-nutrition-text.pdf)3文本切分方法一 from sentence_transformers import SentenceTransformer def split_text(text, chunk_size300): # 简单按字符数切分可用更智能的分句/分段 return [text[i:ichunk_size] for i in range(0, len(text), chunk_size)] chunks [] for page in pages: for chunk in split_text(page[text]): if len(chunk.strip()) 50: # 过滤过短片段 chunks.append({page: page[page], text: chunk}) chunks 方法二 from langchain_text_splitters import CharacterTextSplitter text_splitter CharacterTextSplitter( chunk_size300,#切分长度 chunk_overlap5 #相邻两个chunks之间的重叠token数量 ) all_text \n.join([p[text] for p in pages]) chunks text_splitter.split_text(all_text) chunks4向量化与存储import numpy as np from sentence_transformers import SentenceTransformer #文档向量化 model SentenceTransformer(all-mpnet-base-v2) doc_vectors model.encode([c[text] for c in chunks], normalize_embeddingsTrue) doc_vectors np.array(doc_vectors) # 构建 FAISS 索引 import faiss dim doc_vectors.shape[1] index faiss.IndexFlatIP(dim) # 精确搜索内积 index.add(doc_vectors) #保存索引 faiss.write_index(index, /opt/cyc/rag/simple-local-rag/docs.index)在线阶段问题 → Embedding → 召回 → 重排 → 拼上下文 → LLM 生成答案1问题 → Embedding → 召回# 查询 query 怎么保持饮食健康 query_vector model.encode([query], normalize_embeddingsTrue) query_vector np.asarray(query_vector, dtypefloat32) texts [d[text] for d in pages] # 搜索 top-k k 3 scores, ids index.search(query_vector, k) print(fQuery: {query}\n) for score, idx in zip(scores[0], ids[0]): print(fscore{score:.4f}, doc{texts[idx]})2重排 → 拼上下文 → LLM 生成答案from anthropic import Anthropic import numpy as np client Anthropic() def bge_reranker_matched(text1, text2, threshold0.01, ratio_threshold0.01): 接口仅支持 1 对多返回得分threshold 的候选占比ratio_threshold 的 text1 元素。 bge_urlbge_url bge_tokenbge_token headers { Content-Type: application/json, Authorization: bge_token } matched [] for d in text2: data { model: bge-reranker-v2-m3, text_1: text1, text_2: d } response requests.post(bge_url, jsondata, headersheaders) res response.json().get(data) or [] if not res: continue hit_count sum(item.get(score, 0) threshold for item in res) if hit_count / len(res) ratio_threshold: matched.append(d) return matched def answer_with_rag(query, model, index, pages, k3): # 1. query - vector records chunks texts [d[text] for d in records] doc_vectors model.encode( texts, normalize_embeddingsTrue, show_progress_barTrue ) doc_vectors np.asarray(doc_vectors, dtypefloat32) query_vector model.encode([query], normalize_embeddingsTrue) query_vector np.asarray(query_vector, dtypefloat32) # 2. retrieve k 3 scores, ids index.search(query_vector, k) retrieved [] for score, idx in zip(scores[0], ids[0]): retrieved.append({ score: float(score), page: pages[idx][page], text: pages[idx][text] }) #3.精排 context bge_reranker_matched(query, retrieved) # 4. generate prompt f 你是一个基于资料回答问题的助手。 请只根据“参考资料”回答不要编造。 如果参考资料不足以回答就明确说“参考资料不足”。 用户问题 {query} 参考资料 {context} 请输出 1. 简洁答案 resp client.messages.create( modelclaude-sonnet-5, max_tokens500, messages[{role: user, content: prompt}] ) return { answer: resp.content[0].text, retrieved: retrieved } answer_with_rag(蛋白质的主要功能是什么, model, index, pages)了解SFT、LoRA、RLHF等技术的适用场景对比RAG与微调的成本、效果、维护难度差异。SFTSupervised Fine-Tuning一种训练方式有监督微调用“输入 → 标准输出”的标注数据去教模型LoRA一种参数微调方法不直接全量改模型参数只训练少量“适配层/低秩矩阵”微调(Fine-Tuning)├── SFT│├── DPO│├── PPO(RLHF 里常用的一种优化算法注RLHF是一整套对齐流程)│└── ORPOSFT问题什么是ROI标准答案ROI是投资回报率用于衡量...DPO问题什么商品适合圣诞节促销回答A应该选择具有明显圣诞节消费场景的商品并结合历史销量...回答B所有商品都应该在圣诞节促销。人工标注A BRLHFSFT先做监督微调训练奖励模型Reward Model用强化学习优化策略常见就是PPOPPO(强化学习算法)根据奖励模型的分数继续优化模型输出参数更新方式├── Full Fine-Tuning├── LoRA├── QLoRA├── AdaLoRA└── DoRA适用场景技术主要作用适合场景不适合场景SFT监督微调让模型学会特定任务格式、风格、流程客服话术、结构化输出、领域问答、工具调用格式、固定写作风格频繁变化的知识库、强实时性事实LoRA低成本做微调预算有限、数据量中小、要快速试验、要多个版本并存需要极致性能且愿意全量训练时RLHF对齐人类偏好改善“好不好用”语气、礼貌、安全性、减少胡说、提升回答偏好一致性单纯补知识、快速迭代事实内容DPO/ORPO常见替代更简单的偏好优化想要类似 RLHF 的效果但训练更简化复杂奖励建模场景对比差异维度RAG微调SFT/LoRA成本通常较低到中等中等到高数据准备文档清洗、切分、向量化需要高质量指令数据/对话数据训练成本无训练或很少训练要训练LoRA 比全量便宜很多推理成本通常更高一些检索 生成通常更低、更稳定效果最新知识和可追溯性以及幻觉减少更强、复杂行为学习一般幻觉取决于训练质量、复杂行为学习更好维护难度改知识库、文档/索引版本管理、容易定位是检索还是生成问题、加文档即可扩展需要重新训练、新能力需要加数据重训 RAG与Agent_体系
RAG全链路数据清洗、切分、Embedding、向量库、召回、重排、答案生成的设计逻辑离线阶段文档 → 清洗 → 切分 → Embedding → 向量库1文档加载import os import fitz # PyMuPDF from loguru import logger from tqdm import tqdm log_dir log if not os.path.exists(log_dir): os.makedirs(log_dir) logger.add( f{log_dir}/pdf_load.log, rotation200 MB, retention1, levelINFO, format{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}, encodingutf-8)2数据清洗import re #页级清洗过滤版权/图片授权页 NOISE_KEYWORDS [ unsplash license, creative commons, attribution, photo by, image by, reused, license, copyright, all rights reserved, ] def is_noise_page(text): lower_text text.lower() hit_count sum(keyword in lower_text for keyword in NOISE_KEYWORDS) # 如果一页里多个版权/授权关键词同时出现大概率是噪声页 if hit_count 3: return True # URL 很多的页通常是引用/授权页 url_count len(re.findall(rhttps?://|www\., lower_text)) if url_count 5: return True return False #过滤空格等特殊字符合并多个换行符 def clean_text(text): text text.replace(\xa0, ) text re.sub(r[ \t], , text) text re.sub(r(?!\n)\n(?!\n), , text) text re.sub(r\n{3,}, \n\n, text) return text.strip() def load_pdf(pdf_path): try: pages [] with fitz.open(pdf_path) as doc: logger.info(f开始加载PDF: {pdf_path}, 总页数: {doc.page_count}) for page_index, page in tqdm(enumerate(doc), totaldoc.page_count): text clean_text(page.get_text()) if not text: logger.info(f第{page_index}页为空跳过) continue if is_noise_page(text): logger.info(f第{page_index}页为噪声页跳过) continue pages.append({ page: page_index, text: text }) logger.info(fPDF加载完成有效页数: {len(pages)}) return pages except Exception as error: logger.exception(f加载PDF失败: {pdf_path}, error: {error}) raise pages load_pdf(rag/simple-local-rag/human-nutrition-text.pdf)3文本切分方法一 from sentence_transformers import SentenceTransformer def split_text(text, chunk_size300): # 简单按字符数切分可用更智能的分句/分段 return [text[i:ichunk_size] for i in range(0, len(text), chunk_size)] chunks [] for page in pages: for chunk in split_text(page[text]): if len(chunk.strip()) 50: # 过滤过短片段 chunks.append({page: page[page], text: chunk}) chunks 方法二 from langchain_text_splitters import CharacterTextSplitter text_splitter CharacterTextSplitter( chunk_size300,#切分长度 chunk_overlap5 #相邻两个chunks之间的重叠token数量 ) all_text \n.join([p[text] for p in pages]) chunks text_splitter.split_text(all_text) chunks4向量化与存储import numpy as np from sentence_transformers import SentenceTransformer #文档向量化 model SentenceTransformer(all-mpnet-base-v2) doc_vectors model.encode([c[text] for c in chunks], normalize_embeddingsTrue) doc_vectors np.array(doc_vectors) # 构建 FAISS 索引 import faiss dim doc_vectors.shape[1] index faiss.IndexFlatIP(dim) # 精确搜索内积 index.add(doc_vectors) #保存索引 faiss.write_index(index, /opt/cyc/rag/simple-local-rag/docs.index)在线阶段问题 → Embedding → 召回 → 重排 → 拼上下文 → LLM 生成答案1问题 → Embedding → 召回# 查询 query 怎么保持饮食健康 query_vector model.encode([query], normalize_embeddingsTrue) query_vector np.asarray(query_vector, dtypefloat32) texts [d[text] for d in pages] # 搜索 top-k k 3 scores, ids index.search(query_vector, k) print(fQuery: {query}\n) for score, idx in zip(scores[0], ids[0]): print(fscore{score:.4f}, doc{texts[idx]})2重排 → 拼上下文 → LLM 生成答案from anthropic import Anthropic import numpy as np client Anthropic() def bge_reranker_matched(text1, text2, threshold0.01, ratio_threshold0.01): 接口仅支持 1 对多返回得分threshold 的候选占比ratio_threshold 的 text1 元素。 bge_urlbge_url bge_tokenbge_token headers { Content-Type: application/json, Authorization: bge_token } matched [] for d in text2: data { model: bge-reranker-v2-m3, text_1: text1, text_2: d } response requests.post(bge_url, jsondata, headersheaders) res response.json().get(data) or [] if not res: continue hit_count sum(item.get(score, 0) threshold for item in res) if hit_count / len(res) ratio_threshold: matched.append(d) return matched def answer_with_rag(query, model, index, pages, k3): # 1. query - vector records chunks texts [d[text] for d in records] doc_vectors model.encode( texts, normalize_embeddingsTrue, show_progress_barTrue ) doc_vectors np.asarray(doc_vectors, dtypefloat32) query_vector model.encode([query], normalize_embeddingsTrue) query_vector np.asarray(query_vector, dtypefloat32) # 2. retrieve k 3 scores, ids index.search(query_vector, k) retrieved [] for score, idx in zip(scores[0], ids[0]): retrieved.append({ score: float(score), page: pages[idx][page], text: pages[idx][text] }) #3.精排 context bge_reranker_matched(query, retrieved) # 4. generate prompt f 你是一个基于资料回答问题的助手。 请只根据“参考资料”回答不要编造。 如果参考资料不足以回答就明确说“参考资料不足”。 用户问题 {query} 参考资料 {context} 请输出 1. 简洁答案 resp client.messages.create( modelclaude-sonnet-5, max_tokens500, messages[{role: user, content: prompt}] ) return { answer: resp.content[0].text, retrieved: retrieved } answer_with_rag(蛋白质的主要功能是什么, model, index, pages)了解SFT、LoRA、RLHF等技术的适用场景对比RAG与微调的成本、效果、维护难度差异。SFTSupervised Fine-Tuning一种训练方式有监督微调用“输入 → 标准输出”的标注数据去教模型LoRA一种参数微调方法不直接全量改模型参数只训练少量“适配层/低秩矩阵”微调(Fine-Tuning)├── SFT│├── DPO│├── PPO(RLHF 里常用的一种优化算法注RLHF是一整套对齐流程)│└── ORPOSFT问题什么是ROI标准答案ROI是投资回报率用于衡量...DPO问题什么商品适合圣诞节促销回答A应该选择具有明显圣诞节消费场景的商品并结合历史销量...回答B所有商品都应该在圣诞节促销。人工标注A BRLHFSFT先做监督微调训练奖励模型Reward Model用强化学习优化策略常见就是PPOPPO(强化学习算法)根据奖励模型的分数继续优化模型输出参数更新方式├── Full Fine-Tuning├── LoRA├── QLoRA├── AdaLoRA└── DoRA适用场景技术主要作用适合场景不适合场景SFT监督微调让模型学会特定任务格式、风格、流程客服话术、结构化输出、领域问答、工具调用格式、固定写作风格频繁变化的知识库、强实时性事实LoRA低成本做微调预算有限、数据量中小、要快速试验、要多个版本并存需要极致性能且愿意全量训练时RLHF对齐人类偏好改善“好不好用”语气、礼貌、安全性、减少胡说、提升回答偏好一致性单纯补知识、快速迭代事实内容DPO/ORPO常见替代更简单的偏好优化想要类似 RLHF 的效果但训练更简化复杂奖励建模场景对比差异维度RAG微调SFT/LoRA成本通常较低到中等中等到高数据准备文档清洗、切分、向量化需要高质量指令数据/对话数据训练成本无训练或很少训练要训练LoRA 比全量便宜很多推理成本通常更高一些检索 生成通常更低、更稳定效果最新知识和可追溯性以及幻觉减少更强、复杂行为学习一般幻觉取决于训练质量、复杂行为学习更好维护难度改知识库、文档/索引版本管理、容易定位是检索还是生成问题、加文档即可扩展需要重新训练、新能力需要加数据重训