如今，使用 LLM 和聊天机器人与文本数据聊天变得简单而直接。你可能已经注意到，当涉及到 PDF 文档中的表格数据时，聊天机器人表现不佳。让我们以财务文档为例，例如财务报表，其中包含以表格形式表示的不同表格和会计信息。采用我们大多数人熟悉的典型 RAG 方法实际上不起作用。

在本文中，我们将介绍解决此问题的技术。我们将学习可用于提取和预处理聊天机器人表格数据的不同工具，以便在聊天时获得更好的结果。具体来说，我们将介绍 Unstructured.io、LangChain、ChromaDB 和 MultiVector 检索器。

上图显示了我们用于完成此任务的架构。让我们开始吧。

安装依赖项

我们要做的第一件事是创建一个 Python 虚拟环境。我将使用 Python 的 Poetry 来执行此操作。使用以下命令在你的终端上实现相同的操作。

$ mkdir financial_statement_bot
$ cd financial_statement_bot
$ poetry init
$ poetry install

完成后，打开 VS 代码或任何支持笔记本的编辑器。确保选择刚刚创建的虚拟环境。

你可以使用以下命令打开 VS 代码：

$ code .

确保在 financial_statement_bot 目录下运行上述命令。这将在你的电脑上打开 VS 代码。你不必使用 VS 代码，如果你愿意，也可以使用 Google Colab 或 Conda Jupyter Environment。

在项目文件夹的根目录下创建一个 Jupyter 笔记本，在该文件中将其命名为 main.ipynb，在第一个单元格中添加以下命令并运行以安装必要的依赖项。

%pip install "unstructured[all-docs]" unstructured-client watermark python-dotenv pydantic langchain langchain-community langchain_core langchain_openai chromadb

获取财务报表 PDF 文档

本教程中使用的财务报表 PDF 文档可以从这里的链接获取。

下载完成后，在项目目录根目录下新建一个文件夹，名为 financial_statement_bot/data

$ mkdir data

完成后，添加刚下载的文件。我的情况是：financial_statement_bot/data/Sample-Accounting-Income-Statement-PDF-File.pdf。

机器基本信息

你可以获取你的计算机或硬件的信息，这只是为了确保没有潜在的硬件兼容性问题，或者只是以防万一，你想知道我是在哪个硬件上运行的。总之，下面的命令就可以做到这一点：

# Get installed versions
import watermark
%load_ext watermark
%watermark -n -v -m -g -b

另外，为了忽略警告，请在另一个单元格中添加以下代码：

# Ignore warnings
import warnings
warnings.filterwarnings('ignore')

你还可以使用以下命令查看系统中安装的非结构化软件包列表：

# Get list of installed packages with unstructured
import unstructured.partition
help(unstructured.partition)

加载 PDF 文档

现在我们要加载 PDF 文档了。为此，请创建一个新单元格并添加以下代码：

from unstructured.partition.pdf import partition_pdf
# Path to the PDF file
pdf_path = "./data/Sample-Accounting-Income-Statement-PDF-File.pdf"
# read the file and get list element each of a page of the parsed pdf file
elements = partition_pdf(pdf_path)

上述代码块应返回在文档中使用 Unstructured.io 确定的元素类别列表。你可以将元素变量打印出来，以便对其进行挑选。

我们还可以获得文档中所有已识别元素的数量。

print(f"Lenght of elements: {len(elements)}")

为了便于阅读，让我们把元素对象转换成 JSON 对象。这样我们打印出来时就更容易理解了。

import json
element_dict = [el.to_dict() for el in elements]
output = json.dumps(element_dict, indent=2)

print(output)

我们还可以从 PDF 文档中获取所有唯一标识元素的列表。

unique_element_types = set()
for item in element_dict:
    unique_element_types.add(item['type'])

从 PDF 文件中唯一标识的元素集合中，你可以发现没有表格元素被标识出来。但是，我们的 PDF 文件中有表格。

在这种情况下，我们的表格被识别为普通文本元素。

至少在你的文档中的第 8 单元，也就是我的文档中的第 8 单元。在我们打印输出变量的部分，如果你滚动完成并点击蓝色文本链接写入文本编辑器，你应该会被重定向到另一个文档。

在本文档中，按 CTRL + F 打开查找功能，输入 CURRENT 并启用大小写敏感性。为什么是 “CURRENT”？因为它显示在文档的表格部分，我只是想看看它被分配给了哪个元素，因为还没有表格元素，“CURRENT ”应该是一个表格元素。

从上图可以清楚地看到，“CURRENT ”一词所属的所有列都已被确定为标题类别。

比较上面两张图片中高亮显示的文本。这表明我们的表格只是被提取为文本数据。这不是我们想要的。如果你继续根据这些数据构建聊天机器人，你的机器人将获得错误的信息，从而得到错误的答案，并降低其他下游任务的性能。

