RAG客服系统：文档处理与向量化

可视化演示

展示文档解析分块向量化管道、Embedding模型选择与Top-N检索上下文拼接，理解RAG客服系统从文档到可检索知识库的核心环节。

文档解析——食材的验收与清洗

餐厅后厨每天要收各种包装的食材：真空袋、纸箱、泡沫箱。每一件都有“包装”要拆，有“泥巴”要洗，有“烂叶”要摘。在RAG系统里，这些“包装”就是PDF、Word、HTML等文件格式，“泥巴烂叶”就是页眉、页脚、导航栏、广告等噪音。如果验收不严，脏东西混进来，会影响后续所有环节——就像一块带沙的肉，无论厨艺多好，客人都会吃到沙子。

文档解析（Parsing）的任务，正是把这些非结构化文件转成干净的纯文本。具体包括三大任务。格式识别是判断食材的包装类型：PDF分为文本型（可直接提取）和扫描型（需OCR识别图片中的文字），Word内部有段落与表格结构，HTML有标签嵌套。内容抽取是取出能吃的部分：表格转为Markdown或CSV保留结构语义，多栏布局按阅读顺序拼接，图片文字用OCR补充。数据清洗是扔掉烂叶：去掉页眉页脚、重复空行、特殊控制字符，只保留业务需要的信息。

在实际项目中，一份PDF可能混合了文字层和扫描图，需要逐页判断处理策略；HTML则要先剥离script和style标签，再按正文区域抽取。清洗的严格程度直接决定后续向量化的质量——一段混入“© 2024 版权所有 京ICP备XXXXXXXX号”的文本，在检索时可能被当成有效内容返回给用户。

为了量化解析效果，我们可以用“有效信号占比”来评估：

解析完成，食材验收与清洗结束。如果这段没做好，后面所有环节都会被拖累——这正是很多RAG项目“检索结果乱七八糟”的第一大根因。

文本切分——大块食材切适口小片

一块五斤重的整牛肉直接下锅没法吃。大厨会先切片，太厚咬不动，太薄容易碎。更有经验的师傅切的时候会让相邻两片之间留一点重叠，保证肉的纹理不断开。在RAG系统里，一段5000字的文档包含多个主题，一个向量根本“装不下”，所以必须切成chunk（文本块）。

文本切分由三个要素决定。切分粒度：按固定字符数切（如256或512）实现简单，但容易切断句子；按句子或段落切语义完整，但chunk长度不齐。重叠窗口：相邻chunk保留10%-20%的重叠字符，确保被切断位置在另一侧仍有上下文，避免“断章取义”。语义边界：优先在句子结束、段落结束处下刀，遇到Markdown标题或PDF章节标志时，把标题和正文放进同一个chunk，保持内容归属清晰。

切分的效果直接决定检索召回的质量。如果一刀切断了关键语义——就像把“不要加香菜”切成“不要加”和“香菜”——客户明明说了不要，系统却检索出了加香菜的做法。重叠窗口正是为了缓解这个问题而存在，它相当于给每片肉留了一段“重复的纹理”。

给定文档长度、块大小和重叠长度，chunk数量可以这样计算：

切得好不好，直接决定“下锅”后的味道。好的切分策略能让检索精度提升近一倍，这也是为什么资深工程师会花大量时间测试不同的chunk大小和overlap组合。

向量化模型——给食材贴上味道标签

走进一间黑漆漆的冷库，你看不清每份食材。聪明的仓库管理员会给每个盒子贴二维码，扫码就知道这是“咸鲜五花肉”还是“甜香提拉米苏”。向量化做的就是这件事：把每个文本块变成一个几百维的“二维码”——一串数字向量，让语义相近的文本在向量空间中彼此靠近。

向量化（Embedding）的核心是让计算机理解“语义距离”。“怎么退钱”和“退款流程如何办理”用词完全不同，但语义一致，对应的向量就应该很近。深度学习模型（如BERT系列）通过在海量语料上的预训练，学到了这种语义映射：Transformer编码器将文本压缩成一个固定长度的向量，把“语义”分散编码到每个维度里。

在实际选型中，有几个关键考量。Embedding原理：模型输出的向量维度常见有384、768、1024、1536等，维度越高语义区分度越高，但存储与计算开销也越大。模型选型：通用领域常用BGE、text-embedding-ada-002等开源/商业模型；垂直领域（医疗、法律、客服）建议在领域数据上微调或选择领域特化模型，否则会出现“报修”被理解成“保修”的尴尬。维度权衡：10万条768维向量大约占用3GB内存，如果资源有限，可以考虑384维或降维压缩。

向量之间的相似程度，最常使用余弦相似度来衡量：

