從零到一：大語言模型在知識圖譜構建中的實操指南

?作者|Ninja Geek

來源|神州問學

將你的?Pandas data frame?利用大語言模型轉換為知識圖譜。從零開始構建自己的基于大語言模型的圖譜構建器，實際使用?Langchain?的?LLMGraphTransformer?，并對知識圖譜進行問答。

在當前 RAG 相關技術領域，知識圖譜變得越來越重要，因為該檢索結構支持許多大語言模型背后的知識檢索系統。許多公司、科研院所、高校都在大力投資檢索增強生成技術的實際落地，因為這是提高語言模型輸出準確性，并防止產生幻覺的一種高效手段。

這不僅僅是技術上的進步（相對于較為早期的 RAG 實現而言），從我個人的角度來看，Graph RAG 正在人工智能領域普及化。這是因為在過去，如果我們想將模型定制化以適應不同的場景（無論是處于娛樂場景還是商業場景），通常有以下三種選擇：預訓練模型以提供更好的行業適應性，或者針對特定數據集對模型進行微調，又或者是基于給定的上下文讓模型進行總結性回復。

對于預訓練，這種方式的成本極高并且技術實現上也相對復雜，對于大多數開發者來說并不現實。

模型微調比預訓練模型相對容易些，盡管微調的成本取決于模型和訓練的語料庫，但通常來說是一種更經濟的選擇。這曾是大多數人工智能開發者的首選策略。然而，模型每隔一段時間就會發布新的版本，因此開發者需要不斷的在新模型上進行微調。

第三種選擇是直接在基于提示詞的上下文中提供知識。然而，這僅在所需知識量較小時有效。盡管模型的上下文容量越來越大，但召回特定元素的準確性通常與提供的上下文大小成反比。

這三種技術路徑似乎都不是理想的解決方案。是否存在另外一種方法，讓模型能夠學習完成某個特定任務或主題所需的所有知識？答案是：沒有。

但模型不需要一次性學習所有知識。當我們向大語言模型提問時，通常只需要獲取一部分或幾條信息。在這里，Graph-RAG 提供了幫助。通過基于查詢進行的信息檢索，Graph-RAG 能夠提取所需的信息，而無需進一步訓練模型。

讓我們來看一下 Graph-RAG 的構成：

圖構建：在這一步中，我們從數據源創建節點（有些語境下也叫實體）和邊（有些語境下也叫關系），并將它們加載到知識圖譜（Knowledge Graph）中。這通常是一個更加需要人工介入的步驟，常用的查詢語言（如 OpenCypher）上傳實體并通過“邊”將它們相互連接。

節點索引：此步驟涉及創建一種數據結構以高效檢索數據。在知識圖譜中，這通常包括創建一個向量搜索索引，其中每個索引都與一個向量嵌入相關聯。

圖檢索器：在這里，我們構建一個檢索函數，用于計算相似度分數并檢索最相關的節點，作為大語言模型提供答案的上下文。在最簡單的情況下，我們將查詢轉換為一個向量嵌入，并對其與向量索引中的所有嵌入進行余弦相似度計算。

RAG 評估：最后一步是用于衡量大語言模型的準確性和性能。這對于試驗階段非常有用，可以比較不同的大語言模型和 RAG 框架在特定用例上的表現。

現在我們對 RAG 管道有了總體了解，你可能會急于嘗試復雜的數學函數來優化圖圖檢索，確保信息檢索的最佳準確性。不過請先等等。到目前為止，還沒有知識圖譜。這一步可能看起來像數據科學中的經典數據清洗和預處理過程。但如果告訴你還有更好的選擇呢？一種引入更多科學性和自動化的方法。

確實，最近的研究集中在如何自動構建知識圖譜，因為這一步對于良好的信息檢索至關重要。想一想，如果知識圖譜中的數據質量不好，就不可能讓你的 Graph-RAG 達到最好的性能。

在我的這篇文章中，我們將深入探討第一步：如何在不真正構建知識圖譜的情況下構建知識圖譜。

從 CSV 到知識圖譜

