介绍

在当今数据驱动的世界中，组织坐拥大量信息，这些信息隐藏在无数 PDF 文档中。这些文件虽然易于人类阅读，但对于试图理解和利用其内容的机器来说却是一项重大挑战。无论是研究论文、技术手册还是业务报告，PDF 通常都包含有价值的知识，可以为智能系统提供动力并推动基于数据的决策。

尽管 PDF 包含有价值的信息，但存储其内容的传统方法通常依赖于 Milvus 或 ChromaDB 等向量索引。我之前在关于构建生产级问答系统的博客中探讨了这种技术，其中我演示了如何创建检索增强生成 (RAG) 系统。该系统可以通过搜索存储在向量索引中的上传 PDF 内容来回答查询。

然而，这种方法有很大的局限性。在处理非结构化 PDF 数据时，会出现三个关键挑战：1. 缺乏可解释性——很难追踪系统如何得出特定答案 2. 分析能力有限——非结构化数据限制了复杂分析 3. 精度降低——在处理大量信息时尤其明显

这些局限性凸显了表格和知识图谱等结构化格式为何更为强大。它们将原始信息转化为机器能够更有效地处理的有组织的、可查询的数据。

但是，我们如何弥合这些非结构化文档与高级分析和人工智能应用所需的结构化数据之间的差距呢？

这一挑战是许多现代信息处理系统的核心，尤其是那些旨在构建全面知识图谱的系统。从 PDF 中提取结构化知识的过程不仅仅涉及简单的文本提取——它需要理解上下文、识别关键概念以及识别思想之间的关系。

在这篇博文中，我们将深入探讨将 PDF 内容转换为知识图谱的复杂过程。我们将探索逐页解析文档、提取有意义的节点和关系以及准备这些信息以集成到知识图谱中的技术。最后，您将清楚地了解如何将非结构化文档转换为强大的 AI 驱动应用程序的基础。

让我们开始 PDF 知识提取世界的旅程，看看如何释放这些无处不在的文档中隐藏的潜力。

PDF 解析

提取结构化知识从解析 PDF 的内容开始。虽然有许多库可以促进这一过程，但我选择了PyMuPDF及其扩展PyMuPDF4LLM。我做出这一选择的主要动机是 PyMuPDF4LLM 的 markdown 提取功能。

Markdown 格式的输出保留了标题和列表等关键结构元素。这种结构使大型语言模型 (LLM) 能够更好地识别和解释文档组织，从而显著增强检索增强生成 (RAG) 结果。虽然这种方法会消耗额外的标记，但准确率的提升往往可以证明最低的开销是合理的。

import base64
from typing import Union, List, Dict, Any

import pymupdf
import pymupdf4llm
import numpy as np
from PIL import Image
from langchain_core.language_models import BaseLanguageModel
from langchain_core.messages import SystemMessage, HumanMessage


def ocr_images_pytesseract(images: List[Union[np.ndarray | Image.Image]]) -> str:
    import pytesseract

    all_text: str = ""
    for image in images:
        if isinstance(image, Image.Image):
            image = image.filter(ImageFilter.SMOOTH())
        all_text += '\n' +  pytesseract.image_to_string(image, lang="eng", config='--psm 3 --dpi 300 --oem 1')
    return all_text.strip('\n \t')

# Use PyMuPDF to open the document and PyMuPDF4LLM to get the markdown output
doc = pymupdf.Document(pdf_path)
doc_markdown = pymupdf4llm.to_markdown(doc, page_chunks=True, write_images=False, force_text=True)


