从非结构化文本中构建知识图谱是一项具有挑战性的任务。它通常需要识别关键术语，理清它们之间的相互关系，并借助定制代码或机器学习工具来提取这种结构化信息。
我们将创建一个端到端的管道，该管道由大语言模型（LLM）驱动，能够自动将原始文本转换为交互式知识图谱。
所有代码都可以在我的 GitHub 仓库中找到：https://github.com/FareedKhan-dev/KG-Pipeline
环境配置
正如任何优秀的项目一样，我们需要合适的工具。我们将使用几个关键的 Python 库来完成这项工作。首先，安装它们。
# 安装库（此单元格运行一次即可）
pip install openai networkx "ipycytoscape>=1.3.1" ipywidgets pandas运行安装命令后，你可能需要重启 Jupyter 内核或运行时环境，以使更改生效。
安装完成后，让我们将所有需要的库导入到脚本中。
import openai # 用于与 LLM 交互
import json # 用于解析 LLM 的响应
import networkx as nx # 用于创建和管理图数据结构
import ipycytoscape # 用于在 notebook 中进行交互式图可视化
import ipywidgets # 用于交互式元素
import pandas as pd # 用于以表格形式展示数据
import os # 用于访问环境变量（对 API 密钥更安全）
import math # 用于基本的数学运算
import re # 用于基本的文本清理（正则表达式）
import warnings # 用于抑制潜在的弃用警告我们的工具箱已经准备就绪，所有必要的库都已加载到我们的环境中。
什么是知识图谱？
知识图谱是一种由相互连接的实体和关系构成的网络，它以结构化的方式表示知识，并支持知识的推理和发现。它包含两个主要部分：
• 节点（或实体 Nodes/Entities）： 这些是“事物”——比如‘玛丽·居里’、‘物理学’、‘巴黎’、‘诺贝尔奖’。在我们的项目中，我们提取的每个唯一的主语或宾语都将成为一个节点。 • 边（或关系 Edges/Relationships）： 这些是事物之间的连接，展示了它们如何关联。关键在于，这些连接具有意义，并且通常有方向。例如：‘玛丽·居里’ — 赢得 → ‘诺贝尔奖’。“赢得”这部分就是关系，定义了这条边。
上图展示了一个简单的图示，包含两个节点（例如，“玛丽·居里”，“镭”）通过一条标有“发现”的有向边连接。旁边还有一个小簇（“巴黎” — 位于 → “索邦大学”）。这直观地展示了“节点-边-节点”的概念。
知识图谱之所以强大，是因为它们以一种更接近我们思考事物联系的方式来组织信息，使得发现洞见甚至推断新事实变得更加容易。
主谓宾（SPO）三元组
那么，我们如何从纯文本中获取这些节点和边呢？我们寻找简单的陈述性事实，这些事实通常可以构造成**主语-谓语-宾语（Subject-Predicate-Object, SPO）**三元组。
• 主语（Subject）： 事实所描述的对象（例如，‘玛丽·居里’），对应图谱中的一个节点。 • 谓语（Predicate）： 连接主语和宾语的动作或关系（例如，‘发现’），对应边的标签。 • 宾语（Object）： 与主语相关的事物（例如，‘镭’），对应另一个节点。
示例： 句子 “玛丽·居里发现了镭” 可以完美地分解为三元组：(玛丽·居里, 发现, 镭)。
这直接映射到我们的图结构：
• (玛丽·居里) -[发现]-> (镭)。
大语言模型的任务是从文本中提取这些基本的主谓宾三元组。
配置 LLM 连接
我们需要配置脚本，使其能够与 LLM API 进行通信，包括提供 API 密钥和 API 端点（URL）。
我们将使用 NebiusAI LLM 的 API，但你也可以使用 Ollama 或其他任何兼容 OpenAI 模块的 LLM 提供商。
# 如果使用标准的 OpenAI
export OPENAI_API_KEY='你的_openai_api密钥_放在这里'
# 如果使用本地模型，如 Ollama
export OPENAI_API_KEY='ollama' # 对于 Ollama，可以是任何非空字符串
export OPENAI_API_BASE='http://localhost:11434/v1'
# 如果使用其他提供商，如 Nebius AI
export OPENAI_API_KEY='你的_提供商_api密钥_放在这里'
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") # 如果未设置（例如，对于标准 OpenAI），则为 None现在，客户端已准备好与 LLM 通信。
最后，我们设置几个控制 LLM 行为的参数：
• Temperature（温度）： 控制随机性。较低的值意味着更专注、更确定的输出（非常适合事实提取！）。我们将其设置为 0.0 以获得最大的可预测性。 • Max Tokens（最大令牌数）： 限制 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... (完整文本打印)
# -------------------------
# 总字符数: 1995 (示例值)
# 大致词数: 324 (示例值)
# -------------------------我们有了关于玛丽·居里的文本，大约 324 个词。对于生产环境来说可能不太理想，但足以展示知识图谱的构建过程。
文本分块（Chunking）
LLM 通常对单次处理的文本长度有限制（即上下文长度限制）。
我们关于玛丽·居里的文本相对较短，但对于更长的文档，我们肯定需要将其分解成更小的片段，即块（chunks）。每个块包含一定数量的词语，以便于处理。即使对于这段文本，分块有时也能帮助 LLM 专注于特定部分。
我们将为此定义两个参数：
• 分块大小（Chunk Size）： 我们希望每个块包含的最大词数。 • 重叠词数（Overlap）： 一个块的末尾和下一个块的开头之间应重叠多少个词。这种重叠有助于保持上下文连贯，避免事实在块与块之间被生硬地切断。
上图展示了将完整文本分割成三个重叠片段（块）的过程。清晰地标示了“块 1”、“块 2”、“块 3”。突出显示了块 1 和块 2 之间以及块 2 和块 3 之间的重叠部分。并标示了“分块大小”和“重叠词数”。
让我们设定所需的分块大小和重叠词数。
# --- 分块配置 ---
chunk_size = 150# 每个块的词数（根据需要调整）
overlap = 30 # 重叠的词数（必须小于 chunk_size）
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 = 1
print(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" 已创建块 {chunk_number}: 词语 {start_index} 到 {end_index-1}") # 取消注释以查看详细日志
# 计算下一个块的起始索引
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("警告：分块循环次数超过总词数，已中断。")
break
print(f"\n文本成功分割成 {len(chunks)} 个块。")
#### 预期输出 ####
# 开始分块处理...
#
# 文本成功分割成 3 个块。文本成功分割成 3 个块。让我们使用 Pandas DataFrame 来查看这些块的大小和内容片段。
print("--- 块详情 ---")
if chunks:
# 创建 DataFrame 以便更好地可视化
chunks_df = pd.DataFrame(chunks)
chunks_df['word_count'] = chunks_df['text'].apply(lambda x: len(x.split()))
# 在 Jupyter 环境中，display() 会以更美观的表格形式显示 DataFrame
# 如果在普通 Python 脚本中，可以使用 print(chunks_df[['chunk_number', 'word_count', 'text']])
try:
display(chunks_df[['chunk_number', 'word_count', 'text']])
except NameError: # 'display' 可能未定义在非 Jupyter 环境
print(chunks_df[['chunk_number', 'word_count', 'text']])
else:
print("没有创建任何块（文本可能短于分块大小）。")
print("-" * 25)（此处预期会显示一个包含 3 行的表格，列为 chunk_number, word_count, text。前两个块的 word_count 为 150，最后一个为 84（324 - (150-30) - (150-30) = 84）。）
表格清晰地展示了我们的 3 个块。注意前两个块正好是 150 个词，最后一个包含剩余的 84 个词。现在我们有了可以提交给 LLM 的、易于管理的小片段。
LLM 提示模板
提示工程是构建知识图谱流程中的关键环节。从 LLM 获得良好结果在很大程度上取决于给出清晰、精确的指令——即 LLM 提示模板（LLM Prompt Template）。
我们需要明确告诉它我们想要什么（SPO 三元组），以及我们希望它如何格式化（特定的 JSON 结构）。
我们将为提示创建两个部分：
1. 系统提示（System Prompt）： 为 LLM 设定整体角色和背景（例如，“你是一位知识图谱提取专家”）。 2. 用户提示（User Prompt）： 包含具体的任务指令和待处理的实际文本块。
上图展示了两个框。框 1 标记为“系统提示”，包含类似“你是一位专家…”的文本。
框 2 标记为“用户提示”，包含类似“提取 SPO 三元组… 规则：… 文本：{text_chunk} … 要求 JSON 格式：… 你的 JSON：”的文本。
一个箭头从文本块数据指向用户提示框中的 {text_chunk} 占位符。
以下是我们在用户提示中要强调的关键规则：
• 提取**主语-谓语-宾语（Subject-Predicate-Object）**三元组。 • 仅输出一个有效的 JSON 数组 [...]。不要包含任何额外的文字、解释或 markdown 代码围栏（如json ...）。• 数组中的每个元素必须是一个对象： { "subject": "...", "predicate": "...", "object": "..." }。• 保持谓语简洁（1-3 个词，动词为佳）。 • 所有主语、谓语和宾语的值必须是小写。这有助于后续的标准化。 • 解析代词（如 'she', 'her'）并将其替换为它们所指的具体实体名称（例如，'marie curie'）。 • 要具体（例如，如果文本提到 'nobel prize in physics'，就提取它，而不仅仅是 'nobel prize'）。 • 尽量捕获所有不同的事实。
让我们用 Python 定义这些提示。
# --- System Prompt: Sets the context/role for the LLM ---
extraction_system_prompt = """
You are an AI expert specialized in knowledge graph extraction.
Your task is to identify and extract factual Subject-Predicate-Object (SPO) triples from the given text.
Focus on accuracy and adhere strictly to the JSON output format requested in the user prompt.
Extract core entities and the most direct relationship.
"""
# --- User Prompt Template: Contains specific instructions and the text ---
extraction_user_prompt_template = """
Please extract Subject-Predicate-Object (S-P-O) triples from the text below.
**VERY IMPORTANT RULES:**
1. **Output Format:** Respond ONLY with a single, valid JSON array. Each element MUST be an object with keys "subject", "predicate", "object".
2. **JSON Only:** Do NOT include any text before or after the JSON array (e.g., no 'Here is the JSON:' or explanations). Do NOT use markdown ```json ... ``` tags.
3. **Concise Predicates:** Keep the 'predicate' value concise (1-3 words, ideally 1-2). Use verbs or short verb phrases (e.g., 'discovered', 'was born in', 'won').
4. **Lowercase:** ALL values for 'subject', 'predicate', and 'object' MUST be lowercase.
5. **Pronoun Resolution:** Replace pronouns (she, he, it, her, etc.) with the specific lowercase entity name they refer to based on the text context (e.g., 'marie curie').
6. **Specificity:** Capture specific details (e.g., 'nobel prize in physics' instead of just 'nobel prize' if specified).
7. **Completeness:** Extract all distinct factual relationships mentioned.
**Text to Process:**
{text_chunk}
让我们输出并验证其内容是否符合预期，包括一个当我们插入第一个文本块时用户提示会是什么样子的示例。
print("--- 系统提示 ---")
print(extraction_system_prompt)
print("\n" + "-" * 25 + "\n")
print("--- 用户提示模板（结构） ---")
# 显示结构，替换占位符以便更清晰
print(extraction_user_prompt_template.replace("{text_chunk}", "[... 文本块放在这里 ...]"))
print("\n" + "-" * 25 + "\n")
# 显示将为第一个块发送的 *实际* 提示示例
print("--- 填充后的用户提示示例（针对块 1） ---")
if chunks:
example_filled_prompt = extraction_user_prompt_template.format(text_chunk=chunks[0]['text'])
# 为简洁起见，仅显示一部分
print(example_filled_prompt[:600] + "\n[... 块文本的其余部分 ...]\n" + example_filled_prompt[-200:])
else:
print("没有可用的块来创建填充后的提示示例。")
print("\n" + "-" * 25)
#### 预期输出 ####
# --- 系统提示 ---
# 你是一位专门从事知识图谱提取的 AI 专家... (完整系统提示)
# -------------------------
#
# --- 用户提示模板（结构） ---
# 请从以下文本中提取主谓宾 (S-P-O) 三元组。
# **非常重要的规则：**
# [... 规则打印在这里 ...]
# **待处理文本：**
# [... 文本块放在这里 ...]
# **你的 JSON 输出:**
# -------------------------
#
# --- 填充后的用户提示示例（针对块 1） ---
# 请从以下文本中提取主谓宾 (S-P-O) 三元组。
# ... (规则) ...
# **待处理文本：**
# 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',
# [... 块文本的其余部分 ...]
# **你的 JSON 输出:**
# -------------------------提示信息结构清晰，内容准确。我们准备好将这些发送给 LLM 了。
从 LLM 获取三元组
接下来，我们将从 LLM API 中提取三元组。我们将遍历每个文本块，用块的文本格式化用户提示，通过 API 将系统提示和用户提示都发送给 LLM，然后尝试解析它返回的 JSON 响应。
我们将跟踪成功提取的结果以及任何失败的块。
上图展示了一个循环过程。从“文本块”开始。箭头指向“格式化提示（系统 + 用户 + 块）”。箭头指向“发送到 LLM API”。箭头指向“接收响应”。箭头指向“解析 JSON”。箭头指向“验证三元组”。箭头指向“存储有效三元组”。
一个箭头指回开头，处理下一个块。旁边有一个小框用于“处理错误/失败”。
让我们初始化用于存储结果的列表。
# 初始化列表以存储结果和失败记录
all_extracted_triples = []
failed_chunks = []
# 假设 client 已经根据之前的 'Configuring Our LLM Connection' 部分正确初始化
# 例如: client = openai.OpenAI(api_key=api_key, base_url=base_url)
# 为了代码可运行性，这里添加一个简单的 client 初始化（需要替换为实际的）
try:
client = openai.OpenAI(api_key=api_key, base_url=base_url)
except Exception as e:
print(f"无法初始化 OpenAI 客户端: {e}")
print("请确保 API 密钥和基础 URL 已正确设置。")
client = None# 标记 client 无效
print(f"开始从 {len(chunks)} 个块中提取三元组，使用模型 '{llm_model_name}'...")
# 我们将在接下来的单元格中逐一处理块。开始从 3 个块中提取三元组，使用模型 'deepseek-ai/DeepSeek-V3'...
好的，让我们处理第一个块（请记住，完整的 notebook 会遍历所有块，但为了清晰起见，我们这里只展示一个块的详细步骤）。
chunk_index = 0 # 仅处理第一个块
if client and chunk_index < len(chunks): # 检查 client 是否有效
chunk = chunks[chunk_index]
print(f"\n--- 正在处理块 {chunk['chunk_number']}/{len(chunks)} ---")
prompt = extraction_user_prompt_template.format(text_chunk=chunk['text'])
raw_response = None# 初始化原始响应变量
parsed_data = None# 初始化解析后的数据变量
triples_in_chunk = [] # 初始化当前块的三元组列表
try:
print("1. 格式化用户提示...")
print("2. 向 LLM 发送请求...")
# 调用 LLM，包含系统提示和用户提示
res = client.chat.completions.create(
model=llm_model_name,
messages=[{"role": "system", "content": extraction_system_prompt},
{"role": "user", "content": prompt}],
temperature=llm_temperature,
max_tokens=llm_max_tokens,
# 尝试要求 JSON 输出（如果模型和 API 支持）
response_format={"type": "json_object"},
)
print(" LLM 响应已接收。")
print("3. 提取原始响应内容...")
raw_response = res.choices[0].message.content.strip()
print(f"\n--- 原始 LLM 输出 (块 {chunk['chunk_number']}) ---")
print(raw_response) # 打印原始输出来看看
print("-" * 15)
print("\n4. 尝试从响应中解析 JSON...")
# 尝试直接解析 JSON，因为我们要求了 json_object 格式
try:
parsed_data = json.loads(raw_response)
# 有些模型可能将列表包装在一个键下，尝试提取它
ifisinstance(parsed_data, dict):
# 查找字典中第一个值是列表的项
potential_list = next((v for v in parsed_data.values() ifisinstance(v, list)), None)
if potential_list isnotNone:
parsed_data = potential_list
else:
# 如果没有找到列表，但字典看起来像单个三元组，将其放入列表中
ifall(k in parsed_data for k in ['subject', 'predicate', 'object']):
parsed_data = [parsed_data]
else:
# 如果字典不是三元组也不是包含列表的包装器，则认为无效
print(" 警告：收到字典，但既不是三元组也不是包含三元组列表的包装器。")
parsed_data = [] # 置为空列表
ifnotisinstance(parsed_data, list):
print(f" 警告：解析结果不是列表，而是 {type(parsed_data)}。尝试查找列表。")
parsed_data = [] # 如果不是列表，则认为无效
print(f" 成功解析 JSON 列表（或将其转换/找到）。包含 {len(parsed_data)} 个项目。")
# 可选：打印解析后的数据结构
# print(f" --- 解析后的 JSON 数据 (块 {chunk['chunk_number']}) ---")
# print(json.dumps(parsed_data, indent=2, ensure_ascii=False)) # 使用 ensure_ascii=False 正确显示非 ASCII 字符
# print("-" * 15)
except json.JSONDecodeError as json_e:
print(f" 直接 JSON 解析失败: {json_e}")
print(" 尝试使用正则表达式提取 JSON 数组...")
# 如果直接解析失败（例如，模型在 JSON 前后添加了文本），尝试用正则提取
match = re.search(r'\[.*?\]', raw_response, re.DOTALL) # 查找 [...] 结构
ifmatch:
try:
parsed_data = json.loads(match.group(0))
print(f" 通过正则表达式成功提取并解析了 JSON 列表。包含 {len(parsed_data)} 个项目。")
except json.JSONDecodeError as regex_json_e:
print(f" 从正则表达式匹配中解析 JSON 失败: {regex_json_e}")
parsed_data = [] # 解析仍然失败
else:
print(" 未找到符合 [...] 格式的 JSON 数组。")
parsed_data = [] # 未找到匹配
print("\n5. 验证结构并提取三元组...")
ifisinstance(parsed_data, list):
valid_triples_count = 0
for item in parsed_data:
# 检查是否是字典且包含所有必需的键，并且值是字符串
ifisinstance(item, dict) andall(k in item andisinstance(item[k], str) for k in ['subject', 'predicate', 'object']):
# 添加来源块编号
item_with_chunk = dict(item, chunk=chunk['chunk_number'])
triples_in_chunk.append(item_with_chunk)
valid_triples_count += 1
else:
print(f" 警告：跳过无效项目：{item}")
print(f" 在此块中找到 {valid_triples_count} 个有效三元组。")
if triples_in_chunk:
# 使用 Pandas 显示提取的三元组（如果可用）
try:
print(f" --- 提取的有效三元组 (块 {chunk['chunk_number']}) ---")
display(pd.DataFrame(triples_in_chunk))
except NameError:
print(pd.DataFrame(triples_in_chunk))
all_extracted_triples.extend(triples_in_chunk)
else:
print(" 未能获取有效的 JSON 列表，无法提取三元组。")
failed_chunks.append({'chunk_number': chunk['chunk_number'], 'error': '未能解析出有效的 JSON 列表', 'response': raw_response})
except Exception as e:
print(f"处理块 {chunk['chunk_number']} 时发生错误：{e}")
failed_chunks.append({'chunk_number': chunk['chunk_number'], 'error': str(e), 'response': raw_response or'请求失败，无响应'})
# 打印当前累计结果
print(f"\n--- 当前累计提取的三元组总数: {len(all_extracted_triples)} ---")
print(f"--- 到目前为止失败的块数: {len(failed_chunks)} ---")
print(f"\n完成处理此块。")
elifnot client:
print("错误：LLM 客户端未初始化。无法处理块。")
else:
print("块索引超出范围或没有块。")运行上述循环后，程序就会开始提取实体及其他用于构建知识图谱的信息，从而构建知识图谱。
以下是处理单个数据块时的过程示例：
===== 正在处理数据块 1/3 =====
1. 格式化用户提示...
2. 向大语言模型发送请求...
已收到大语言模型响应。
3. 提取原始响应内容...
=====
===== 大语言模型原始输出 (数据块 1) =====
[
{ "subject": "marie curie", "predicate": "born as", "object": "maria skłodowska" },
{ "subject": "marie curie", "predicate": "born in", "object": "warsaw, poland" },
{ "subject": "marie curie", "predicate": "was", "object": "physicist" },
# [... 更多原始三元组 ...]
{ "subject": "marie curie", "predicate": "born to", "object": "family of teachers" }
]
=====
4. 尝试从响应中解析 JSON...
成功直接解析 JSON 列表。
===== 已解析的 JSON 数据 (数据块 1) =====
[
{
"subject": "marie curie",
"predicate": "born as",
"object": "maria sk\u0142odowska"
},
# [... 更多已解析的三元组 ...]
{
"subject": "marie curie",
"predicate": "born to",
"object": "family of teachers"
}
]
=====
5. 验证结构并提取三元组...
在此数据块中找到 18 个有效三元组。
===== 已提取的有效三元组 (数据块 1) =====
subject predicate object chunk
0 marie curie born as maria skłodowska 1
1 marie curie born in warsaw, poland 1
2 marie curie was physicist 1
# [... 在数据框中显示的更多三元组 ...]
## 17 marie curie born to family of teachers 1
===== 已提取的三元组总数：18 =====
===== 截至目前失败的数据块数：0 =====
该数据块处理完毕。我们发送了第一个数据块并收到响应，随后成功将其解析为 JSON 格式。原始输出显示，大语言模型（LLM）很好地遵循了指令，返回了一个字典列表。接着，我们验证了这些数据，并将其清晰地展示在表格中 —— 仅这第一个数据块就提取出了 18 条事实！
（请注意：上述代码仅运行了第一个数据块。完整的运行过程会处理所有数据块，并累积更多的三元组。）
接下来，我们汇总处理所有数据块后的整体结果（基于完整运行的 Notebook）。
# ===== 提取过程总结 (反映单数据块演示后 / 或完整运行后的状态) =====
print(f"\n===== 整体提取总结 =====\n")
print(f"定义的总数据块数: {len(chunks)}")
print(f"已处理 (尝试处理) 的数据块数: {len(chunks)}") # 我们循环遍历的数据块
print(f"所有已处理数据块中提取的有效三元组总数: {len(all_extracted_triples)}")
print(f"API 调用或解析失败的数据块数量: {len(failed_chunks)}")
if failed_chunks:
print("\n失败数据块详情:")
failed_df = pd.DataFrame(failed_chunks)
display(failed_df[['chunk_number', 'error']]) # 清晰展示失败的数据块
# for failure in failed_chunks:
# print(f" 数据块 {failure['chunk_number']}: 错误: {failure['error']}")
print("-" * 25)
# 使用 Pandas 展示所有提取的三元组
print("\n===== 所有提取的三元组 (规范化之前) =====\n")
if all_extracted_triples:
all_triples_df = pd.DataFrame(all_extracted_triples)
display(all_triples_df)
else:
print("未能成功提取任何三元组。")
print("-" * 25)输出结果如下：
===== 整体提取总结 =====
定义的总数据块数: 3
已处理 (尝试处理) 的数据块数: 3
所有已处理数据块中提取的有效三元组总数: 45 # <--- 示例总数
API 调用或解析失败的数据块数量: 0
-------------------------
===== 所有提取的三元组 (规范化之前) =====
subject predicate object chunk
0 marie curie born as maria skłodowska 1
1 marie curie born in warsaw, poland 1
# [... 来自所有数据块的更多三元组 ...]
## 44 marie curie had daughters irène 3好的，在处理完所有数据块（在完整运行中）后，我们得到了一个包含 LLM 找到的所有三元组的合并列表。这是一个不错的开始，不过你可能会注意到其中存在一些潜在的重叠或表述上的细微差异。现在，是时候进行数据清理了！
规范化与去重
来自 LLM 的原始输出已经很出色，但通常需要进一步优化。我们将执行以下几个简单的清理步骤：
1. 规范化 (Normalize): 清除主语（subject）、谓语（predicate）和宾语（object）开头和结尾的多余空格。我们之前已要求 LLM 输出小写，但在此再次强制执行小写转换，以防万一。 2. 过滤 (Filter): 移除清理后主语、谓语或宾语为空的三元组（例如，LLM 返回了空白内容）。 3. 去重 (De-duplicate): 移除完全相同的三元组。这些重复项可能来自重叠的数据块，或是文本中不同的表述方式。
下面我们开始进行初始化。
# 初始化列表和跟踪变量
normalized_triples = []
seen_triples = set() # 用于跟踪 (subject, predicate, object) 元组
original_count = len(all_extracted_triples)
empty_removed_count = 0
duplicates_removed_count = 0
print(f"开始对 {original_count} 个三元组进行规范化和去重处理...")
#### 输出 ####
开始对 45 个三元组进行规范化和去重处理... # <--- 示例总数现在，我们将遍历原始的 all_extracted_triples 列表，应用清理步骤，并只保留唯一的、有效的三元组。
我们将打印出前几个转换过程，以展示具体的操作。
print("正在处理三元组 (展示前 5 个):")
for i, t inenumerate(all_extracted_triples):
# 提取主语、谓语和宾语；去除首尾空格并转换为小写；如果不是字符串，则设置为空字符串
s, p, o = [t.get(k, '').strip().lower() ifisinstance(t.get(k), str) else''for k in ['subject', 'predicate', 'object']]
# 将谓语中的多个空格替换为单个空格
p = re.sub(r'\s+', ' ', p)
# 确保主语、谓语、宾语都不为空
ifall([s, p, o]):
key = (s, p, o) # 创建用于检查重复的键
if key notin seen_triples: # 如果这个三元组是新的
normalized_triples.append({'subject': s, 'predicate': p, 'object': o, 'source_chunk': t.get('chunk', '?')}) # 添加到结果列表
seen_triples.add(key) # 记录下来，避免重复
if i < 5: # 打印前 5 个的处理信息
print(f"\n#{i+1}: {key}\n状态: 保留")
else: # 如果是重复的
duplicates_removed_count += 1
if i < 5: print(f"\n#{i+1}: 重复 - 跳过")
else: # 如果清理后有空的部分
empty_removed_count += 1
if i < 5: print(f"\n#{i+1}: 无效 - 跳过")
print(f"\n处理完成。总计: {len(all_extracted_triples)}, 保留: {len(normalized_triples)}, 重复: {duplicates_removed_count}, 空值: {empty_removed_count}")
运行三元组生成的循环后，得到的输出如下：
正在处理三元组以进行规范化 (展示前 5 个示例):
===== 示例 1 =====
原始三元组 (数据块 1): {'subject': 'marie curie', 'predicate': 'born as', 'object': 'maria skłodowska', 'chunk': 1}
规范化后: 主语='marie curie', 谓语='born as', 宾语='maria skłodowska'
状态: 保留 (新的唯一三元组)
===== 示例 2 =====
原始三元组 (数据块 1): {'subject': 'marie curie', 'predicate': 'born in', 'object': 'warsaw, poland', 'chunk': 1}
规范化后: 主语='marie curie', 谓语='born in', 宾语='warsaw, poland'
状态: 保留 (新的唯一三元组)
... 完成处理 45 个三元组。 # <--- 示例总数从以上示例可以看出，该过程会对每个三元组进行检查。如果它有效（清理后不为空）并且我们之前没有记录过这个确切的事实，我们就会保留它。
让我们总结一下最初有多少三元组，移除了多少，并展示最终清理后的列表。
# ===== 规范化总结 =====
print(f"\n===== 规范化与去重总结 =====\n")
print(f"原始提取的三元组数量: {original_count}\n")
print(f"因包含空/无效部分而被移除的三元组数量: {empty_removed_count}\n")
print(f"被移除的重复三元组数量: {duplicates_removed_count}\n")
final_count = len(normalized_triples)
print(f"最终唯一的、规范化后的三元组数量: {final_count}\n")
print("-" * 25)
# 使用 Pandas 展示规范化后三元组的样本
print("\n===== 最终规范化后的三元组 =====\n")
if normalized_triples:
normalized_df = pd.DataFrame(normalized_triples)
display(normalized_df)
else:
print("规范化后没有剩余的有效三元组。")
print("-" * 25)
#### 输出 ####
===== 规范化与去重总结 =====
原始提取的三元组数量: 45
因包含空/无效部分而被移除的三元组数量: 0
被移除的重复三元组数量: 3# <--- 示例：发现了一些重复项
最终唯一的、规范化后的三元组数量: 42# <--- 示例最终数量
-------------------------
===== 最终规范化后的三元组 =====
subject predicate object source_chunk
0 marie curie born as maria skłodowska 1
1 marie curie born in warsaw, poland 1
# [... 仅显示唯一的、干净的三元组 ...]
41 marie curie had daughter named eve 3
-------------------------现在，我们得到了一个干净、唯一的事实列表（三元组），可以用来构建图结构了。
使用 NetworkX 创建图
现在开始组装知识图谱！我们将使用 networkx 这个 Python 库来创建一个有向图（Directed Graph, DiGraph）。我们清理后的三元组将按以下方式映射到图结构：
• 每个唯一的主语（subject）成为一个节点（node）。 • 每个唯一的宾语（object）成为一个节点（node）。 • 每个三元组（主语, 谓语, 宾语）成为一条从主语节点指向宾语节点的有向边（directed edge），谓语（predicate）作为这条边的标签（label）。
这张图展示了左侧的 2-3 个 SPO 三元组（例如，(玛丽·居里, 发现, 镭), (玛丽·居里, 获得, 诺贝尔物理学奖)）。右侧则显示了对应的图结构元素：代表“玛丽·居里”、“镭”、“诺贝尔物理学奖”的节点。一条从“玛丽·居里”指向“镭”并标记为“发现”的边。一条从“玛丽·居里”指向“诺贝尔物理学奖”并标记为“获得”的边。箭头清晰地展示了三元组各部分到图元素的映射关系。
首先，让我们创建一个空的图对象。
# 创建一个空的有向图
knowledge_graph = nx.DiGraph()
print("已初始化一个空的 NetworkX DiGraph。")
# 可视化初始空图状态
print("===== 初始图信息 =====\n")
try:
# 尝试使用较新版本的方法
print(nx.info(knowledge_graph))
except AttributeError:
# 兼容不同 NetworkX 版本的备选方法
print(f"类型: {type(knowledge_graph).__name__}")
print(f"节点数: {knowledge_graph.number_of_nodes()}")
print(f"边数: {knowledge_graph.number_of_edges()}")
print("-" * 25)
不出所料，我们的图目前是空的。
现在，我们将遍历 normalized_triples 列表，并将每个三元组作为一条边（及其对应的节点）添加到图中。
我们会定期打印更新信息，以展示图的增长过程。
print("正在将三元组添加到 NetworkX 图中...")
added_edges_count = 0
update_interval = 10# 打印图信息更新的频率
ifnot normalized_triples:
print("警告：没有规范化后的三元组可添加到图中。")
else:
for i, triple inenumerate(normalized_triples):
subject_node = triple['subject']
object_node = triple['object']
predicate_label = triple['predicate']
# 添加边时会自动添加节点，但显式调用 add_node 也可以
# knowledge_graph.add_node(subject_node)
# knowledge_graph.add_node(object_node)
# 添加带有谓语作为 'label' 属性的有向边
knowledge_graph.add_edge(subject_node, object_node, label=predicate_label)
added_edges_count += 1
# ===== 可视化图的增长 =====
if (i + 1) % update_interval == 0or (i + 1) == len(normalized_triples):
print(f"\n===== 添加第 {i+1} 个三元组后的图信息 ===== ({subject_node} -> {object_node})")
try:
# 尝试使用较新版本的方法
print(nx.info(knowledge_graph))
except AttributeError:
# 兼容不同 NetworkX 版本的备选方法
print(f"类型: {type(knowledge_graph).__name__}")
print(f"节点数: {knowledge_graph.number_of_nodes()}")
print(f"边数: {knowledge_graph.number_of_edges()}")
# 对于非常大的图，过于频繁地打印信息可能会很慢，请调整 update_interval。
print(f"\n完成添加三元组。共处理了 {added_edges_count} 条边。")
#### 输出 ####
正在将三元组添加到 NetworkX 图中...")
===== 添加第 10 个三元组后的图信息 ===== (marie curie -> only woman to win nobel prize twice)
类型: DiGraph
节点数: 11
边数: 10
===== 添加第 20 个三元组后的图信息 ===== (pierre curie -> was professor of physics)
类型: DiGraph
节点数: 24
边数: 20我们遍历了清理后的三元组列表，并将每一个都作为一条边添加到了 networkx 图中。
输出显示，图的节点和边的数量在稳步增加。
让我们看一下最终的统计数据，并查看一些我们构建的图中的节点和边的样本。
# ===== 最终图统计 =====
num_nodes = knowledge_graph.number_of_nodes()
num_edges = knowledge_graph.number_of_edges()
print(f"\n===== 最终 NetworkX 图总结 =====\n")
print(f"总唯一节点数 (实体): {num_nodes}")
print(f"总唯一边数 (关系): {num_edges}")
if num_edges != added_edges_count andisinstance(knowledge_graph, nx.DiGraph):
print(f"注意: 添加了 {added_edges_count} 条边，但图中只有 {num_edges} 条。DiGraph 会覆盖具有相同源节点和目标节点的边。如果需要保留多条相同方向的边，请使用 MultiDiGraph。")
if num_nodes > 0:
try:
density = nx.density(knowledge_graph) # 图密度：衡量图的连接紧密程度
print(f"图密度: {density:.4f}")
if nx.is_weakly_connected(knowledge_graph): # 弱连通：忽略边的方向，图中所有节点是否都相互可达？
print("该图是弱连通的 (忽略方向，所有节点都可达)。")
else:
num_components = nx.number_weakly_connected_components(knowledge_graph) # 弱连通分量的数量
print(f"该图包含 {num_components} 个弱连通分量。")
except Exception as e:
print(f"无法计算某些图指标: {e}") # 处理空图或小图可能出现的错误
else:
print("图为空，无法计算指标。")
print("-" * 25)
# ===== 节点样本 =====
print("\n===== 节点样本 (前 10 个) =====\n")
if num_nodes > 0:
nodes_sample = list(knowledge_graph.nodes())[:10]
display(pd.DataFrame(nodes_sample, columns=['节点样本']))
else:
print("图中没有节点。")
# ===== 边样本 =====
print("\n===== 边样本 (前 10 个，带标签) =====\n")
if num_edges > 0:
edges_sample = []
# 提取前10条边及其数据（包括标签）
for u, v, data inlist(knowledge_graph.edges(data=True))[:10]:
edges_sample.append({'源节点': u, '目标节点': v, '标签': data.get('label', 'N/A')})
display(pd.DataFrame(edges_sample))
else:
print("图中没有边。")
print("-" * 25)
我们已经使用 networkx 在内存中成功构建了图结构。可以看到图中唯一实体（节点）和关系（边）的总数，并大致了解了它们的样子。
使用 ipycytoscape 实现交互式图谱
现在，到了很酷的环节——可视化我们的图！我们将使用 ipycytoscape 在这个 notebook 中直接创建一个交互式可视化。
首先，快速检查一下我们是否确实有一个可以可视化的图。
print("准备交互式可视化...")
# ===== 检查图是否有效以进行可视化 =====
can_visualize = False
if'knowledge_graph'notinlocals() ornotisinstance(knowledge_graph, nx.Graph):
print("错误: 未找到 'knowledge_graph' 或其不是 NetworkX 图对象。")
elif knowledge_graph.number_of_nodes() == 0:
print("NetworkX 图为空，无法进行可视化。")
else:
print(f"图似乎有效，可以进行可视化 ({knowledge_graph.number_of_nodes()} 个节点, {knowledge_graph.number_of_edges()} 条边)。")
can_visualize = True
#### 输出 ####
准备交互式可视化...")
图似乎有效，可以进行可视化 (35 个节点, 42 条边)。ipycytoscape 需要特定格式的图数据（一个包含节点信息的字典列表，以及另一个包含边信息的字典列表，类似于 JSON）。
我们现在就来转换 networkx 图的数据。同时，我们还会计算节点的度（degree，即每个节点有多少连接），以便后续可能根据度数调整节点大小。
cytoscape_nodes = []
cytoscape_edges = []
if can_visualize:
print("正在转换节点...")
# 计算节点度数用于调整节点大小
node_degrees = dict(knowledge_graph.degree())
max_degree = max(node_degrees.values()) if node_degrees else1# 找到最大度数，避免除零
for node_id in knowledge_graph.nodes():
degree = node_degrees.get(node_id, 0)
# 简单的节点大小缩放逻辑 (可根据需要调整)
node_size = 15 + (degree / max_degree) * 50if max_degree > 0else15
cytoscape_nodes.append({
'data': {
'id': str(node_id), # ID 必须是字符串
'label': str(node_id).replace(' ', '\n'), # 显示的标签 (将空格替换为换行)
'degree': degree, # 存储度数信息
'size': node_size, # 存储计算出的节点大小，用于样式设置
'tooltip_text': f"实体: {str(node_id)}\n度数: {degree}"# 鼠标悬停时显示的提示信息
}
})
print(f"已转换 {len(cytoscape_nodes)} 个节点。")
print("正在转换边...")
edge_count = 0
for u, v, data in knowledge_graph.edges(data=True):
edge_id = f"edge_{edge_count}"# 唯一的边 ID
predicate_label = data.get('label', '') # 获取边的标签 (谓语)
cytoscape_edges.append({
'data': {
'id': edge_id,
'source': str(u), # 源节点 ID (必须是字符串)
'target': str(v), # 目标节点 ID (必须是字符串)
'label': predicate_label, # 显示在边上的标签
'tooltip_text': f"关系: {predicate_label}"# 鼠标悬停时显示的提示信息
}
})
edge_count += 1
print(f"已转换 {len(cytoscape_edges)} 条边。")
# 组合成最终的数据结构
cytoscape_graph_data = {'nodes': cytoscape_nodes, 'edges': cytoscape_edges}
# 可视化转换后的结构 (前几个节点/边)
print("\n===== Cytoscape 节点数据样本 (前 2 个) =====\n")
# 使用 json.dumps 美化打印
import json
print(json.dumps(cytoscape_graph_data['nodes'][:2], indent=2, ensure_ascii=False))
print("\n===== Cytoscape 边数据样本 (前 2 个) =====\n")
print(json.dumps(cytoscape_graph_data['edges'][:2], indent=2, ensure_ascii=False))
print("-" * 25)
else:
print("由于图无效，跳过数据转换。")
cytoscape_graph_data = {'nodes': [], 'edges': []} # 确保变量存在数据转换完成。我们已经将节点和边的数据整理成了 ipycytoscape 所需的格式，其中包含了计算好的节点大小和有用的悬停提示信息。
接下来，创建实际的可视化小部件对象，并将我们的图数据加载进去。
我们可以通过类似于 CSS 的样式语法来定义节点和边的外观。让我们定义一个美观、色彩丰富且具有交互性的样式。我们将根据节点的度数调整其大小，在鼠标悬停或选中时改变颜色，并为边添加标签。
现在来渲染这个小部件。你应该能在这个单元格下方的输出区域看到交互式的知识图谱。
好的，关于这张知识图谱图像的关键信息如下：
• 中心实体: “marie curie”（玛丽·居里）是位于中心的主要节点。 • 关系: 橙色的箭头（边）显示了从“marie curie”出发的连接。 • 谓语标签: 边上的文字（例如，“discovered”发现，“won”获得，“was”是）定义了关系的类型。 • 连接的实体: 箭头末端的节点是宾语或相关的实体（例如，“radium”镭，“polonium”钋，“physicist”物理学家）。 • SPO 三元组: 每个箭头代表一个由 LLM 提取的（主语, 谓语, 宾语）事实。 • 可视化总结: 该图提供了一个关于玛丽·居里相关事实的快速、结构化的概览。 • 图结构: 这是一个以玛丽·居里为中心的辐射状结构。
未来可探索的方向
这个流程是一个很棒的起点，但总有改进和进一步探索的空间：
• 更强大的错误处理: 通过引入重试机制或改进对连续失败数据块的处理，增强 LLM 调用的稳定性。 • 更高级的规范化技术: 不仅仅是简单的字符串匹配。可以实现实体链接（将“Marie Curie”和“M. Curie”关联到同一个现实世界实体的 ID）或关系聚类（将相似的谓语，如“was born in”和“born at”，归为一类）。 • 深入探索提示工程: 进行实验！尝试不同的 LLM 模型，调整提示指令，改变温度（temperature）参数（用于控制模型输出的随机性），观察结果如何变化。 • 建立评估体系: 提取出的三元组质量如何？可以实现评估方法来衡量精确率（precision）和召回率（recall）。 • 实现更丰富的可视化效果: 为不同类型的节点（人物、地点、概念）使用不同的颜色或形状。在可视化中直接添加更多交互功能或图分析结果。 • 利用图分析挖掘更深层信息: 发挥 networkx的强大功能，找出最重要的节点（通过中心性分析，衡量节点在网络中的重要程度），发现实体之间的路径，或识别图中的社群结构。• 数据持久化存储: 将图进行保存！对于大型项目和更复杂的查询，可以将提取的三元组或图结构存储在专门的图数据库中，如 Neo4j。
希望本文能为您提供有价值的参考。
