AI系统中PDF表格内容提取与向量数据库检索实战指南

1次阅读

没有评论

共计 3050 个字符，预计需要花费 8 分钟才能阅读完成。

在 AI 系统中处理 PDF 表格时，经常会遇到以下问题：

表格格式复杂，包含合并单元格、嵌套表格等情况
内容提取不完整，尤其是跨页表格
提取后的数据结构混乱，难以直接使用
检索效率低下，无法快速找到相关内容

这些问题导致后续的分析和应用变得困难，因此需要一套完整的解决方案。

在处理 PDF 表格内容时，常用的 Python 库有以下几种：

PyPDF2：轻量级，但表格处理能力有限
pdfplumber：专注于文本和表格提取，支持复杂表格结构
Camelot：专门用于表格提取，但依赖较重

对于初学者来说，pdfplumber 是一个不错的选择，它提供了直观的 API 和良好的错误处理机制。

首先安装必要的库：

pip install pdfplumber sentence-transformers faiss-cpu

然后使用 pdfplumber 提取表格：

import pdfplumber

with pdfplumber.open("example.pdf") as pdf:
    for page in pdf.pages:
        tables = page.extract_tables()
        for table in tables:
            for row in table:
                print(row)

提取的表格数据通常是二维列表，我们可以将其转换为字典列表：

def convert_to_dicts(table_data, headers=None):
    if headers is None:
        headers = table_data[0]
        table_data = table_data[1:]

    result = []
    for row in table_data:
        row_dict = {}
        for i, cell in enumerate(row):
            row_dict[headers[i]] = cell
        result.append(row_dict)
    return result

安装 Sentence Transformers 并生成向量：

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')

def generate_embeddings(texts):
    return model.encode(texts)

FAISS 是 Facebook 开源的向量相似度搜索库：

import faiss
import numpy as np

dimension = 384  # all-MiniLM-L6-v2 的向量维度
index = faiss.IndexFlatL2(dimension)

# 假设我们已经有了向量数据
vectors = np.array([...]).astype('float32')
index.add(vectors)

在检索时，可以设置相似度阈值来过滤结果：

def search(query, index, threshold=0.7):
    query_vector = generate_embeddings([query])[0]
    D, I = index.search(np.array([query_vector]), k=5)

    results = []
    for distance, idx in zip(D[0], I[0]):
        if 1 - distance > threshold:  # 余弦相似度
            results.append((idx, 1 - distance))
    return results

对于多列表格，可以将多列内容拼接后进行向量化：

def combine_columns(row_dict, columns):
    return " ".join([str(row_dict[col]) for col in columns])

表格跨页问题：
检查表格是否跨页，如果是，可能需要手动合并
使用 pdfplumber 的 page.bbox 和table.bbox属性判断表格边界
特殊字符处理：
使用正则表达式清理特殊字符
处理 Unicode 编码问题
空单元格处理：
为空的单元格填充默认值
或者在生成向量时跳过这些行

对于大批量 PDF 文件处理，可以考虑以下优化：

使用多进程并行处理
预处理 PDF 文件，提取所有表格后再统一向量化
使用 GPU 加速向量生成
分批写入向量数据库

import pdfplumber
from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

# 1. 提取表格
def extract_tables_from_pdf(pdf_path):
    all_tables = []
    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            tables = page.extract_tables()
            for table in tables:
                all_tables.append(table)
    return all_tables

# 2. 转换为结构化数据
def convert_to_dicts(tables, headers=None):
    all_dicts = []
    for table in tables:
        if headers is None:
            current_headers = table[0]
            rows = table[1:]
        else:
            current_headers = headers
            rows = table

        for row in rows:
            row_dict = {}
            for i, cell in enumerate(row):
                if i < len(current_headers):
                    row_dict[current_headers[i]] = cell
            all_dicts.append(row_dict)
    return all_dicts

# 3. 生成向量
def setup_model():
    return SentenceTransformer('all-MiniLM-L6-v2')

# 4. 构建索引
def build_index(vectors):
    dimension = vectors.shape[1]
    index = faiss.IndexFlatL2(dimension)
    index.add(vectors)
    return index

# 主流程
if __name__ == "__main__":
    # 提取表格
    tables = extract_tables_from_pdf("example.pdf")

    # 转换为结构化数据
    data = convert_to_dicts(tables)

    # 生成文本表示
    texts = [" ".join([str(v) for v in row.values()]) for row in data]

    # 生成向量
    model = setup_model()
    vectors = model.encode(texts)

    # 构建索引
    index = build_index(vectors)

    # 保存索引
    faiss.write_index(index, "pdf_table.index")