简化物业数据管理：使用 Indexify 进行高级数据提取和检索

使用 Indexify 进行文档查询的分步指南。欢迎来到雲闪世界。

添加图片注释，不超过 140 字（可选）

总结：

传统的数据提取方法通常会错过非结构化内容的更深层次的见解，尤其是在房地产领域。
本文探讨使用 Indexify（一种用于实时、多模式数据提取的开源框架）来更好地分析财产文件。
您将学习设置 Indexify，包括服务器设置、提取图形创建、文档提取和数据查询，以及如何创建自定义提取器。
实施 Indexify 可以增强财产文件分析，从而获得更准确的见解、更好的决策和简化的管理。

传统的数据提取方法通常无法挖掘隐藏在非结构化内容中的更深层次、更复杂的见解。大多数方法只能捕获表面信息，无法揭示更深层次的见解。随着企业的发展和文档收集的扩大，对高级工具的需求变得至关重要，这些工具可以解析、分析和理解这一浩瀚的信息海洋。这种深度对于推动明智的决策、识别趋势和保持竞争优势至关重要。

房地产行业已准备好接受人工智能驱动的转型。通过集成先进的人工智能解决方案，组织可以增强文档分析、解析房地产记录并简化管理流程。更好的数据提取策略可以改善房地产数据分析，并提供更准确、更可操作的房地产文档洞察，从而实现高效的维护、销售等。

在本文中，我将探索使用Tensorlake的开源框架 Indexify 进行数据提取和检索。我还将展示这个可扩展系统在处理和分析与财产相关的文书工作方面的能力。

Indexify 概述

Indexify是一个开源提取和提取引擎，旨在实时为 LLM 应用程序提供支持。它能够以极低的延迟从非结构化源实时提取数据。它还支持可应用于各种用例的多模式提取工作流，包括从文档中提取和嵌入实体、音频转录、摘要以及从图像甚至视频中检测对象。

它还支持高效的索引、存储和检索数据，使其成为可扩展实时 RAG 系统的有力候选者。

任何工作流程都可以通过 4 个基本步骤轻松实现：

启动 Indexify 服务器和提取器。
创建提取图。
以您所需的格式（视频、图像、音频、pdf 等）提取数据。
检索提取的数据。

提取器

提取器模块是 Indexify 的核心功能。提取器可以从任何形式的非结构化数据中提取并返回结构化信息。例如，从 JSON 格式的 PDF 中获取特定信息、将数据转换为嵌入，以及识别视频中的面部或物体。

添加图片注释，不超过 140 字（可选）

提取器通常接收非结构化数据作为输入，并生成内容对象和特征列表作为输出。非结构化数据的原始字节存储在 blob 存储中，提取的特征作为索引存储在向量存储和结构化存储中以供检索。任何用于处理非结构化数据的模型或算法都可以通过扩展提取器 SDK 中提供的抽象类来实现为提取器。

协调员

这些是 Indexify 中使用的高性能任务调度程序。它们在开始提取数据时将任务分配给提取器，从而帮助实现卓越的速度和性能。

有关 Indexify 的更多深入信息，请查看其文档。

本教程的先决条件

对于本教程，请确保您拥有 Python 版本 3.11 或更高版本以获得最佳性能。其他安装说明将在稍后提供。

我将使用Groq作为 LLM 引擎。首先，请转到此页面并创建 API 密钥。

添加图片注释，不超过 140 字（可选）

另外，安装库。

pip install groq

对于数据，我准备了一些房地产宣传册的 PDF 版本。您可以随意使用自己收集的与物业管理相关的文档。

设置 Indexify 进行高级文档分析

安装和配置 Indexify

在 Linux 系统上设置 Indexify 很容易。对于开发，您需要运行 3 个终端窗口。

终端 1：用于下载和运行 Indexify 服务器。
终端 2：用于运行 Indexify 提取器，处理结构化提取、分块和嵌入。
终端 3：用于运行 Python 脚本以从 Indexify 服务器加载和查询数据。

您可以使用以下命令启动并运行 Indexify 服务器。记得检查命令在哪些终端上运行。

TERMINAL 1

curl https://getindexify.ai | sh
./indexify server -d.