def extract_page_info(page_num: int, doc_metadata: dict, toc: list = None) -> Dict[str, Any]:
    """Extracts text and metadata from a single page of a PDF document.

    Args:
        page_num (int): The page number
        doc_metadata (dict): The whole document metadata to store along with each page metadata
        toc (list | None): The list that represents the table-of-contents of the document
    
    Returns:
        A dictionary with the following keys:
            1) text: Text content extracted from the page
            2) page_metadata: Metadata specific to the page only like page number, chapter number etc.
            3) doc_metadata: Metadata common to the whole document like filename, author etc.
    
    """
    page_info = {}

    # Read the page of 
    page = doc[page_num]

    # doc_markdown stores the page-by-page markdown output of the document
    text_content: str = self.doc_markdown[page_num]['text']

    # Get a list of all the images on this page - automatically, perform OCR on all these images and store the output as page text
    # There are 3 options available: each with different preprocessing steps and all of them implemented in the self._ocr_func method
    images_list: list[list] = page.get_images()
    if len(images_list) > 0:
        imgs = []
        for img in images_list:
            xref = img[0]
            pix = pymupdf.Pixmap(doc, xref)
            # using PyTesseract requires storing the image as PIL.Image though it could've been bytes or np.ndarray as well
            cspace = pix.colorspace
            if cspace is None:
                mode: str = "L"
            elif cspace.n == 1:
                mode = "L" if pix.alpha == 0 else "LA"
            elif cspace.n == 3:
                mode = "RGB" if pix.alpha == 0 else "RGBA"
            else:
                mode = "CMYK"
            img = Image.frombytes(mode, (pix.width, pix.height), pix.samples)
            if mode != "L":
                img = img.convert("L")
            imgs.append(img)
            
        text_content += '\n' + ocr_images_pytesseract(imgs)
    text_content = text_content.strip(' \n\t')
    page_info["text"] = text_content
    page_info["page_metadata"] = get_page_metadata(page_num=page_num+1, page_text=text_content, toc=toc)
    page_info["doc_metadata"] = doc_metadata
    return page_info

为了解析 PDF，我使用 PyMuPDF4LLM 的 markdown 输出作为每个文档页面的文本内容。同时，我提取图像中存在的所有图像，并将其 OCR 输出附加到同一个 markdown 输出。这样，我们就可以自动处理不同类型的 PDF 的情况：仅包含图像的扫描 PDF、包含文本的 PDF 以及同时包含图像和文本的 PDF。至于 OCR，我使用的是PyTesseract — Google Tesseract-OCR Engine的 Python 包装器

节点和关系架构

要开始从 PDF 中提取结构化内容，我们需要首先定义输出的结构。为此，我利用了大多数现代 LLM（大型语言模型）的“结构化输出”功能。在我的代码中，我使用LangChain来处理与 LLM 相关的所有流程。鉴于这一事实，我使用with_structured_output方法。请注意，并非所有模型都支持此方法，因此您应该首先检查哪些模型支持此方法。

from pydantic import BaseModel, Field

class Property(BaseModel):
    """A single property consisting of key and value"""
    key: str = Field(..., description="key")
    value: str = Field(..., description="value")

class Node(BaseModel):
    id: str = Field(..., description="The identifying property of the node in Title Case")
    type: str = Field(..., description="The entity type / label of the node in PascalCase.")
    properties: Optional[List[Property]] = Field(default=[], description="Detailed properties of the node")
    aliases: List[str] = Field(default=[], description="Alternative names or identifiers for the entity in Title Case")
    definition: Optional[str] = Field(None, description="A concise definition or description of the entity")

class Relationship(BaseModel):
    start_node_id: str = Field(..., description="The id of the first node in the relationship")
    end_node_id: str = Field(..., description="The id of the second node in the relationship")
    type: str = Field(..., description="TThe specific, descriptive label of the relationship in SCREAMING_SNAKE_CASE")
    properties: List[Property] = Field(default=[], description="Detailed properties of the relationship")
    context: Optional[str] = Field(None, description="Additional contextual information about the relationship")
        
class KnowledgeGraph(BaseModel):
    """Generate a knowledge graph with entities and relationships."""
    nodes: list[Node] = Field(
        ..., description="List of nodes in the knowledge graph")
    rels: list[Relationship] = Field(
        ..., description="List of relationships in the knowledge graph"
    )

使用 Pydantic 实现的知识图谱模式由两个主要组件组成：节点和关系。

Node 类使用以下方式定义单个实体：

? 识别属性

? 实体类型

? 附加属性（可选）

? 别名：替代标识符（可选）

? 文本衍生定义（可选）

虽然别名和定义标记为可选，但它们却发挥着至关重要的作用：

1. 在后续处理步骤中启用节点消歧义

