基于多模态LLM的PDF处理管道

APPLICATION Dec 17, 2024

推荐： ComfyUI-PandasAI | ComfyUI-Autodistill | ComfyUI节点速查手册

自动化文档处理是 ChatGPT 革命的最大赢家之一，因为 LLM 能够在零样本设置中处理广泛的主题和任务，这意味着无需域内标记的训练数据。这使得构建 AI 驱动的应用程序来处理、解析和自动理解任意文档变得更加容易。虽然使用 LLM 的简单方法仍然受到非文本上下文（例如图形、图像和表格）的阻碍，但这正是我们将在本篇博文中尝试解决的问题，特别关注 PDF。

从根本上讲，PDF 只是字符、图像和线条及其精确坐标的集合。它们没有固有的“文本”结构，也不是为作为文本处理而构建的，而只是按原样查看。这就是使用它们变得困难的原因，因为纯文本方法无法捕获这些类型文档中的所有布局和视觉元素，从而导致上下文和信息的大量丢失。

绕过此“纯文本”限制的一种方法是，在将文档输入 LLM 之前，通过检测表格、图像和布局对文档进行大量预处理。表格可以解析为 Markdown 或 JSON，图像和图形可以用其标题表示，文本可以按原样输入。但是，这种方法需要自定义模型，并且仍会导致一些信息丢失，那么我们能做得更好吗？

1、多模态 LLM

大多数最新的大型模型现在都是多模态的，这意味着它们可以处理文本、代码和图像等多种模态。这为我们的问题提供了一种更简单的解决方案，即一个模型可以同时完成所有工作。因此，我们不必为图像添加标题和解析表格，而是可以将页面作为图像输入并按原样处理。我们的管道将能够加载 PDF，将每个页面提取为图像，将其拆分为块（使用 LLM），并索引每个块。如果检索到块，则将整个页面包含在 LLM 上下文中以执行任务。接下来，我们将详细介绍如何在实践中实现这一点。

2、处理管道

我们正在实施的管道是一个两步过程。首先，我们将每个页面分割成重要的块并总结每个块。其次，我们对块进行一次索引，然后在每次收到请求时搜索这些块，并在 LLM 上下文中包含每个检索到的块的完整上下文。

2.1 页面分割和摘要

我们将页面提取为图像，并将它们中的每一个传递给多模态 LLM 进行分割。像 Gemini 这样的模型可以轻松理解和处理页面布局：

表格被识别为一个块。
图形形成另一个块。
文本块被分割成单独的块。
...

对于每个元素，LLM 都会生成一个摘要，然后可以嵌入并索引到矢量数据库中。

2.2 嵌入和上下文检索

在本教程中，我们将仅使用文本嵌入以简化操作，但一个改进是直接使用视觉嵌入。

数据库中的每个条目包括：

区块摘要。
找到它的页码。
指向完整页面图像表示的链接，用于添加上下文。

此架构允许在本地级别搜索（在区块级别）的同时跟踪上下文（通过链接回完整页面）。例如，如果搜索查询检索到某个项目，则代理可以包含整个页面图像，以便为 LLM 提供完整布局和额外上下文，从而最大限度地提高响应质量。

通过提供完整图像，在生成响应时，LLM 可以使用所有视觉提示和重要布局信息（如图像、标题、项目符号……）和相邻项目（表格、段落……）。

3、代理

我们将每个步骤实现为单独的可重复使用的代理：

第一个代理用于解析、分块和摘要。这涉及将文档分割成重要的块，然后为每个块生成摘要。此代理只需在每个 PDF 中运行一次即可预处理文档。

第二个代理管理索引、搜索和检索。这包括将块嵌入到矢量数据库中以实现高效搜索。每个文档执行一次索引，而搜索可以根据不同查询的需要重复多次。

对于这两个代理，我们使用 Gemini，这是一种具有强大视觉理解能力的多模 LLM。

3.1 解析和分块代理

第一个代理负责将每个页面分割成有意义的块并总结每个块，遵循以下步骤：

步骤 1：将 PDF 页面提取为图像

我们使用 pdf2image 库。然后以 Base64 格式对图像进行编码，以简化将它们添加到 LLM 请求的过程。

以下是实现：

from document_ai_agents.document_utils import extract_images_from_pdf
from document_ai_agents.image_utils import pil_image_to_base64_jpeg
from pathlib import Path

class DocumentParsingAgent:
    @classmethod
    def get_images(cls, state):
        """
        Extract pages of a PDF as Base64-encoded JPEG images.
        """
        assert Path(state.document_path).is_file(), "File does not exist"
        # Extract images from PDF
        images = extract_images_from_pdf(state.document_path)
        assert images, "No images extracted"
        # Convert images to Base64-encoded JPEG
        pages_as_base64_jpeg_images = [pil_image_to_base64_jpeg(x) for x in images]
        return {"pages_as_base64_jpeg_images": pages_as_base64_jpeg_images}

