使用 Unstructured.io 和 Elasticsearch 向量数据库搜索复杂文档

原创

点火三周

发布于 2024-07-12 14:12:30

89002

代码可运行

文章被收录于专栏：Elastic Stack专栏Elastic Stack专栏

运行总次数：2

代码可运行

在构建真实的 RAG（检索增强生成）应用时，解析文档以使信息可搜索是重要的一步。Unstructured.io 和 Elasticsearch 在这个场景中有效地协同工作，为开发者提供了互补的工具来构建 RAG 应用。

Unstructured.io 提供了一组工具库，可以提取、清理和转换不同格式和不同内容来源的文档。一旦文档被添加到 Elasticsearch 索引中，开发者可以选择许多 Elastic 的功能，包括聚合、过滤、RBAC（基于角色的访问控制）工具以及 BM25 或向量搜索功能，将复杂的业务逻辑实现到 RAG 应用中。

在这篇博客中，我们将研究一个相当常见的用例，即解析并导入一个包含文本、表格和图像的 PDF 文档。我们将使用 Elastic 的 ELSER 模型创建稀疏向量嵌入，然后使用 Elasticsearch 作为向量数据库存储和搜索这些嵌入。

Unstructured 的强大之处在于其模型能够识别文档的独特组成部分并将其提取为“文档元素”。Unstructured 还具有使用不同策略对分块进行分区的能力，而不仅仅是按字符数分块。这些“智能分区和分块”策略可以提高搜索相关性并减少 RAG 应用中的幻觉。

在解析数据后，我们将其存储为 Elasticsearch 向量数据库中的向量嵌入并运行搜索操作。我们使用 Elasticsearch 向量数据库连接器将这些数据发送到 Elastic。我们还将一个管道附加到流程中，以便在导入时创建 ELSER（一种开箱即用的稀疏编码模型，用于语义搜索）嵌入。

高级流程

在 Elastic 平台上部署 ELSER 模型
创建一个导入管道，该管道将为导入的分块创建嵌入。字段 text 将存储分块文本，text_embeddings 将存储嵌入。我们将使用 ELSER v2 模型。

PUT _ingest/pipeline/chunks-to-elser
{
  "processors": [
    {
      "inference": {
        "model_id": ".elser_model_2_linux-x86_64",
        "input_output": [
          {
            "input_field": "text",
            "output_field": "text_embedding"
          }
        ]
      }
    }
  ]
}

下一步是创建一个名为 unstructured-demo 的索引，并为 ELSER 嵌入创建必要的映射。我们还将我们在上一步中创建的管道附加到该索引上。其他所有字段都将动态映射。

PUT unstructured-demo
{
  "settings": {
    "default_pipeline": "chunks-to-elser"
  },
  "mappings": {
    "properties": {
      "text_embedding": {
        "type": "sparse_vector"
      },
      "text": {
        "type": "text"
      }
    }
  }
}

最后一步是使用 Unstructured 的代码示例，利用 Elasticsearch 连接器来创建分区和分块。请按照安装依赖项的说明操作。

import os
from unstructured.ingest.connector.elasticsearch import (
    ElasticsearchAccessConfig,
    ElasticsearchWriteConfig,
    SimpleElasticsearchConfig,
)
from unstructured.ingest.connector.local import SimpleLocalConfig
from unstructured.ingest.interfaces import (
    ChunkingConfig,
    PartitionConfig,
    ProcessorConfig,
    ReadConfig,
)
from unstructured.ingest.runner import LocalRunner
from unstructured.ingest.runner.writers.base_writer import Writer
from unstructured.ingest.runner.writers.elasticsearch import (
    ElasticsearchWriter,
)

我们将主机设置为 Elastic Cloud（Elasticsearch 服务）。我们设置用户名和密码，并设置要写入的索引：

def get_writer() -> Writer:
    return ElasticsearchWriter(
        connector_config=SimpleElasticsearchConfig(
            access_config=ElasticsearchAccessConfig(
                hosts="https://unstructured-demo.es.us-central1.gcp.cloud.es.io",
                username="elastic",
                password=<insert password>
            ),
            index_name="unstructured-demo",
        ),
        write_config=ElasticsearchWriteConfig(
            batch_size_bytes=15_000_000,
            num_processes=2,
        ),
    )

在下一步中，注册一个 Unstructured API 端点和密钥。Unstructured 中的分区功能从非结构化文档中提取结构化内容。partition 函数检测文档类型并自动确定适当的分区函数。如果用户知道他们的文件类型，也可以指定特定的分区函数。在分区步骤中，我们指示 Unstructured 通过传入 pdf_infer_table_structure=True 并将分区策略设置为 hi_res 来推断表结构，自动识别文档的布局。你可以了解各种 Unstructured 分区策略。我们将分块策略设置为 by_title，它“保留章节和页面边界”。分块策略对 RAG 应用的性能和质量有重大影响。你可以在 Unstructured 的有效检索增强生成分块论文中了解更多。

writer = get_writer()
runner = LocalRunner(
    processor_config=ProcessorConfig(
        verbose=True,
        output_dir="local-output-to-elasticsearch",
        num_processes=2,
    ),
    connector_config=SimpleLocalConfig(
        input_path=<path to PDF>,
    ),
    read_config=ReadConfig(),
    partition_config=PartitionConfig(
        pdf_infer_table_structure=True,
        strategy='hi_res',
        partition_by_api=True,
        partition_endpoint=<your partition endpoint>',
        api_key=<your api key>
    ),
    chunking_config=ChunkingConfig(
        chunk_elements=True,
        max_characters=500,
        chunking_strategy="by_title"
    ),
    writer=writer,
    writer_kwargs={},
)
runner.run()

在 Elasticsearch 向量数据库中的结果文档中，你会看到一些由 Unstructured API 生成的有趣的元数据。如果元素是一个表格，你会看到表格的 HTML 结构以及有关其外观的信息。如果它是文本块并且是早期块的延续，你会看到 is_continuation，这在 RAG 场景中很有价值，当你想将段落的整个上下文传递给 LLM 时。如果你想知道哪些单独的分区组成了一个块，你可以在 base-64 编码的 orig_elements 字段中找到它。在上面的示例中，我们使用了 Unstructured 的 API 服务。这些 API 服务可以通过三种不同方式使用：

有限试用的 Unstructured API
SaaS Unstructured API
AWS/Azure Marketplace Unstructured API

试用版本的处理能力限制为 1000 页，并且你的文档可以用于专有模式的训练和评估目的。对于快速原型设计，你也可以查看 Unstructured 的开源版本。unstructured 库提供了使用其 Python 安装程序运行的选项。如果你想避免处理多个依赖项，可以使用捆绑了所有必需库的 Docker 容器。与开源版本相比，Unstructured API 提供了以下附加功能：