End-to-End从混沌到秩序：基于LLM的Pipeline将非结构化数据转化为知识图谱

摘要：本文介绍了一种将非结构化数据转换为知识图谱的端到端方法。通过使用大型语言模型（LLM）和一系列数据处理技术，我们能够从原始文本中自动提取结构化的知识。这一过程包括文本分块、LLM 提示设计、三元组提取、归一化与去重，最终利用 NetworkX 和 ipycytoscape 构建并可视化知识图谱。该方法不仅能够高效地从文本中提取知识，还能通过交互式可视化帮助用户更好地理解和分析数据。通过实验，我们展示了该方法在处理复杂文本时的有效性和灵活性，为知识图谱的自动化构建提供了新的思路。

🧠 向所有学习者致敬！

“学习不是装满一桶水，而是点燃一把火。” —— 叶芝

我的博客主页： https://lizheng.blog.csdn.net

🌐 欢迎点击加入AI人工智能社区！

🚀 让我们一起努力，共创AI未来！ 🚀

从纯文本构建知识图谱是一项挑战。它通常需要识别重要术语，弄清楚它们之间的关系，并使用自定义代码或机器学习工具来提取这种结构。我们将创建一个由大型语言模型（LLM）驱动的端到端流水线，自动将原始文本转换为交互式知识图谱。
我们将使用几个关键的 Python 库来完成这项工作。让我们先安装它们。

# 安装库（只运行一次）

pip install openai networkx "ipycytoscape>=1.3.1" ipywidgets pandas

安装完成后，你可能需要重启 Jupyter 内核或运行时，以便更改生效。

现在我们已经安装好了，让我们将所有内容导入到脚本中。

import openai             # 用于 LLM 交互
import json               # 用于解析 LLM 响应
import networkx as nx     # 用于创建和管理图数据结构
import ipycytoscape       # 用于交互式笔记本中的图可视化
import ipywidgets         # 用于交互式元素
import pandas as pd       # 用于以表格形式显示数据
import os                 # 用于访问环境变量（API 密钥更安全）
import math               # 用于基本数学运算
import re                 # 用于基本文本清理（正则表达式）
import warnings           # 抑制潜在的弃用警告

完美！我们的工具箱已经准备好了。所有必要的库都已加载到我们的环境中。

什么是知识图谱？

想象一个网络，有点像社交网络，但除了人之外，它还连接事实和概念。这基本上就是一个知识图谱（KG）。它有两个主要部分：

节点（或实体）：这些是“事物”——比如“居里夫人”、“物理学”、“巴黎”、“诺贝尔奖”。在我们的项目中，我们提取的每个独特的主语或宾语都将变成一个节点。
边（或关系）：这些是事物之间的连接，显示它们如何关联。关键在于，这些连接有意义，并且通常有方向。例如：“居里夫人” — 获得 → “诺贝尔奖”。其中的“获得”部分是关系，定义了边。

最简单的知识图谱示例

一个简单的图，显示两个节点（例如，“居里夫人”和“镭”）通过一个标记为“发现”的有向边连接。再添加一个小集群，如（“巴黎” — 位于 → “索邦大学”）。这可视化了节点-边-节点的概念。

知识图谱之所以强大，是因为它们以更接近我们思考连接的方式结构化信息，使我们更容易发现见解，甚至推断出新的事实。

主语-谓语-宾语（SPO）三元组

那么，我们如何从纯文本中获取这些节点和边呢？我们寻找简单的事实陈述，通常以 主语-谓语-宾语（SPO） 三元组的形式出现。

主语：事实是关于谁或什么的（例如，“居里夫人”）。将成为一个节点。
谓语：连接主语和宾语的动作或关系（例如，“发现”）。将成为边的标签。
宾语：主语相关的事物（例如，“镭”）。将成为另一个节点。

示例：句子 “居里夫人发现了镭” 完美地分解为三元组：(居里夫人, 发现, 镭)。