二维码打得准不准，决定了冷库取菜快不快。如果你用的Embedding模型是“通用型”，它扫出来的标签可能与你的行业“方言”对不上。选择或微调领域模型，就是让标签系统学会你的行业语言——这是向量化效果提升最立竿见影的一步。

向量存储——井井有条的冷库与智能货架

冷库里光有标签还不够，还得有货架。如果食材随便堆在地上，客人点菜时配菜师傅要满仓库翻找。有了按类别分区、按热度排序的智能货架，就能快速锁定位置。向量数据库就是这间智能冷库——它不仅“存”向量，更负责“找”向量。

生产环境中，向量数据库（如Milvus、Qdrant、FAISS）主要做三件事。存向量：把每一条文本块的向量连同元数据（文档ID、章节、时间戳）一起存入。建索引：暴力扫描在百万级数据下太慢，所以引入ANN（近似最近邻）索引，其中HNSW（层级小世界图）兼顾召回率与速度，是当前主流；IVF（倒排文件）则适合超大规模场景。算相似度：常用余弦相似度、内积和欧氏距离，归一化后余弦与内积等价。

单靠语义向量并不万能。用户问“订单号20240815001的退款状态”，这种精确匹配场景下，关键词检索（BM25/倒排索引）准确率更高。因此生产系统普遍采用混合召回：语义向量通道负责泛化理解，关键词通道负责精确命中，同时结合元数据过滤（如先限定在“退款政策”章节下检索），把搜索范围缩小一个数量级。

混合召回的核心是给两个通道的分数量一个权重：

冷库布局决定了出菜速度。HNSW索引结构选得好，百万级向量几十毫秒就能返回结果；混合召回则像一位经验老到的配菜师傅——“你给我订单号，我按编号找；你描述口感，我按味道标签找”，两手都硬才是生产级方案。

检索增强——配菜师傅的高阶智慧

客人一句“上次那个有点辣但也不能太辣的菜再来一份”，普通服务员听得一头雾水。经验丰富的配菜师傅会先确认：“您说的是水煮鱼吧？微辣，去香菜，多加豆芽？”然后迅速下单。RAG的检索增强环节正是如此：理解Query、优化Query、精排结果，把最相关的内容送到大模型手里。

Query改写是第一步。用户真实输入往往口语化、有指代：“那个怎么弄？”“我需要它尽快到。”如果直接拿去检索，效果很差。改写组件会把query补全为信息更充分的检索语句，如“如何申请退款”；也可以利用多轮对话历史，把“它”替换成之前提到的商品名。改写可以用大模型完成，也可以用规则+模板降低延迟。

重排序（Rerank）是第二步。初召回可能拿到50条候选，但大模型上下文窗口最多放5~10条。Rerank模型（如CrossEncoder）将query与每条候选拼接后精细打分，比Embedding的粗粒度相似度更准。一个典型的线上配置是：向量检索召回50条 → Rerank精排取Top-5 → 拼接成上下文送大模型。

结果融合是第三步。多路召回（向量、关键词、元数据）的结果需要合并去重。这里常用RRF（Reciprocal Rank Fusion），它不看绝对分数大小，只看排名，因此对不同通道的分数尺度天然鲁棒：

配菜师傅不会把客人的模糊需求原封不动转给后厨，而是先确认、再挑菜、最后按最合适的顺序递给大厨。Query改写、重排序、结果融合，就是这套“确认—挑菜—排队”的流程，让大模型拿到的永远是“精挑细选过的食材”。

系统优化——厨房的持续运营

开业三个月后，菜谱更新了，新食材不断到货，有些食材过期要下架；午市高峰顾客排队，厨师忙不过来。餐厅若不优化运营流程，好菜也会上得慢。RAG系统上线后同样要面对数据时效、查询成本和资源效率三大挑战。

增量更新解决数据时效问题。当文档新增、修改或删除时，没必要全量重建向量库——只对变化的文档重新解析→切分→向量化，更新对应条目即可。实现上可以用文档哈希（如MD5）判断内容是否变化；加上版本号字段，查询时按“最新版本”过滤，避免新旧内容混淆。产品手册每周更新，增量更新让新知识几秒内生效，而全量重建可能要跑几个小时。

缓存策略解决查询成本问题。热门问题（“怎么退货”“运费谁出”）命中率极高，直接把答案缓存起来，或者至少缓存检索结果，可以大幅降低大模型调用费用和时延。常用LRU淘汰策略，把不常问的旧缓存让位给新热点。性能调优解决资源效率问题。向量维度可以用PCA降维，存储精度可以从FP32降到INT8，HNSW的M和efConstruction参数也要在召回率与延迟之间调平衡。

缓存策略的效果用命中率来度量：