服务器将在http://localhost:8900上运行。

接下来，创建一个 Python 环境并安装所有必要的库和提取器。我将详细讨论本教程所需的特定提取器。

TERMINAL 2

pip3 install indexify-extractor-sdk indexify wikipedia
indexify-extractor download tensorlake/paddleocr_extractor
indexify-extractor download tensorlake/minilm-l6
indexify-extractor download tensorlake/chunk-extractor

下载后，使用以下命令运行提取服务器：

TERMINAL 2

indexify-extractor join-server

系统现已准备好进行开发。我将在整个教程中保持这两个终端运行。

准备文件集

第一步是整理您的文档集合。我将在此工作流程中使用 PDF 文档。对于多个文档，您可以按如下方式构造目录：将所有文档作为 PDF 添加到数据目录。对于其他数据类型，您必须使用其他提取器或定义自定义提取器，我将在本博客的后面部分讨论。

└──  data
 ├── doc1
 ├── doc2
 ├── doc3
 ├── doc4
├── venv
├── indexify file
├── ingest_document.py
├──  query_doc.py
└──  setup_extraction_graph.py

使用 Indexify 提取和处理文档

设置提取图表

Indexify 可以使用其核心功能“提取器”来处理和存储来自任何模态的数据。这些提取器旨在从大量非结构化数据中提取内容。通过将不同的提取器链接在一起，您可以创建一个简化整个数据提取和存储过程的管道。此过程通过创建提取图来管理。

对于本教程，我将构建以下提取图：

paddle_ocr：从房地产宣传册中识别并提取文本。
chunk_extract：将数据分成用于 RAG 的块。
minilm-l6：将数据转换为嵌入。

添加图片注释，不超过 140 字（可选）

配置 Indexify 客户端并以 YAML 格式定义流程，如下所示。

# setup_extraction_graph.py


from indexify import ExtractionGraph, IndexifyClient
client = IndexifyClient()
extraction_graph_spec = """
  name: 'propertyQA'
  extraction_policies:
    - extractor: 'tensorlake/paddleocr_extractor'
      name: 'textextract'
    - extractor: 'tensorlake/chunk-extractor'
      name: 'chunker'
      input_params:
          chunk_size: 1000
          overlap: 100
      content_source: 'textextract'
    - extractor: 'tensorlake/minilm-l6'
      name: 'pdfembedding'
      content_source: 'chunker'
"""
extraction_graph = ExtractionGraph.from_yaml(extraction_graph_spec)
client.create_extraction_graph(extraction_graph)

运行命令来创建提取图。

TERMINAL 3

python3 ./setup_extraction_graph.py

自定义提取器

为了捕获基本字段以外的复杂数据点，我可以创建自定义提取器并将其添加到我们的提取图中。您可以使用如下所示的模板定义它们。

from pydantic import BaseModel
from typing import List
from indexify_extractor_sdk import Extractor, Content, Feature
from indexify_extractor_sdk.base_extractor import Content
import json


class InputParams(BaseModel):
    a: int = 0
    b: str = ""


class MyExtractor(Extractor):
    name = "your-docker-hub-username/MyExtractor"
    description = "Description of the extractor goes here."
 
    # Any python dependencies included in the extractor must be listed here.
    python_dependencies = ["torch", "transformers"]
    
    # Any system dependencies that the python code here depends on needs to be listed here. We use Ubuntu base images, so any ubuntu package can be installed here.


    system_dependencies = []


    input_mime_types = ["text/plain"]


    def __init__(self):
        super().__init__()


    def extract(self, content: Content, params: InputParams) -> List[Content]:
        return [
            Content.from_text(
                text="Hello World", feature=Feature.embedding(values=[1, 2, 3])
            ),
            Content.from_text(
                text="Pipe Baz", feature=Feature.embedding(values=[1, 2, 3])
            ),
            Content.from_text(
                text="Hello World",
                feature=Feature.metadata(value=json.dumps({"key": "value"})),
            ),
        ]


    def sample_input(self) -> Content:
        Content.from_text(text="Hello World")


if __name__ == "__main__":
    MyExtractor().extract_sample_input()