为了解决这个问题，我们需要将表格提取为表格，而不是 Unstructure.io 提供的任何其他元素。

提取表格

要提取表格，我们可以在本地运行，也可以使用 Unstructured 自带的免费或付费 API。

在本地运行时，你需要在机器上安装 Tesseract，还需要一些非常好的 GPU 或强大的 CPU。但不是每个人都能做到这一点。因此，我将使用他们的免费 API。

该 API 使用 OCR 和其他技术从 PDF 文档中提取表格数据。

要开始使用，首先需要获取一个 API 密钥，可以通过此处的链接获取。

获得 API 密钥后，在项目根目录下新建一个名为 .env 的文件。在该文件中按以下格式添加 API 密钥：

UNSTRUCTURED_API_KEY=Your_APPI_key
UNSTRUCTURED_API_URL=https://api.unstructured.io/general/v0/general

加载环境变量

在 .env 文件中写入 API 密钥后，运行以下命令加载 API 密钥。

import os
from dotenv import load_dotenv, find_dotenv
# Load environment variables from .env file
load_dotenv(find_dotenv())

然后，你可以使用以下代码访问你的 API 密钥：

unstructured_api_key = os.environ.get('UNSTRUCTURED_API_KEY')
unstructured_api_url = os.environ.get('UNSTRUCTURED_API_URL')

创建非结构化客户端

我们需要创建一个Unstructured客户端，使用上面载入的API密钥和URL，连接Unstructured的免费API，以执行提取。

使用下面的代码就可以做到这一点：

from unstructured_client import UnstructuredClient
client = UnstructuredClient(
    api_key_auth=unstructured_api_key, 
    server_url=unstructured_api_url
)

使用非结构化 API 客户端提取表格和文本元素

建立连接后，让我们开始提取信息！

首先，让我们引入必要的导入：

from unstructured_client.models import shared
from unstructured_client.models.errors import SDKError
from unstructured.staging.base import dict_to_elements

现在，让我们提取信息！

with open(pdf_path, "rb") as f:
    files=shared.Files(
        content=f.read(),
        file_name=pdf_path,
    )
req = shared.PartitionParameters(
    files=files,
    strategy="hi_res",
    hi_res_model_name="yolox",
    skip_infer_table_types=[],
    pdf_infer_table_structure=True,
)
try:
    resp = client.general.partition(req)
    elements = dict_to_elements(resp.elements)
except SDKError as e:
    print(e)

我们可以获得提取出的所有唯一元素的列表。

# Unique element types
unique_elements_set = set()
for el in elements:
    unique_elements_set.add(el.category)

现在你可以看到我们能够提取表格数据了。

检查所有提取的表格

既然我们已经能够提取表格数据，那么让我们来检查一下我们能够提取的不同类别的元素。

tables = [el for el in elements if el.category == "Table"]

print(tables)

print(f"Number of tables: {len(tables)}")

从上图我们可以看到，我们的文档中有 8 张表格。是的，没错。

你可以检查给定表格中的文本。

tables[0].text

你还可以使用以下方法查看表对象的元数据：

tables[0].metadata

表格 HTML 数据

在 Unstructured 中提取的每个表都会以 HTML 文档的形式返回。让我们查看提取的第一个表的 HTML。

first_table_html = tables[0].metadata.text_as_html
first_table_html

这看起来不太友好，不是吗？让我们让它看起来更像 HTML 格式：

from io import StringIO 
from lxml import etree
parser = etree.XMLParser(remove_blank_text=True)
file_obj = StringIO(first_table_html)
tree = etree.parse(file_obj, parser)
print(etree.tostring(tree, pretty_print=True).decode())

让我们把它转换成一个实际的表格：

from IPython.core.display import HTML
HTML(first_table_html)

你现在明白上面的两张图片了吗？好了，我们提取并转换成 HTML 的 HTML 表格就是财务报表 PDF 文档中的表格。现在明白了吗？

将提取的非结构化表格转换为 Pandas DataFrame

你也可以将提取的表格转换为 Pandas DataFrame 对象，下面是具体方法：

import pandas as pd
dfs = pd.read_html(first_table_html)
df = dfs[0]
df.head()

当我们提取出表格并将提取出的表格可视化后，我们就可以继续处理提取出的其他类别的元素，而这些元素并不是表格元素。

检查所有提取的文本数据元素

要获取提取的文本元素，可以使用下图所示的方法：

texts = [el for el in elements if el.category != "Table"]
texts[0].text

基本上，我们正在获取所有不属于 “表格 ”类别的元素。这些元素都是文本元素。这些元素包括标题、电子邮件地址等。你可以在这里阅读更多关于元素所属类别的信息。