現在，讓我們通過一個實際的例子來具體說明。我們來解決一個重要的日常問題：看什么電影？我相信在你打開優酷、愛奇藝、騰訊視頻等視頻平臺 App 后會漫無目的的瀏覽可能感興趣的電影，直到意識到時間已經過去半個小時了。

為了解決這個問題，我們可以使用來自 Kaggle 的關于維基百科中統計的電影數據集來創建一個知識圖譜，并與知識圖譜對話。首先，我們將使用大語言模型實現一個“從零開始”的解決方案。然后，我們會探討通過 Langchain 這一流行且強大的大語言模型框架，以及另一個常用的解決方案 LlamaIndex 的最新實現。

讓我們從 Kaggle 下載這個公開的數據集：

Wikipedia Movie Plots

先決條件

在開始之前，我們需要下載?Neo4j?桌面版和一個商業模型的 API 秘鑰，或者你用本地模型也可以。如果這些都準備好了，可以跳過這部分直接進入實操環節。如果沒有，也不用擔心，我會帶著你完成設置。

使用 Neo4j 有多種方式，但為了簡化流程，我們將使用 Neo4j 桌面版，因此數據庫將本地托管。由于數據集較小，運行該應用程序不會對你的電腦性能造成影響。

要安裝 Neo4j，只需要訪問 Neo4j 桌面版下載頁面并點擊“下載”。安裝完成后打開 Neo4j Desktop，登錄或創建一個 Neo4j 的賬號（激活軟件的時候需要用到）。

登錄后，創建一個新項目：

1. 點擊左上角的 ? 按鈕。

2. 為項目命名（例如：Wiki-Movie-KG）

3. 在項目內，點擊 Add Database，選擇 Local DBMS，最后點擊 Create a Local Graph。

配置數據庫：

1. 名稱：輸入一個數據庫名稱（例如：neo4j）

2. 密碼：設置一個密碼（例如：mysecret）。請記住該密碼，后面會用到。

3.點擊 Create 初始化數據庫。

接下來，讓我們配置大語言模型。運行該 notebook 的推薦方式是使用?Ollama。Ollama 是一種本地托管的大語言模型解決方案，可以非常簡單地在你的電腦上下載并設置大語言模型。它支持許多開源大語言模型，例如 Qwen2.5、GLM、Llama 和 Gemma。我認為 Ollama 是非常棒的本地模型解決方案，因為這樣一來我可以在本地使用大語言模型而不用將個人數據暴露給服務提供商。

要下載 Ollama，請訪問 Ollama 官網，下載適合你操作系統的安裝程序。安裝完成后，打開 Ollama 應用。

打開終端，使用以下命令列出可用模型：

解釋

ollama list

安裝并運行某個模型。這里，我們將使用 qwen2.5-coder:latest ，這是一款在編碼任務上進行過微調的 7B 模型。

解釋

ollama run qwen2.5-coder:latest

最后來驗證安裝：

解釋

ollama show qwen2.5-coder:latest

您會看到如下的輸出：

解釋

Modelarchitecture        qwen2parameters          7.6Bcontext length      32768embedding length    3584quantization        Q4_K_MSystemYou are Qwen, created by Alibaba Cloud. You are a helpful assistant.LicenseApache LicenseVersion 2.0, January 2004

另一個免費的方案是使用 Google的大模型服務平臺，要獲取 api key，請訪問這里（假設你已經擁有一個 Google 的賬號并且已經登錄到管理控制臺），然后按照開發平臺的引導創建 api key 并將生成的 key 復制并保存起來，后面會用到。

從零開始構建圖譜工具

讓我們先導入項目所需的一些庫：

解釋

