为结构化和非结构化数据构建图形 RAG。

Rifx.Online
Programming , Natural Language Processing , Data Science
14 Jan, 2025

RAG 架构迄今为止是解决 LLM 缺乏上下文化的最适应和复杂的解决方案。通过 RAG，几乎不需要微调，就在很大程度上解决了使用未训练知识库的 LLM 所面临的问题。

尽管向量 RAG 可以建立上下文化，但其能力是有限的。在复杂的关系和高度互联的数据中，向量 RAG 的召回率并不令人印象深刻。其主要原因之一是构成知识库的简单向量嵌入，仅考虑几何接近性。

另一方面，图形天生结构化，以捕捉数据中的复杂关系，从而导致更长的上下文性。因此，基于图形的 RAG 已成为利用 LLM 能力的最佳手段。

构建知识图谱

知识图谱可以通过非结构化数据（如文本、PDF）和结构化数据（如表格/CSV）进行填充。手动识别和提取跨多个页面的节点、关系和节点属性是一项非人力的任务，需要大量的领域专业知识。Langchain通过使用LLM进行图形实体提取，将这一艰巨的任务变得轻而易举。

让我们来看看如何将非结构化和结构化数据转换为知识图谱。

非结构化数据

在进一步操作之前，让我们进行所有必要的导入

from langchain_community.graphs import Neo4jGraph
from langchain_experimental.graph_transformers import LLMGraphTransformer
from langchain_openai import AzureChatOpenAI
import os
from langchain_core.documents import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter, CharacterTextSplitter
from langchain_community.vectorstores.neo4j_vector import Neo4jVector
from langchain_openai import AzureOpenAIEmbeddings
import fitz
import logging 
import ast
from tqdm.notebook import tqdm
from concurrent.futures import ThreadPoolExecutor, as_completed
from langchain.memory import ConversationBufferMemory

在本文中，我们将使用 Neo4j Desktop 应用程序作为图数据库，并使用 AzureOpenAI LLM。

#Azure opena ai Gpt 模型初始化
gpt_llm = AzureChatOpenAI(temperature=0, azure_deployment= OPENAI_GPT_DEPLOYMENT_NAME, api_key= OPENAI_API_KEY, api_version= OPENAI_API_VERSION, azure_endpoint= AZURE_ENDPOINT)
## 使用有效的 neo4j 服务器凭据初始化 neo4j 对象 
graph_db = Neo4jGraph()

首先，我们需要从 PDF 中提取所有文本，并用这些文本实例化一个 langchain Document 对象。

def convert_pdf_to_text (file_paths):

    """
    file_paths = list[str]
    rtype: Document object 
    实现从 pdf 中提取文本，并用文档文本形成一个 langchain 对象。
    """
  
    doc_text = []
    #遍历每个文档
    for i in file_paths:
        logging.info("正在读取 ..."+str(i))
        doc = fitz.open(i)
            
        # 遍历文档中的每一页并提取文本
        try:
            text = ""
            for page_num in range(len(doc)):
                page = doc.load_page(page_num)
                text += page.get_text()
            # 将元数据与文档文本结合  
            doc_text.append([Document(page_content=str(text))])
        
        except Exception as e:
            logging.info("读取文档时出错 : "+str(e))

    return(doc_text)

在文本提取之后，我们现在需要分块文档。每个新块将是一个 Document 对象。

def split_text(docs):

    """
    docs: list of document objects 
    rtype: list[Document]
    实现通过将给定文档分割成块以及元数据的分块机制 
    """
    #初始化文本分割器，设置适当的块大小和块重叠
    text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=1000, chunk_overlap=100 )
    documents = []
    #遍历所有文档并实现分块
    for i in docs:
        documents.append(text_splitter.split_documents(i))
    # 将所有分块文档合并为一个单一列表
    merged_docs = [item for sublist in documents for item in sublist]

    return(merged_docs)

块和重叠大小可以视为超参数。

我们现在需要通过从文档文本中提取节点、节点属性和关系来填充图形。随着文档数量的增加，手动提取将变得不可能。因此，我们将使用 LLM 进行提取。LLM 将遍历所有文档块，并填充那些具有相互关联关系的节点。LLMGraphTransformer 函数来自 langchain，可以为我们完成这项工作。

def construct_graph(doc):
    """ 
    doc: Document object
    rtype: Graph Document
    实现使用 llm 的图形构建功能。
    """
    #初始化图形转换器对象
    llm_transformer = LLMGraphTransformer(llm= gpt_llm)
    
    #通过使用 LLM 从文档中提取节点和关系来构建图形
    graph_doc = llm_transformer.convert_to_graph_documents([doc])
    return(graph_doc)

虽然这可能减少了大量的手工劳动，但在实体提取上仍然消耗了大量时间。因此，我们将对相同的过程进行多线程处理，以处理多个文档并汇总结果。以下是上述函数的多线程实现。

def  thread_construct_graph(merged_docs):
  
    """
    merged_docs: list
    实现使用 LLM 在多线程方法中从文档构建图形，并将提取的节点和关系推送到提供的 GraphDB 中
    """
    MAX_WORKERS = 20 # 可根据需要更改
    # NUM_ARTICLES = len(merged_docs)
    graph_documents = []

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
        # 提交所有任务并创建未来对象列表
        futures =  [pool.submit(construct_graph, merged_docs[i]) for i in range(len(merged_docs)) ]
    
        #tqdm 显示线程执行可视化的进度条 
        for future in tqdm(as_completed(futures), total=len(futures), desc="处理文档"):
            
            #捕获线程状态的结果
            graph_document = future.result()
            graph_documents.extend(graph_document) #列表扩展方法将新图文档添加到现有列表中

    # 将构建的图形添加到图形数据库中，包含提取的节点和关系 
    graph_db.add_graph_documents(graph_documents,baseEntityLabel = True, include_source = True)
    logging.info("构建的图形数据库...")
    logging.info("模式: "+str(graph_db.get_schema))

