知识图谱技术架构
知识图谱技术架构详解
🏛️ 整体技术架构全景图
┌─────────────────────────────────────────────────────────────┐
│ 应用层 │
│ 智能搜索 | 问答系统 | 推荐引擎 | 风控系统 | 决策支持 │
└─────────────────────────┬───────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 知识计算层 │
│ 知识推理 | 图谱补全 | 实体链接 | 语义计算 │
└─────────────────────────┬───────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 知识管理层 │
│ 知识融合 | 质量评估 | 版本管理 | 知识更新 │
└─────────────────────────┬───────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 知识获取层 │
│ 信息抽取 | 知识挖掘 | 众包标注 | 结构化数据导入 │
└─────────────────────────┬───────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 数据存储层 │
│ 图数据库 | RDF存储 | 向量数据库 | 缓存层 │
└─────────────────────────┬───────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 数据源层 │
│ 结构化DB | 网页文本 | 学术论文 | 企业文档 | API数据 │
└─────────────────────────────────────────────────────────────┘
1️⃣ 数据源层
数据来源分类
数据源
├── 结构化数据
│ ├── 关系型数据库（MySQL、PostgreSQL）
│ ├── Excel / CSV 文件
│ └── 业务系统导出数据
│
├── 半结构化数据
│ ├── 维基百科（Infobox）
│ ├── HTML 网页表格
│ ├── JSON / XML 数据
│ └── 学术数据库
│
└── 非结构化数据
├── 新闻文章
├── 学术论文
├── 企业文档
├── 社交媒体
└── 对话记录
2️⃣ 知识获取层
核心任务架构
原始文本："马云1999年在杭州创立了阿里巴巴公司"
↓
┌───────────────────────────────────────────┐
│ NLP 处理流水线 │
│ │
│ 分词 → 词性标注 → 句法分析 → 语义分析 │
│ │
│ 马云/1999年/杭州/创立/阿里巴巴/公司 │
└────────────────────┬──────────────────────┘
↓
┌────────────┴────────────┐
↓ ↓
┌───────────────┐ ┌──────────────────┐
│ 实体识别 │ │ 关系抽取 │
│ (NER) │ │ (RE) │
│ │ │ │
│ 马云 → 人物 │ │ 马云─[创立]→阿里 │
│ 1999年 → 时间 │ │ 马云─[出生地]→杭州│
│ 杭州 → 地点 │ │ │
│ 阿里巴巴→ 组织│ └──────────────────┘
└───────────────┘
↓
┌───────────────────────────────────────────┐
│ 属性抽取 │
│ 阿里巴巴.成立时间 = 1999年 │
│ 阿里巴巴.创始人 = 马云 │
│ 阿里巴巴.总部 = 杭州 │
└───────────────────────────────────────────┘
关键技术模型
# 实体识别示例（基于 BERT）
from transformers import AutoTokenizer, AutoModelForTokenClassification
import torch
class 实体识别器:
def __init__(self):
self.tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese")
self.model = AutoModelForTokenClassification.from_pretrained(
"chinese-bert-ner"
)
def 识别实体(self, 文本: str):
# 分词
inputs = self.tokenizer(文本, return_tensors="pt")
# 模型推理
outputs = self.model(**inputs)
predictions = torch.argmax(outputs.logits, dim=2)
# 解析实体
实体列表 = self._parse_entities(文本, predictions)
return 实体列表
def _parse_entities(self, 文本, predictions):
# B-PER: 人名开始, I-PER: 人名中间
# B-ORG: 组织开始, B-LOC: 地名开始
实体 = []
# ... 解析逻辑
return 实体
# 关系抽取示例
class 关系抽取器:
def __init__(self):
self.model = self._load_relation_model()
def 抽取关系(self, 文本: str, 实体对: tuple):
主体, 客体 = 实体对
# 构造输入：[CLS] 文本 [SEP] 主体 [SEP] 客体
input_text = f"[CLS]{文本}[SEP]{主体}[SEP]{客体}"
# 预测关系类型
关系类型 = self.model.predict(input_text)
置信度 = self.model.confidence(input_text)
return {
"主体": 主体,
"关系": 关系类型,
"客体": 客体,
"置信度": 置信度
}
3️⃣ 数据存储层
存储架构设计
┌─────────────────────────────────────────────────────┐
│ 多层存储架构 │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 热数据缓存 │ │ 图数据库 │ │
│ │ Redis │ │ Neo4j │ │
│ │ 高频查询 │ │ 核心图谱 │ │
│ └──────────────┘ └──────────────┘ │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 向量数据库 │ │ 关系型数据库 │ │
│ │ Milvus │ │ MySQL │ │
│ │ 语义检索 │ │ 元数据存储 │ │
│ └──────────────┘ └──────────────┘ │
│ │
│ ┌──────────────────────────────────┐ │
│ │ 对象存储 / 文件系统 │ │
│ │ 原始文档 | 图谱备份 | 日志 │ │
│ └──────────────────────────────────┘ │
└─────────────────────────────────────────────────────┘
Neo4j 图数据库详解
// ========== 数据模型设计 ==========
// 节点标签定义
(:人物 {
id: "P001",
姓名: "马云",
出生日期: "1964-09-10",
国籍: "中国",
简介: "阿里巴巴集团创始人"
})
(:公司 {
id: "C001",
名称: "阿里巴巴",
成立时间: "1999",
行业: "互联网",
上市地: "纽约证券交易所"
})
(:地点 {
id: "L001",
名称: "杭州",
类型: "城市",
所属省份: "浙江省"
})
// 关系定义
(马云)-[:创立 {时间: "1999", 角色: "CEO"}]->(阿里巴巴)
(马云)-[:出生于]->(杭州)
(阿里巴巴)-[:总部位于]->(杭州)
// ========== 常用查询 ==========
// 1. 查找某人创立的所有公司
MATCH (p:人物 {姓名: "马云"})-[:创立]->(c:公司)
RETURN p.姓名, c.名称, c.成立时间
ORDER BY c.成立时间
// 2. 多跳查询：与马云有2度关系的人物
MATCH (p:人物 {姓名: "马云"})-[*1..2]-(related:人物)
WHERE related.姓名 <> "马云"
RETURN DISTINCT related.姓名, related.职业
// 3. 最短路径查询
MATCH path = shortestPath(
(马云:人物 {姓名: "马云"})-[*]-(任正非:人物 {姓名: "任正非"})
)
RETURN path
// 4. 图谱统计分析
MATCH (n)
RETURN labels(n)[0] AS 节点类型, count(n) AS 数量
ORDER BY 数量 DESC
// 5. 社区发现：找出关系最密集的实体群
MATCH (n)-[r]-(m)
WITH n, count(r) AS 关系数
WHERE 关系数 > 10
RETURN n.姓名, 关系数
ORDER BY 关系数 DESC
LIMIT 20
向量存储（语义检索）
from pymilvus import Collection, FieldSchema, CollectionSchema, DataType
from sentence_transformers import SentenceTransformer
class 知识向量库:
def __init__(self):
self.encoder = SentenceTransformer("paraphrase-multilingual-mpnet-base-v2")
self.collection = self._init_collection()
def _init_collection(self):
# 定义字段结构
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
FieldSchema(name="实体名称", dtype=DataType.VARCHAR, max_length=200),
FieldSchema(name="描述文本", dtype=DataType.VARCHAR, max_length=2000),
FieldSchema(name="向量", dtype=DataType.FLOAT_VECTOR, dim=768),
]
schema = CollectionSchema(fields)
return Collection("知识图谱实体", schema)
def 存入实体(self, 实体列表: list):
# 批量向量化
文本列表 = [实体["描述"] for 实体 in 实体列表]
向量列表 = self.encoder.encode(文本列表).tolist()
# 写入向量库
self.collection.insert([
[实体["id"] for 实体 in 实体列表],
[实体["名称"] for 实体 in 实体列表],
文本列表,
向量列表
])
def 语义检索(self, 查询文本: str, top_k: int = 10):
# 查询向量化
查询向量 = self.encoder.encode([查询文本]).tolist()
# 相似度搜索
结果 = self.collection.search(
data=查询向量,
anns_field="向量",
param={"metric_type": "COSINE", "params": {"nprobe": 16}},
limit=top_k,
output_fields=["实体名称", "描述文本"]
)
return 结果
4️⃣ 知识管理层
知识融合流程
多源异构数据输入：
数据源A：马云 → 企业家
数据源B：Jack Ma → Businessman
数据源C：马云 → 出生于1964年
↓
┌───────────────────────────────────┐
│ 实体对齐 │
│ │
│ 马云 = Jack Ma = 马云 │
│ （通过姓名/别名/属性特征匹配） │
└─────────────────┬─────────────────┘
↓
┌───────────────────────────────────┐
│ 冲突检测与消解 │
│ │
│ 冲突示例： │
│ A说：马云出生于1964年 │
│ B说：马云出生于1965年 │
│ │
│ 消解策略： │
│ ① 可信度加权（选高可信来源） │
│ ② 时效性优先（选最新数据） │
│ ③ 人工审核（重要冲突） │
└─────────────────┬─────────────────┘
↓
┌───────────────────────────────────┐
│ 知识补全 │
│ 填充缺失属性和关系 │
└───────────────────────────────────┘
质量评估体系
class 知识质量评估:
def 评估三元组质量(self, 三元组: dict) -> dict:
scores = {}
# 1. 准确性评分
scores["准确性"] = self._check_accuracy(三元组)
# 2. 完整性评分
scores["完整性"] = self._check_completeness(三元组)
# 3. 一致性评分
scores["一致性"] = self._check_consistency(三元组)
# 4. 时效性评分
scores["时效性"] = self._check_timeliness(三元组)
# 综合评分
scores["综合"] = sum(scores.values()) / len(scores)
return scores
def _check_accuracy(self, 三元组):
# 与权威数据源交叉验证
主体, 谓语, 客体 = 三元组["主体"], 三元组["谓语"], 三元组["客体"]
# 检查类型约束
if 谓语 == "出生地" and not self._is_location(客体):
return 0.0 # 出生地必须是地点实体
# 检查值域范围
if 谓语 == "出生年份":
年份 = int(客体)
if not (1800 <= 年份 <= 2024):
return 0.0
return 1.0
def _check_consistency(self, 三元组):
# 检查是否与已有知识矛盾
现有知识 = self.graph.query(三元组["主体"], 三元组["谓语"])
if 现有知识 and 现有知识 != 三元组["客体"]:
return 0.5 # 存在冲突，需人工审核
return 1.0
5️⃣ 知识计算层
知识推理架构
推理方式分类：
┌─────────────────────────────────────────────┐
│ 知识推理 │
│ │
│ ┌──────────────┐ ┌───────────────────┐ │
│ │ 符号推理 │ │ 统计推理 │ │
│ │ ────────── │ │ ────────────── │ │
│ │ 规则引擎 │ │ 知识图谱嵌入 │ │
│ │ 本体推理 │ │ TransE/RotatE │ │
│ │ 一阶逻辑 │ │ 链接预测 │ │
│ └──────────────┘ └───────────────────┘ │
│ │
│ ┌──────────────────────────────────────┐ │
│ │ 神经符号融合推理 │ │
│ │ 深度学习 + 逻辑规则 结合 │ │
│ └──────────────────────────────────────┘ │
└─────────────────────────────────────────────┘
知识图谱嵌入（KGE）
将实体和关系映射到低维向量空间：
$$\text{TransE 模型：} \quad \vec{h} + \vec{r} \approx \vec{t}$$
其中 $\vec{h}$ 是头实体向量，$\vec{r}$ 是关系向量，$\vec{t}$ 是尾实体向量。
import torch
import torch.nn as nn
class TransE(nn.Module):
"""
TransE 知识图谱嵌入模型
核心思想：h + r ≈ t
"""
def __init__(self, 实体数量, 关系数量, 嵌入维度=100):
super().__init__()
# 实体嵌入矩阵
self.实体嵌入 = nn.Embedding(实体数量, 嵌入维度)
# 关系嵌入矩阵
self.关系嵌入 = nn.Embedding(关系数量, 嵌入维度)
# 初始化
nn.init.uniform_(self.实体嵌入.weight, -1, 1)
nn.init.uniform_(self.关系嵌入.weight, -1, 1)
def forward(self, 头实体, 关系, 尾实体):
# 获取向量表示
h = self.实体嵌入(头实体)
r = self.关系嵌入(关系)
t = self.实体嵌入(尾实体)
# 计算得分：|h + r - t|
# 得分越低 → 三元组越可能为真
得分 = torch.norm(h + r - t, p=2, dim=1)
return 得分
def 链接预测(self, 头实体_id, 关系_id, top_k=10):
"""预测：给定头实体和关系，找最可能的尾实体"""
h = self.实体嵌入(torch.tensor([头实体_id]))
r = self.关系嵌入(torch.tensor([关系_id]))
# 计算与所有实体的距离
所有实体 = self.实体嵌入.weight
得分 = torch.norm(h + r - 所有实体, p=2, dim=1)
# 返回得分最低的 top_k 个实体
_, top_indices = torch.topk(得分, k=top_k, largest=False)
return top_indices.tolist()
# 模型对比
嵌入模型对比 = {
"TransE": {"优点": "简单高效", "缺点": "不能处理1-N关系"},
"TransR": {"优点": "关系特定空间", "缺点": "参数量大"},
"RotatE": {"优点": "处理复杂关系模式", "缺点": "复数计算复杂"},
"ComplEx": {"优点": "处理非对称关系", "缺点": "可解释性差"},
"DistMult": {"优点": "计算简单", "缺点": "只能处理对称关系"},
}
多跳推理
class 多跳推理引擎:
"""
实现多步推理：
问题："马云的母校在哪个城市？"
推理链：马云 → 毕业于 → 杭州师范大学 → 位于 → 杭州
"""
def __init__(self, 知识图谱, llm_model):
self.图谱 = 知识图谱
self.llm = llm_model
async def 推理(self, 问题: str) -> str:
# 第一步：LLM 分解问题为推理链
推理链 = await self.llm.decompose_question(问题)
# ["马云的母校", "母校的城市"]
# 第二步：逐步在图谱中查询
当前实体 = "马云"
推理历史 = []
for 子问题 in 推理链:
# 图谱查询
关系 = await self.llm.extract_relation(子问题)
下一实体 = self.图谱.query(当前实体, 关系)
推理历史.append({
"实体": 当前实体,
"关系": 关系,
"结果": 下一实体
})
当前实体 = 下一实体
# 第三步：汇总推理路径生成答案
答案 = await self.llm.generate_answer(问题, 推理历史)
return 答案, 推理历史
6️⃣ 应用层接口
RESTful API 设计
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI(title="知识图谱 API")
# ========== 实体查询接口 ==========
@app.get("/entity/{entity_id}")
async def 查询实体(entity_id: str):
"""查询单个实体的详细信息"""
实体 = graph_db.get_entity(entity_id)
if not 实体:
raise HTTPException(status_code=404, detail="实体不存在")
return 实体
# ========== 关系查询接口 ==========
@app.get("/relations/{entity_id}")
async def 查询关系(
entity_id: str,
relation_type: str = None,
depth: int = 1
):
"""查询实体的关系网络"""
关系图 = graph_db.get_relations(
entity_id,
relation_type=relation_type,
max_depth=depth
)
return 关系图
# ========== 语义搜索接口 ==========
class 搜索请求(BaseModel):
query: str
top_k: int = 10
entity_type: str = None
@app.post("/search")
async def 语义搜索(请求: 搜索请求):
"""自然语言语义搜索"""
# 向量检索
向量结果 = vector_db.search(请求.query, 请求.top_k)
# 图谱扩展
扩展结果 = graph_db.expand(向量结果)
return 扩展结果
# ========== 智能问答接口 ==========
class 问答请求(BaseModel):
question: str
context: str = None
@app.post("/qa")
async def 智能问答(请求: 问答请求):
"""基于知识图谱的智能问答"""
# 实体识别
实体列表 = ner_model.extract(请求.question)
# 图谱查询
相关知识 = graph_db.query_subgraph(实体列表)
# LLM 生成答案
答案 = await llm.answer(
问题=请求.question,
知识上下文=相关知识
)
return {
"答案": 答案,
"推理路径": 相关知识["推理链"],
"置信度": 答案["confidence"]
}
# ========== 知识写入接口 ==========
class 三元组(BaseModel):
主体: str
谓语: str
客体: str
来源: str
置信度: float = 1.0
@app.post("/triple")
async def 写入三元组(triple: 三元组):
"""写入新的知识三元组"""
# 质量检测
质量分数 = quality_checker.evaluate(triple)
if 质量分数 < 0.6:
raise HTTPException(status_code=400, detail="知识质量不达标")
# 冲突检测
冲突 = conflict_detector.check(triple)
if 冲突:
return {"status": "待审核", "conflict": 冲突}
# 写入图谱
graph_db.insert_triple(triple)
return {"status": "成功", "quality_score": 质量分数}
🔄 知识图谱 + LLM 融合架构
GraphRAG 完整架构
用户提问
↓
┌─────────────────────────────────────────────┐
│ 查询理解层 │
│ LLM 分析问题意图 → 提取关键实体和关系 │
└────────────────────┬────────────────────────┘
↓
┌────────────┴────────────┐
↓ ↓
┌──────────────┐ ┌─────────────────┐
│ 图谱精确查询 │ │ 向量语义检索 │
│ Neo4j │ │ Milvus │
│ 结构化知识 │ │ 非结构化知识 │
└──────┬───────┘ └────────┬────────┘
└──────────┬──────────────┘
↓
┌─────────────────────────────────────────────┐
│ 知识融合与排序 │
│ 合并图谱结果 + 向量结果 → 相关度排序 │
└────────────────────┬────────────────────────┘
↓
┌─────────────────────────────────────────────┐
│ LLM 答案生成 │
│ 基于检索到的知识 → 生成准确流畅的答案 │
└────────────────────┬────────────────────────┘
↓
┌─────────────────────────────────────────────┐
│ 答案验证层 │
│ 知识图谱验证事实准确性 → 防止幻觉 │
└─────────────────────────────────────────────┘
↓
最终答案输出
（含推理路径 + 来源引用）
class GraphRAG系统:
def __init__(self):
self.图谱 = Neo4jClient()
self.向量库 = MilvusClient()
self.llm = OpenAIClient()
self.ner = NERModel()
async def 回答问题(self, 问题: str) -> dict:
# ===== Step 1: 理解问题 =====
实体列表 = self.ner.extract(问题)
查询意图 = await self.llm.classify_intent(问题)
# ===== Step 2: 双路检索 =====
# 路径A：图谱结构化查询
图谱结果 = self.图谱.query_subgraph(
entities=实体列表,
depth=3
)
# 路径B：向量语义检索
向量结果 = self.向量库.search(
query=问题,
top_k=5
)
# ===== Step 3: 知识融合 =====
融合知识 = self._merge_results(图谱结果, 向量结果)
# ===== Step 4: 生成答案 =====
prompt = f"""
基于以下知识图谱信息回答问题：
问题：{问题}
相关知识：
{融合知识}
要求：
- 只使用提供的知识作答
- 如果知识不足，明确说明
- 提供推理步骤
"""
答案 = await self.llm.generate(