从10万份文档中更快、更准确地找到信息，还能理解语义！试试ElasticSearch+RAG

想象一下这样的场景：你在一个满是10万份文档的数字仓库里，目标是快速、准确地找到某一条关键信息。如果文档内容还涉及复杂的法律术语和严密的逻辑结构，那难度直接翻倍。这正是我们这次要解决的问题——用ElasticSearch和大模型相关技术（检索增强生成，RAG）来处理和检索一个超大规模的德语法律文档库。

（编者注：Elasticsearch是一个开源全文搜索引擎，每个字段均可被索引，能在极短时间内存储、搜索和分析TB级数据）

处理海量文本的挑战

法律文件的复杂程度，是想象一下就知道的——密密麻麻的条款、环环相扣的逻辑、咬文嚼字的表述。核心难点在于，如何构建一个能高效处理这套庞大文档库的系统，让它在用户提问时，瞬间给出既精准又全面的答案。

这个问题其实不新，但解决办法的成败，往往出在分块策略上。

我们最初在探讨时，想到的是LangChain的递归分块策略。它是个不错的起点，但面对巨量文档，它有一个致命短板：无法很好地保持语义完整性。简单来说，就是把一段逻辑严密的论述，强行切成了互不相关的碎块。

来看一个使用句子转换器进行文本分割的示例代码。

from langchain.text_splitter import SentenceTransformersTokenTextSplitter
text_splitter = SentenceTransformersTokenTextSplitter(
                                 tokens_per_chunk = 480,
                                 chunk_overlap = 50,
                                 model_name = "intfloat/multilingual-e5-large-instruct"
                                 )
texts = text_splitter.split_documents(data)

确实，分块重叠是一种有效保持上下文的手段，但它也有不小的代价：

• 存储需求暴增
  ：重叠意味着文本重复，10万份文档产生的冗余，对存储来说是不小的压力。

• 处理时间延长
  ：更多数据意味着计算开销扩大，每个分块都需要单独处理和索引，拖慢整体进度。

• 上下文依旧碎片化
  ：重叠并不能100%解决语境断裂问题。对逻辑严密的法律文本而言，关键信息仍可能散落各块，导致检索不完整。

考虑到我们有超过10万份文档，单份文档最多可达1000多页，这条路径并不适用。

解决方案

数据存储在云端的多个文件夹中，每个文件夹里都有大量文件。一个一个手动下载显然不现实，所以选择了并发下载（concurrent futures），一次性拽下来多个文件。

from concurrent import futures
from concurrent.futures import ProcessPoolExecutor

def download_parallel_multiprocessing():
    with ProcessPoolExecutor() as executor:
        future_to_key = {executor.submit(download_one_file, key): key for key in list_key_to_download}

        for future in futures.as_completed(future_to_key):
            key = future_to_key[future]
            exception = future.exception()

            if not exception:
                yield key, future.result()
            else:
                yield key, exception

for key, result in download_parallel_multiprocessing():
#     print(f"{key}: {result}")
    pass

下载后的文件先存放在临时文件夹里。接下来，就得确定用于嵌入的模型了。因为文档是德语，自然需要一个多语言编码器模型——于是选择了intfloat/multilingual-e5-large。

from langchain_community.llms import WatsonxLLM
import os
from ibm_watson_machine_learning.metanames import GenTextParamsMetaNames as GenParams

api_key = "Your Key Here"
ibm_cloud_url = "Write your URL"
project_id = "The project id"

params = {
    GenParams.DECODING_METHOD: "sample",
    GenParams.MIN_NEW_TOKENS: 50,
    GenParams.MAX_NEW_TOKENS: 430,
    GenParams.RANDOM_SEED: 42,
    GenParams.TEMPERATURE: 0,
    GenParams.TOP_K: 20,
    GenParams.TOP_P:1
}

llm = WatsonxLLM(
            model_id='intfloat/multilingual-e5-large',
            url=ibm_cloud_url,
            apikey=api_key,
            project_id=project_id,
            params=params,
            )

