利用Gemini构建PDF文档AI管道的原理、实现与代码示例

文档处理的自动化需求正在快速增长，尤其是在PDF这类顽固格式面前，很多传统方案显得力不从心。大型语言模型的爆发，尤其是像ChatGPT这类模型的走红，确实给自动化文档处理带来了新的解题思路。不过，PDF的复杂性——那些嵌入的图表、表格、复杂的排版——总让纯文本方法栽跟头。今天就来拆解一下，如何用Gemini这样的多模态大模型，搭建一条真正能用的PDF文档AI处理管道，把信息提取的效率和精度都提上去。

PDF文档处理的挑战

PDF的设计初衷是“所见即所得”，保证文档在不同设备上显示一致。但问题在于，它本质上是一堆字符、图像、线条和坐标的集合，并没有真正的“文本流”结构。这意味着，你没法像处理Word文档那样直接提取文字——纯文本方式会丢掉大量布局和视觉信息，而表格、图表、图片里的关键数据，正是理解文档的核心。传统工具处理这些非文本元素时，要么直接忽略，要么识别得一塌糊涂，最终导致信息残缺或误读。

Gemini多模态LLM的优势

针对上述痛点，多模态大模型给出了更优雅的解法。Gemini的能力边界不止于文本和代码，还能理解图像。这意味着，你不再需要拼凑一堆OCR、表格识别、图像分类的独立工具，一个模型就能通吃。它能直接“看懂”页面布局，区分表格、图片和文本块，并将它们转换成下游任务可用的结构化数据。这不仅提升了处理精度，也让整个管线的设计和维护变得简洁得多。

构建文档AI管道的具体步骤

（一）页面分割与总结（Agent 1）

整个管线的第一步，是把PDF拆解成模型能“消化”的单元。具体来说，就是先把PDF的每一页转成图片，再交给Gemini去做布局分析和总结。

1. 提取PDF页面为图像
   这一步很直接：用pdf2image库把每一页转为PIL图像，再编码成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"
        images = extract_images_from_pdf(state.document_path)
        assert images, "No images extracted"
        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}

2. 使用LLM进行分割和总结
   把Base64图片发给Gemini，同时定义好输出结构，告诉模型要识别哪种布局元素以及如何总结。通过LayoutElements和DetectedLayoutItem这类类定义，明确每个元素的类型（表格、图片、文本块等）及其描述。这样，模型就能准确区分“这里是一张表”、“那边是一张图”，并生成对应的内容摘要。

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"):
        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 ha ve 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},
        ]
        result = self.model.generate_content(messages)
        data = json.loads(result.text)

        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}

3. 并行处理页面
   处理一本数百页的PDF时，逐页串行显然不现实。所以这里用并行机制，每页作为一个独立任务发给find_layout_items函数。LangGraph的Send对象能很好地管理这类fan-out模式，显著缩短总处理时间。

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)
        ]

完整的工作流可以这样搭建：

from langgraph.graph import StateGraph, START, END

class DocumentParsingAgent:
    def build_agent(self):
        """Build the agent workflow using a state graph."""
        builder = StateGraph(DocumentLayoutParsingState)

        builder.add_node("get_images", self.get_images)
        builder.add_node("find_layout_items", self.find_layout_items)

        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()

运行测试：

if __name__ == "__main__":
    _state = DocumentLayoutParsingState(document_path="path/to/document.pdf")
    agent = DocumentParsingAgent()

    result_images = agent.get_images(_state)
    _state.pages_as_base64_jpeg_images = result_images["pages_as_base64_jpeg_images"]

    result_layout = agent.find_layout_items(
        FindLayoutItemsInput(
            base64_jpeg=_state.pages_as_base64_jpeg_images[0],
            page_number=0,
            document_path=_state.document_path,
        )
    )
    for item in result_layout["documents"]:
        print(item.page_content)
        print(item.metadata["element_type"])

（二）嵌入和上下文检索（Agent 2）

Agent 1产出的文档摘要，需要被高效地索引和检索，才能支撑后续的问答任务。

1. 索引分割后的文档
   用ChromaDB这类向量数据库，把Agent 1生成的文档总结连同文档路径、页面编号等元数据一起存储。索引前先做去重检查，避免重复写入。这样当用户提问时，系统能快速定位到最相关的文档块。

class DocumentRAGAgent:
    def index_documents(self, state: DocumentRAGState):
        """Index the parsed documents into the vector store."""
        assert state.documents, "Documents should ha ve at least one element"
        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
        self.vector_store.add_documents(state.documents)
        logger.info(f"Indexed {len(state.documents)} documents for {state.document_path}")

2. 处理用户问题
   用户提问后，Agent 2先在向量库中检索最相关的文档块，然后根据这些块的页面编号，取出对应的页面图像。把图像和总结组合成上下文，连同问题一并发给Gemini生成答案。这种方式既保留了文本的精确性，又让模型能“看到”原始页面的布局和视觉信息，回答自然更准确。

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

        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}")
        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}",
            ]
        )
        response = self.model.generate_content(messages)
        return {"response": response.text, "relevant_documents": relevant_documents}

借助Gemini这类多模态大模型，我们确实可以搭建一条高效、完整的PDF文档AI处理管线。它绕过了传统文本方法在布局、表格、图像上的短板，让信息提取的精度和可用性上了一个台阶。

你是一名 AI 行业编辑，请围绕下面这条热点输出一份资讯解读：
热点：利用Gemini构建PDF文档AI管道的原理、实现与代码示例要求：
1. 先用一句话解释这条热点在讲什么
2. 再总结它为什么重要
3. 说明会影响哪些 AI 产品或内容方向
4. 最后给出 3 个适合资讯站使用的标题