使用大语言模型从零构建知识图谱（上）

©作者|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

您会看到如下的输出：

Model    architecture        qwen2    parameters          7.6B    context length      32768    embedding length    3584    quantization        Q4_K_M
  System    You are Qwen, created by Alibaba Cloud. You are a helpful assistant.
  License    Apache License    Version 2.0, January 2004

另一个免费的方案是使用 Google的大模型服务平台，要获取 api key，请访问这里（假设你已经拥有一个 Google 的账号并且已经登录到管理控制台），然后按照开发平台的引导创建 api key 并将生成的 key 复制并保存起来，后面会用到。

从零开始构建图谱工具

让我们先导入项目所需的一些库：

# 提供类型提示from typing import Any, Dict, List, Tuple
# 标准库import astimport loggingimport reimport warnings
# 第三方包 - 数据操作import pandas as pdfrom tqdm import tqdm
# 第三方包 - 环境配置和数据库from dotenv import load_dotenvfrom neo4j import GraphDatabase
# 第三方包 - 错误处理与重试逻辑from tenacity import retry, stop_after_attempt, wait_exponential
# Langchain - 核心库from langchain.chains import GraphCypherQAChainfrom langchain.prompts import PromptTemplatefrom langchain_core.documents import Document
# Langchain - 模型和连接器from langchain_google_genai import ChatGoogleGenerativeAI, GoogleGenerativeAIfrom langchain_ollama.llms import OllamaLLM
# Langchain - 图相关库与体验包from langchain_community.graphs import Neo4jGraphfrom 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 whitespace        df[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 df
movies = 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 None    try:        cleaned = str(value).strip().replace(',', '')        return target_type(cleaned)    except (ValueError, TypeError):        return None
def 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 directors            for 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)}")            continue
    logger.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.nameLIMIT 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 pLIMIT 100;

你现在应该会看到类似这样的内容：

是不是看起来很厉害！

然而，这确实还需要一些时间来探索数据集、进行数据清理，并手动编写 Cypher 查询。不过，现在大家不都在使用人工智能了嘛，当然我们不再需要通过人工方式这样做了，除非你想这么做。现在已经有多种方法可以自动化此过程。在该教程的下半部分中，我们将利用 LLM 创建一个基本的自动化解决方案。敬请期待！