如果你愿意，还可以将所有提取的元素组合起来，同时保持元素的顺序与实际文档的顺序一致：

extracted_text = ""
for cat in elements:
    if cat.category == "Formula":
        extracted_text += cat.text + "\n"
    if cat.category == "FigureCaption":
        extracted_text += cat.text + "\n"
    if cat.category == "NarrativeText":
        extracted_text += cat.text + "\n"
    if cat.category == "ListItem":
        extracted_text += cat.text + "\n"
    if cat.category == "Title":
        extracted_text += cat.text + "\n"
    if cat.category == "Address":
        extracted_text += cat.text + "\n"
    if cat.category == "EmailAddress":
        extracted_text += cat.text + "\n"
    if cat.category == "Table":
        extracted_text += cat.metadata.text_as_html + "\n"
    if cat.category == "Header":
        extracted_text += cat.text + "\n"
    if cat.category == "Footer":
        extracted_text += cat.text + "\n"
    if cat.category == "CodeSnippet":
        extracted_text += cat.text + "\n"
    if cat.category == "UncategorizedText":
        extracted_text += cat.text + "\n"

print(extracted_text)

你可以将其显示为markdown：

# Display the extracted text in markdown format
from IPython.display import Markdown
Markdown(extracted_text)

对提取的表格和文本进行预处理

让我们继续前进，将提取的表格和文本数据预处理成我们自己的文档类型。这将用于创建嵌入和多向量检索存储。

from typing import Any
from pydantic import BaseModel
class Element(BaseModel):
    type: str
    page_content: Any
# Categorize by type
categorized_elements = []
for element in elements:
    if "unstructured.documents.elements.Table" in str(type(element)):
        categorized_elements.append(Element(type="table", page_content=str(element.metadata.text_as_html)))
    elif "unstructured.documents.elements.NarrativeText" in str(type(element)):
        categorized_elements.append(Element(type="text", page_content=str(element)))
    elif "unstructured.documents.elements.ListItem" in str(type(element)):
        categorized_elements.append(Element(type="text", page_content=str(element)))
    elif "unstructured.documents.elements.Title" in str(type(element)):
        categorized_elements.append(Element(type="text", page_content=str(element)))
    elif "unstructured.documents.elements.Address" in str(type(element)):
        categorized_elements.append(Element(type="text", page_content=str(element)))
    elif "unstructured.documents.elements.EmailAddress" in str(type(element)):
        categorized_elements.append(Element(type="text", page_content=str(element)))
    elif "unstructured.documents.elements.Header" in str(type(element)):
        categorized_elements.append(Element(type="CodeSnippet", page_content=str(element)))
    elif "unstructured.documents.elements.CodeSnippet" in str(type(element)):
        categorized_elements.append(Element(type="text", page_content=str(element)))
    elif "unstructured.documents.elements.UncategorizedText" in str(type(element)):
        categorized_elements.append(Element(type="text", page_content=str(element)))

现在，让我们把所有表格元素和文本元素放在一起。

# Tables
table_elements = [e for e in categorized_elements if e.type == "table"]
print(len(table_elements))
# Text
text_elements = [e for e in categorized_elements if e.type == "text"]
print(len(text_elements))

根据上述信息，我们有 8 种不同的表格元素和 44 种不同的非表格元素类别。

你可以使用以下方法访问非表格元素（又称文本元素）的文本：

for text_element in text_elements:
    print(text_element.page_content)

表格元素类别也可以这样做：

all_tables_html = ""
for table in table_elements:
    all_tables_html += table.page_content + "</br></br>"
    print(table.page_content)

Markdown(all_tables_html)

文本预处理：生成摘要

现在我们已经能够提取表格和文本内容。我们需要分别创建摘要。我们可以将原始表格和文本存储到矢量数据库中，但如果你真正理解了矢量数据库在检索类似内容时的工作原理，这就不是一个好主意了。

因此，最好对每个表格的摘要或文本内容（只是文本嵌入）进行相似性搜索。从这里我们可以知道我们需要检索哪个表。

在摘要内容的相似性搜索中获得更高的匹配度意味着原始数据将提供更多的上下文数据来回答给定的查询。

这就是我们现在的情况，与我之前向大家展示的图表有关。

让我们继续创建摘要。

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

让我们使用 LCEL（LangChain Expression Language）来创建摘要链。

summary_chain = (
    {"doc": lambda x: x}
    | ChatPromptTemplate.from_template("Summarize the following html tables or text given to you:\n\n{doc}")
    | ChatOpenAI(max_retries=0)
    | StrOutputParser()
)

五批的表格摘要