该类InputParams使用 Pydantic 定义可用于配置提取器行为的参数。

MyExtractor是实现提取器的主类。它指定名称、描述、依赖项以及它可以处理的输入类型。任何提取到 Indexify 中的内容，如果与这些 MIME 类型之一不匹配，都不会发送到此提取器。

该extract方法为核心功能。它处理输入内容并返回具有元数据等特征的转换内容对象列表。该sample_input方法提供了用于测试的示例输入。

例如，这是一个自定义提取器，以 JSON 格式返回属性详细信息。

# custom_extractor.py


from pydantic import BaseModel
from typing import List
from indexify_extractor_sdk import Extractor, Content, Feature
from indexify_extractor_sdk.base_extractor import Content
import json
import re


class InputParams(BaseModel):
    price_regex: str = r"\$(\d{1,3}(,\d{3})*(\.\d+)?)"
    sqft_regex: str = r"(\d+)\s*sq\s*ft"
    location_regex: str = r"located\s+in\s+([^\.]+)"


class PropertyExtractor(Extractor):
    name = "your-docker-hub-username/PropertyExtractor"
    description = "Extracts property price, square footage, and location from text."
    python_dependencies = ["re"]
    system_dependencies = []
    input_mime_types = ["text/plain"]


    def __init__(self):
        super().__init__()


    def extract(self, content: Content, params: InputParams) -> List[Content]:
        text = content.text()
       
        price_match = re.search(params.price_regex, text)
        sqft_match = re.search(params.sqft_regex, text)
        location_match = re.search(params.location_regex, text)


        property_info = {
            "property_price": price_match.group(1) if price_match else "",
            "property_square_feet": sqft_match.group(1) if sqft_match else "",
            "property_location": location_match.group(1).strip() if location_match else ""
        }


        return [
            Content.from_text(
                text=json.dumps(property_info),
                feature=Feature.metadata(value=json.dumps(property_info))
            )
        ]


    def sample_input(self) -> Content:
        return Content.from_text(text="Priced at $500,000 with 2,000 sq ft of living space, located in San Francisco.")


if __name__ == "__main__":
    PropertyExtractor().extract_sample_input()

您可以将提取器打包到容器中进行生产，也可以使用以下命令在本地安装提取器并在提取图中使用它。

indexify-extractor install-local custom_extractor:PropertyExtractor

上传文件

设置提取图后，数据目录中的每个文档都应通过管道并作为嵌入存储在向量存储中。Indexify 有一个内置的向量存储，可用于存储、查询和检索数据。

# ingest_document.py

import os
import requests
from indexify import IndexifyClient


# Initialize IndexifyClient
client = IndexifyClient()
folder_path = "data"


for filename in os.listdir(folder_path):
    if filename.endswith(".pdf"):
        # Construct the full file path
        file_path = os.path.join(folder_path, filename)
       
        # Upload the PDF to Indexify
        client.upload_file("propertyQA", file_path)
        print(f"Uploaded: {filename}")

可以使用 Indexify UI ( http://localhost:8900/ui)查看向量存储和索引。

添加图片注释，不超过 140 字（可选）

您还可以使用客户端访问索引：

from indexify import IndexifyClient
client = IndexifyClient()


content_ids = [content.id for content in client.list_content("propertyQA")]


extract = client.get_extracted_content(content_ids[1], "propertyQA", "textextract")
embedding = client.get_extracted_content(content_ids[1], "propertyQA", "pdfembedding")

在 Indexify 中，提取的数据也存储在 SQL 表中，允许您使用内置 API 和 SQL 查询来查询数据。

result = client.sql_query("select * from propertyQA;")"select * from propertyQA;")

输出

SqlQueryResult(result=[{'content_id': 'd6e584685d74a21d', 'type': 'text'}, {'content_id': 'e32fd65fc2bbebf3', 'type': 'text'}])

使用 Indexify 提出复杂问题

语义搜索和制定查询

Indexify 的高级提取引擎使用户能够制定和执行超出基本数据提取的复杂查询。Indexify 可以解释文档的上下文和语义，从而解决更复杂的问题。

