这是一份关于如何利用大语言模型构建知识图谱的实用指南。核心内容覆盖三个方面：自定义流程构建知识图谱的方法、不同方案的性能对比与选择、以及本地计算机配置对模型选择的影响。
通过创建一个自定义流程来自动上传业务数据
在本节中，我们将构建一个专属自定义流程，借助大语言模型自动生成节点定义、关系以及Cypher查询，并基于数据集运行。该方法同样适用于其他DataFrame，并且能够自动识别其Schema结构。
需要注意的是，这种方法在性能上可能存在瓶颈，尤其是与Langchain的LLMGraphTransformer相比，下一节会进行详细对比。而本节的核心目标是帮助你理解如何从零开始构建这一流程，从原理出发，让你有机会设计自己的Graph-Builder。事实上，当前主流方法的主要局限在于它对数据的天然含义和模式高度敏感。正因如此，突破固有思维模式就显得至关重要，这样才能从零开始设计GraphRAG，或者利用现有的最佳实践来满足业务需求。
现在，我们进入实操环节，先设置后续练习要用的大语言模型。你可以使用Langchain支持的任何模型，只要其性能满足真实业务需要即可。
这里有两个免费的备选方案：DeepSeek-V3（注册后赠送10元额度，有效期一个月）和Ollama（可让你轻松在本地运行开源模型）。两种方案都经过测试——尽管DeepSeek-V3提供了与GPT-4o相近的性能，但还是推荐使用Ollama进行学习，这样可以更深入地了解从模型下载到运行的完整流程。
在Ollama示例中，我们将使用Qwen2.5-Coder:7B，该模型针对代码任务进行了微调，在代码生成、推理和修复代码错误方面表现出色。你可以根据本地计算机配置选择更高参数量的版本，如14B或32B。
先初始化模型：
llm = OllamaLLM(model="qwen2.5-coder:latest")接着提取数据集的结构，并定义节点及其属性：
node_structure = "\n".join([
f"{col}: {', '.join(map(str, movies[col].unique()[:3]))}..."
for col in movies.columns
])
print(node_structure)对数据集中的每一列（例如：电影类型、导演），展示部分样本值。这有助于大语言模型理解数据格式以及每一列的典型值。
Release Year: 1907, 1908, 1909...
Title: Daniel boone, Laughing gas, The adventures of dollie...
Origin/Ethnicity: American...
Director: Wallace mccutcheon and ediwin s. porter, Edwin stanton porter, D. w. griffith...
Cast: William cra ven, florence lawrence, Bertha regustus, edward boulden, Arthur v. johnson, linda arvidson...
Genre: Biographical, Comedy, Drama...
Plot: Boone's daughter befriends an indian maiden as boone and his companion start out on a hunting expedition. while he is away, boone's cabin is attacked by the indians, who set it on fire and abduct boone's daughter. boone returns, swears vengeance, then heads out on the trail to the indian camp. his daughter escapes but is chased. the indians encounter boone, which sets off a huge fight on the edge of a cliff. a burning arrow gets shot into the indian camp. boone gets tied to the stake and tortured. the burning arrow sets the indian camp on fire, causing panic. boone is rescued by his horse, and boone has a knife fight in which he kills the indian chief.[2], The plot is that of a black woman going to the dentist for a toothache and being given laughing gas. on her way walking home, and in other situations, she can't stop laughing, and everyone she meets "catches" the laughter from her, including a vendor and police officers., On a beautiful summer day a father and mother take their daughter dollie on an outing to the river. the mother refuses to buy a gypsy's wares. the gypsy tries to rob the mother, but the father drives him off. the gypsy returns to the camp and devises a plan. they return and kidnap dollie while her parents are distracted. a rescue crew is organized, but the gypsy takes dollie to his camp. they gag dollie and hide her in a barrel before the rescue party gets to the camp. once they lea ve the gypsies and escapes in their wagon. as the wagon crosses the river, the barrel falls into the water. still sealed in the barrel, dollie is swept downstream in dangerous currents. a boy who is fishing in the river finds the barrel, and dollie is reunited safely with her parents...生成节点
接下来，使用大语言模型的提示词模板来引导模型提取节点及其属性。先看完整代码：
# 设置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def validate_node_definition(node_def: Dict) -> bool:
"""验证节点结构定义"""
if not isinstance(node_def, dict):
return False
return all(
isinstance(v, dict) and all(isinstance(k, str) for k in v.keys())
for v in node_def.values()
)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def get_node_definitions(chain, structure: str, example: Dict) -> Dict[str, Dict[str, str]]:
"""使用重试逻辑来获取节点定义"""
try:
# 从大语言模型获得响应
response = chain.invoke({"structure": structure, "example": example})
# 解析响应
node_defs = ast.literal_eval(response)
# 验证结构
if not validate_node_definition(node_defs):
raise ValueError("无效的节点结构定")
return node_defs
except (ValueError, SyntaxError) as e:
logger.error(f"解析节点定义时出错: {e}")
raise
# 更新节点定义模板
node_example = {
"NodeLabel1": {"property1": "row['property1']", "property2": "row['property2']"},
"NodeLabel2": {"property1": "row['property1']", "property2": "row['property2']"},
"NodeLabel3": {"property1": "row['property1']", "property2": "row['property2']"},
}
define_nodes_prompt = PromptTemplate(
input_variables=["example", "structure"],
template=("""
分析以下数据集结构并提取节点的实体标签及其属性。\n
节点属性应基于数据集列和它们的值。\n
返回的结果应为一个字典，其中键是节点标签，值是节点属性。\n\n
示例: {example}\n\n
数据集结构:\n{structure}\n\n
确保包括所有可能的节点标签及其属性。\n
如果某个属性可以是其自己的节点，请将其作为单独的节点标签。\n
请不要使用三重反引号标识代码块，只需返回元组的列表。\n
仅返回包含节点标签和属性的字典，不要包含任何其他文本或引号。
"""
),
)
# 带有错误处理机制的执行过程
try:
node_chain = define_nodes_prompt | llm
node_definitions = get_node_definitions(node_chain, structure=node_structure, example=node_example)
logger.info(f"节点定义: {node_definitions}")
except Exception as e:
logger.error(f"获取节点定义失败: {e}")
raise代码中先使用logging库设置日志记录，这是一个Python模块，用于跟踪执行过程中的事件（如错误或状态更新）：
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)通过basicConfig配置日志，显示INFO级别或更高的消息，并初始化日志记录器实例，用于在代码中记录消息。这一步并非强制，也可以用print语句替代，但良好的工程实践往往倾向于使用日志。
接着创建一个函数来验证大语言模型生成的节点：
def validate_node_definition(node_def: Dict) -> bool:
"""验证节点结构定义"""
if not isinstance(node_def, dict):
return False
return all(
isinstance(v, dict) and all(isinstance(k, str) for k in v.keys())
for v in node_def.values()
)该函数接收一个字典，键是节点标签（如Movie），值是属性的字典（如title、year）。首先检查node_def是否为字典，然后验证字典中的每个值是否也是字典，且这些字典中的所有键都是字符串。若结构有效则返回True。
再创建一个函数来调用LLM链并生成节点：
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def get_node_definitions(chain, structure: str, example: Dict) -> Dict[str, Dict[str, str]]:
"""获取带有重试逻辑的节点定义"""
try:
# 从大语言模型获取响应
response = chain.invoke({"structure": structure, "example": example})
# 解析响应
node_defs = ast.literal_eval(response)
# 验证结构
if not validate_node_definition(node_defs):
raise ValueError("无效的节点结构定义")
return node_defs@retry装饰器意味着如果发生错误，函数会自动重试。其中stop_after_attempt(3)表示最多重试3次，wait_exponential则在重试之间递增延迟（如4秒、8秒、16秒等）。函数的输入包括chain（LangChain管道：提示+LLM）、structure（数据集结构和示例值）以及example（引导LLM的示例节点定义）。chain.invoke将结构和示例发送给LLM，返回字符串响应，用ast.literal_eval将其转为Python字典，再通过validate_node_definition检查格式，无效则抛出ValueError。
except (ValueError, SyntaxError) as e:
logger.error(f"Error parsing node definitions: {e}")
raise若响应无法解析或验证，记录错误并抛出异常。
接着为LLM提供提示词模板来引导节点生成：
define_nodes_prompt = PromptTemplate(
input_variables=["example", "structure"],
template=("""
分析以下数据集结构并提取节点的实体标签及其属性。\n
节点属性应基于数据集列和它们的值。\n
返回的结果应为一个字典，其中键是节点标签，值是节点属性。\n\n
示例: {example}\n\n
数据集结构:\n{structure}\n\n
确保包括所有可能的节点标签及其属性。\n
如果某个属性可以是其自己的节点，请将其作为单独的节点标签。\n
请不要使用三重反引号标识代码块，只需返回元组的列表。\n
仅返回包含节点标签和属性的字典，不要包含任何其他文本或引号。
"""),
)注意这里提供了本节开始时定义的节点结构，以及如何生成节点字典的示例：
node_example = {
"NodeLabel1": {"property1": "row['property1']", "property2": "row['property2']"},
"NodeLabel2": {"property1": "row['property1']", "property2": "row['property2']"},
"NodeLabel3": {"property1": "row['property1']", "property2": "row['property2']"},
}示例中键是节点标签，值是映射到数据集列的属性字典。
然后执行链：
try:
node_chain = define_nodes_prompt | llm
node_definitions = get_node_definitions(node_chain, structure=node_structure, example=node_example)
logger.info(f"节点定义: {node_definitions}")
except Exception as e:
logger.error(f"获取节点定义失败: {e}")
raise在LangChain中，使用结构化提示词|LLM|…创建链，将提示词模板与LLM结合形成管道。通过get_node_definitions获取并验证节点定义。如果过程成功，将生成类似下面的结果：
INFO:__main__:Node Definitions: {'Movie': {'Release Year': "row['Release Year']", 'Title': "row['Title']"}, 'Director': {'Name': "row['Director']"}, 'Cast': {'Actor': "row['Cast']"}, 'Genre': {'Type': "row['Genre']"}, 'Plot': {'Description': "row['Plot']"}}生成关系
节点定义完成后，就可以识别它们之间的关系了。接下来看完整代码：
class RelationshipIdentifier:
"""识别图数据库中节点之间的关系。"""
RELATIONSHIP_EXAMPLE = [
("NodeLabel1", "RelationshipLabel", "NodeLabel2"),
("NodeLabel1", "RelationshipLabel", "NodeLabel3"),
("NodeLabel2", "RelationshipLabel", "NodeLabel3"),
]
PROMPT_TEMPLATE = PromptTemplate(
input_variables=["structure", "node_definitions", "example"],
template="""
考虑以下数据集结构：\n{structure}\n\n
考虑以下节点定义：\n{node_definitions}\n\n
根据数据集结构和节点定义，识别节点之间的关系（边）。\n
以三元组的形式返回关系，其中每个三元组包含起始节点标签、关系标签和结束节点标签，每个三元组是一个元组。\n
请仅返回元组列表。请不要使用三重反引号标识代码块，只返回元组列表。\n\n
示例：\n{example}
"""
)
def __init__(self, llm: Any, logger: logging.Logger = None):
self.llm = llm
self.logger = logger or logging.getLogger(__name__)
self.chain = self.PROMPT_TEMPLATE | self.llm
def validate_relationships(self, relationships: List[Tuple]) -> bool:
"""验证关系结构"""
return all(
isinstance(rel, tuple) and
len(rel) == 3 and
all(isinstance(x, str) for x in rel)
for rel in relationships
)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def identify_relationships(self, structure: str, node_definitions: Dict) -> List[Tuple]:
"""识别关系并应用重试逻辑"""
try:
response = self.chain.invoke({
"structure": structure,
"node_definitions": str(node_definitions),
"example": str(self.RELATIONSHIP_EXAMPLE)
})
relationships = ast.literal_eval(response)
if not self.validate_relationships(relationships):
raise ValueError("无效的关系结构")
self.logger.info(f"已验证 {len(relationships)} 个关系")
return relationships
except Exception as e:
self.logger.error(f"验证关系时出现错误：{e}")
raise
def get_relationship_types(self) -> List[str]:
"""提取唯一的关系类型。"""
return list(set(rel[1] for rel in self.identify_relationships()))
# 用法
identifier = RelationshipIdentifier(llm=llm)
relationships = identifier.identify_relationships(node_structure, node_definitions)
print("关系:", relationships)由于这段代码比节点生成更复杂，我们将逻辑封装在RelationshipIdentifier类中。关系示例采用三元组形式：
RELATIONSHIP_EXAMPLE = [
("NodeLabel1", "RelationshipLabel", "NodeLabel2"),
("NodeLabel1", "RelationshipLabel", "NodeLabel3"),
("NodeLabel2", "RelationshipLabel", "NodeLabel3"),
]每个元组包含起始节点标签、关系标签和结束节点标签。
实际提示词模板如下：
PROMPT_TEMPLATE = PromptTemplate(
input_variables=["structure", "node_definitions", "example"],
template="""
考虑以下数据集结构：\n{structure}\n\n
考虑以下节点定义：\n{node_definitions}\n\n
根据数据集结构和节点定义，识别节点之间的关系（边）。\n
以三元组的形式返回关系，其中每个三元组包含起始节点标签、关系标签和结束节点标签，每个三元组是一个元组。\n
请仅返回元组列表。请不要使用三重反引号标识代码块，只返回元组列表。\n\n
示例：\n{example}
"""
)这里使用了三个输入变量：structure（数据集结构）、node_definitions（节点标签及其属性）、example（示例关系）。
类的初始化方法：
def __init__(self, llm: Any, logger: logging.Logger = None):
self.llm = llm
self.logger = logger or logging.getLogger(__name__)
self.chain = self.PROMPT_TEMPLATE | self.llmllm是语言模型，logger可选用于记录进度，self.chain则创建可重用的管道。
验证关系的函数：
def validate_relationships(self, relationships: List[Tuple]) -> bool:
"""验证关系结构。"""
return all(
isinstance(rel, tuple) and
len(rel) == 3 and
all(isinstance(x, str) for x in rel)
for rel in relationships
)检查每个项目是否为三元组且所有元素都是字符串。
识别关系的方法带有重试逻辑：
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def identify_relationships(self, structure: str, node_definitions: Dict) -> List[Tuple]:
"""识别关系并应用重试逻辑。"""
try:
response = self.chain.invoke({
"structure": structure,
"node_definitions": str(node_definitions),
"example": str(self.RELATIONSHIP_EXAMPLE)
})
relationships = ast.literal_eval(response)
if not self.validate_relationships(relationships):
raise ValueError("无效的关系结构")
self.logger.info(f"已验证 {len(relationships)} 个关系")
return relationships失败时记录错误并最多重试3次。
最后一个方法提取唯一的关系标签：
def get_relationship_types(self) -> List[str]:
"""Extract unique relationship types."""
return list(set(rel[1] for rel in self.identify_relationships()))现在生成关系：
identifier = RelationshipIdentifier(llm=llm)
relationships = identifier.identify_relationships(node_structure, node_definitions)
print("Relationships:", relationships)成功时返回类似下面的结果：
INFO:__main__:Identified 4 relationships
Relationships: [('Movie', 'Directed By', 'Director'), ('Movie', 'Starring', 'Cast'), ('Movie', 'Has Genre', 'Genre'), ('Movie', 'Contains Plot', 'Plot')]生成Cypher查询
节点和关系定义完成后，创建Cypher查询将数据加载到Neo4j。这个过程与节点生成和关系生成逻辑相似，但增加了额外的验证步骤，因为输出将直接用于加载到知识图谱中，所以需要尽可能提高成功的概率。先看完整代码：
class CypherQueryBuilder:
"""构建用于 Neo4j 图数据库的 Cypher 查询。"""
INPUT_EXAMPLE = """
NodeLabel1: value1, value2
NodeLabel2: value1, value2
"""
EXAMPLE_CYPHER = example_cypher = """
CREATE (n1:NodeLabel1 {property1: "row['property1']", property2: "row['property2']"})
CREATE (n2:NodeLabel2 {property1: "row['property1']", property2: "row['property2']"})
CREATE (n1)-[:RelationshipLabel]->(n2);
"""
PROMPT_TEMPLATE = PromptTemplate(
input_variables=["structure", "node_definitions", "relationships", "example"],
template="""
考虑以下节点定义：\n{node_definitions}\n\n
考虑以下关系：\n{relationships}\n\n
生成 Cypher 查询以创建节点和关系，使用下面的节点定义和关系。记得用数据集中的实际数据替换占位符值。\n
包括每个节点的所有属性，按照节点定义，并创建关系。\n
返回一个包含每个查询用分号分隔的单个字符串。\n
请不要在响应中包含任何其他文本或引号。\n
请仅返回包含 Cypher 查询的字符串。请不要使用三重反引号标识代码块。\n\n
示例输入：\n{input}\n\n
示例输出Cypher查询：\n{cypher}
"""
)
def __init__(self, llm: Any, logger: logging.Logger = None):
self.llm = llm
self.logger = logger or logging.getLogger(__name__)
# self.chain = LLMChain(llm=llm, prompt=self.PROMPT_TEMPLATE)
self.chain = self.PROMPT_TEMPLATE | self.llm
def validate_cypher_query(self, query: str) -> bool:
"""使用 LLM 和正则表达式模式验证 Cypher 查询语法。"""
VALIDATION_PROMPT = PromptTemplate(
input_variables=["query"],
template="""
验证此Cypher查询并返回 TRUE 或 FALSE：
查询: {query}
检查规则：
1. 有效的 CREATE 语句
2. 正确的属性格式
3. 有效的关系语法
4. 无缺失的括号
5. 有效的属性名称
6. 有效的关系类型
如果查询有效，返回 TRUE；如果无效，返回 FALSE。
"""
)
try:
# 基本模式验证
basic_valid = all(re.search(pattern, query) for pattern in [
r'CREATE \(',
r'{.*?}',
r'\)-\[:.*?\]->'
])
if not basic_valid:
return False
# LLM 验证
validation_chain = VALIDATION_PROMPT | self.llm
result = validation_chain.invoke({"query": query})
# 解析结果
is_valid = "TRUE" in result.upper()
if not is_valid:
self.logger.warning(f"LLM 验证查询失败: {query}")
return is_valid
except Exception as e:
self.logger.error(f"验证错误: {e}")
return False
def sanitize_query(self, query: str) -> str:
"""清理并格式化 Cypher 查询"""
return (query
.strip()
.replace('\n', ' ')
.replace(' ', ' ')
.replace("'row[", "row['")
.replace("]'", "']"))
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def build_queries(self, node_definitions: Dict, relationships: List) -> str:
"""构建带有重试逻辑的 Cypher 查询。"""
try:
response = self.chain.invoke({
"node_definitions": str(node_definitions),
"relationships": str(relationships),
"input": self.INPUT_EXAMPLE,
"cypher": self.EXAMPLE_CYPHER
})
# 获取位于三重反引号内的响应。
if '```' in response:
response = response.split('```')[1]
# 清理响应
queries = self.sanitize_query(response)
# 验证查询
if not self.validate_cypher_query(queries):
raise ValueError("无效的 Cypher 查询语法")
self.logger.info("成功生成 Cypher 查询")
return queries
except Exception as e:
self.logger.error(f"构建 Cypher 查询出错: {e}")
raise
def split_queries(self, queries: str) -> List[str]:
"""将组合的查询拆分为单独的语句。"""
return [q.strip() for q in queries.split(';') if q.strip()]
# 用法
builder = CypherQueryBuilder(llm=llm)
cypher_queries = builder.build_queries(node_definitions, relationships)
print("Cypher 查询:", cypher_queries)提示词模板接收四个变量：structure（数据集结构）、node_definitions（生成的节点及其属性）、relationships（生成的关系）、example（示例查询）。
初始化方式与关系类相同。
验证方法执行两个步骤：先用正则表达式进行基本验证，确保包含必要的Cypher语法（CREATE、属性括号、关系箭头）。然后使用LLM进行高级验证，要求LLM检查6个规则。
def validate_cypher_query(self, query: str) -> bool:
"""使用 LLM 和正则表达式模式验证 Cypher 查询语法。"""
...
# 基本模式验证
basic_valid = all(re.search(pattern, query) for pattern in [
r'CREATE \(',
r'{.*?}',
r'\)-\[:.*?\]->'
])
if not basic_valid:
return False
# LLM 验证
validation_chain = VALIDATION_PROMPT | self.llm
result = validation_chain.invoke({"query": query})
is_valid = "TRUE" in result.upper()
...sanitize_query方法用于清理生成输出：移除多余空格和换行符，修复可能的格式问题（如row['property1']的引号位置）。注意，对于较小参数量的模型，可能需要更多的数据清理操作。
build_queries方法带有重试逻辑，并处理LLM可能输出的Markdown代码块标记：如果响应中有三重反引号，只提取内部内容。
split_queries方法将一个包含多个查询的字符串拆分为单独的语句列表，如将“CREATE (n1:Movie {title: "Inception"}); CREATE (n2:Director {name: "Nolan"});”拆分为两个独立字符串。
最后，初始化类并生成查询：
builder = CypherQueryBuilder(llm=llm)
cypher_queries = builder.build_queries(node_definitions, relationships)
print("Cypher 查询:", cypher_queries)成功时输出如下：
INFO:__main__:Successfully generated Cypher queries
Cypher Queries: CREATE (m:Movie {Release_Year: "row['Release Year']", Title: "row['Title']"}) CREATE (d:Director {Name: "row['Director']"}) CREATE (c:Cast {Actor: "row['Cast']"}) CREATE (g:Genre {Type: "row['Genre']"}) CREATE (p:Plot {Description: "row['Plot']"}) CREATE (m)-[:Directed_By]->(d) CREATE (m)-[:Starring]->(c) CREATE (m)-[:Has_Genre]->(g) CREATE (m)-[:Contains_Plot]->(p)最后，遍历数据集，为每一行执行生成的Cypher查询：
logs = ""
total_rows = len(df)
def sanitize_value(value):
if isinstance(value, str):
return value.replace('"', '')
return str(value)
for index, row in tqdm(df.iterrows(),
total=total_rows,
desc="正在加载数据到 Neo4j",
position=0,
lea ve=True):
# 将占位符替换为实际的值
cypher_query = cypher_queries
for column in df.columns:
cypher_query = cypher_query.replace(
f"row['{column}']",
f'{sanitize_value(row[column])}'
)
try:
# 执行查询并更新进度
conn.execute_query(cypher_query)
except Exception as e:
logs += f"在行 {index+1}: {str(e)} 出现错误\n"这里定义了一个空字符串变量logs用于捕获失败信息，并添加了一个清理函数sanitize_value，防止字符串中的双引号破坏查询语法。
遍历数据集时，使用tqdm添加进度条，total_rows用于计算进度，desc设置标签，position和lea ve确保进度条在控制台中保持可见。然后将占位符如row['column_name']替换为实际值，并执行查询。
检查数据集是否上传成功：切换到Neo4j，运行以下Cypher查询：
MATCH p=(m:Movie)-[r]-(n)
RETURN p
LIMIT 100;在测试机器上，LLM生成了以下图表：
这与手动上传的知识图谱非常相似。对于一个简单的LLM来说，结果已经相当不错了。虽然需要不少编码工作，但现在可以重复用于多个数据集，更重要的是，它成为了构建更复杂LLM图构建器的基础。
在本示例中，还没有通过提供实体、关系和属性示例来帮助LLM。但实际应用中，可以考虑添加示例来提高性能。此外，更现代的方法会利用思维链来提出额外的节点和关系，让模型能够顺序推理并进一步改进结果。另一种策略是提供行样本，以更好地适应每行中提供的具体值。