持续优化的厨房永远不会“开一次张就关门”。增量更新像每日补货，缓存像招牌菜提前备好，性能调优像高峰期加派人手。RAG系统的生命力，就在这一天天细水长流的运营中积累出来。

AI 应用场景

Python 演示

import mathimport refrom collections import Counter  class DocProcessor:    """RAG客服系统文档处理与向量化完整流程演示"""     def __init__(self, chunk_size=40, overlap=8):        self.chunk_size = chunk_size        self.overlap = overlap        self.vocab = {}        self.chunk_vectors = []        self.chunk_texts = []     def parse_document(self, raw_text):        # 步骤一：文档解析,去掉HTML标签和多余空白        text = re.sub(r'<[^>]+>', '', raw_text)        text = re.sub(r'\s+', ' ', text).strip()        return text     def chunk_text(self, text):        # 步骤二：文本切分,固定大小加重叠窗口        chunks = []        step = self.chunk_size - self.overlap        for i in range(0, len(text), step):            chunk = text[i:i + self.chunk_size]            if len(chunk) >= self.chunk_size // 2:                chunks.append(chunk)        return chunks     def build_vocab(self, chunks):        # 构建词表,给每个词一个唯一编号        words = set()        for chunk in chunks:            words.update(re.findall(r'[\w\u4e00-\u9fff]+', chunk.lower()))        self.vocab = {w: idx for idx, w in enumerate(words)}        return self.vocab     def vectorize(self, chunks):        # 步骤三：向量化,词袋模型加归一化        vectors = []        for chunk in chunks:            vec = [0.0] * len(self.vocab)            counter = Counter(re.findall(r'[\w\u4e00-\u9fff]+', chunk.lower()))            for w, cnt in counter.items():                if w in self.vocab:                    vec[self.vocab[w]] = cnt            norm = math.sqrt(sum(v * v for v in vec))            if norm > 0:                vec = [v / norm for v in vec]            vectors.append(vec)        return vectors     def retrieve(self, query, top_k=3):        # 步骤五：检索,查询向量化加余弦相似度        q_vec = [0.0] * len(self.vocab)        for w in set(re.findall(r'[\w\u4e00-\u9fff]+', query.lower())):            if w in self.vocab:                q_vec[self.vocab[w]] = 1.0        norm = math.sqrt(sum(v * v for v in q_vec))        if norm > 0:            q_vec = [v / norm for v in q_vec]         scores = []        for vec in self.chunk_vectors:            sim = sum(a * b for a, b in zip(q_vec, vec))            scores.append(sim)         top_indices = sorted(range(len(scores)),                             key=lambda i: scores[i], reverse=True)[:top_k]        return [(idx, scores[idx]) for idx in top_indices]     def process(self, raw_text, query):        # 完整流程：解析 -> 切分 -> 向量化 -> 存储 -> 检索        text = self.parse_document(raw_text)        chunks = self.chunk_text(text)        self.build_vocab(chunks)        self.chunk_texts = chunks        self.chunk_vectors = self.vectorize(chunks)        return self.retrieve(query)  if __name__ == "__main__":    doc = (        "<html><body>"        "尊敬的客户，欢迎使用智能客服系统。本系统提供退款、换货、物流查询服务。"        "如需退款，请在订单页点击退款按钮，填写原因并提交。"        "物流查询请提供订单号，系统将自动匹配物流状态。"        "换货申请需在签收后7天内完成，逾期无法受理。"        "</body></html>"    )    processor = DocProcessor(chunk_size=30, overlap=5)    query = "如何申请退款"    results = processor.process(doc, query)    print("查询：", query)    for idx, score in results:        print(f"[相似度 {score:.4f}] {processor.chunk_texts[idx][:40]}...")

1. text = re.sub(r'<[^>]+>', '', raw_text)用正则表达式剥离HTML标签，对应文档解析中的数据清洗
2. step = self.chunk_size - self.overlap计算切分步长，每次只向前移动“新增部分”，重叠处自然覆盖
3. self.vocab = {w: idx for idx, w in enumerate(words)}构建词表，给每个词一个编号，向量中该位置的数字代表这个词的权重
4. vec = [0.0] * len(self.vocab)创建一个与词表等长的全零向量，等待填充词频
5. counter = Counter(re.findall(r'[\w\u4e00-\u9fff]+', chunk.lower()))统计当前文本块中每个词的出现次数，中文按连续汉字片段切词
6. norm = math.sqrt(sum(v * v for v in vec))计算向量的模长，用于归一化，让不同长度文本的向量可比
7. sim = sum(a * b for a, b in zip(q_vec, vec))计算余弦相似度分子（向量已归一化，省略了分母）
8. top_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:top_k]按分数从高到低排序，取前top_k个结果，这就是最原始的“召回”

常见误区

面试题