1. 从“玩具”到“工具”为什么你的Agent需要RAG上一关我们搓了个能聊天的Agent感觉挺酷但聊深了就会发现它像个健忘的“天才”——知识面广但记不住你给的资料。你跟它讨论昨天上传的《2024年Q3产品规划.pdf》它要么胡编乱造要么一脸无辜地说“根据我的训练数据……”。这感觉就像找了个博学的朋友聊天但他对你公司内部的事情一无所知。这种Agent充其量是个“玩具”。要让Agent真正成为能处理你私有知识、解决具体业务问题的“工具”RAG检索增强生成是必须迈过的一道坎。简单说RAG就是给Agent装上一个“外部记忆库”和“实时搜索引擎”。当Agent需要回答问题时它不再仅仅依赖训练时学到的、可能过时的通用知识而是先去你的专属知识库比如公司文档、产品手册、聊天记录里检索最相关的信息片段然后结合这些“新鲜出炉”的上下文再生成精准、可靠的回答。这不仅仅是“查资料”那么简单。一个设计良好的RAG系统决定了你的Agent是“一本正经地胡说八道”还是“有理有据地对答如流”。今天我们就来深入RAG的核心环节文档处理与向量化索引。这是整个RAG流水线的基石也是最容易“埋坑”的地方。很多人以为把文档扔进向量数据库就完事了结果召回的全是无关内容生成答案的质量惨不忍睹。问题往往就出在这最初的一公里。2. 文档接入与清洗给“原材料”做预处理想象一下你要教一个新人熟悉公司业务你不会直接把一堆混乱的邮件、PDF、网页链接和会议记录碎片扔给他说“自己看吧”。你会先整理、归类甚至提炼重点。对RAG系统来说文档接入与清洗就是这一步。2.1 支持多种格式用统一“语言”对话你的知识来源必然是多样的。LangChain等框架提供了丰富的文档加载器Document Loaders这是我们的第一道工具。from langchain_community.document_loaders import PyPDFLoader, TextLoader, UnstructuredMarkdownLoader, WebBaseLoader import os # 示例加载不同格式的文档 documents [] # 1. 加载PDF产品手册 pdf_loader PyPDFLoader(path/to/product_manual.pdf) pdf_docs pdf_loader.load() # 返回Document对象列表 documents.extend(pdf_docs) # 2. 加载纯文本日志文件 text_loader TextLoader(path/to/meeting_notes.txt) text_docs text_loader.load() documents.extend(text_docs) # 3. 加载Markdown项目README md_loader UnstructuredMarkdownLoader(path/to/README.md) md_docs md_loader.load() documents.extend(md_docs) # 4. 加载网页技术博客 web_loader WebBaseLoader([https://example.com/tech-blog]) web_docs web_loader.load() documents.extend(web_docs) print(f总共加载了 {len(documents)} 个文档片段原始。)注意Unstructured系列加载器功能强大能处理复杂格式如PPT、Word但可能需要额外安装系统依赖如pandoc、libreoffice。对于生产环境建议将文档格式标准化如统一转为Markdown或纯文本后再处理能极大减少后续的解析错误。加载后的每个Document对象通常包含page_content文本内容和metadata元数据如来源、页码。元数据至关重要它是在后续回答中引用来源的依据。2.2 文本清洗剔除噪音保留精华原始文档文本往往包含大量对语义理解无益甚至有害的“噪音”。直接向量化这些噪音会污染你的向量空间。清洗操作通常在加载后、切片前进行。我们可以定义一个简单的清洗管道import re from typing import List from langchain.schema import Document def clean_document_text(docs: List[Document]) - List[Document]: 对文档内容进行基础清洗。 cleaned_docs [] for doc in docs: text doc.page_content # 1. 移除多余的换行符和空白字符保留段落间的单个换行 text re.sub(r\n{3,}, \n\n, text) # 将3个以上换行替换为2个 text re.sub(r[ \t], , text) # 合并多个空格/制表符 # 2. 移除常见的无意义字符或模式根据你的文档特点调整 # 例如移除单独的页码标记如“- 1 -” text re.sub(r^\s*[-–—]\s*\d\s*[-–—]\s*$, , text, flagsre.MULTILINE) # 移除URL如果不需要的话 # text re.sub(rhttps?://\S, , text) # 3. 统一字符编码问题如全角字符、乱码 # 这里可以引入更复杂的处理如使用ftfy库修复乱码 # import ftfy # text ftfy.fix_text(text) # 更新文档内容 cleaned_doc Document(page_contenttext.strip(), metadatadoc.metadata) cleaned_docs.append(cleaned_doc) return cleaned_docs # 应用清洗 cleaned_documents clean_document_text(documents) print(f文档清洗完成。)实操心得清洗规则需要根据你的文档集特点“量身定制”。最好先抽样检查原始文本找出共性的噪音模式如公司特定的页眉页脚、水印、扫描OCR产生的特殊错误字符等再编写针对性的正则表达式或处理逻辑。这是一个迭代的过程。3. 文档切片Chunking把握信息的“黄金尺寸”这是RAG系统中最核心、最微妙的环节之一。切片的目标是将长文档分解成大小适中、语义相对完整的片段chunks。切片太大会包含无关信息稀释核心语义导致检索精度下降切片太小会割裂完整的语义单元让模型无法理解上下文。3.1 常见的切片策略与陷阱固定长度重叠切片最常用、最基础的方法。按固定字符数切片并设置一个重叠区overlap以避免在句子中间切断。from langchain.text_splitter import RecursiveCharacterTextSplitter # 使用递归字符文本分割器推荐 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个chunk的最大字符数 chunk_overlap50, # chunk之间的重叠字符数 length_functionlen, # 计算长度的方法 separators[\n\n, \n, 。, , , , , , ] # 分割优先级 ) splits text_splitter.split_documents(cleaned_documents) print(f文档被切分为 {len(splits)} 个片段。)为什么选RecursiveCharacterTextSplitter它尝试按分隔符优先级如双换行、单换行、句号来分割尽可能在自然语义边界处断开比简单的固定长度切片更合理。chunk_size设多少这不是固定的。它需要匹配你使用的嵌入模型Embedding Model的上下文长度以及大语言模型LLM的输入窗口。例如如果你的嵌入模型最大处理512个token那么chunk_size对应的字符数应远小于512中文字符约1-2个token。通常从300-800字符开始实验。overlap设多少通常为chunk_size的10%-20%。重叠是为了防止一个完整的句子或概念被切到两个chunk的边缘导致检索时丢失关键信息。基于语义的切片更高级的方法使用NLP模型如句子分割器在句子边界处切割。这能更好地保证chunk的语义完整性但对技术栈要求更高且处理速度可能较慢。# 示例使用NLTK进行句子分割需先安装nltk并下载punkt # from nltk.tokenize import sent_tokenize # 更适用于英文中文需要专门的分句模型。基于章节/标题的切片对于结构清晰的文档如Markdown、有明确标题的PDF可以按标题层级进行切片。这能产生语义高度凝聚的chunk。from langchain.text_splitter import MarkdownHeaderTextSplitter headers_to_split_on [ (#, Header 1), (##, Header 2), (###, Header 3), ] markdown_splitter MarkdownHeaderTextSplitter(headers_to_split_onheaders_to_split_on) # 适用于Markdown文档能保留标题结构作为元数据。3.2 切片策略的实战选择与调优没有“一招鲜”的切片策略。你需要根据文档类型和查询模式来决定。技术文档/知识库结构清晰概念独立。适合按标题切片或中等尺寸如500字符的递归切片重叠区可适当减小。长篇小说/连贯性文本上下文依赖性强。适合较大尺寸如1000字符的递归切片并需要较大的重叠区如150-200字符以保持情节或论述的连贯。对话记录/会议纪要每条记录相对独立。可以按对话者或时间点进行分割或者使用较小的固定尺寸切片。踩坑实录我曾在一个法律合同分析的Agent项目中最初使用固定长度切片chunk_size400。结果在检索“争议解决条款”时经常只召回条款的前半部分关于仲裁而漏掉了后半部分关于管辖法院因为条款本身超过了400字符。后来改为按“条款标题”如“第十条 争议解决”进行语义切片并适当增大尺寸召回准确率立刻大幅提升。给你的建议永远不要假设一种切片策略对所有文档都有效。在构建索引前务必人工抽查不同文档类型切片后的结果。问自己如果我是用户问这个问题这个chunk能提供完整答案吗相邻的chunk有必要的上下文吗4. 向量化与嵌入Embedding将文本映射到“语义空间”切片后的文本是计算机无法直接理解的。我们需要通过嵌入模型将每一段文本chunk转换成一个高维度的向量一组数字。这个向量的神奇之处在于语义相似的文本其向量在空间中的距离如余弦相似度会很接近。4.1 嵌入模型选型开源 vs. 闭源通用 vs. 领域开源本地模型代表BAAI/bge-small-zh-v1.5,moka-ai/m3e-base,text2vec系列。优点数据隐私有保障无需网络调用成本固定可定制微调。缺点需要本地GPU或CPU资源性能可能低于顶级闭源模型需要自己维护。适用场景对数据安全要求极高、离线环境、预算有限、愿意投入运维精力。from langchain.embeddings import HuggingFaceEmbeddings model_name BAAI/bge-small-zh-v1.5 model_kwargs {device: cpu} # 或 cuda encode_kwargs {normalize_embeddings: True} # 归一化方便计算余弦相似度 embeddings HuggingFaceEmbeddings( model_namemodel_name, model_kwargsmodel_kwargs, encode_kwargsencode_kwargs ) # 测试 text 什么是机器学习 vector embeddings.embed_query(text) print(f向量维度{len(vector)})闭源API模型代表OpenAI的text-embedding-3-small/ada-002 百度的Embedding-V1 阿里的text-embedding-v2。优点开箱即用性能稳定且通常很强尤其是对通用语料无需关心底层基础设施。缺点按调用次数收费有网络延迟数据需传输到厂商服务器需关注合规性。适用场景快速原型验证、对通用语义理解要求高、无严格数据本地化要求。from langchain_openai import OpenAIEmbeddings # 需要设置环境变量 OPENAI_API_KEY embeddings OpenAIEmbeddings(modeltext-embedding-3-small)如何选择如果你的领域非常垂直如医学、法律且有充足的领域文本微调一个开源小模型的效果可能远超通用的闭源大模型。可以先从闭源API开始快速验证效果如果效果满意且成本可控可以持续使用如果对效果、成本或隐私有更高要求再迁移到开源模型。4.2 嵌入过程的优化细节批量处理嵌入生成是计算密集型或网络IO密集型操作。务必使用批量处理来提升效率。# LangChain的向量存储库通常自带批量嵌入功能 # 例如在使用Chroma时add_documents方法内部会调用嵌入模型失败重试与速率限制调用API时必须实现重试逻辑和遵守速率限制否则极易因网络波动或超限导致索引构建失败。from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def safe_embed_query(text): return embeddings.embed_query(text)向量归一化大多数相似度计算如余弦相似度要求向量是归一化的长度为1。确保你的嵌入模型输出或你在存入数据库前进行了归一化。上文HuggingFaceEmbeddings中的normalize_embeddingsTrue就是做这个。5. 向量数据库选型与索引构建存放与检索我们的“记忆”生成向量后我们需要一个专门的数据来高效存储和检索它们这就是向量数据库。5.1 主流向量数据库对比与选型考虑特性ChromaFAISS(Facebook)Milvus/Zilliz CloudPGVector(PostgreSQL扩展)核心定位轻量级开发者友好内置嵌入功能高性能向量检索库需自行管理数据持久化云原生分布式功能全面的数据库基于PostgreSQLSQL向量二合一部署复杂度极简可内存/本地文件单机低库但持久化方案需自研较高分布式架构有运维成本中等需有PostgreSQL基础持久化支持本地目录不支持需自己保存向量到磁盘加载时重建索引支持分布式存储支持依托PG查询功能基础相似度搜索支持元数据过滤极速相似度搜索算法丰富丰富混合搜索、标量过滤、时间旅行等支持向量搜索可与SQL联查生产就绪适合原型、中小项目适合研究、对检索速度要求极高的嵌入应用适合大规模、高并发生产环境适合已用PG、需强事务和复杂查询的场景社区/生态活跃与LangChain集成极好非常活跃AI研究领域标准活跃商业支持完善依托庞大的PG生态选型思考路径你的场景是什么是快速验证想法的原型还是即将上线的生产系统数据量和QPS每秒查询量预估是多少百万级以下、低并发Chroma、FAISS够用千万级以上、高并发考虑Milvus。你的技术栈是什么团队熟悉Python和简单架构Chroma是快乐首选。团队有运维能力且需要极致性能和控制力FAISS自定义持久化是利器。系统已基于PostgreSQL引入PGVector是平滑扩展。需要复杂的元数据过滤吗如果需要“找到去年发布的、属于A部门的、关于财务的文档”Chroma、Milvus、PGVector的元数据过滤会更方便。对于大多数从0到1的Agent项目我的建议是从Chroma开始。它让你在5分钟内跑通整个流程把精力集中在RAG逻辑和效果调优上而不是数据库部署上。5.2 使用Chroma构建你的第一个向量索引让我们用Chroma把前面所有步骤串起来。import chromadb from chromadb.config import Settings from langchain_chroma import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import PyPDFLoader from langchain.embeddings import HuggingFaceEmbeddings import os # 1. 准备嵌入模型 embeddings HuggingFaceEmbeddings(model_nameBAAI/bge-small-zh-v1.5) # 2. 加载并处理文档以单个PDF为例 loader PyPDFLoader(path/to/your/document.pdf) raw_docs loader.load() # 3. 文本切片 text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) all_splits text_splitter.split_documents(raw_docs) print(f共生成 {len(all_splits)} 个文本片段。) # 4. 初始化Chroma向量存储并构建索引 # persist_directory 指定持久化目录否则数据仅在内存中 persist_directory ./chroma_db # 方法一使用 LangChain 的 Chroma 包装器推荐更简洁 vectorstore Chroma.from_documents( documentsall_splits, # 切片后的文档列表 embeddingembeddings, # 嵌入模型 persist_directorypersist_directory, # 持久化路径 collection_namemy_knowledge_base # 集合名称 ) print(向量索引构建并持久化完成) # 现在可以进行检索测试 query 文档中提到了哪些主要功能 docs vectorstore.similarity_search(query, k3) # 检索最相似的3个片段 print(f针对查询 {query} 检索到 {len(docs)} 个相关片段) for i, doc in enumerate(docs): print(f\n--- 片段 {i1} (相关性得分估算) ---) print(f内容预览{doc.page_content[:200]}...) print(f元数据{doc.metadata})5.3 索引构建的进阶考量增量更新知识库不是一成不变的。Chroma支持增量添加文档。关键是要处理好去重问题。简单的做法是在元数据中记录文档的唯一ID如MD5值添加前先检查是否存在。# 假设 new_docs 是新加载并切片后的文档 # 计算每个doc的ID简单示例生产环境需更严谨 for doc in new_docs: doc.metadata[doc_id] hash(doc.page_content) # 添加到已有集合 vectorstore.add_documents(new_docs) # Chroma 内部会根据文档内容去重吗不会。需要你自己管理。元数据策略元数据是精准过滤的钥匙。除了自动生成的source、page你应该主动添加业务相关的元数据如document_type合同/邮件/手册、department、author、create_date。# 在加载或切片时丰富元数据 for split in all_splits: split.metadata[document_type] 产品手册 split.metadata[year] 2024 split.metadata[section] infer_section(split.page_content) # 自定义函数推断章节这样后续你可以进行如下检索# 伪代码检索2024年产品手册中关于“安装”的内容 docs vectorstore.similarity_search( query安装步骤, k5, filter{document_type: 产品手册, year: 2024} # 元数据过滤 )索引参数对于Chroma默认的索引方法通常是HNSW已适合大部分场景。如果数据量极大100万可以关注集合的hnsw:space距离度量余弦相似度用cosine等配置。6. 避坑指南向量索引构建中的典型问题检索结果完全不相关可能原因1嵌入模型不匹配。你用英文模型处理了中文文档。检查用embeddings.embed_query(“测试”)看向量维度是否正常并与已知正确的向量简单计算相似度。可能原因2Chunk尺寸过大或过小。检查打印出被检索到的chunk全文看它是否真的包含了查询答案。可能原因3文本清洗过度或不足。噪音污染了语义或关键信息被洗掉。检查对比清洗前后的文本特别是被检索到的chunk。构建索引速度极慢可能原因1未使用批量嵌入。如果是API模型网络IO是瓶颈如果是本地模型频繁加载模型或单条推理是瓶颈。确保使用框架提供的批量方法如from_documents。可能原因2硬件限制。本地小模型在CPU上跑万级文档就会很慢。考虑使用性能更好的模型如bge-small比bge-large快很多或租用GPU实例或换用API模型。内存/磁盘占用过大可能原因1Chunk尺寸太小导致数量爆炸。10万字的文档按100字符切就有1000个chunk每个chunk生成384维的向量float32仅向量存储就约1000 * 384 * 4 bytes ≈ 1.5MB加上文本和索引体积不小。优化在不损失语义的前提下适当增大chunk_size。可能原因2保留了原始文本的过多副本。检查Chroma持久化目录的大小。无法实现准确的元数据过滤可能原因元数据格式或查询语法错误。Chroma的过滤语法是特定的。示例filter{year: {$eq: 2024}}。务必查阅所用向量数据库的过滤文档。构建一个高质量的向量索引更像是一门实验科学而非纯工程。它需要你不断地观察、假设、实验和调整调整切片策略、尝试不同的嵌入模型、设计有效的元数据。这个过程没有银弹但每一次调试都让你更了解你的数据和你的需求。在下一关我们将进入RAG的下半场召回与重排序策略。当你的Agent从向量库中检索出多个相关片段后如何从中挑选出最相关、最精华的部分组合成最终的提示词Prompt送给LLM这里面同样充满了学问和技巧。我们会探讨如何通过重排序Re-ranking模型、上下文压缩等技术让Agent的回答更加精准、凝练真正发挥出“增强生成”的威力。