LLM-RAG项目细节-数据处理、分块..

LLM-RAG项目细节-数据处理、分块与向量化实战指南大家好我是一名资深技术博主。今天我们来深入聊聊LLM-RAG检索增强生成项目中的关键环节数据处理、分块与向量化。如果你已经熟悉RAG的基本概念但卡在“如何把文档变成模型能理解的结构”这一步那么这篇文章就是为你准备的。我会用通俗的语言、可运行的代码带你从原始文档一路走到向量数据库。### 为什么数据处理和分块是RAG的基石RAG的核心思想是“先检索再生成”。LLM大语言模型在回答问题时先从一个外部知识库中检索相关片段然后基于这些片段生成答案。如果知识库里的数据是乱七八糟的文本块或者分块方式不合理检索到的内容就会牛头不对马嘴生成的结果自然拉胯。所以数据处理和分块策略直接决定了RAG系统的上限。简单来说我们需要-数据清洗去除无关噪音比如HTML标签、多余空格、乱码字符。-分块策略决定如何把长文档切成小片段既要保持语义完整又要适配向量模型的输入长度。-向量化把文本片段转换成数值向量让机器能计算相似度。下面我们一步步用代码实现。### 第一步数据清洗——从脏数据到干净文本假设我们有一堆从网站爬下来的Markdown文档里面夹杂着代码块、空行和特殊字符。我们需要一个通用清洗函数。pythonimport redef clean_text(raw_text: str) - str: 清洗原始文本移除多余空格、特殊字符、空行。 参数 raw_text (str): 原始文本 返回 str: 清洗后的文本 # 移除HTML标签如果有 text re.sub(r[^], , raw_text) # 移除多个连续空格 text re.sub(r\s, , text) # 移除首尾空白 text text.strip() # 移除空行通过保留非空行 lines [line for line in text.split(\n) if line.strip()] text \n.join(lines) return text# 示例使用raw_doc 这是 一个测试文档。它包含 多余空格。以及空行。cleaned clean_text(raw_doc)print(cleaned)# 输出这是 一个测试文档。 它包含 多余空格。 以及 空行。清洗后文本变得干净、紧凑为分块打好基础。注意这里我们保留了换行符因为它在后续分块中可能作为自然边界。### 第二步分块策略——文本切割的艺术分块没有银弹常见策略有-固定长度分块按字符或token数切分简单但易割裂语义。-递归分块先按段落切如果段落过长再按句子切最后按固定长度补刀。-语义分块用模型判断语义边界代价高但效果最好。这里我用递归分块演示因为它平衡了效果和效率。我们使用LangChain的RecursiveCharacterTextSplitter但为了理解原理手写一个简化版。pythondef recursive_chunk(text: str, chunk_size: int 500, chunk_overlap: int 50) - list: 递归分块优先按段落两个换行切再按句子句号切最后按字符切。 参数 text (str): 待分块文本 chunk_size (int): 每个块的最大字符数 chunk_overlap (int): 相邻块重叠字符数保持上下文连贯 返回 list: 分块列表 # 分隔符优先级段落 句子 字符 separators [\n\n, \n, 。, , , , , ] def split_by_separator(text, sep): 按指定分隔符切分文本 parts text.split(sep) return [p for p in parts if p.strip()] # 去除空片段 chunks [] current_chunk for sep in separators: if sep : # 最后一级按字符切 while current_chunk: chunks.append(current_chunk[:chunk_size]) current_chunk current_chunk[chunk_size:] break parts split_by_separator(text, sep) for part in parts: if len(current_chunk) len(part) len(sep) chunk_size: current_chunk part sep else: if current_chunk: chunks.append(current_chunk.strip()) # 保留重叠部分 if chunk_overlap 0 and len(current_chunk) chunk_overlap: current_chunk current_chunk[-chunk_overlap:] part sep else: current_chunk part sep if current_chunk: chunks.append(current_chunk.strip()) return chunks# 示例使用sample_text 这是第一段。它包含两个句子。\n\n这是第二段它只有一个长句子但后面有更多内容。\n\n第三段很短。chunks recursive_chunk(sample_text, chunk_size50, chunk_overlap10)for i, chunk in enumerate(chunks): print(f块{i}: {chunk})# 输出块0: 这是第一段。它包含两个句子。# 块1: 这是第二段它只有一个长句子但后面有更多内容。# 块2: 第三段很短。注意chunk_overlap参数很关键它让相邻块有重复内容避免在边界丢失上下文。比如一个句子被分到两个块里重叠部分能保持连贯性。### 第三步向量化——把文本变成数字分块完成后我们需要用嵌入模型Embedding Model把每个块转成向量。常用的有OpenAI的text-embedding-ada-002或开源的BGE系列。这里我们用Sentence-Transformers库它提供预训练模型本地运行免费。pythonfrom sentence_transformers import SentenceTransformer# 加载轻量级中文嵌入模型model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2)def vectorize_chunks(chunks: list) - list: 将文本块列表转换为向量列表。 参数 chunks (list): 文本块列表 返回 list: 向量列表每个向量是numpy数组或列表 # encode方法返回numpy数组每个元素对应对应块的向量 embeddings model.encode(chunks, show_progress_barFalse) return embeddings.tolist() # 转成列表便于存储# 示例使用sample_chunks [今天天气真好, 适合出去散步]vectors vectorize_chunks(sample_chunks)print(f向量维度: {len(vectors[0])}) # 输出向量维度: 384print(f第一个向量前5个值: {vectors[0][:5]}) # 输出第一个向量前5个值: [...] 随机数字得到向量后你可以把它们存到向量数据库如FAISS、Pinecone中。查询时把用户问题也向量化然后计算余弦相似度找到最相关的块。### 性能优化与常见坑-分块大小太小会丢失上下文太大则浪费token。一般512-1024字符中文或200-400token。根据你的嵌入模型输入长度调整。-重叠比例10%-20%的重叠通常够用。太多会增加存储和检索开销。-清洗细节不要过度清洗。比如代码示例中的换行符如果全删了分块时可能割裂代码块。-向量模型选择中文场景推荐使用专门的中文模型如BAAI/bge-large-zh效果更好。### 总结数据处理、分块和向量化是RAG项目中最容易被低估的环节。一个糟糕的分块策略会让最强大的LLM也束手无策而精心设计的流程则能大幅提升检索质量。本文从实战出发带你用Python实现了从清洗到向量化的完整链路。记住没有最好的分块只有最适合你数据的分块。动手尝试不断调参你就能找到那个“黄金分割点”。希望这篇文章能帮你少走弯路。如果有什么疑问欢迎在评论区交流