发布于 由admin

复杂结构化文档进入 RAG 训练集后导致召回偏移的工程化清洗策略

复杂结构化文档RAG训练集清洗策略：避免召回偏移

大家好，今天我们来探讨一个在构建基于复杂结构化文档的RAG（Retrieval-Augmented Generation）系统时经常遇到的问题：复杂结构化文档进入训练集后导致的召回偏移。这个问题直接影响RAG系统的性能，轻则召回结果不相关，重则导致生成内容错误。

RAG系统依赖于一个有效的检索机制，从大量文档中找到与用户查询最相关的片段，并将这些片段作为上下文提供给生成模型。如果训练集中的文档结构复杂，例如包含表格、列表、嵌套段落等，未经处理直接用于索引，就会导致检索系统难以准确理解文档的语义，从而产生召回偏移。

本文将以讲座的形式，深入分析复杂结构化文档导致召回偏移的原因，并提出一系列工程化的清洗策略，帮助大家构建更可靠的RAG系统。

一、召回偏移的原因分析

在深入清洗策略之前，我们需要理解为什么复杂结构化文档会导致召回偏移。主要原因有以下几个方面：

  1. 语义理解困难： 传统的文本检索方法（如基于关键词匹配的BM25、基于向量相似度的 embeddings 等）在处理结构化文档时，难以捕捉文档内部的语义关系。例如，表格中的单元格与标题的关联、列表项之间的层次关系等，都会被忽略。

  2. 信息冗余与噪声： 结构化文档中常常包含大量的重复信息、格式化的标记，以及与核心内容无关的辅助信息（如页眉页脚、版权声明等）。这些信息会干扰检索系统的判断，降低召回的准确率。

  3. 索引粒度不合理： 如果将整个文档作为一个索引单元，会导致检索结果过于宽泛，包含大量无关信息。反之，如果将文档拆分成过小的单元（如每个句子），又可能丢失上下文信息，导致语义不完整。

  4. Embedding空间错位： 如果使用预训练的语言模型生成文档片段的 embeddings，而这些片段的结构与预训练数据分布差异较大，就会导致 embeddings 空间错位，从而影响相似度计算的准确性。例如，预训练模型可能不擅长处理表格数据，因此生成的表格片段 embeddings 质量不高。

二、工程化清洗策略：分而治之

针对以上问题，我们需要采取一系列工程化的清洗策略，对复杂结构化文档进行预处理，提高检索系统的召回准确率。整体思路是分而治之，将文档拆解成更易于理解和处理的单元，并针对不同类型的结构化元素，采用不同的清洗方法。

  1. 文档解析与结构化表示：

    首先，我们需要将原始文档解析成结构化的表示形式，例如使用 BeautifulSoup 或 lxml 解析 HTML 文档，使用 pdfminer 或 PyPDF2 解析 PDF 文档，使用 python-docx 解析 Word 文档。解析完成后，我们可以得到文档的层次结构、文本内容、表格、列表等信息。

    from bs4 import BeautifulSoup
import pdfminer.high_level
import docx

def parse_html(html_file):
    with open(html_file, 'r', encoding='utf-8') as f:
        soup = BeautifulSoup(f, 'html.parser')
    return soup

def parse_pdf(pdf_file):
    text = pdfminer.high_level.extract_text(pdf_file)
    return text

def parse_docx(docx_file):
    doc = docx.Document(docx_file)
    text = 'n'.join([paragraph.text for paragraph in doc.paragraphs])
    return text

# 示例
# html_content = parse_html("example.html")
# pdf_content = parse_pdf("example.pdf")
# docx_content = parse_docx("example.docx")

  2. 文本清洗与标准化：

    对解析后的文本内容进行清洗和标准化，包括：

    • 去除 HTML 标签、XML 标签等格式化标记。
    • 去除特殊字符、控制字符、多余的空格和换行符。
    • 转换大小写、统一编码格式。
    • 进行词干提取或词形还原，将单词转换为其基本形式。
    import re
import nltk
from nltk.stem import PorterStemmer
from nltk.stem import WordNetLemmatizer
from nltk.corpus import stopwords

nltk.download('stopwords')
nltk.download('wordnet')

def clean_text(text):
    # 去除 HTML 标签
    text = re.sub(r'<[^>]+>', '', text)
    # 去除特殊字符和控制字符
    text = re.sub(r'[^a-zA-Z0-9s]', '', text)
    # 去除多余空格
    text = re.sub(r's+', ' ', text).strip()
    # 转换为小写
    text = text.lower()
    return text