2. 通过防止重复创建节点来指导大型语言模型 (LLM) 保持一致性

3. 通过有价值的上下文丰富知识图谱，增强检索增强生成 (RAG) 的性能

关系类通过以下方式捕获节点之间的连接：

? 源节点和目标节点 ID

? 关系类型

? 属性（可选）

? 提取上下文（可选）

与节点别名和定义类似，关系上下文保留了有关如何以及在何处识别连接的宝贵信息，从而提高了图形对下游任务的实用性。

from langchain_core.prompts import ChatPromptTemplate 

DATA_EXTRACTION_SYSTEM = """# 知识图谱提取用于丰富信息检索
## 1. 概述
你是一个先进的算法，旨在从各种类型的信息内容中提取知识。\
你的任务是构建一个知识图谱，为下游任务提供丰富的上下文信息。

## 2. 内容焦点
- 从给定的文本中提取关于概念、实体、流程及其关系的详细信息。
- 优先考虑提供丰富上下文并可能有助于回答广泛问题的信息。
- 包含每个提取实体的相关属性、属性和描述性信息。

## 3. 节点提取
- **节点 ID**：使用清晰、明确的标题大小写标识符。避免使用整数、缩写和首字母缩略词。
- **节点类型**：使用 PascalCase。尽可能具体和描述性以帮助 Wikidata 匹配。
- 在节点属性中包含实体的所有相关属性。
- 当文本中存在实体时，提取并包含实体的替代名称或别名。

## 4. 关系提取
- 对关系类型使用 SCREAMING_SNAKE_CASE。-
创建详细的、信息丰富的关系类型，清楚地描述连接的性质。-
在适用的情况下包括方向关系（例如，PRECEDED_BY、FOLLOWED_BY，而不仅仅是 RELATED_TO）。

## 5. 上下文信息
- 对于每个节点和关系，努力捕获可能对回答问题有用的上下文信息。-
在可用时包括时间信息（例如，日期、时间段、事件顺序）。-
如果相关，捕获地理或空间信息。

## 6. 处理定义和描述
- 对于关键概念，包括简洁的定义或描述作为节点属性。-
捕获实体的任何显着特征、功能或用例。

## 7. 共指和一致性
- 在整个图中保持一致的实体引用。-
将共指解析为最完整的形式，包括潜在的别名或替代名称。

## 8. 粒度
- 在详细提取和维护连贯的图形结构之间取得平衡。
- 为不同的概念创建单独的节点，即使它们密切相关。
"""


 prompt = ChatPromptTemplate.from_messages( 
    [ 
        ( "system" , DATA_EXTRACTION_SYSTEM), 
        ( "human" ,“使用给定的格式从以下输入中提取信息，该输入是来自属于同一主题的更大文本的一小部分样本：{input}”），
        （“human”，“提示：确保以正确的格式回答”），
    ]）

data_ext_chain = prompt | llm.with_structured_output（KnowledgeGraph）

接下来，我定义了提示，其中包含有关数据提取的详细说明。此提示包括对 LLM 的以下说明：

1.最终目标：构建具有丰富的上下文信息的知识图谱。

2. 节点ID、节点类型和关系类型的输出格式

3. 保持节点一致性并将共指解析为其最完整的形式。

使用这个提示和结构化模式，我创建一个数据提取链，它将在后面的步骤中用于从文本中提取知识图谱。

节点和关系提取

最后，我们来到了这篇博文最有趣的部分。这里是大部分魔法发生的地方。给定一个文档列表（每个文档的类型为 langchain_core.documents.base.Document），我们使用上面描述的模式和提示提取文本中存在的节点和关系。

为了简单地从文本中提取节点和关系，我们在每个提供的文档上调用文档提取链并获取输出节点和关系的列表。

nodes = []
rels = []

for doc in docs:
    output: KnowledgeGraph = data_ext_chain.invoke(
        {
            "input": doc.page_content
        }
    )
    nodes.extend(format_nodes(output.nodes))
    rels.extend(format_rels(output.rels))

然而，尽管该解决方案可行，但也存在 3 个主要挑战：

1. 节点类型激增：对创建许多不同节点类型没有限制。拥有几种不同类型的节点会使生成的图形“分散”，从而降低其对下游任务的有效性。

2. 重复实体管理：没有针对跨文档的重复节点和关系的保护，从而导致后续处理步骤中出现潜在的可扩展性问题。

3. 来源可追溯性：我们没有追踪每个节点的来源，这降低了信息本身的可靠性，因为我们无法验证信息。

对于第一个挑战，我们可以通过更新提示来接受现有节点类型列表和“主题”字符串来解决，以作为提取实体类型的指南和限制。对于第二个问题，我们可以用集合替换列表，并在代码中添加检查以确保只提取唯一的节点和关系。对于第三个问题，我们可以将从每个文档中提取的节点列表添加到文档的元数据中。这样，在创建知识图谱时，我们还可以为每个文档创建节点，并在文档节点和从该文档中提取的每个实体节点之间创建关系，可以表示如下：(:Document)-[:MENTIONS]->(:Entity)

进行这些更改后，提示和代码现在如下所示。

DATA_EXTRACTION_SYSTEM = """# 知识图谱提取用于丰富信息检索
## 1. 概述
你是一个先进的算法，旨在从各种类型的信息内容中提取知识。\
你的任务是构建一个知识图谱，为下游任务提供丰富的上下文信息。
{subject} 
## 2. 内容焦点
- 从给定的文本中提取关于概念、实体、流程及其关系的详细信息。
- 优先考虑提供丰富上下文并可能有助于回答广泛问题的信息。
- 包含每个提取实体的相关属性、特性和描述性信息。

## 3. 节点提取
- **节点 ID**：使用清晰、明确的标题大小写标识符。避免使用整数、缩写和首字母缩略词。
- **节点类型**：使用 PascalCase。尽可能具体和描述性以帮助 Wikidata 匹配。
- 在节点属性中包含实体的所有相关属性。
- 当文本中存在实体时，提取并包含实体的替代名称或别名。
- 以下是从同一文档的先前样本中提取的一些现有节点类型：\n{node_types} 

## 4. 关系提取
- 对关系类型使用 SCREAMING_SNAKE_CASE。
- 创建详细的、信息丰富的关系类型，清楚地描述连接的性质。
- 在适用的情况下包括方向关系（例如，PRECEDED_BY、FOLLOWED_BY 而不仅仅是 RELATED_TO）。

## 5. 上下文信息
- 对于每个节点和关系，努力捕获可能对回答问题有用的上下文信息。
- 在可用时包括时间信息（例如，日期、时间段、事件序列）。
- 如果相关，捕获地理或空间信息。

## 6. 处理定义和描述
- 对于关键概念，包括简洁的定义或描述作为节点属性。
- 捕获实体的任何显着特征、功能或用例。

## 7. 共指和一致性
- 在整个图中保持一致的实体引用。
- 将共指解析为最完整的形式，包括潜在的别名或替代名称。

## 8. 粒度
- 在详细提取和维护连贯的图形结构之间取得平衡。
- 为不同的概念创建单独的节点，即使它们密切相关。
""" 
from langchain_core.documents.base import Document 


def  merge_nodes ( nodes: List[Node] ) -> Node: 
    '''
    将所有具有相同 ID 和类型的节点合并为一个节点

    这些节点的唯一区别是它们的属性、别名和定义列表，因此
    我从所有这些节点中提取并组合一个唯一属性和别名列表
    ，并选择最长的定义作为合并节点的定义

    Args: 
        nodes (List[Node]):- 要合并为一个的节点列表
    
    返回：通过合并所有节点创建的单个 Node 对象
    '''
     props: List [Property] = [] 
    definition = ""
     aliases = set ([]) 
    max_def_len = - 1 
    for node in nodes: 
        props.extend([ 
            p for p in node.properties 
            if  not p in props 
        ]) 
        if  len (node.definition) >= max_def_len: 
            # 根据不准确的假设选择最长的定义：
            # 最长的描述 = 最具描述性的定义
            definition = node.definition 
            max_def_len = len (node.definition) 
        aliases.update( set (node.aliases)) 
    return Node( id =nodes[ 0 ]. id , type =nodes[ 0 ]. type , properties=props, definition=definition, aliases=aliases) 

def  docs2nodes_and_rels ( docs: List [Document], text_subject: str = '' , use_existing_node_types: bool = False ) -> Tuple [ List [Document], List [Node], List [Relationship]]: 
    '''
    从文档列表中提取节点和关系列表

    Args: 
        docs (List[Document]): 文档列表
        text_subject (str): 给定文档所属的文本的总体主题。默认值为 '' 
        use_existing_node_types (bool): 是否使用 KG 中已经存在的节点的节点类型
                                        作为 LLM 的指南。默认值为 False。

    返回：
        包含以下列表的元组：
            docs（List[Document]）：略微修改的输入文档列表
            nodes（List[Node]）：唯一提取的节点列表
            rels（List[Relationship]）：提取的关系列表
    ''' 
    # 仅存储唯一节点 - 稍后添加到 KG 时有帮助
    nodes_dict：Dict [ str，list ] = dict（{}）
    rels：list = [] 

    ex_node_types = set（{}）
    if use_existing_node_types：
        # get_existing_labels 是一种返回 KG 中已经存在的节点的标签集的方法
        ex_node_types = get_existing_labels（）

    for i，doc in  enumerate（docs）：
        
        output：KnowledgeGraph = data_ext_chain.invoke（
            { 
                “subject”：text_subject，
                “input”：doc.page_content，
                “node_types”：'，' .join（list（ex_node_types））
            } 
        ) 

        output_nodes: List [Node] = format_nodes(output.nodes) 
        output_rels: List [Relationship] = format_rels(output.rels) 

        ntypes = set ([n. type  for n in output_nodes]) 

        # 将相同类型和 ID 的节点存储在一起，以便稍后合并在一起
        dnodes_dict: Dict [ str , List [ str ]] = {} 
        for n in output_nodes: 
            nk = f"{n.id}::{n.type}"
            if nk in dnodes_dict: 
                if  not n in dnodes_dict[nk]: 
                    dnodes_dict[nk].append                 (n) 
            else : 
                dnodes_dict[nk] = [n] 

        for r in output_rels: 
            if  not r in rels: rels.append(r) # 在文档元数据中存储此文档/块中提到的节点的节点 ID + 节点类型        doc.metadata[ 'mentions' ] = list (dnodes_dict.keys())         nodes_dict = {**nodes_dict, **dnodes_dict}


        


        # 更新现有节点类型以供下次迭代
        ex_node_types：set[str] = ex_node_types.union(ntypes) 

    # 将所有重复节点（具有相同 ID 和类型的节点）合并为一个节点
    # 这涉及组合它们的属性、别名和定义
    nodes：List [Node] = [merge_nodes(nds) for _, nds in nodes_dict.items()] 

    return docs, nodes, list (rels)

如上所示，提取过程已得到增强，有两个关键改进：

1.扩展提示变量：

– 添加主题：提供有关文档整体主题的背景信息

– 添加了 node_types：利用知识图谱中现有的节点类型

– 链的调用语句已更新以使用这些新参数

2.节点管理：

– 实现了一个动态集（ex_node_types）来跟踪所有提取的节点类型

– 每次后续提取都受益于先前发现的节点类型

– 通过组合具有相同 ID 和类型的节点来确保节点的唯一性

– 开发了一种合并机制，以合并重复节点，同时保留其独特属性

– 在每个文档的元数据中添加一个提及键，用于存储在该文档中找到的每个节点的列表

结论

本文探讨了如何将非结构化的 PDF 内容转换为结构化、有意义的数据。我们首先研究了如何利用 LLM 的结构化输出功能来可靠地提取信息。对于没有此功能的 LLM，您可以使用 JSON 输出，尽管它的可靠性相对较低。

架构设计不仅包括基本要素（节点和关系 ID 和类型），还包括一些附加属性。这种更全面的架构不仅可以创建更丰富的知识表示，还可以通过谨慎使用别名和定义来减少歧义。

实施过程中展示了强大的提取技术，同时通过重复数据删除确保了数据质量并保持了源可追溯性。