extract_images_from_pdf：将 PDF 的每一页提取为 PIL 图像。
pil_image_to_base64_jpeg：将图像转换为 Base64 编码的 JPEG 格式。

步骤 2：分块和汇总

然后将每幅图像发送到 LLM 进行分割和汇总。我们使用结构化输出来确保我们得到预测我们期望的格式：

from pydantic import BaseModel, Field
from typing import Literal
import json
import google.generativeai as genai
from langchain_core.documents import Document

class DetectedLayoutItem(BaseModel):
    """
    Schema for each detected layout element on a page.
    """
    element_type: Literal["Table", "Figure", "Image", "Text-block"] = Field(
        ..., 
        description="Type of detected item. Examples: Table, Figure, Image, Text-block."
    )
    summary: str = Field(..., description="A detailed description of the layout item.")

class LayoutElements(BaseModel):
    """
    Schema for the list of layout elements on a page.
    """
    layout_items: list[DetectedLayoutItem] = []

class FindLayoutItemsInput(BaseModel):
    """
    Input schema for processing a single page.
    """
    document_path: str
    base64_jpeg: str
    page_number: int

class DocumentParsingAgent:
    def __init__(self, model_name="gemini-1.5-flash-002"):
        """
        Initialize the LLM with the appropriate schema.
        """
        layout_elements_schema = prepare_schema_for_gemini(LayoutElements)
        self.model_name = model_name
        self.model = genai.GenerativeModel(
            self.model_name,
            generation_config={
                "response_mime_type": "application/json",
                "response_schema": layout_elements_schema,
            },
        )
    def find_layout_items(self, state: FindLayoutItemsInput):
        """
        Send a page image to the LLM for segmentation and summarization.
        """
        messages = [
            f"Find and summarize all the relevant layout elements in this PDF page in the following format: "
            f"{LayoutElements.schema_json()}. "
            f"Tables should have at least two columns and at least two rows. "
            f"The coordinates should overlap with each layout item.",
            {"mime_type": "image/jpeg", "data": state.base64_jpeg},
        ]
        # Send the prompt to the LLM
        result = self.model.generate_content(messages)
        data = json.loads(result.text)
        
        # Convert the JSON output into documents
        documents = [
            Document(
                page_content=item["summary"],
                metadata={
                    "page_number": state.page_number,
                    "element_type": item["element_type"],
                    "document_path": state.document_path,
                },
            )
            for item in data["layout_items"]
        ]
        return {"documents": documents}

LayoutElements 模式定义输出的结构，包括每个布局项类型（表格、图形等）及其摘要。

步骤 3：页面的并行处理

页面处理并行处理以提高速度。由于处理是 io 绑定的，因此以下方法会创建一个任务列表来一次性处理所有页面图像：

from langgraph.types import Send

class DocumentParsingAgent:
    @classmethod
    def continue_to_find_layout_items(cls, state):
        """
        Generate tasks to process each page in parallel.
        """
        return [
            Send(
                "find_layout_items",
                FindLayoutItemsInput(
                    base64_jpeg=base64_jpeg,
                    page_number=i,
                    document_path=state.document_path,
                ),
            )
            for i, base64_jpeg in enumerate(state.pages_as_base64_jpeg_images)
        ]

每个页面都作为独立任务发送到 find_layout_items 函数。

完整工作流程

代理的工作流程是使用 StateGraph 构建的，将图像提取和布局检测步骤链接到统一的管道 ->

from langgraph.graph import StateGraph, START, END

class DocumentParsingAgent:
    def build_agent(self):
        """
        Build the agent workflow using a state graph.
        """
        builder = StateGraph(DocumentLayoutParsingState)
        
        # Add nodes for image extraction and layout item detection
        builder.add_node("get_images", self.get_images)
        builder.add_node("find_layout_items", self.find_layout_items)
        # Define the flow of the graph
        builder.add_edge(START, "get_images")
        builder.add_conditional_edges("get_images", self.continue_to_find_layout_items)
        builder.add_edge("find_layout_items", END)
        
        self.graph = builder.compile()

要在示例 PDF 上运行代理，我们需要执行以下操作：

if __name__ == "__main__":
    _state = DocumentLayoutParsingState(
        document_path="path/to/document.pdf"
    )
    agent = DocumentParsingAgent()
    
    # Step 1: Extract images from PDF
    result_images = agent.get_images(_state)
    _state.pages_as_base64_jpeg_images = result_images["pages_as_base64_jpeg_images"]
    
    # Step 2: Process the first page (as an example)
    result_layout = agent.find_layout_items(
        FindLayoutItemsInput(
            base64_jpeg=_state.pages_as_base64_jpeg_images[0],
            page_number=0,
            document_path=_state.document_path,
        )
    )
    # Display the results
    for item in result_layout["documents"]:
        print(item.page_content)
        print(item.metadata["element_type"])