def standardize_text(text):
    # 停用词去除
    stop_words = set(stopwords.words('english'))
    words = text.split()
    words = [word for word in words if word not in stop_words]

    # 词干提取
    stemmer = PorterStemmer()
    stemmed_words = [stemmer.stem(word) for word in words]

    # 或者 词形还原
    # lemmatizer = WordNetLemmatizer()
    # lemmatized_words = [lemmatizer.lemmatize(word) for word in words]

    return ' '.join(stemmed_words) # 或 ' '.join(lemmatized_words)

# 示例
# cleaned_text = clean_text("<h1>This is a sample text with <b>HTML</b> tags.</h1>")
# standardized_text = standardize_text(cleaned_text)

  3. 表格处理：

    表格是结构化文档中常见的元素，处理表格需要特别注意以下几点：

    • 识别表格结构： 使用 pandas 等库识别表格的行、列、标题等信息。
    • 填充缺失值： 对表格中的缺失值进行填充，可以使用平均值、中位数、众数等方法。
    • 提取表格元数据： 提取表格的标题、描述等信息，作为表格内容的补充。
    • 将表格转换为文本： 将表格转换为文本形式，方便检索系统处理。常用的方法有：
      • 线性化： 将表格的每一行或每一列转换为一个字符串。
      • 关系抽取： 提取表格中单元格之间的关系，例如 "单元格 A 包含 属性 B 的值 C"。
    import pandas as pd

def process_table(table):
    # 将 BeautifulSoup 的 table 对象转换为 pandas DataFrame
    df = pd.read_html(str(table))[0]

    # 填充缺失值
    df = df.fillna('')

    # 将表格转换为文本 (线性化)
    text = ''
    for index, row in df.iterrows():
        text += ' '.join(row.astype(str).tolist()) + 'n'

    return text

# 示例 (需要先用 BeautifulSoup 解析 HTML)
# soup = BeautifulSoup(html_content, 'html.parser')
# tables = soup.find_all('table')
# for table in tables:
#     table_text = process_table(table)
#     print(table_text)

    更高级的表格处理：关系抽取示例

    def extract_table_relations(df):
    """
    从 pandas DataFrame 中提取表格单元格之间的关系。
    返回一个包含关系描述的列表。
    """
    relations = []
    header = df.columns.tolist()

    for index, row in df.iterrows():
        for i, value in enumerate(row):
            relation = f"The {header[i]} is {value}"
            relations.append(relation)

    return relations

# 示例
# relations = extract_table_relations(df)
# for relation in relations:
#     print(relation)

  4. 列表处理：

    列表也是结构化文档中常见的元素，处理列表需要注意以下几点：

    • 识别列表结构： 识别列表的类型（有序列表、无序列表）、层级关系。
    • 提取列表项内容： 提取每个列表项的文本内容。
    • 保留列表上下文： 将列表的标题、描述等信息与列表项内容关联起来。
    def process_list(list_element):
    """
    处理 HTML 列表元素 (<ul> 或 <ol>)。
    返回一个包含列表项文本的字符串。
    """
    list_items = list_element.find_all('li')
    text = ''
    for item in list_items:
        text += item.text.strip() + 'n'
    return text

# 示例
# soup = BeautifulSoup(html_content, 'html.parser')
# lists = soup.find_all(['ul', 'ol'])
# for list_element in lists:
#     list_text = process_list(list_element)
#     print(list_text)

  5. 文档分块与索引：

    在完成文本清洗和结构化元素处理后，我们需要将文档分割成更小的块，作为索引单元。分块策略的选择至关重要，它直接影响检索系统的召回准确率。常用的分块策略有：

    • 固定大小分块： 将文档按照固定的字符数或单词数进行分割。
    • 基于句子分块： 将文档按照句子进行分割。
    • 基于段落分块： 将文档按照段落进行分割。
    • 语义分块： 使用语义分割模型（例如 SentenceTransformers）将文档分割成语义相关的块。

    选择合适的分块策略需要根据文档的特点和应用场景进行权衡。一般来说，对于结构复杂的文档，语义分块的效果更好，但计算成本也更高。

    from sentence_transformers import SentenceTransformer, util

def semantic_chunking(text, model_name='all-MiniLM-L6-v2', chunk_size=500):
    """
    使用 SentenceTransformers 进行语义分块。
    返回一个包含文本块的列表。
    """
    model = SentenceTransformer(model_name)
    sentences = nltk.sent_tokenize(text)
    chunks = []
    current_chunk = ""

    for sentence in sentences:
        if len(current_chunk) + len(sentence) + 1 <= chunk_size:
            current_chunk += sentence + " "
        else:
            chunks.append(current_chunk.strip())
            current_chunk = sentence + " "

    if current_chunk:
        chunks.append(current_chunk.strip())

    return chunks