过滤异常文件和去重

原始数据里混着一些异常文件。以下代码会扫描文件名中的特定模式，把匹配到的模式移除后，再检查列表中是否存在同名文件。如果有，说明存在重复，需要标记出来。

import re

def matching_files_with_names(file_names):
    matching_files = []
    for file_name in file_names:
        pattern_match = re.search(r'[d+]', file_name)
        if pattern_match:
            pattern_removed_name = re.sub(r'[d+].', '', file_name)
            if pattern_removed_name in file_names and pattern_removed_name != file_name:
                matching_files.append([file_name, pattern_removed_name])
    return matching_files

matching_files = matching_files_with_names(matching_pairs)

做完这一步，仍剩下超过10.2万个文件需要处理。

提取元数据

接下来是对文件进行预处理——把它拆解成更小的、可操作的块。文件内容是xml格式，而实际正文是html格式。一个自然的想法是：根据标题把文档切开。这样一来，每个标题和它下方的文本都能保持完整，整段语义也就被保留下来了。

第一步是加载和解析xml文件，提取所有元数据。用for循环遍历所有文件即可。

import lxml
import pickle
from lxml import etree
from langchain.docstore.document import Document

dir_list_pre = os.listdir("temp_pre/")
full_doc = []

for file in tqdm(list(matching_pairs_og.keys())):
    file_path = "temp_pre/" + file
    with open(file_path, "r") as f:
        xml_content = f.read()

    try:
        with open(file_path, "rb") as f:
            xml_content_tag_reader = f.read()
        parser = etree.XMLParser(recover=True)
        root = etree.fromstring(xml_content_tag_reader, parser)
        tree = root.getroottree()
        meta_tags = {tree.getpath(d): d.text for d in root.iterdescendants() if '/txt' not in tree.getpath(d)}

        soup = BeautifulSoup(xml_content, "xml")
        text_tags = soup.find_all('txt')
        html_content = ''.join([' ' + tag.get_text() for tag in text_tags])

        if len(html_content)>0:
            html_soup = BeautifulSoup(html_content, "html.parser")
            paragraphs_with_strong = html_soup.find_all(lambda tag: tag.name == 'p' and tag.strong)

            for p_tag in paragraphs_with_strong:
                strong_tag = p_tag.strong
                if strong_tag:
                    strong_text = strong_tag.text.strip()
                    if re.match(r'^d+. ', strong_text):
                        previous_header = p_tag.find_previous_sibling(lambda tag: tag.name.startswith('h') and len(tag.name) == 2)
                        if previous_header:
                            next_header_level = 'h6'
                            new_header_tag = html_soup.new_tag(next_header_level)
                            new_header_tag.string = strong_text
                            p_tag.replace_with(new_header_tag)

            modified_html = str(html_soup)
            headers_to_split_on = [
                ("h1", "Header 1"),
                ("h2", "Header 2"),
                ("h3", "Header 3"),
                ("h4", "Header 4"),
                ("h5", "Header 5"),
                ("h6", "Header 6")
            ]
            html_splitter = HTMLHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
            html_header_splits = html_splitter.split_text(modified_html)
            for i, split in enumerate(html_header_splits):
                tokens = llm.get_num_tokens(split.page_content)
                html_header_splits[i].metadata["token_count"] = tokens
                html_header_splits[i].metadata["source"] = file
                html_header_splits[i].metadata["index"] = i

            full_doc.append(html_header_splits)

        else:
            soup = BeautifulSoup(xml_content, "xml")
            text_tags = soup.find_all('txtascii')
            html_content = ''.join([' ' + tag.get_text() for tag in text_tags])
            lc_doc = Document(page_content=html_content)
            tokens = llm.get_num_tokens(html_content)
            lc_doc.metadata["token_count"] = tokens
            lc_doc.metadata["source"] = file
            lc_doc.metadata["index"] = 1
            full_doc.append(lc_doc)

    except lxml.etree.XSLTApplyError as e:
        continue

with open('full_xml.pkl', 'wb') as file:
    pickle.dump(full_doc, file)