Indexify 从创建的索引中检索相关上下文并返回 Content 对象。此对象中的信息可用于构建复杂的提示，这些提示可传递给 LLM 以生成适当的响应。让我们看看它是如何工作的。

results = client.search_index(name=index, query=question, top_k=3)3)
context = ""
for result in results:
        context = context + f"content id: {result['content_id']} \n \npassage: {result['text']}\n"

在这里，Indexify 通过索引执行简单的语义搜索，并返回最相关的段落以及内容 ID 等元数据，让您了解结果的背景和重要性。参数中的数据text可用于构建上下文。

检索和分析结果

语义搜索和复杂查询使您能够更深入地研究文档集合，提取不立即显现的有意义的见解。

复杂问题的例子：

“根据历史数据，哪些房产出现维护问题的风险最高？”
“地震破坏的风险有哪些？采取了哪些减灾措施？”

LLM 可以帮助描述性地、准确地回答此类查询。从上一步检索到的上下文可用于构建复杂的提示。

"You are a real estate expert and you have to help me manage my property. Answer the question, based on the context. Answer \"Information not found\" if there is no context. Do not hallucinate.  \nquestion: {question}  \ncontext: {context}"

然后使用构造的提示来查询 LLM。对于本教程，我使用了Groq 的gemma-7b模型。以下是工作流程的完整代码。

#query_doc.py


from indexify import IndexifyClient
from groq import Groq


client = IndexifyClient()
groq_client = Groq(
    api_key="API_KEY",
)
def get_context(question: str, index: str, top_k=3):
    results = client.search_index(name=index, query=question, top_k=3)
    context = ""
    for result in results:
        context = context + f"content id: {result['content_id']} \n \npassage: {result['text']}\n"
    return context


def create_prompt(question, context):
    return f"You are a real estate expert and you have to help me manage my property. Answer the question, based on the context. Answer \"Information not found\" if there is no context. Do not hallucinate.  \nquestion: {question}  \ncontext: {context}"


def generate_response(prompt):
    chat_completion = groq_client.chat.completions.create(
        messages=[
            {
                "role": "user",
                "content": prompt,
            }
        ],
        model="gemma-7b-it",
    )
    return chat_completion.choices[0].message.content


question = "What are the risks of earthquake damage and what mitigation measures are taken?"
context = get_context(question, "propertyQA.pdfembedding.embedding")
prompt = create_prompt(question, context)


response = generate_response(prompt)
print(response)

运行此文件以获取响应。

TERMINAL 3

python3 ./query_doc.py

**Risks of Earthquake Damage:**
The provided text does not include information regarding the specific risks of earthquake damage, so this information cannot be extracted from the given context.
**Mitigation Measures:**
The provided text mentions that the structure is earthquake resistance as per IS Standards. Specific mitigation measures mentioned include:
- Metal/Fibre strip reinforcement on all joints to avoid cracks.
- Sharp edges avoided on fabrications to provide safety especially for children.

检索到数据后，您可以对其进行分析，以大规模识别物业管理中的趋势、异常值和可操作见解。可视化工具是绘制趋势和传达发现的关键。例如，在房地产数据分析中，您可能会发现某些物业的维修成本较高，这表明需要主动维护。

extractor: 'tensorlake/chunk-extractor'
      name: 'chunker'
      input_params:
          chunk_size: 1000
          overlap: 100
      content_source: 'textextract'

优化您的系统以实现高效的索引和检索，从而管理大规模物业数据和房地产分析。这涉及配置向量存储以高效管理大量嵌入。在生产环境中，Indexify 可以水平扩展到多个服务器实例和协调器，以并行简化实时数据提取和存储。您还可以微调提取图中的参数（如和），chunk_size以overlap在粒度和性能之间取得平衡。

extractor: 'tensorlake/chunk-extractor'
      name: 'chunker'
      input_params:
          chunk_size: 1000
          overlap: 100
      content_source: 'textextract'

Indexify 与流行的LangChain框架完美集成，后者是用于构建 AI 应用程序的高度可扩展框架。首先，安装库。

TERMINAL 3

pip install indexify-langchain langchain-groq

我正在使用 LangChain 的基于 Indexify 的检索器。我将传递我准备的索引和参数top_k。