这直接翻译到我们的图中：

(居里夫人) -[发现]-> (镭)。

我们的 LLM 的工作就是读取文本并为我们识别这些基本的 SPO 事实。

配置我们的 LLM 连接

要使用 LLM，我们需要告诉脚本如何与之通信。这意味着提供一个 API 密钥，有时还需要一个特定的 API 端点（URL）。

我们将使用 NebiusAI LLM 的 API，但你可以使用 Ollama 或任何其他在 OpenAI 模块下工作的 LLM 提供商。

# 如果使用标准 OpenAI
export OPENAI_API_KEY='your_openai_api_key_here'# 如果使用本地模型，如 Ollama
export OPENAI_API_KEY='ollama' # 对于 Ollama，可以是任何非空字符串
export OPENAI_API_BASE='http://localhost:11434/v1'# 如果使用其他提供商，如 Nebius AI
export OPENAI_API_KEY='your_provider_api_key_here'
export OPENAI_API_BASE='https://api.studio.nebius.com/v1/' # 示例 URL

首先，让我们指定我们想要使用的 LLM 模型。这取决于你的 API 密钥和端点可用的模型。

# --- 定义 LLM 模型 ---

# 选择在你的配置端点可用的模型。

# 示例：'gpt-4o', 'gpt-3.5-turbo', 'llama3', 'mistral', 'deepseek-ai/DeepSeek-Coder-V2-Lite-Instruct', 'gemma'

llm_model_name = "deepseek-ai/DeepSeek-V3" # <-- **_ 更改为你使用的模型 _**

好的，我们已经记录下了目标模型。现在，让我们从之前设置的环境变量中获取 API 密钥和基础 URL（如果需要）。

# --- 获取凭据 ---

api_key = os.getenv("OPENAI_API_KEY")
base_url = os.getenv("OPENAI_API_BASE") # 如果未设置，则为 None（例如，对于标准 OpenAI）

客户端已准备好与 LLM 通信。

最后，让我们设置一些控制 LLM 行为的参数：

温度：控制随机性。较低的值意味着更专注、更确定性的输出（适合事实提取！）。我们将温度设置为 0.0，以实现最大的可预测性。
最大令牌数：限制 LLM 响应的长度。

# --- 定义 LLM 调用参数 ---

llm_temperature = 0.0 # 较低的温度用于更确定性、事实性的输出。0.0 是提取的最佳选择。
llm_max_tokens = 4096 # LLM 响应的最大令牌数（根据模型限制调整）

定义我们的输入文本（原材料）

现在，我们需要将要转换为知识图谱的文本。我们将使用居里夫人的传记作为示例。

unstructured_text = """
Marie Curie, born Maria Skłodowska in Warsaw, Poland, was a pioneering physicist and chemist.
She conducted groundbreaking research on radioactivity. Together with her husband, Pierre Curie,
she discovered the elements polonium and radium. Marie Curie was the first woman to win a Nobel Prize,
the first person and only woman to win the Nobel Prize twice, and the only person to win the Nobel Prize
in two different scientific fields. She won the Nobel Prize in Physics in 1903 with Pierre Curie
and Henri Becquerel. Later, she won the Nobel Prize in Chemistry in 1911 for her work on radium and
polonium. During World War I, she developed mobile radiography units, known as 'petites Curies',
to provide X-ray services to field hospitals. Marie Curie died in 1934 from aplastic anemia, likely
caused by her long-term exposure to radiation."""

让我们打印出来，看看它的长度。

print("--- 输入文本已加载 ---")
print(unstructured_text)
print("-" \* 25)

# 基本统计信息可视化

char_count = len(unstructured_text)
word_count = len(unstructured_text.split())
print(f"总字符数：{char_count}")
print(f"大约单词数：{word_count}")
print("-" \* 25)#### 输出结果 ####
--- 输入文本已加载 ---
Marie Curie, born Maria Skłodowska in Warsaw, Poland, was a pioneering physicist and chemist.
She conducted groundbreaking research on radioactivity. Together with her husband, Pierre Curie,