一旦图形被填充，add_graph_documents 方法将把填充的图形推送到 neo4j 数据库中。

结构化数据

从结构化数据（如CSV）构建知识图谱相较于非结构化数据来说更为繁琐，但可以通过编程方式填充。可以使用neomodel库，该库可以从python连接到neo4j，用于定义节点及其属性的结构。

所需导入：

from neomodel import db, config, StructuredNode, RelationshipTo, RelationshipFrom, StringProperty

config.DATABASE_URL = "bolt://neo4j:#PASSWORD@localhost:7687"

为了定义节点的结构，我们需要定义一个节点类，实施StructuredNode。类的成员变量将成为节点的属性和关系。

下面的代码演示了一个员工节点，它具有诸如Emp_ID、Emp_Fname、Emp_Lname、Emp_Location、Emp_Manager_Name、Emp_Rank等属性，并通过“reports to”关系与经理节点相关联，通过“belongs to”关系与部门节点相关联。

class Employee (StructuredNode):
    
    __label__ = "Employee"
    Emp_ID= StringProperty(unique_index = True)
    Emp_Fname= StringProperty()
    Emp_Lname= StringProperty()
    Emp_Location= StringProperty()
    Emp_Manager_Name= StringProperty()
    Emp_Rank= StringProperty()
    
    reports_to= RelationshipTo(Manager, "reports to")
    belongs_to= RelationshipTo(Department, "belongs to")


class Manager (StructuredNode):
    
    __label__ = "Manager"
    Emp_id= StringProperty(unique_index = True)
    M_Fname= StringProperty()
    M_Lname= StringProperty()
    dept_id=  StringProperty()
    dept_name = StringProperty()

class Department (StructuredNode):
  
    __label__ = "Department"
    Dept_ID= StringProperty(unique_index = True)
    Dept_name= StringProperty()
    Emp_count = StringProperty()
    Manager_emp_id = StringProperty()

在定义了节点属性和关系后，我们现在可以实例化节点并为所有节点属性分配值。例如，如果我需要定义一个新员工节点并在其经理和部门节点之间建立关系，可以这样做。

emp_node = Employee.get_or_create({
    Emp_ID= #employee_id
    Emp_Fname= #employee_fname
    Emp_Lname= #employee_lname
    Emp_Location= #employee_location
    Emp_Manager_Name= .#employee_manager_name
    Emp_Rank= #employee_rank
    })

Manager = Manager.get_or_create({
    Emp_id= #manager_emp_id
    M_Fname= #manager_fname
    M_Lname= #manager_lname
    dept_id= #dept_id
    dept_name = #dept_name 
})

dept = Department.get_or_create({
    Dept_ID= #dept_id
    Dept_name= #dept_name
    Emp_count = #emp_count
    Manager_emp_id =#manager_emp_id
})


emp_node.reports_to.connect(Manager)
emp_node.belongs_to.connect(dept)

get_or_create函数会获取已存在的节点（如果存在），或者创建一个新节点。如果员工节点的经理或部门节点已经存在，那么它会将员工节点与相应的关系分配给它们。

LLM 响应

一旦知识图谱开发完成，我们将通过开发最后一个模块来完成图形 RAG 循环，该模块将图形数据库与 LLM 连接并回答用户查询。

Langchain 有几个库，它们与 neo4j 结合提供了一个接口，以帮助构建 QA 链。让我们探索下面的其中一个函数，它使用用户查询查询图形数据库。

def query_db_with_llm(existing_graph_index,  metadata, query):

    """
    existing_graph_index = Neo4j Object
    metadata= dict
    query= str
    Implements querying the Graph DB using an LLM, alongside implicit metadata filtering
    """
  
    
    logging.info("Metadata = ", metadata)
    logging.info("Query = ", query)
    
    #Initialise Neo4j object from the existing kowledge graph
    existing_graph_index = Neo4jVector.from_existing_graph(
    az_embeddings, # use the same embedding model as used for the embedding creations
    )
    # Intialise langchain querying object
    qa_chain = ConversationalRetrievalChain.from_llm(return_source_documents= True, # True when the source of answers is required else False
    llm= #LLM of your choice, 
    retriever = existing_graph_index.as_retriever(search_kwargs={'filter': metadata}), #does a vector embedding semantic search along with metadata filtering on node properties using neo4j vector indexing 
    verbose = True,
    )
    #query the Graph db 
    answer = qa_chain({"question": query, "chat_history": []}) 
    
    return(answer)

上述代码还演示了额外的 元数据过滤 功能，通过将其作为参数传递给 index.as_retriever 函数，过滤节点和关系的属性。

嵌入

正如许多人所注意到的，我们在将非结构化文本数据插入图形数据库之前，并没有明确创建嵌入。LLMGraphTransformer 默认会创建一个名为 Document 的节点标签，其中的节点将包含块及其嵌入作为节点属性。当调用 Neo4jVector.from_existing_graph 方法时，嵌入会被生成，如果尚不存在的话。

*在我的下一篇文章中，我将讨论并提出一种增强的图形 RAG 查询方法。敬请关注！！！

希望你觉得这些信息有用。