# 提供類型提示
from typing import Any, Dict, List, Tuple# 標準庫
import ast
import logging
import re
import warnings# 第三方包 - 數據操作
import pandas as pd
from tqdm import tqdm# 第三方包 - 環境配置和數據庫
from dotenv import load_dotenv
from neo4j import GraphDatabase# 第三方包 - 錯誤處理與重試邏輯
from tenacity import retry, stop_after_attempt, wait_exponential# Langchain - 核心庫
from langchain.chains import GraphCypherQAChain
from langchain.prompts import PromptTemplate
from langchain_core.documents import Document# Langchain - 模型和連接器
from langchain_google_genai import ChatGoogleGenerativeAI, GoogleGenerativeAI
from langchain_ollama.llms import OllamaLLM# Langchain - 圖相關庫與體驗包
from langchain_community.graphs import Neo4jGraph
from langchain_experimental.graph_transformers import LLMGraphTransformer# 抑制警告
warnings.filterwarnings('ignore')# 加載環境變量
load_dotenv()

正如你所看到的，Langchain 在代碼組織方面表現不佳，導致導入了相當多的庫。讓我們逐一解析這些庫的用途：

●os 和 dotenv：幫助我們管理環境變量（例如數據庫憑證）。

● pandas：用于處理電影數據集。

● neo4j：這個庫將 Python 連接到 Neo4j 圖數據庫。

●langchain：提供工具，用于處理語言模型和圖譜。

● tqdm：為打印語句添加一個友好的用戶界面。我們會用它在循環中顯示進度條，以便知道剩余的處理量。

● warnings：抑制不必要的警告信息，使輸出的內容更干凈。

我們將加載電影數據集，該數據集包含來自世界各地的大約 34000 多部電影的信息。此數據集可在 Kaggle 上公開獲取。

解釋

movies = pd.read_csv('data/wiki_movies.csv') # 按照你自己實際保存該數據集的路徑進行修改
movies.head()

這里我們可以看到以下特征：

● Release Year（上映年份）：電影的上映年份

● Title（標題）：電影的標題

●Origin/Ethnicity（起源/民族性）：電影的起源（例如：中國、美國、好萊塢、寶萊塢）

●Director（導演）：導演（一部電影可能由多名導演共同執導）

●Plot（劇情）：主要演員

●Genre（類型）：電影的類型

●Wiki Page（電影在維基百科上的頁面）：提供劇情描述的維基百科頁面的 URL 地址

● Plot（劇情描述）：電影劇情的長篇描述

●通過觀察這些特征，我們可以快速設計一些希望在知識圖譜中看到的標簽和關系。由于這是一個電影數據集，“電影”會是一個顯而易見的標簽。此外，我們可能對查詢特定的演員和導演會感興趣。因此，我們最終為節點確定了三個標簽：Movie（電影）、Actor（演員）和 Director（導演）。當然，我們還可以添加更多標簽，但為了簡單起見，先到此為止。

● 同樣的，為了簡化操作，我們對數據集稍作清理，僅提取前 1000 行：

解釋

def clean_data(df: pd.DataFrame) -> pd.DataFrame:"""清理并預處理 DataFrame.參數:data: 輸入 DataFrame返回值:清理后的 DataFrame"""df.drop(["Wiki Page"], axis=1, inplace=True)# 刪除重復項df = df.drop_duplicates(subset='Title', keep='first')# 獲取對象類型的列col_obj = df.select_dtypes(include=["object"]).columns# 清理字符串列for col in col_obj:# Strip whitespacedf[col] = df[col].str.strip()# 替換未知/空值df[col] = df[col].apply(lambda x: None if pd.isna(x) or x.lower() in ["", "unknown"] else x.capitalize())# 刪除包含任何空值的行df = df.dropna(how="any", axis=0)return dfmovies = clean_data(movies).head(1000)
movies.head()

在這里，我們刪除了包含維基百科頁面鏈接的 Wiki Page 列。不過，如果你愿意，也可以保留這一列，因為它可以作為 Movie（電影）節點的一個屬性。接下來，我們根據標題刪除所有重復項，并清理所有字符串（對象）列。最后，僅保留前 1000 部電影。

由于我們的知識圖譜將托管在 Neo4j 上，接下來設置一個輔助類來建立連接并提供一些實用的方法：

解釋

class Neo4jConnection:def __init__(self, uri, user, password):self.driver = GraphDatabase.driver(uri, auth=(user, password))def close(self):self.driver.close()print("連接已關閉。")def reset_database(self):with self.driver.session() as session:session.run("MATCH (n) DETACH DELETE n")print("數據庫重置成功！")def execute_query(self, query, parameters=None):with self.driver.session() as session:result = session.run(query, parameters or {})return [record for record in result]