# [... 文本其余部分在此打印 ...]

includes not only her scientific discoveries but also her role in breaking gender barriers in academia
and science.-------------------------
总字符数：1995
大约单词数：324

---

所以，我们有大约 324 个单词的居里夫人的文本，虽然在生产环境中不太理想，但足以看到知识图谱构建的过程。

切分它：文本分块

LLM 通常有一个处理文本的限制（它们的“上下文窗口”）。

我们的居里夫人文本相对较短，但对于更长的文档，我们肯定需要将它们分解为更小的部分，或分块。即使是这种文本，分块有时也能帮助 LLM 更专注于特定部分。

我们将定义两个参数：

分块大小：每个分块中我们希望的最大单词数。
重叠：一个分块的末尾与下一个分块的开头之间应该有多少单词重叠。这种重叠有助于保留上下文，避免事实被尴尬地截断在分块之间。

文本分块过程

我们的完整文本被划分为三个重叠的段（分块）。清晰标记“分块 1”、“分块 2”、“分块 3”。突出显示分块 1 与 2 之间以及分块 2 与 3 之间的重叠部分。标记“分块大小”和“重叠大小”。

让我们设置我们期望的大小和重叠。

# --- 分块配置 ---

chunk_size = 150 # 每个分块的单词数（根据需要调整）
overlap = 30 # 分块之间的重叠单词数（必须小于分块大小）print(f"分块大小设置为：{chunk_size} 单词")
print(f"重叠设置为：{overlap} 单词")# --- 基本验证 ---
if overlap >= chunk_size and chunk_size > 0:
print(f"错误：重叠 ({overlap}) 必须小于分块大小 ({chunk_size})。")
raise SystemExit("分块配置错误。")
else:
print("分块配置有效。")### 输出结果 ###
分块大小设置为：150 单词
重叠设置为：30 单词
分块配置有效。

好的，计划是将分块大小设置为 150 个单词，分块之间的重叠为 30 个单词。

首先，我们需要将文本拆分为单独的单词。

words = unstructured_text.split()
total_words = len(words)print(f"文本拆分为 {total_words} 个单词。")

# 显示前 20 个单词

print(f"前 20 个单词：{words[:20]}")### 输出结果 ###
文本拆分为 324 个单词。
前 20 个单词：['Marie', 'Curie,', 'born', 'Maria', 'Skłodowska', 'in',
'Warsaw,', 'Poland,', 'was', 'a', 'pioneering', 'physicist', 'and',
'chemist.', 'She', 'conducted', 'groundbreaking', 'research', 'on',
'radioactivity.']

输出确认我们的文本有 324 个单词，并显示了前几个单词。现在，让我们应用我们的分块逻辑。

我们将逐步遍历单词列表，每次抓取 chunk_size 个单词，然后回退 overlap 个单词以开始下一个分块。

chunks = []
start_index = 0
chunk_number = 1print(f"开始分块过程...")
while start_index < total_words:
    end_index = min(start_index + chunk_size, total_words)
    chunk_text = " ".join(words[start_index:end_index])
    chunks.append({"text": chunk_text, "chunk_number": chunk_number})    # print(f"  Created chunk {chunk_number}: words {start_index} to {end_index-1}") # Uncomment for detailed log    # 计算下一个分块的起始位置
    next_start_index = start_index + chunk_size - overlap    # 确保有进展
    if next_start_index <= start_index:
        if end_index == total_words:
             break # 已经处理了最后一部分
        next_start_index = start_index + 1    start_index = next_start_index
    chunk_number += 1    # 安全中断（可选）
    if chunk_number > total_words: # 简单安全检查
        print("警告：分块循环超出总单词数，中断。")
        breakprint(f"\n文本成功拆分为 {len(chunks)} 个分块。")#### 输出结果 ####
开始分块过程...文本成功拆分为 3 个分块。