# 示例
# chunks = semantic_chunking(cleaned_text)
# for chunk in chunks:
#     print(chunk)

    索引建立：

    在完成分块后，我们需要将文本块及其元数据（例如文档 ID、标题、段落号等）存储到向量数据库中，并建立索引。常用的向量数据库有：

    • FAISS: Facebook AI Similarity Search，高性能的向量相似度搜索库。
    • Annoy: Approximate Nearest Neighbors Oh Yeah， Spotify 开源的近似最近邻搜索库。
    • Milvus: 开源的向量数据库，支持多种索引类型和查询方式。
    • Pinecone: 云原生的向量数据库，提供高可用性和可扩展性。
    from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

def create_index(chunks, model_name='all-MiniLM-L6-v2', index_path='my_index.faiss'):
    """
    创建 FAISS 索引。
    """
    model = SentenceTransformer(model_name)
    embeddings = model.encode(chunks)
    dimension = embeddings.shape[1]
    index = faiss.IndexFlatL2(dimension)  # L2距离
    index.add(embeddings)
    faiss.write_index(index, index_path)
    return index_path

def load_index(index_path):
    """
    加载 FAISS 索引。
    """
    index = faiss.read_index(index_path)
    return index

def search_index(query, index, model_name='all-MiniLM-L6-v2', top_k=5):
    """
    在 FAISS 索引中搜索与查询最相关的文本块。
    """
    model = SentenceTransformer(model_name)
    query_embedding = model.encode(query)
    query_embedding = np.expand_dims(query_embedding, axis=0) # Reshape for FAISS
    distances, indices = index.search(query_embedding, top_k)
    return distances, indices

# 示例
# index_path = create_index(chunks)
# index = load_index(index_path)
# distances, indices = search_index("What is the capital of France?", index)
# print(distances, indices)

  6. 元数据增强：

    为了提高检索的准确率，我们可以为每个文本块添加元数据，例如：

    • 文档标题： 文本块所属的文档标题。
    • 段落标题： 文本块所属的段落标题。
    • 关键词： 文本块的关键词。
    • 实体： 文本块中包含的实体。
    • 文档结构信息： 文本块在文档中的位置、层级关系等。

    这些元数据可以帮助检索系统更好地理解文本块的语义，从而提高召回的准确率。例如，我们可以使用关键词或实体作为过滤条件，缩小检索范围。

  7. 评估与优化：

    完成以上步骤后，我们需要对清洗后的训练集进行评估，并根据评估结果进行优化。常用的评估指标有：

    • 召回率： 检索系统返回的相关文档占所有相关文档的比例。
    • 准确率： 检索系统返回的文档中，相关文档占所有返回文档的比例。
    • F1 值： 召回率和准确率的调和平均值。

    如果评估结果不理想，我们需要回过头来检查清洗策略的各个环节，例如：

    • 文档解析是否正确？
    • 文本清洗是否彻底？
    • 分块策略是否合理？
    • 元数据是否有效？

    通过不断迭代和优化，我们可以构建一个高质量的 RAG 训练集，提高 RAG 系统的性能。

三、代码示例：一个完整的流程

下面是一个完整的代码示例，演示了如何使用上述策略清洗一个 HTML 文档，并建立 FAISS 索引：

from bs4 import BeautifulSoup
import re
import nltk
from nltk.stem import PorterStemmer
from nltk.corpus import stopwords
from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

nltk.download('stopwords')
nltk.download('wordnet')
nltk.download('punkt') # 下载 sent_tokenize 所需资源

def parse_html(html_file):
    with open(html_file, 'r', encoding='utf-8') as f:
        soup = BeautifulSoup(f, 'html.parser')
    return soup

def clean_text(text):
    text = re.sub(r'<[^>]+>', '', text)
    text = re.sub(r'[^a-zA-Z0-9s]', '', text)
    text = re.sub(r's+', ' ', text).strip()
    text = text.lower()
    return text

def standardize_text(text):
    stop_words = set(stopwords.words('english'))
    words = text.split()
    words = [word for word in words if word not in stop_words]

    stemmer = PorterStemmer()
    stemmed_words = [stemmer.stem(word) for word in words]

    return ' '.join(stemmed_words)

def process_table(table):
    df = pd.read_html(str(table))[0]
    df = df.fillna('')
    text = ''
    for index, row in df.iterrows():
        text += ' '.join(row.astype(str).tolist()) + 'n'
    return text

def process_list(list_element):
    list_items = list_element.find_all('li')
    text = ''
    for item in list_items:
        text += item.text.strip() + 'n'
    return text