这将产生 PDF 的解析、分段和汇总表示，这是我们接下来要构建的第二个代理的输入。

3.2 RAG 代理

第二个代理处理索引和检索部分。它将前一个代理的文档保存到矢量数据库中，并使用结果进行检索。这可以分为两个独立的步骤，即索引和检索。

步骤 1：索引拆分文档

使用生成的摘要，我们将其矢量化并保存在 ChromaDB 数据库中：

class DocumentRAGAgent:
    def index_documents(self, state: DocumentRAGState):
        """
        Index the parsed documents into the vector store.
        """
        assert state.documents, "Documents should have at least one element"
        # Check if the document is already indexed
        if self.vector_store.get(where={"document_path": state.document_path})["ids"]:
            logger.info(
                "Documents for this file are already indexed, exiting this node"
            )
            return  # Skip indexing if already done
        # Add parsed documents to the vector store
        self.vector_store.add_documents(state.documents)
        logger.info(f"Indexed {len(state.documents)} documents for {state.document_path}")

index_documents 方法嵌入块摘要存入向量存储。我们保留文档路径和页码等元数据以供日后使用。

步骤 2 ：处理问题

当用户提出问题时，代理会在向量存储中搜索最相关的块。它会检索摘要和相应的页面图像以进行上下文理解。

class DocumentRAGAgent:
    def answer_question(self, state: DocumentRAGState):
        """
        Retrieve relevant chunks and generate a response to the user's question.
        """
        # Retrieve the top-k relevant documents based on the query
        relevant_documents: list[Document] = self.retriever.invoke(state.question)

        # Retrieve corresponding page images (avoid duplicates)
        images = list(
            set(
                [
                    state.pages_as_base64_jpeg_images[doc.metadata["page_number"]]
                    for doc in relevant_documents
                ]
            )
        )
        logger.info(f"Responding to question: {state.question}")
        # Construct the prompt: Combine images, relevant summaries, and the question
        messages = (
            [{"mime_type": "image/jpeg", "data": base64_jpeg} for base64_jpeg in images]
            + [doc.page_content for doc in relevant_documents]
            + [
                f"Answer this question using the context images and text elements only: {state.question}",
            ]
        )
        # Generate the response using the LLM
        response = self.model.generate_content(messages)
        return {"response": response.text, "relevant_documents": relevant_documents}

检索器查询向量存储以找到与用户问题最相关的块。然后，我们为 LLM (Gemini) 构建上下文，它将文本块和图像组合在一起以生成响应。

完整的代理工作流程

代理工作流程有两个阶段，一个索引阶段和一个问题回答阶段：

class DocumentRAGAgent:
    def build_agent(self):
        """
        Build the RAG agent workflow.
        """
        builder = StateGraph(DocumentRAGState)
        # Add nodes for indexing and answering questions
        builder.add_node("index_documents", self.index_documents)
        builder.add_node("answer_question", self.answer_question)
        # Define the workflow
        builder.add_edge(START, "index_documents")
        builder.add_edge("index_documents", "answer_question")
        builder.add_edge("answer_question", END)
        self.graph = builder.compile()

示例运行：

if __name__ == "__main__":
    from pathlib import Path

  # Import the first agent to parse the document
    from document_ai_agents.document_parsing_agent import (
        DocumentLayoutParsingState,
        DocumentParsingAgent,
    )
    # Step 1: Parse the document using the first agent
    state1 = DocumentLayoutParsingState(
        document_path=str(Path(__file__).parents[1] / "data" / "docs.pdf")
    )
    agent1 = DocumentParsingAgent()
    result1 = agent1.graph.invoke(state1)
    # Step 2: Set up the second agent for retrieval and answering
    state2 = DocumentRAGState(
        question="Who was acknowledged in this paper?",
        document_path=str(Path(__file__).parents[1] / "data" / "docs.pdf"),
        pages_as_base64_jpeg_images=result1["pages_as_base64_jpeg_images"],
        documents=result1["documents"],
    )
    agent2 = DocumentRAGAgent()
    # Index the documents
    agent2.graph.invoke(state2)
    # Answer the first question
    result2 = agent2.graph.invoke(state2)
    print(result2["response"])
    # Answer a second question
    state3 = DocumentRAGState(
        question="What is the macro average when fine-tuning on PubLayNet using M-RCNN?",
        document_path=str(Path(__file__).parents[1] / "data" / "docs.pdf"),
        pages_as_base64_jpeg_images=result1["pages_as_base64_jpeg_images"],
        documents=result1["documents"],
    )
    result3 = agent2.graph.invoke(state3)
    print(result3["response"])

通过此实现，文档处理、检索和问答的管道已完成。