在初始化方法（__init__）中，我們使用數據庫的 URL（uri）、用戶名和密碼來設置與 Neo4j 數據庫的連接。在初始化類時會傳遞這些變量。

close 方法：終止與數據庫的連接。

reset_database 方法：通過 Cypher 命令 MATCH (n) DETACH DELETE n 刪除數據庫中的所有節點和關系。

execute_query 方法：運行給定的查詢（如添加電影或獲取關系）并返回結果。

接下來，我們使用這個輔助類連接到數據庫：

解釋

uri = "bolt://localhost:7687"
user = "neo4j"
password = "mysecret"
conn = Neo4jConnection(uri, user, password)
conn.reset_database()

默認情況下，uri 和 user 將與上面提供的內容一致。至于 password ，它將是你在創建數據庫時定義的密碼。此外，我們使用 reset_database 來確保從干凈的數據庫開始，移除任何現有數據。

如果你遇到與數據庫中未安裝 APOC 相關的錯誤，請按照以下步驟操作：

1.打開 Neo4j

2.點擊數據庫

3.轉到 Plugins（插件）

4.安裝 APOC 插件

現在我們需要將數據集中的每部電影轉換成圖中的一個節點。在本節中，我們將手動完成此操作，而在接下來的章節中，我們將利用大語言模型來為我們完成這一任務。

解釋

def parse_number(value: Any, target_type: type) -> Optional[float]:"""將字符串解析為數字，并進行適當的錯誤處理。"""if pd.isna(value):return Nonetry:cleaned = str(value).strip().replace(',', '')return target_type(cleaned)except (ValueError, TypeError):return Nonedef clean_text(text: str) -> str:"""清理并規范化文本字段。"""if pd.isna(text):return ""return str(text).strip().title()

讓我們創建兩個簡短的函數 —— parse_number 和 clean_text ，分別用于將數值列的數據轉換為數字，并正確格式化文本列。如果轉換失敗（比如值為空），對于數值列，這些函數返回 None ，對于對象列，返回一個空字符串。

接下來，讓我們創建一個方法，用于將數據迭代加載到我們的知識圖譜中：

解釋

def load_movies_to_neo4j(movies_df: pd.DataFrame, connection: GraphDatabase) -> None:"""將電影數據加載到 Neo4j 中，并進行進度跟蹤和錯誤處理。"""logger = logging.getLogger(__name__)logger.setLevel(logging.INFO)# 查詢模板MOVIE_QUERY = """MERGE (movie:Movie {title: $title})SET movie.year = $year,movie.origin = $origin,movie.genre = $genre,movie.plot = $plot"""DIRECTOR_QUERY = """MATCH (movie:Movie {title: $title})MERGE (director:Director {name: $name})MERGE (director)-[:DIRECTED]->(movie)"""ACTOR_QUERY = """MATCH (movie:Movie {title: $title})MERGE (actor:Actor {name: $name})MERGE (actor)-[:ACTED_IN]->(movie)"""# 處理每部電影for _, row in tqdm(movies_df.iterrows(), total=len(movies_df), desc="正在加載電影數據"):try:# 準備電影相關實體的參數movie_params = {"title": clean_text(row["Title"]),"year": parse_number(row["Release Year"], int),"origin": clean_text(row["Origin/Ethnicity"]),"genre": clean_text(row["Genre"]),"plot": str(row["Plot"]).strip()}# 創建電影節點connection.execute_query(MOVIE_QUERY, parameters=movie_params)# Process directorsfor director in str(row["Director"]).split(" and "):director_params = {"name": clean_text(director),"title": movie_params["title"]}connection.execute_query(DIRECTOR_QUERY, parameters=director_params)# 處理演員相關數據if pd.notna(row["Cast"]):for actor in row["Cast"].split(","):actor_params = {"name": clean_text(actor),"title": movie_params["title"]}connection.execute_query(ACTOR_QUERY, parameters=actor_params)except Exception as e:logger.error(f"加載 {row['Title']} 錯誤: {str(e)}")continuelogger.info("加載電影數據到 Neo4j 已完成。")