def semantic_chunking(text, model_name='all-MiniLM-L6-v2', chunk_size=500):
    model = SentenceTransformer(model_name)
    sentences = nltk.sent_tokenize(text)
    chunks = []
    current_chunk = ""

    for sentence in sentences:
        if len(current_chunk) + len(sentence) + 1 <= chunk_size:
            current_chunk += sentence + " "
        else:
            chunks.append(current_chunk.strip())
            current_chunk = sentence + " "

    if current_chunk:
        chunks.append(current_chunk.strip())

    return chunks

def create_index(chunks, model_name='all-MiniLM-L6-v2', index_path='my_index.faiss'):
    model = SentenceTransformer(model_name)
    embeddings = model.encode(chunks)
    dimension = embeddings.shape[1]
    index = faiss.IndexFlatL2(dimension)
    index.add(embeddings)
    faiss.write_index(index, index_path)
    return index_path

def load_index(index_path):
    index = faiss.read_index(index_path)
    return index

def search_index(query, index, model_name='all-MiniLM-L6-v2', top_k=5):
    model = SentenceTransformer(model_name)
    query_embedding = model.encode(query)
    query_embedding = np.expand_dims(query_embedding, axis=0)
    distances, indices = index.search(query_embedding, top_k)
    return distances, indices

# 假设我们有一个名为 example.html 的 HTML 文件
if __name__ == '__main__':
    html_file = "example.html"  # 替换为你的 HTML 文件名

    # 1. 解析 HTML 文档
    soup = parse_html(html_file)

    # 2. 提取文本内容、表格和列表
    text_content = clean_text(soup.get_text())
    tables = soup.find_all('table')
    lists = soup.find_all(['ul', 'ol'])

    # 3. 处理表格和列表
    table_texts = [process_table(table) for table in tables]
    list_texts = [process_list(list_element) for list_element in lists]

    # 4. 合并所有文本内容
    all_text = text_content + 'n'.join(table_texts) + 'n'.join(list_texts)

    # 5. 标准化文本
    standardized_text = standardize_text(all_text)

    # 6. 语义分块
    chunks = semantic_chunking(standardized_text)

    # 7. 创建 FAISS 索引
    index_path = create_index(chunks)

    # 8. 加载 FAISS 索引
    index = load_index(index_path)

    # 9. 搜索索引
    query = "What are the key features?"
    distances, indices = search_index(query, index)

    # 10. 打印搜索结果
    print("Query:", query)
    for i, index_val in enumerate(indices[0]):
        print(f"Rank {i+1}: Distance = {distances[0][i]}, Chunk = {chunks[index_val]}")

# 创建一个示例 HTML 文件 (example.html)
with open("example.html", "w", encoding="utf-8") as f:
    f.write("""
    <!DOCTYPE html>
    <html>
    <head>
        <title>Example Document</title>
    </head>
    <body>
        <h1>Introduction</h1>
        <p>This is a sample HTML document with tables and lists.</p>

        <h2>Table Example</h2>
        <table>
            <thead>
                <tr>
                    <th>Header 1</th>
                    <th>Header 2</th>
                </tr>
            </thead>
            <tbody>
                <tr>
                    <td>Row 1, Cell 1</td>
                    <td>Row 1, Cell 2</td>
                </tr>
                <tr>
                    <td>Row 2, Cell 1</td>
                    <td>Row 2, Cell 2</td>
                </tr>
            </tbody>
        </table>

        <h2>List Example</h2>
        <ul>
            <li>Item 1</li>
            <li>Item 2</li>
            <li>Item 3</li>
        </ul>

        <h2>Another Section</h2>
        <p>This section contains more text and information about key features.</p>
        <p>Key features include: functionality, scalability, and ease of use.</p>
    </body>
    </html>
    """)

这段代码提供了一个完整的示例，从HTML解析，数据清洗，到向量索引建立和搜索，展示了整个流程。确保安装了所需的库： pip install beautifulsoup4 pandas nltk sentence-transformers faiss-cpu

四、一些额外的建议

除了以上策略，还有一些额外的建议可以帮助大家提高 RAG 系统的性能：

  • 领域知识融合： 将领域知识融入到清洗策略中，例如使用领域词典进行实体识别和关键词提取。
  • 数据增强： 使用数据增强技术扩充训练集，例如对文本进行同义词替换、句子重组等。
  • 持续学习： 随着用户查询的积累，我们可以使用持续学习的方法不断优化 RAG 系统，提高其适应性和准确性。
  • Prompt 工程： 好的 prompt 可以引导模型更好地利用检索到的上下文信息。

提升召回率和准确率的工程化思路

复杂结构化文档的清洗是一个复杂而精细的过程，需要根据文档的特点和应用场景进行定制化的设计。希望本文提出的策略能够帮助大家构建更可靠的 RAG 系统，解决实际问题。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注