知识图谱构建——从非结构化文本自动提取实体关系
2026/4/30大约 7 分钟
知识图谱构建——从非结构化文本自动提取实体关系
适读人群：做企业知识管理或 RAG 系统的工程师 | 阅读时长：约 15 分钟 | 核心价值：掌握用 LLM 从文档构建知识图谱的完整工程方案，包括 Neo4j 集成和准确率提升技巧
去年我接了一个企业知识管理项目，甲方是一家做工程咨询的公司，有几千份技术文档积累了十几年，查什么东西都得靠人工翻。
最初方案是做 RAG，把文档向量化之后搜索。但做到一半发现一个问题：他们的很多知识是关系性的——"A 项目用了 B 技术，B 技术和 C 供应商有合作，C 供应商在 D 地区有服务能力"——这类多跳的关系查询，纯向量检索做不好。
于是我在 RAG 的基础上加了一层知识图谱。
这篇文章写的是那个知识图谱的完整实现，有真实代码，有踩坑记录，有准确率数据。
整体架构
文档 (PDF/Word/TXT)
|
v
[文本预处理] -- 分块、清洗
|
v
[LLM 实体关系抽取] -- 每个 chunk 抽取实体和关系
|
v
[实体消歧] -- 合并指向同一实体的不同表述
|
v
[Neo4j 图数据库] -- 存储实体节点和关系边
|
v
[查询接口] -- Cypher 查询 + LLM 自然语言转换第一步：文本预处理和分块
这步看起来简单，实际上对后续质量影响很大。
import re
from typing import Generator
def clean_text(text: str) -> str:
"""清洗文本，去掉影响理解的噪音"""
# 去掉多余空白
text = re.sub(r'\n{3,}', '\n\n', text)
text = re.sub(r' {2,}', ' ', text)
# 去掉页码（常见格式）
text = re.sub(r'\n第\s*\d+\s*页\s*\n', '\n', text)
text = re.sub(r'\n- \d+ -\n', '\n', text)
# 去掉页眉页脚（通常短且重复）
lines = text.split('\n')
# 过滤掉极短的行（可能是页眉页脚）
lines = [line for line in lines if len(line.strip()) > 5 or line.strip() == '']
return '\n'.join(lines)
def split_by_semantic_boundary(text: str, max_chunk_size: int = 1000) -> Generator[str, None, None]:
"""
按语义边界分块，优先在段落边界切分
而不是机械地按字符数切
"""
paragraphs = text.split('\n\n')
current_chunk = []
current_size = 0
for para in paragraphs:
para = para.strip()
if not para:
continue
para_size = len(para)
if current_size + para_size > max_chunk_size and current_chunk:
yield '\n\n'.join(current_chunk)
current_chunk = [para]
current_size = para_size
else:
current_chunk.append(para)
current_size += para_size
if current_chunk:
yield '\n\n'.join(current_chunk)第二步：LLM 实体关系抽取
这是核心步骤，也是最容易出问题的地方。
import json
from openai import OpenAI
from dataclasses import dataclass, field
@dataclass
class Entity:
name: str
type: str # PERSON, ORGANIZATION, TECHNOLOGY, PROJECT, LOCATION, CONCEPT
aliases: list[str] = field(default_factory=list)
properties: dict = field(default_factory=dict)
@dataclass
class Relation:
source: str # 实体名
target: str # 实体名
relation_type: str # 关系类型
properties: dict = field(default_factory=dict)
evidence: str = "" # 原文证据（用于验证）
def extract_entities_and_relations(text_chunk: str, domain_hint: str = "") -> dict:
"""
从文本块中抽取实体和关系
"""
client = OpenAI()
system_prompt = f"""你是一个知识图谱构建系统，专门从{domain_hint or '工程技术'}文档中抽取结构化知识。
抽取规则：
1. 实体类型包括：PERSON（人员）、ORGANIZATION（组织/公司）、TECHNOLOGY（技术/产品）、PROJECT（项目）、LOCATION（地点）、CONCEPT（概念/方法）
2. 关系必须有文本证据支撑，不要推断文中没有明确表达的关系
3. 实体名称使用文中出现的最完整表述
4. 如果同一实体有多种叫法，在 aliases 中列出
返回严格的 JSON 格式。"""
user_prompt = f"""请从以下文本中抽取所有实体和关系：
{text_chunk}
返回格式：
{{
"entities": [
{{
"name": "实体名称",
"type": "实体类型",
"aliases": ["别名1", "别名2"],
"properties": {{"key": "value"}}
}}
],
"relations": [
{{
"source": "源实体名称",
"target": "目标实体名称",
"relation_type": "关系类型",
"evidence": "原文中支撑这个关系的句子",
"properties": {{}}
}}
]
}}
如果文本中没有明确的实体或关系，返回空列表。不要编造文中没有的关系。"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
response_format={"type": "json_object"},
temperature=0.1
)
result = json.loads(response.choices[0].message.content)
return result关键设计决策：为什么要保留 evidence 字段？
evidence 字段是我踩坑之后加上去的。LLM 在抽取关系时有一个毛病：它会用"常识"去补全文本里没有明确表达的关系。比如文本里说"张工程师负责 A 项目"，它可能会自动添加"张工程师精通 Java"（因为它觉得工程师通常会 Java）。
要求 LLM 把原文证据也一起输出，一方面可以在验证时快速找到来源，更重要的是：强制 LLM 把注意力放在文本上，而不是靠常识推断。
第三步：实体消歧
这是最影响最终质量的步骤。同一家公司可能被叫做"阿里"、"阿里巴巴"、"Alibaba"、"阿里云"（等等，这个又不一样了）。
def merge_entities(all_entities: list[dict]) -> dict[str, str]:
"""
实体消歧：返回一个映射，把各种别名映射到标准名称
格式：{"别名": "标准名称"}
"""
# 先做简单的字符串相似度合并
from difflib import SequenceMatcher
entity_groups = [] # 每个元素是一组相似实体的名称列表
all_names = [e["name"] for e in all_entities]
processed = set()
for name in all_names:
if name in processed:
continue
group = [name]
processed.add(name)
# 收集显式的 aliases
for entity in all_entities:
if entity["name"] == name:
group.extend(entity.get("aliases", []))
# 字符串相似度匹配
for other_name in all_names:
if other_name in processed:
continue
ratio = SequenceMatcher(None, name, other_name).ratio()
if ratio > 0.85: # 85% 相似度阈值
group.append(other_name)
processed.add(other_name)
entity_groups.append(group)
# 对有歧义的组，用 LLM 做最终判断
ambiguous_groups = [g for g in entity_groups if len(g) > 1]
if ambiguous_groups:
name_map = llm_resolve_entity_ambiguity(ambiguous_groups)
else:
name_map = {}
return name_map
def llm_resolve_entity_ambiguity(groups: list[list[str]]) -> dict[str, str]:
"""
用 LLM 解决实体消歧
"""
client = OpenAI()
groups_text = json.dumps(groups, ensure_ascii=False, indent=2)
prompt = f"""以下是从文档中抽取的可能指向同一实体的名称组。
请判断每组中哪些名称确实指向同一实体，哪些其实是不同实体。
{groups_text}
返回格式：
{{
"merges": [
{{
"canonical_name": "标准名称（最完整、最正式的叫法）",
"aliases": ["别名1", "别名2"]
}}
],
"splits": [
{{
"group_index": 0,
"reason": "这组里的X和Y其实是不同实体，因为..."
}}
]
}}"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0.1
)
result = json.loads(response.choices[0].message.content)
# 构建别名到标准名的映射
name_map = {}
for merge in result.get("merges", []):
canonical = merge["canonical_name"]
for alias in merge.get("aliases", []):
name_map[alias] = canonical
return name_map第四步：写入 Neo4j
from neo4j import GraphDatabase
class KnowledgeGraphDB:
def __init__(self, uri: str, user: str, password: str):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def close(self):
self.driver.close()
def create_entity(self, entity: dict, name_map: dict):
"""创建实体节点"""
# 应用消歧映射
canonical_name = name_map.get(entity["name"], entity["name"])
with self.driver.session() as session:
session.run(
"""
MERGE (e:Entity {name: $name})
SET e.type = $type
SET e += $properties
WITH e
UNWIND $aliases AS alias
MERGE (a:EntityAlias {name: alias})
MERGE (a)-[:ALIAS_OF]->(e)
""",
name=canonical_name,
type=entity["type"],
properties=entity.get("properties", {}),
aliases=[name_map.get(a, a) for a in entity.get("aliases", [])]
)
def create_relation(self, relation: dict, name_map: dict):
"""创建关系边"""
source = name_map.get(relation["source"], relation["source"])
target = name_map.get(relation["target"], relation["target"])
# 关系类型转成大写下划线格式
rel_type = relation["relation_type"].upper().replace(" ", "_").replace("-", "_")
with self.driver.session() as session:
session.run(
f"""
MERGE (s:Entity {{name: $source}})
MERGE (t:Entity {{name: $target}})
MERGE (s)-[r:{rel_type}]->(t)
SET r.evidence = $evidence
SET r += $properties
""",
source=source,
target=target,
evidence=relation.get("evidence", ""),
properties=relation.get("properties", {})
)
def query_natural_language(self, question: str) -> str:
"""
把自然语言问题转成 Cypher 查询
"""
# 先获取图谱的 schema
schema = self.get_schema_summary()
client = OpenAI()
prompt = f"""你是一个 Neo4j Cypher 查询专家。
图谱的节点类型和关系类型如下：
{schema}
请把以下自然语言问题转成 Cypher 查询：
"{question}"
只返回 Cypher 查询语句，不要解释。
查询结果限制在 20 条以内。"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)
cypher = response.choices[0].message.content.strip()
# 去掉可能的代码块标记
cypher = cypher.replace("```cypher", "").replace("```", "").strip()
with self.driver.session() as session:
result = session.run(cypher)
records = [dict(record) for record in result]
return records
def get_schema_summary(self) -> str:
"""获取图谱 schema 的文字描述"""
with self.driver.session() as session:
# 获取节点标签
node_result = session.run("CALL db.labels()")
labels = [record["label"] for record in node_result]
# 获取关系类型
rel_result = session.run("CALL db.relationshipTypes()")
rel_types = [record["relationshipType"] for record in rel_result]
return f"节点类型: {', '.join(labels)}\n关系类型: {', '.join(rel_types)}"准确率数据和常见问题
在这个项目里，我人工抽样验证了 200 条关系抽取结果：
| 问题类型 | 比例 |
|---|---|
| 正确的关系 | 71% |
| 推断出的（文中没有明确表达） | 18% |
| 实体边界错误（切得太细或太粗） | 7% |
| 完全错误 | 4% |
71% 的准确率在行业里属于中等水平。提高准确率最有效的方法是：
方法 1：加 few-shot 示例
在 prompt 里加 3-5 个领域相关的例子，比通用 prompt 准确率提升大约 10-15%。
方法 2：双遍验证
让 LLM 抽取一遍，再让它对自己的抽取结果做一遍验证（类似 chain-of-thought 的思路），把"没有文本证据的关系"标记出来然后删除。
方法 3：控制 chunk 大小
chunk 太长，LLM 容易在末尾的实体里混入前面实体的关系。我测试下来 800-1200 字的 chunk 效果最好。
知识图谱不是 RAG 的替代品，而是补充。复杂的多跳关系查询，图谱比向量检索准确得多；但模糊的语义相似查询，向量检索更好用。两者结合，才是完整的企业知识系统。