ToB企服应用市场:ToB评测及商务社交产业平台

标题: 利用大模型实时提取和检索多模态数据探索-利用 Indexify 进行文档分析 [打印本页]

作者: 曹旭辉    时间: 2024-9-6 03:49
标题: 利用大模型实时提取和检索多模态数据探索-利用 Indexify 进行文档分析
概览

传统的文本提取方法常常无法明白非结构化内容,因此提取数据的数据往往是错误的。本文将探究使用 Indexify,一个用于实时多模态数据提取的开源框架,来更好地分析pdf等非结构化文件。我将介绍怎样设置 Indexify,包括服务器设置、提取graph创建、文档导入和数据查询,以及怎样创建自界说提取器。通过 Indexify 可以加强文档分析,从而获得更准确的见解、做出更好的决策并简化管理流程。
Indexify 介绍

Indexify 是一个开源的导入和提取引擎,旨在为实时 LLM 应用提供支持。它能够以极低的延迟从非结构化来源进行实时数据提取。它还支持可应用于各种用例的多模态提取工作流程,包括从文档中提取实体和嵌入、音频转录、摘要以及从图像甚至视频中检测对象。
它还支持高效地索引、存储和检索数据,使其成为可扩展实时 RAG 体系的工具。
任何工作流程都可以通过 4 个根本步骤轻松实现:
提取器
提取器模块是 Indexify 的焦点功能。提取器可以从任何模态的非结构化数据中提取和返回结构化信息。例如,从 PDF 中以 JSON 格式获取特定信息,将数据转换为嵌入,以及识别视频中的面部或物体。
提取器通常接收非结构化数据作为输入,并天生内容对象列表和特性作为输出。来自非结构化数据的原始字节存储在 Blob 存储中,而提取的特性存储在向量数据库和结构化存储中以供检索。任何用于处理非结构化数据的模型或算法都可以通过扩展提取器 SDK 中提供的抽象类来实现为提取器。
和谐器
Indexify中有一个高性能任务调度器。当数据开始被导入时,它们将任务分配给提取器,从而帮助实现卓越的速率和性能。
开始教程

对于本教程,需要有 Python 3.11 或更高版本以获得最佳性能。其他安装说明将在背面给出。
我将使用 Groq 作为 LLM 引擎。要开始,请访问此页面并创建一个 API 密钥。

然后安装sdk
  1. pip install groq
复制代码
对于数据,我已经预备了一些经典论文作为 PDF 文件。固然你也可以使用自己的文档集。
安装和设置 Indexify

在 Linux 体系上设置 Indexify 很轻易。只需要 3 个运行的终端窗口。

这里可以使用以下命令启动和运行 Indexify 服务。
终端 1
  1. curl https://getindexify.ai | sh
  2. ./indexify server -d
复制代码
服务将在 http://localhost:8900 上运行。接下来,创建一个 Python 环境并安装所有须要的库和提取器。
终端 2
  1. pip3 install indexify-extractor-sdk indexify wikipedia
  2. indexify-extractor download tensorlake/paddleocr_extractor
  3. indexify-extractor download tensorlake/minilm-l6
  4. indexify-extractor download tensorlake/chunk-extractor
复制代码
下载完成后,使用以下命令运行提取器服务:
终端 2
  1. indexify-extractor join-server
复制代码
颠末以上两个步骤环境就预备好了。在本教程中,我将保持这两个终端运行。
预备文档集

第一步是整理文档集。在本工作流程中,我将使用 PDF 文档。对于多个文档,你可以像这样构建目次:将所有文档作为 PDF 添加到 data 目次。对于其他文件范例,你必须使用其他提取器或界说自界说提取器,我将在本文中稍后讨论。
  1. └──  data
  2. ├── doc1
  3. ├── doc2
  4. ├── doc3
  5. ├── doc4
  6. ├── venv
  7. ├── indexify file
  8. ├── ingest_document.py
  9. ├──  query_doc.py
  10. └──  setup_extraction_graph.py
复制代码
使用 Indexify 导入和处理文档

设置提取graph

Indexify 可以使用其焦点功能(提取器)处理和存储来自任何模态的数据。这些提取器旨在从大量非结构化数据中提取内容。通过将不同的提取器链接在一起,我们可以创建一个pipeline,简化整个数据提取和存储过程。此过程通过创建提取graph来管理。
在本教程中,我将构建以下提取graph:


设置 Indexify 客户端并以 YAML 格式界说流程,如下所示。
  1. # setup_extraction_graph.py
  2. from indexify import ExtractionGraph, IndexifyClient
  3. client = IndexifyClient()
  4. extraction_graph_spec = """
  5.   name: 'propertyQA'
  6.   extraction_policies:
  7.     - extractor: 'tensorlake/paddleocr_extractor'
  8.       name: 'textextract'
  9.     - extractor: 'tensorlake/chunk-extractor'
  10.       name: 'chunker'
  11.       input_params:
  12.           chunk_size: 1000
  13.           overlap: 100
  14.       content_source: 'textextract'
  15.     - extractor: 'tensorlake/minilm-l6'
  16.       name: 'pdfembedding'
  17.       content_source: 'chunker'
  18. """
  19. extraction_graph = ExtractionGraph.from_yaml(extraction_graph_spec)
  20. client.create_extraction_graph(extraction_graph)
复制代码
终端 3
运行命令以创建提取graph。
  1. python3 ./setup_extraction_graph.py
复制代码
自界说提取器

为了捕获复杂数据,我可以创建自界说提取器并将其添加到我们的提取graph中。你可以使用以下模板界说它们。
  1. from pydantic import BaseModel
  2. from typing import List
  3. from indexify_extractor_sdk import Extractor, Content, Feature
  4. from indexify_extractor_sdk.base_extractor import Content
  5. import json
  6. class InputParams(BaseModel):
  7.     a: int = 0
  8.     b: str = ""
  9. class MyExtractor(Extractor):
  10.     name = "your-docker-hub-username/MyExtractor"
  11.     description = "Description of the extractor goes here."
  12.     # Any python dependencies included in the extractor must be listed here.
  13.     python_dependencies = ["torch", "transformers"]
  14.     # 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.
  15.     system_dependencies = []
  16.     input_mime_types = ["text/plain"]
  17.     def __init__(self):
  18.         super().__init__()
  19.     def extract(self, content: Content, params: InputParams) -> List[Content]:
  20.         return [
  21.             Content.from_text(
  22.                 text="Hello World", feature=Feature.embedding(values=[1, 2, 3])
  23.             ),
  24.             Content.from_text(
  25.                 text="Pipe Baz", feature=Feature.embedding(values=[1, 2, 3])
  26.             ),
  27.             Content.from_text(
  28.                 text="Hello World",
  29.                 feature=Feature.metadata(value=json.dumps({"key": "value"})),
  30.             ),
  31.         ]
  32.     def sample_input(self) -> Content:
  33.         Content.from_text(text="Hello World")
  34. if __name__ == "__main__":
  35.     MyExtractor().extract_sample_input()`
复制代码
InputParams 类使用 Pydantic 来界说可用于设置提取器行为的参数。
MyExtractor 是实现提取器的主类。它指定了它可以处理的名称、描述、依靠项和输入范例。导入到 Indexify 的任何与这些 MIME 范例不匹配的内容都不会发送到此提取器。
extract 方法是焦点功能。它处理输入内容并返回具有特性(如元数据)的转换后的内容对象列表。sample_input 方法提供了一个示例输入以供测试。
例如,以下是一个自界说提取器,它以 JSON 格式返回论文概要信息。
  1. # custom_extractor.py
  2. from pydantic import BaseModel
  3. from typing import List
  4. from indexify_extractor_sdk import Extractor, Content, Feature
  5. from indexify_extractor_sdk.base_extractor import Content
  6. import json
  7. import re
  8. class InputParams(BaseModel):
  9.     author_regex: str = r"^([\w\s]+)"
  10.     team_regex: str = r"(Google Brain)"
  11.     email_regex: str = r"([\w.-]+@[\w.-]+\.[\w]+)"
  12. class PropertyExtractor(Extractor):
  13.     name = "your-docker-hub-username/PropertyExtractor"
  14.     description = "Extract author,team and email from this article."
  15.     python_dependencies = ["re"]
  16.     system_dependencies = []
  17.     input_mime_types = ["text/plain"]
  18.     def __init__(self):
  19.         super().__init__()
  20.     def extract(self, content: Content, params: InputParams) -> List[Content]:
  21.         text = content.text()
  22.         author_match = re.search(params.author_regex, text)
  23.         team_match= re.search(params.team_regex, text)
  24.         email_match = re.search(params.email_regex, text)
  25.         property_info = {
  26.             "author": author_match.group(1) if author_match else "",
  27.             "team": team_match.group(1) if team_match else "",
  28.             "team": email_match.group(1) if email_match else "",
  29.         }
  30.         return [
  31.             Content.from_text(
  32.                 text=json.dumps(property_info),
  33.                 feature=Feature.metadata(value=json.dumps(property_info))
  34.             )
  35.         ]
  36.     def sample_input(self) -> Content:
  37.         return Content.from_text(text="Ashish Vaswani Google Brain avaswani@google.com")
  38. if __name__ == "__main__":
  39.     PropertyExtractor().extract_sample_input()
复制代码
你可以将提取器打包到容器中以用于生产,或者使用以下命令在本地安装提取器,并在提取graph中使用它。
  1. indexify-extractor install-local custom_extractor:PropertyExtractor
复制代码
上传文档

设置提取graph后,data 目次中的每个文档都应该通过pipeline并作为嵌入存储在向量数据库中。Indexify 具有一个内置的向量数据库,你可以使用它来存储、查询和检索数据。
  1. # ingest_document.py
  2. import os
  3. import requests
  4. from indexify import IndexifyClient
  5. # Initialize IndexifyClient
  6. client = IndexifyClient()
  7. folder_path = "data"
  8. for filename in os.listdir(folder_path):
  9.     if filename.endswith(".pdf"):
  10.         # Construct the full file path
  11.         file_path = os.path.join(folder_path, filename)
  12.         # Upload the PDF to Indexify
  13.         client.upload_file("propertyQA", file_path)
  14.         print(f"Uploaded: {filename}")
复制代码
可以使用 Indexify UI 在 http://localhost:8900/ui 上查看向量数据库和索引。

你也可以使用客户端访问索引:
  1. from indexify import IndexifyClient
  2. client = IndexifyClient()
  3. content_ids = [content.id for content in client.list_content("propertyQA")]
  4. extract = client.get_extracted_content(content_ids[1], "propertyQA", "textextract")
  5. embedding = client.get_extracted_content(content_ids[1], "propertyQA", "pdfembedding")
复制代码
在 Indexify 中,导入的数据也存储在 SQL 表中,答应你使用内置 API 和 SQL 查询来查询数据。
  1. result = client.sql_query("select * from propertyQA;")
复制代码
输出:
  1. `SqlQueryResult(result=[{'content_id': 'd6e584685d74a21d', 'type': 'text'}, {'content_id': 'e32fd65fc2bbebf3', 'type': 'text'}])`
复制代码
使用 Indexify 提出复杂题目

语义搜索和查询公式
Indexify 的高级提取引擎使用户能够制定和执行超出根本数据提取的复杂查询。Indexify 可以表明文档的上下文和语义,从而答应更复杂的查询。
Indexify 从创建的索引中检索相关上下文并返回一个 Content 对象。来自此对象的信息可用于构建复杂提示,这些提示可以通报给 LLM 以天生得当的响应。让我们看看它是怎样工作的。
  1. results = client.search_index(name=index, query=question, top_k=3)
  2. context = ""
  3. for result in results:
  4.         context = context + f"content id: {result['content_id']} \n \npassage: {result['text']}\n"
复制代码
在这里,Indexify 通过索引执行简单的语义搜索,并返回最相关的段落以及元数据(如内容 ID),使你能够明白结果的上下文和紧张性。来自 text 参数的数据可用于构建上下文。
检索和分析结果
语义搜索和复杂查询使你能够更深入地挖掘文档集,提取更加深入的答案。
复杂题目的示例:

LLM 可以帮助描述性且准确地答复此类查询。从上一步检索到的上下文可用于构建复杂提示。
  1. "You are an expert in paper interpretation,You must help me interpret the core content of the paper. 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 模型。以下是工作流程的完备代码。
  1. `#query_doc.pyfrom indexify import IndexifyClientfrom groq import Groqclient = 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 contextdef create_prompt(question, context):    return f"You are an expert in paper interpretation,You must help me interpret the core content of the paper. Answer the question, based on the context. Answer "Information not found" if there is no context. Do not hallucinate.  \nquestion: {question}  \ncontext: {context}"
  2. 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.contentquestion = "How does the Transformer model use self-attention mechanisms to improve the efficiency and effectiveness of sequence-to-sequence tasks compared to traditional RNN-based models?"context = get_context(question, "propertyQA.pdfembedding.embedding")prompt = create_prompt(question, context)response = generate_response(prompt)print(response)`
复制代码
运行此文件以获取响应。
终端 3
  1. python3 ./query_doc.py
复制代码
  1. The Transformer model replaces recurrent layers with self-attention mechanisms, allowing for parallelization and capturing dependencies between words regardless of their distance in the sequence. This improves efficiency by reducing the training time and enhances effectiveness by better handling long-range dependencies.
复制代码
大规模存储和查询数据

使用 LangChain 进行扩展
优化你的体系以实现高效的索引和检索,以管理大量的论文数据和论文分析数据。这涉及设置向量数据库以有用地管理大量嵌入。在生产环境中,Indexify 可以程度扩展到多个服务器实例和和谐器,以并行简化实时数据提取和存储。你还可以微调提取graph中的参数(如 chunk_size 和 overlap),以在粒度和性能之间取得平衡。
  1. extractor: 'tensorlake/chunk-extractor'
  2.       name: 'chunker'
  3.       input_params:
  4.           chunk_size: 1000
  5.           overlap: 100
  6.       content_source: 'textextract'
复制代码
Indexify 与流行的 LangChain 框架很好地集成,LangChain 框架是一个高度可扩展的框架,用于构建 AI 应用。
终端 3
  1. pip install indexify-langchain langchain-groq
复制代码
我正在使用来自 LangChain 的基于 Indexify 的检索器。我将通报我预备好的索引和 top_k 参数。
  1. from indexify import IndexifyClient
  2. client = IndexifyClient()
  3. from indexify_langchain import IndexifyRetriever
  4. params = {"name": "propertyQA.pdfembedding.embedding", "top_k": 2}
  5. retriever = IndexifyRetriever(client=client, params=params)
复制代码
以下是完备代码:
  1. # langchain_query_doc.pyimport requestsimport dotenv# Setup retrieverfrom indexify import IndexifyClientfrom indexify_langchain import IndexifyRetrieverclient = IndexifyClient()params = {"name": "propertyQA.pdfembedding.embedding", "top_k": 2}retriever = IndexifyRetriever(client=client, params=params)from langchain_groq import ChatGroqllm = ChatGroq(    model="gemma-7b-it",    temperature=0,    max_tokens=None,    timeout=None,    max_retries=2,    api_key="API_KEY",)# Setup Chat Prompt Templatefrom langchain.prompts import ChatPromptTemplatetemplate = """You are an expert in paper interpretation,You must help me interpret the core content of the paper. Answer the question, based on the context. Answer "Information not found" if there is no context. Do not hallucinate.  \nquestion: {question}  \ncontext: {context}"
  2. ""prompt = ChatPromptTemplate.from_template(template)from langchain.schema.runnable import RunnablePassthroughfrom langchain.schema.output_parser import StrOutputParserrag_chain = (    {"context": retriever, "question": RunnablePassthrough()}    | prompt    | llm    | StrOutputParser())# Queryquery = "How does the Transformer model use self-attention mechanisms to improve the efficiency and effectiveness of sequence-to-sequence tasks compared to traditional RNN-based models?"print(rag_chain.invoke(query))
复制代码
这使用 IndexifyRetriever 根据查询检索最相关的段落。然后,它使用 ChatPromptTemplate 创建一个针对上下文的提示。该提示通报给 LLM,模型在其中处理信息并天生响应。
将查询结果存储在数据库中
执行查询并检索结果后,将这些数据存储在结构化数据库中可以确保你可以参考它并执行进一步的分析。
起首,要将查询结果存储在数据库中,请设置一个关系数据库并界说一个模式,该模式可以容纳 Indexify 返回的复杂数据结构。以下是怎样将查询结果存储在 PostgreSQL 数据库中的示例:
  1. import psycopg2
  2. # Connect to your PostgreSQL database
  3. conn = psycopg2.connect(
  4.     dbname="indexify_results",
  5.     user="your_username",
  6.     password="your_password",
  7.     host="localhost",
  8.     port="5432"
  9. )
  10. cursor = conn.cursor()
  11. # Create a table to store query results
  12. cursor.execute("""
  13.     CREATE TABLE IF NOT EXISTS query_results (
  14.         id SERIAL PRIMARY KEY,
  15.         query TEXT,
  16.         content_id VARCHAR(255),
  17.         passage TEXT
  18.     );
  19. """)
  20. conn.commit()
  21. # Example function to store results
  22. def store_results(question: str, results: list):
  23.     for result in results:
  24.         cursor.execute("""
  25.             INSERT INTO query_results (query, content_id, passage)
  26.             VALUES (%s, %s, %s);
  27.         """, (question, result['content_id'], result['text']))
  28.     conn.commit()
  29. # Retrieve and store query results
  30. question = "How does the Transformer model use self-attention mechanisms to improve the efficiency and effectiveness of sequence-to-sequence tasks compared to traditional RNN-based models?"
  31. results = client.search_index(name="propertyQA.pdfembedding.embedding", query=question, top_k=5)
  32. store_results(question, results)
  33. # Close the database connection
  34. cursor.close()
  35. conn.close()
复制代码
在界说数据库模式时,请思量数据的关联关系和复杂性。该模式应该容纳内容 ID 和段落文本以及元数据,例如时间戳、文档来源以及任何相关的标签或类别。
以下是一个你可以思量的示例模式。
  1. CREATE TABLE query_results (
  2.     id SERIAL PRIMARY KEY,
  3.     query TEXT,
  4.     content_id VARCHAR(255),
  5.     passage TEXT,
  6.     timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  7.     document_source VARCHAR(255),
  8.     tags TEXT[]
  9. );
复制代码

总结
我们已经介绍了 Indexify 怎样通过解读论文的示例,为需要更深入的实时分析的应用提供高效的数据提取和检索。我们具体介绍了 Indexify 的工作原理,包括怎样构建用于数据检索的知识库。别的,我们还观察了怎样扩展体系以处理更大的数据集和更高的吞吐量。最后,我们讨论了将结果存储在结构化数据库中以供将来参考和深入分析的方法。

免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。




欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4