cos_client.upload_file('./full_xml.pkl', bucket_name, 'full_xml.pkl')

第一步完成后，调用lxml库中的etree解析器，提取所有元数据：

with open(file_path, "rb") as f:
    xml_content_tag_reader = f.read()
parser = etree.XMLParser(recover=True)
root = etree.fromstring(xml_content_tag_reader, parser)
tree = root.getroottree()
meta_tags = {tree.getpath(d): d.text for d in root.iterdescendants() if '/txt' not in tree.getpath(d)}

提取txt标签

接下来，从xml内容中读取txt标签。每个页面有两个xml文件：一个包含html标签，另一个包含高级元数据。因此，先检查是否有html_content可用——如果有，就用LangChain的HTMLHeaderTextSplitter。

from langchain_text_splitters import HTMLHeaderTextSplitter
headers_to_split_on = [
    ("h1", "Header 1"),
    ("h2", "Header 2"),
    ("h3", "Header 3"),
    ("h4", "Header 4"),
    ("h5", "Header 5"),
    ("h6", "Header 6")
]

html_splitter = HTMLHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
html_header_splits = html_splitter.split_text(modified_html)

不过分完块后发现，某些块依然太大。深入探查后发现，部分子标题被嵌套在

标签的下，但完全可以作为标题使用。

def extract_headers_and_content(html):
    html_soup = BeautifulSoup(html, 'html.parser')
    headers = ['h1', 'h2', 'h3', 'h4', 'h5', 'h6']
    result = []
    current_metadata = {}

    for tag in html_soup.find_all(headers):
        level = headers.index(tag.name)
        current_metadata = {h: current_metadata[h] for h in headers[:level] if h in current_metadata}
        current_metadata[tag.name] = tag.get_text(strip=True)

        content = tag.find_next_sibling()
        content_text = ""
        while content and content.name not in headers:
            content_text += content.get_text(strip=True) + " "
            content = content.find_next_sibling()

        if content_text.strip():
            result.append({'page_content': content_text.strip(), 'metadata': current_metadata.copy()})

    return result

exception_file_chunking = []
for key, value in tqdm(full_doc_v2.items()):
    extracted_data = extract_headers_and_content(value)
    extracted_data_v2 = [i for i in extracted_data if i['page_content'] != '']
    for idx, i in enumerate(extracted_data_v2):
        i['metadata']['source'] = key
        tokens = llm.get_num_tokens(i['page_content'])
        i['metadata']['token_count'] = tokens
        i['metadata']['index'] = idx
    exception_file_chunking.append(extracted_data_v2)

for k, v in tqdm(tags.items()):
    new_dict = {}
    for key, value in v.items():
        if key == '/Segmente/Segment':
            continue
        parts = key.rsplit('/', 2)
        text = '_'.join([part.lower() for part in parts[-2:]])
        new_dict[text] = value
    tags[k] = new_dict

from langchain.embeddings import HuggingFaceEmbeddings

model_name = "intfloat/multilingual-e5-large"
model_kwargs = {'device': 'cuda'}
encode_kwargs = {'normalize_embeddings': False}
hf = HuggingFaceEmbeddings(
    model_name=model_name,
    model_kwargs=model_kwargs,
    encode_kwargs=encode_kwargs
)

from elasticsearch import Elasticsearch
from langchain_elasticsearch import ElasticsearchStore

es_client = Elasticsearch(
    es_url,
    basic_auth=(es_user, es_password),
    verify_certs=False,
    request_timeout = 1200
)

elastic_vector_search = ElasticsearchStore(
    index_name="index_e5_multiprocessing_htmlsplitter_gpu_test_v2",
    es_connection=es_client,
    embedding=hf,
)

db = ElasticsearchStore.from_documents(
 documents=texts,
    embedding=hf,
    index_name="index_e5_multiprocessing-v6",
    strategy=ElasticsearchStore.ApproxRetrievalStrategy(
        hybrid = True
    ),
    es_connection=es_client,
)