理解上述 Cypher 查詢的兩個重要關鍵詞是 MERGE 和 SET。

MERGE 確保節點或關系的存在，如果不存在，則創建它。因此，它結合了 MATCH 和 CREATE 子句的功能，其中 MATCH 用于在圖中搜索特定結構，CREATE 用于創建節點和關系。因此，MERGE 會首先檢查我們要創建的節點/邊是否已存在，如果不存在，則創建它。

在上述函數中，我們使用 MERGE 為每部電影（Movie）、導演（Director）和演員（Actor）創建節點。特別地，由于我們有關于演員的特性（Star1、Star2、Star3 和 Star4），我們為每一列創建一個演員節點。

接下來，我們創建兩個關系：

1.從導演到電影的關系（DIRECTED)。

2.從演員到電影的關系（ACTED_IN)。

SET 用于更新節點或邊的屬性。在這里，我們為電影節點提供了 Year（年份）、Rating（評分）、Genre（電影風格）、Runtime（電影時長）和 Overview（電影概述）屬性。為導演和演員節點添加了 Name（姓名）屬性。

另外，需要注意的是，我們使用了特殊符號 $ 來定義參數。

接下來，讓我們調用該函數并加載所有電影數據：

解釋

load_movies_to_neo4j(movies, conn)

加載所有 1000 部電影大約需要一分鐘。（當然，這個視你的電腦配置而定）

執行完成后，我們運行一個 Cypher 查詢來檢查電影是否已正確上傳：

解釋

query = """
MATCH (m:Movie)-[:ACTED_IN]-(a:Actor)
RETURN m.title, a.name
LIMIT 10;
"""
results = conn.execute_query(query)
for record in results:print(record)

該查詢現在應該看起來很熟悉了。我們使用 MATCH 來查找圖中的模式。

●(m:Movie): 匹配標簽為 “Movie” （電影）的節點。

● (a:Actor): 匹配標簽為 “Actor” （演員）的節點。

此外，我們使用 RETURN 指定要顯示的內容 —— 在這個例子中是電影標題和演員姓名，并使用 LIMIT 限制結果為前 10 個匹配項。

你應該會得到類似以下的輸出：

解釋

[<Record m.title='Daniel Boone' a.name='William Craven'>,<Record m.title='Daniel Boone' a.name='Florence Lawrence'>,<Record m.title='Laughing Gas' a.name='Bertha Regustus'>,<Record m.title='Laughing Gas' a.name='Edward Boulden'>,<Record m.title="A Drunkard'S Reformation" a.name='Arthur V. Johnson'>,<Record m.title='The Adventures Of Dollie' a.name='Arthur V. Johnson'>,<Record m.title='A Calamitous Elopement' a.name='Linda Arvidson'>,<Record m.title='The Adventures Of Dollie' a.name='Linda Arvidson'>,<Record m.title='The Black Viper' a.name='D. W. Griffith'>,<Record m.title='A Calamitous Elopement' a.name='Harry Solter'>]

這個包含 10 條記錄（代表電影和演員）的列表確認了電影已經上傳到知識圖譜中了。

接下來，讓我們看看實際的圖譜。切換到 Neo4j Desktop 應用，選擇為本次練習創建的電影數據庫（Movie Database），并點擊 Open with Neo4j Browser。這將打開一個新標簽頁，你可以在其中運行 Cypher 查詢。然后，運行以下查詢：

解釋

MATCH p=(m:Movie)-[r]-(n)
RETURN p
LIMIT 100;

你現在應該會看到類似這樣的內容：

是不是看起來很厲害！

然而，這確實還需要一些時間來探索數據集、進行數據清理，并手動編寫 Cypher 查詢。不過，現在大家不都在使用人工智能了嘛，當然我們不再需要通過人工方式這樣做了，除非你想這么做。現在已經有多種方法可以自動化此過程。在該教程的下半部分中，我們將利用 LLM 創建一個基本的自動化解決方案。敬請期待！