from indexify import IndexifyClient
client = IndexifyClient()
from indexify_langchain import IndexifyRetriever
params = {"name": "propertyQA.pdfembedding.embedding", "top_k": 2}
retriever = IndexifyRetriever(client=client, params=params)

完整代码如下：

# langchain_query_doc.py

import requests
import dotenv


# Setup retriever
from indexify import IndexifyClient
from indexify_langchain import IndexifyRetriever
client = IndexifyClient()
params = {"name": "propertyQA.pdfembedding.embedding", "top_k": 2}
retriever = IndexifyRetriever(client=client, params=params)


from langchain_groq import ChatGroq
llm = ChatGroq(
    model="gemma-7b-it",
    temperature=0,
    max_tokens=None,
    timeout=None,
    max_retries=2,
    api_key="API_KEY",
)


# Setup Chat Prompt Template
from langchain.prompts import ChatPromptTemplate


template = """
You are a real estate expert and you have to help me manage my property. Answer the question, based on the context. Answer \"Information not found\" if there is no context.
Do not hallucinate.  \nquestion: {question}  \ncontext: {context}
"""
prompt = ChatPromptTemplate.from_template(template)


from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser
rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)


# Query


query = "What are the risks of earthquake damage and what mitigation measures are taken?"
print(rag_chain.invoke(query))

这将使用检索基于查询的最相关段落IndexifyRetriever。然后，它使用创建适合上下文的提示ChatPromptTemplate。提示将传递到 LLM，模型将在其中处理信息并生成响应。

将查询结果存储在数据库中

执行查询并检索结果后，将这些数据存储在结构化数据库中可确保您可以引用它并执行进一步的分析。

首先，为了将查询结果存储在数据库中，请设置一个关系数据库并定义一个可以容纳 Indexify 返回的复杂数据结构的模式。

以下是如何在 PostgreSQL 数据库中存储查询结果的示例：

import psycopg2
# Connect to your PostgreSQL database
conn = psycopg2.connect(
    dbname="indexify_results",
    user="your_username",
    password="your_password",
    host="localhost",
    port="5432"
)
cursor = conn.cursor()


# Create a table to store query results
cursor.execute("""
    CREATE TABLE IF NOT EXISTS query_results (
        id SERIAL PRIMARY KEY,
        query TEXT,
        content_id VARCHAR(255),
        passage TEXT
    );
""")
conn.commit()


# Example function to store results
def store_results(question: str, results: list):
    for result in results:
        cursor.execute("""
            INSERT INTO query_results (query, content_id, passage)
            VALUES (%s, %s, %s);
        """, (question, result['content_id'], result['text']))
    conn.commit()


# Retrieve and store query results
question = "Which properties are likely to require renovation next year?"
results = client.search_index(name="propertyQA.pdfembedding.embedding", query=question, top_k=5)
store_results(question, results)
# Close the database connection
cursor.close()
conn.close()

定义数据库架构时，请考虑数据的关系和复杂性。架构应容纳内容 ID 和段落文本以及元数据（例如时间戳、文档来源以及任何相关标签或类别）。

这是一个您可以考虑的示例模式。

CREATE TABLE query_results (
    id SERIAL PRIMARY KEY,
    query TEXT,
    content_id VARCHAR(255),
    passage TEXT,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    document_source VARCHAR(255),
    tags TEXT[]
);

添加图片注释，不超过 140 字（可选）

总结

我们以房地产行业为例，介绍了 Indexify 如何为需要更深入实时分析的应用程序实现高效的数据提取和检索。我们详细介绍了 Indexify 的工作原理，包括如何构建数据检索知识库。此外，我们还研究了如何扩展系统以获得更大的数据集和更高的吞吐量。最后，我们讨论了将结果存储在结构化数据库中以供将来参考和深入分析的方法。完整代码可在此处的GitHub 链接中找到。

下一步，您可以探索并构建自定义提取器，利用自定义模型和 LLM 来处理各种用例。此外，了解 Indexify 如何与其他数据库和向量存储一起使用。

感谢关注雲闪世界。（Aws解决方案架构师vs开发人员&GCP解决方案架构师vs开发人员）