# Table summaries
tables_content = [i.page_content for i in table_elements]
table_summaries = summary_chain.batch(tables_content, {"max_concurrency": 5})

table_summaries[:2]

下图是第一张表格，可以看到摘要中的正确描述：

现在，如果我们提出任何有关资产和负债或股东的问题，将对生成的摘要进行相似性搜索，匹配度最高的将是第一个摘要。然后，与该摘要相关的表格将被返回并传递给 LLM，LLM 可以检查原始表格数据（HTML）并适当地回答问题。

文本元素摘要生成

现在我们也来为文本元素生成摘要，与表格和原始表格数据摘要相同的逻辑也适用于文本元素。

# Text summaries
texts_content = [i.page_content for i in text_elements]
text_summaries = summary_chain.batch(texts_content, {"max_concurrency": 5})

text_summaries[:2]

创建嵌入

现在我们有了摘要和原始数据，我们只需嵌入摘要并将其存储在矢量存储中。对于每个嵌入，我们将附加一个元数据，将每个摘要嵌入链接到其实际的原始数据组件。

这样，一旦我们检索到与查询最相似的嵌入，我们就会使用与实际原始数据组件关联 ID 的元数据，然后返回原始数据组件。这就是多向量检索器的工作原理。

让我们为此编写代码：

import uuid
from langchain.vectorstores import Chroma
from langchain.storage import InMemoryStore
from langchain.schema.document import Document
from langchain.embeddings import OpenAIEmbeddings
from langchain.retrievers.multi_vector import MultiVectorRetriever

# The storage layer for the parent documents
store = InMemoryStore()
id_key = "doc_id"
# The vectorstore to use to index the child chunks
vectorstore = Chroma(
    collection_name="financials",
    embedding_function=OpenAIEmbeddings(),
    persist_directory="./chroma_data",
)
# The retriever (empty to start)
retriever = MultiVectorRetriever(
    vectorstore=vectorstore,
    docstore=store,
    id_key=id_key,
)
# Table summaries embeddings and storage
doc_ids = [str(uuid.uuid4()) for _ in table_elements]
summary_tables = [Document(page_content=s,metadata={id_key: doc_ids[i]}) for i, s in enumerate(table_summaries)]
retriever.vectorstore.add_documents(summary_tables)
retriever.docstore.mset(list(zip(doc_ids, table_elements)))
# Text summaries embeddings and storage
doc_ids = [str(uuid.uuid4()) for _ in text_elements]
summary_texts = [Document(page_content=s,metadata={id_key: doc_ids[i]}) for i, s in enumerate(text_summaries)]
retriever.vectorstore.add_documents(summary_texts)
retriever.docstore.mset(list(zip(doc_ids, text_elements)))

我们可以通过检索一些信息来测试它。

retriever_first_response = retriever.invoke("Give me a summary of the CASH FLOWS FROM OPERATING ACTIVITIES in a table format for the year 2001 and 2002")

print(retriever_first_response)
retriever_first_response_page_content = retriever_first_response[0].page_content
print(retriever_first_response_page_content)

Markdown(retriever_first_response_page_content)

从上图可以看出，我们已经检索到了正确的表格来回答用户的问题。

从上图检索到的表中可以看到，整个表都被传给了 LLM，但 LLM 只利用了它所需的部分来生成最终的响应答案。

创建财务报表聊天机器人本身

现在，我们已经完成了读取 PDF 文件、提取文本和表格并将其存储到多向量检索器中的所有工作。现在，让我们来创建一个机器人，它将接收从多向量存储中提取的信息，并利用这些信息回答查询。

对于这个机器人，我将使用 LCEL（LangChain 表达式语言）。

from operator import itemgetter
from langchain.schema.runnable import RunnablePassthrough
# Prompt template
template = """Answer the question based only on the following context, which can include text and tables:
{context}
Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
# LLM
model = ChatOpenAI(temperature=0,model="gpt-4")
#model = ChatOpenAI(temperature=0,model="gpt-3.5-turbo")
# RAG pipeline
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

好了，是时候测试一下了：

response = chain.invoke("What is the Leasehold improvements cost")
print(response)

Bot 的答复： 租赁改进费用为 37 350 美元。

让我们看看有哪些信息（表格）被检索出来并传递给了LLM

table_respones = retriever.invoke("What is the Leasehold improvements cost")

Markdown(table_respones[0].page_content)

让我们再试一个问题：

response = chain.invoke("Give me a summary of the CASH FLOWS FROM OPERATING ACTIVITIES in a table format for the year 2001 and 2002")
print(response)

Markdown(response)

这个回答听起来相当准确。

结论

在本文中，我们探索了使用 Unstructured.io 构建数据提取管道、预处理提取的数据、创建摘要并将数据存储在多向量检索器中，然后在此基础上构建数据库。