AI知识图谱与智能知识管理完全教程

教程简介

知识图谱（Knowledge Graph）是将现实世界中的实体、概念及其关系以图结构组织和表示的技术。结合大语言模型（LLM），知识图谱正在成为企业智能知识管理的核心基础设施。本教程系统讲解AI驱动的知识图谱构建、查询、推理与应用。

第一章：知识图谱基础概念

1.1 什么是知识图谱

知识图谱以三元组（Triple）为基本单元：

(主语 Subject, 谓语 Predicate, 宾语 Object)
例如：(爱因斯坦, 出生于, 乌尔姆)

# 知识图谱的基本数据结构
class KnowledgeGraph:
    """知识图谱基础类"""
    
    def __init__(self):
        self.entities = {}      # 实体集合
        self.relations = {}     # 关系类型
        self.triples = []       # 三元组列表
    
    def add_entity(self, entity_id: str, name: str, entity_type: str, 
                   properties: dict = None):
        """添加实体"""
        self.entities[entity_id] = {
            "name": name,
            "type": entity_type,
            "properties": properties or {}
        }
    
    def add_relation(self, relation_id: str, name: str, 
                     domain: str, range_type: str):
        """添加关系类型"""
        self.relations[relation_id] = {
            "name": name,
            "domain": domain,
            "range": range_type
        }
    
    def add_triple(self, subject: str, predicate: str, obj: str, 
                   weight: float = 1.0, source: str = None):
        """添加三元组"""
        self.triples.append({
            "subject": subject,
            "predicate": predicate,
            "object": obj,
            "weight": weight,
            "source": source
        })
    
    def get_neighbors(self, entity_id: str, direction: str = "both") -> list:
        """获取实体的邻居"""
        neighbors = []
        for triple in self.triples:
            if direction in ("out", "both") and triple["subject"] == entity_id:
                neighbors.append({
                    "relation": triple["predicate"],
                    "entity": triple["object"],
                    "direction": "outgoing"
                })
            if direction in ("in", "both") and triple["object"] == entity_id:
                neighbors.append({
                    "relation": triple["predicate"],
                    "entity": triple["subject"],
                    "direction": "incoming"
                })
        return neighbors

1.2 知识图谱 vs 向量数据库

第二章：LLM驱动的知识抽取

2.1 实体与关系抽取

class LLMKnowledgeExtractor:
    """LLM驱动的知识抽取器"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
    
    async def extract_triples(self, text: str) -> list:
        """从文本中抽取三元组"""
        prompt = f"""从以下文本中抽取知识三元组（主语-谓语-宾语）。

文本：
{text}

要求：
1. 每个三元组格式：(主语, 谓语, 宾语)
2. 实体使用标准名称，不要使用代词
3. 关系使用简洁的动词短语
4. 抽取尽可能多的有效三元组

请以JSON数组格式返回：
[
  {{"subject": "主语", "predicate": "谓语", "object": "宾语"}},
  ...
]"""
        
        result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
        import json
        try:
            return json.loads(result.strip().strip("```json").strip("```"))
        except:
            return []
    
    async def extract_entities(self, text: str) -> list:
        """从文本中抽取实体"""
        prompt = f"""从以下文本中识别所有命名实体。

文本：{text[:2000]}

实体类型：
- PERSON（人物）
- ORG（组织）
- LOCATION（地点）
- TECHNOLOGY（技术/产品）
- CONCEPT（概念）
- EVENT（事件）
- DATE（日期）

请以JSON数组格式返回：
[
  {{"name": "实体名", "type": "实体类型", "description": "简短描述"}},
  ...
]"""
        
        result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
        import json
        try:
            return json.loads(result.strip().strip("```json").strip("```"))
        except:
            return []
    
    async def extract_from_documents(self, documents: list, 
                                      batch_size: int = 10) -> dict:
        """批量从文档中抽取知识"""
        all_triples = []
        all_entities = []
        
        for i in range(0, len(documents), batch_size):
            batch = documents[i:i+batch_size]
            
            for doc in batch:
                triples = await self.extract_triples(doc["text"])
                entities = await self.extract_entities(doc["text"])
                
                # 添加来源信息
                for t in triples:
                    t["source"] = doc.get("id", "")
                
                all_triples.extend(triples)
                all_entities.extend(entities)
        
        # 去重
        unique_triples = self._deduplicate_triples(all_triples)
        unique_entities = self._deduplicate_entities(all_entities)
        
        return {
            "entities": unique_entities,
            "triples": unique_triples,
            "stats": {
                "total_entities": len(unique_entities),
                "total_triples": len(unique_triples)
            }
        }
    
    def _deduplicate_triples(self, triples: list) -> list:
        """三元组去重"""
        seen = set()
        unique = []
        for t in triples:
            key = (t["subject"], t["predicate"], t["object"])
            if key not in seen:
                seen.add(key)
                unique.append(t)
        return unique
    
    def _deduplicate_entities(self, entities: list) -> list:
        """实体去重"""
        seen = {}
        for e in entities:
            name = e["name"].lower()
            if name not in seen:
                seen[name] = e
        return list(seen.values())

2.2 实体消歧与对齐

class EntityDisambiguator:
    """实体消歧器"""
    
    def __init__(self, llm_client, embedding_model):
        self.llm = llm_client
        self.embedding = embedding_model
    
    async def disambiguate(self, entity_mentions: list, 
                           knowledge_base: dict) -> list:
        """实体消歧：将不同表述映射到同一实体"""
        results = []
        
        for mention in entity_mentions:
            # 候选实体
            candidates = self._find_candidates(mention, knowledge_base)
            
            if len(candidates) == 0:
                # 新实体
                results.append({"mention": mention, "action": "create_new"})
            elif len(candidates) == 1:
                # 直接匹配
                results.append({
                    "mention": mention, 
                    "action": "link",
                    "linked_to": candidates[0]["id"]
                })
            else:
                # 需要消歧
                best = await self._select_best(mention, candidates)
                results.append({
                    "mention": mention,
                    "action": "link",
                    "linked_to": best["id"]
                })
        
        return results
    
    def _find_candidates(self, mention: str, kb: dict) -> list:
        """查找候选实体"""
        candidates = []
        mention_embedding = self.embedding.encode([mention])[0]
        
        for entity_id, entity in kb.items():
            # 名称相似度
            name_sim = self._string_similarity(mention, entity["name"])
            
            # 语义相似度
            entity_embedding = self.embedding.encode([entity["name"]])[0]
            semantic_sim = float(
                (mention_embedding * entity_embedding).sum() /
                (np.linalg.norm(mention_embedding) * np.linalg.norm(entity_embedding))
            )
            
            combined_score = 0.5 * name_sim + 0.5 * semantic_sim
            
            if combined_score > 0.6:
                candidates.append({
                    "id": entity_id,
                    "name": entity["name"],
                    "score": combined_score
                })
        
        return sorted(candidates, key=lambda x: x["score"], reverse=True)[:5]
    
    async def _select_best(self, mention: str, candidates: list) -> dict:
        """使用LLM选择最佳匹配"""
        candidate_list = "\n".join([
            f"{i+1}. {c['name']} (相似度: {c['score']:.2f})"
            for i, c in enumerate(candidates)
        ])
        
        prompt = f"""判断以下提及最可能指的是哪个实体。

提及：{mention}

候选实体：
{candidate_list}

请只返回最匹配的候选编号。"""
        
        result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
        try:
            idx = int(result.strip()) - 1
            return candidates[idx]
        except:
            return candidates[0]

第三章：图数据库实战

3.1 Neo4j实战

from neo4j import GraphDatabase

class Neo4jKnowledgeGraph:
    """Neo4j知识图谱操作"""
    
    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_id: str, name: str, 
                      entity_type: str, properties: dict = None):
        """创建实体节点"""
        with self.driver.session() as session:
            props = {"id": entity_id, "name": name, "type": entity_type}
            if properties:
                props.update(properties)
            
            query = """
            MERGE (e:{type} {{id: $id}})
            SET e += $props
            RETURN e
            """.format(type=entity_type)
            
            session.run(query, id=entity_id, props=props)
    
    def create_relation(self, subject_id: str, predicate: str, 
                        object_id: str, properties: dict = None):
        """创建关系"""
        with self.driver.session() as session:
            query = """
            MATCH (a {{id: $subj}}), (b {{id: $obj}})
            MERGE (a)-[r:{rel}]->(b)
            SET r += $props
            RETURN type(r)
            """.format(rel=predicate.upper().replace(" ", "_"))
            
            session.run(query, subj=subject_id, obj=object_id, 
                       props=properties or {})
    
    def query_neighbors(self, entity_id: str, depth: int = 1) -> list:
        """查询实体的邻居"""
        with self.driver.session() as session:
            query = """
            MATCH (e {{id: $id}})-[r*1..{depth}]-(neighbor)
            RETURN DISTINCT neighbor.id AS id, neighbor.name AS name, 
                   neighbor.type AS type, length(r) AS distance
            LIMIT 50
            """.format(depth=depth)
            
            result = session.run(query, id=entity_id)
            return [dict(record) for record in result]
    
    def shortest_path(self, start_id: str, end_id: str) -> list:
        """查找两个实体之间的最短路径"""
        with self.driver.session() as session:
            query = """
            MATCH path = shortestPath(
                (a {{id: $start}})-[*..10]-(b {{id: $end}})
            )
            RETURN [n IN nodes(path) | n.name] AS nodes,
                   [r IN relationships(path) | type(r)] AS relations
            """
            
            result = session.run(query, start=start_id, end=end_id)
            record = result.single()
            if record:
                return {
                    "nodes": record["nodes"],
                    "relations": record["relations"]
                }
            return None
    
    def cypher_query(self, query: str, params: dict = None) -> list:
        """执行自定义Cypher查询"""
        with self.driver.session() as session:
            result = session.run(query, params or {})
            return [dict(record) for record in result]

3.2 知识图谱可视化

class KnowledgeGraphVisualizer:
    """知识图谱可视化"""
    
    def __init__(self, kg: Neo4jKnowledgeGraph):
        self.kg = kg
    
    def to_pyvis(self, output_path: str, center_node: str = None, 
                 depth: int = 2):
        """使用PyVis生成交互式图谱"""
        from pyvis.network import Network
        
        net = Network(height="800px", width="100%", bgcolor="#222222", 
                     font_color="white")
        
        if center_node:
            # 获取中心节点的邻居
            neighbors = self.kg.query_neighbors(center_node, depth)
            
            # 添加节点
            for n in neighbors:
                color = self._get_color(n["type"])
                net.add_node(n["id"], label=n["name"], color=color, 
                           title=f"类型: {n['type']}")
            
            # 添加边（需要从图数据库获取关系信息）
            # ... 省略边的添加逻辑
        
        net.show(output_path)
    
    def to_d3_json(self, entity_id: str, depth: int = 2) -> dict:
        """生成D3.js兼容的JSON数据"""
        neighbors = self.kg.query_neighbors(entity_id, depth)
        
        nodes = []
        links = []
        
        for n in neighbors:
            nodes.append({
                "id": n["id"],
                "name": n["name"],
                "type": n["type"],
                "group": n["type"]
            })
        
        # 从图数据库获取边信息
        # ... 省略边的获取逻辑
        
        return {"nodes": nodes, "links": links}
    
    def _get_color(self, entity_type: str) -> str:
        """根据实体类型返回颜色"""
        colors = {
            "PERSON": "#ff7675",
            "ORG": "#74b9ff",
            "LOCATION": "#55efc4",
            "TECHNOLOGY": "#a29bfe",
            "CONCEPT": "#fdcb6e",
            "EVENT": "#e17055"
        }
        return colors.get(entity_type, "#b2bec3")

第四章：GraphRAG架构

4.1 GraphRAG核心原理

GraphRAG将知识图谱与传统RAG结合，利用图结构增强检索效果。

class GraphRAG:
    """GraphRAG：知识图谱增强的检索系统"""
    
    def __init__(self, kg: Neo4jKnowledgeGraph, vector_store, 
                 embedding_model, llm_client):
        self.kg = kg
        self.vector_store = vector_store
        self.embedding = embedding_model
        self.llm = llm_client
    
    async def query(self, question: str) -> dict:
        """GraphRAG查询流程"""
        
        # 1. 实体识别：从问题中提取关键实体
        entities = await self._extract_entities(question)
        
        # 2. 图检索：从知识图谱中获取相关子图
        graph_context = self._graph_retrieval(entities)
        
        # 3. 向量检索：从文档中获取相关段落
        vector_context = await self._vector_retrieval(question)
        
        # 4. 上下文融合
        combined_context = self._merge_context(graph_context, vector_context)
        
        # 5. 生成回答
        answer = await self._generate(question, combined_context)
        
        return {
            "answer": answer,
            "entities": entities,
            "graph_context": graph_context,
            "vector_context": vector_context
        }
    
    async def _extract_entities(self, question: str) -> list:
        """从问题中提取实体"""
        prompt = f"""从以下问题中提取关键实体（人名、组织、技术、概念等）。

问题：{question}

请以JSON数组格式返回实体列表：
["实体1", "实体2", ...]"""
        
        result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
        import json
        try:
            return json.loads(result.strip())
        except:
            return []
    
    def _graph_retrieval(self, entities: list) -> str:
        """从知识图谱中检索相关子图"""
        context_parts = []
        
        for entity_name in entities:
            # 在知识图谱中查找实体
            entity_id = self._find_entity_id(entity_name)
            if entity_id:
                # 获取2跳邻居
                neighbors = self.kg.query_neighbors(entity_id, depth=2)
                
                # 构建上下文
                context_parts.append(f"关于「{entity_name}」的知识：")
                for n in neighbors[:10]:
                    context_parts.append(f"  - {n['name']} ({n['type']})")
        
        return "\n".join(context_parts)
    
    async def _vector_retrieval(self, question: str) -> str:
        """向量检索相关文档"""
        query_embedding = self.embedding.encode([question])[0]
        results = self.vector_store.search(query_embedding.tolist(), top_k=5)
        return "\n\n".join([r["text"] for r in results])
    
    def _merge_context(self, graph_ctx: str, vector_ctx: str) -> str:
        """融合图和向量上下文"""
        return f"""## 知识图谱上下文
{graph_ctx}

## 文档上下文
{vector_ctx}"""
    
    async def _generate(self, question: str, context: str) -> str:
        """生成回答"""
        prompt = f"""基于以下知识图谱和文档上下文，回答用户问题。

{context}

问题：{question}

请给出准确、详细的回答。"""
        
        return await self.llm.chat(messages=[{"role": "user", "content": prompt}])

4.2 微软GraphRAG实现

# 使用微软的GraphRAG框架
# pip install graphrag

"""
微软GraphRAG的核心流程：
1. 文本分块
2. 实体和关系抽取
3. 社区检测（Leiden算法）
4. 社区摘要生成
5. 查询时使用社区摘要增强生成
"""

# 索引构建
"""
# 初始化
python -m graphrag.index --init --root ./ragtest

# 配置 settings.yaml
llm:
  type: openai
  model: gpt-4o-mini
  api_key: ${GRAPHRAG_API_KEY}

embeddings:
  llm:
    type: openai
    model: text-embedding-3-small

# 运行索引
python -m graphrag.index --root ./ragtest
"""

# 查询
"""
from graphrag.query.factory import get_local_search_engine, get_global_search_engine

# 局部搜索：基于实体和关系
local_engine = get_local_search_engine(
    config=config,
    llm=llm,
    embedder=embedder
)
result = await local_engine.asearch("什么是量子计算？")

# 全局搜索：基于社区摘要
global_engine = get_global_search_engine(
    config=config,
    llm=llm
)
result = await global_engine.asearch("总结量子计算领域的主要进展")
"""

第五章：知识图谱嵌入与推理

5.1 知识图谱嵌入模型

import torch
import torch.nn as nn

class TransE(nn.Module):
    """TransE知识图谱嵌入模型"""
    
    def __init__(self, num_entities: int, num_relations: int, 
                 embedding_dim: int = 128):
        super().__init__()
        self.entity_embeddings = nn.Embedding(num_entities, embedding_dim)
        self.relation_embeddings = nn.Embedding(num_relations, embedding_dim)
        
        nn.init.xavier_uniform_(self.entity_embeddings.weight)
        nn.init.xavier_uniform_(self.relation_embeddings.weight)
    
    def forward(self, head, relation, tail):
        """计算三元组分数"""
        h = self.entity_embeddings(head)
        r = self.relation_embeddings(relation)
        t = self.entity_embeddings(tail)
        
        # TransE: h + r ≈ t
        score = torch.norm(h + r - t, p=2, dim=-1)
        return -score  # 返回负分数，分数越高越好
    
    def predict(self, head, relation, tail):
        """预测三元组是否成立"""
        with torch.no_grad():
            score = self.forward(head, relation, tail)
            return torch.sigmoid(score)


class RotatE(nn.Module):
    """RotatE知识图谱嵌入模型（复数空间旋转）"""
    
    def __init__(self, num_entities: int, num_relations: int, 
                 embedding_dim: int = 128):
        super().__init__()
        self.entity_embeddings = nn.Embedding(num_entities, embedding_dim * 2)
        self.relation_embeddings = nn.Embedding(num_relations, embedding_dim)
        
    def forward(self, head, relation, tail):
        """RotatE: 在复数空间中 h ∘ r = t"""
        h = self.entity_embeddings(head)
        t = self.entity_embeddings(tail)
        r = self.relation_embeddings(relation)
        
        # 分离实部和虚部
        h_re, h_im = h.chunk(2, dim=-1)
        t_re, t_im = t.chunk(2, dim=-1)
        
        # 关系作为旋转角度
        r_re = torch.cos(r)
        r_im = torch.sin(r)
        
        # 旋转操作
        rot_re = h_re * r_re - h_im * r_im
        rot_im = h_re * r_im + h_im * r_re
        
        # 计算距离
        score = torch.stack([rot_re - t_re, rot_im - t_im], dim=-1)
        score = torch.norm(score, p=2, dim=-1).sum(dim=-1)
        
        return -score

5.2 图神经网络推理

import torch
from torch_geometric.nn import GCNConv, GATConv

class GNNReasoner(nn.Module):
    """基于GNN的知识图谱推理"""
    
    def __init__(self, in_dim: int, hidden_dim: int, out_dim: int):
        super().__init__()
        self.conv1 = GATConv(in_dim, hidden_dim, heads=4)
        self.conv2 = GATConv(hidden_dim * 4, hidden_dim, heads=4)
        self.fc = nn.Linear(hidden_dim * 4, out_dim)
    
    def forward(self, x, edge_index, edge_type):
        """前向传播"""
        # 多层图注意力
        x = torch.relu(self.conv1(x, edge_index))
        x = torch.relu(self.conv2(x, edge_index))
        
        # 预测
        out = self.fc(x)
        return out
    
    def predict_link(self, head_emb, relation_emb, tail_emb):
        """链接预测"""
        combined = torch.cat([head_emb, relation_emb, tail_emb], dim=-1)
        return torch.sigmoid(self.fc(combined))

第六章：多源知识融合

6.1 知识融合策略

class KnowledgeFusion:
    """多源知识融合"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
    
    async def fuse_knowledge(self, sources: list) -> dict:
        """融合多个知识源"""
        
        # 1. 收集所有三元组
        all_triples = []
        for source in sources:
            all_triples.extend(source["triples"])
        
        # 2. 实体对齐
        aligned = await self._align_entities(all_triples)
        
        # 3. 冲突检测
        conflicts = self._detect_conflicts(aligned)
        
        # 4. 冲突解决
        resolved = await self._resolve_conflicts(conflicts)
        
        # 5. 合并知识
        merged = self._merge_triples(aligned, resolved)
        
        return {
            "triples": merged,
            "conflicts_found": len(conflicts),
            "conflicts_resolved": len(resolved)
        }
    
    def _detect_conflicts(self, triples: list) -> list:
        """检测知识冲突"""
        # 按(主语, 谓语)分组
        grouped = {}
        for t in triples:
            key = (t["subject"], t["predicate"])
            if key not in grouped:
                grouped[key] = []
            grouped[key].append(t)
        
        # 找出有多个不同宾语的组
        conflicts = []
        for key, group in grouped.items():
            objects = set(t["object"] for t in group)
            if len(objects) > 1:
                conflicts.append({
                    "subject": key[0],
                    "predicate": key[1],
                    "conflicting_values": list(objects),
                    "sources": [t.get("source", "unknown") for t in group]
                })
        
        return conflicts
    
    async def _resolve_conflicts(self, conflicts: list) -> list:
        """使用LLM解决冲突"""
        resolved = []
        
        for conflict in conflicts:
            values = ", ".join(conflict["conflicting_values"])
            prompt = f"""关于"{conflict['subject']}"的"{conflict['predicate']}"存在以下冲突值：
{values}

来源：{', '.join(conflict['sources'])}

请判断哪个值最可能是正确的，或是否需要合并。返回格式：
{{"correct_value": "正确值", "reason": "判断理由"}}"""
            
            result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
            import json
            try:
                resolution = json.loads(result.strip())
                resolved.append({
                    **conflict,
                    "resolution": resolution
                })
            except:
                pass
        
        return resolved

第七章：知识图谱驱动的智能问答

7.1 KBQA系统

class KBQASystem:
    """基于知识图谱的问答系统"""
    
    def __init__(self, kg: Neo4jKnowledgeGraph, llm_client):
        self.kg = kg
        self.llm = llm_client
    
    async def answer(self, question: str) -> dict:
        """知识图谱问答"""
        
        # 1. 意图识别
        intent = await self._classify_intent(question)
        
        # 2. 实体链接
        entities = await self._link_entities(question)
        
        # 3. 查询构建
        query = await self._build_query(question, intent, entities)
        
        # 4. 执行查询
        results = self.kg.cypher_query(query["cypher"], query.get("params"))
        
        # 5. 生成回答
        answer = await self._generate_answer(question, results)
        
        return {
            "answer": answer,
            "intent": intent,
            "entities": entities,
            "query": query["cypher"],
            "results_count": len(results)
        }
    
    async def _classify_intent(self, question: str) -> str:
        """分类查询意图"""
        prompt = f"""对以下问题进行意图分类：
问题：{question}

分类：
- factoid: 简单事实查询（谁、什么、哪里）
- count: 数量查询
- list: 列举查询
- comparison: 比较查询
- path: 路径/关系查询

只返回分类名称。"""
        
        result = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
        return result.strip().lower()
    
    async def _build_query(self, question: str, intent: str, 
                           entities: list) -> dict:
        """构建图查询"""
        prompt = f"""根据以下信息构建Neo4j Cypher查询。

问题：{question}
意图：{intent}
实体：{entities}

图谱结构：
- 节点属性：id, name, type
- 关系类型：各种语义关系

请生成Cypher查询语句。只返回Cypher代码。"""
        
        cypher = await self.llm.chat(messages=[{"role": "user", "content": prompt}])
        return {"cypher": cypher.strip().strip("```cypher").strip("```")}
    
    async def _generate_answer(self, question: str, results: list) -> str:
        """基于查询结果生成回答"""
        results_text = str(results[:10])  # 限制结果数量
        
        prompt = f"""基于以下知识图谱查询结果，回答用户问题。

查询结果：
{results_text}

问题：{question}

请给出准确、自然的回答。"""
        
        return await self.llm.chat(messages=[{"role": "user", "content": prompt}])

最佳实践总结

1. 质量优先：知识图谱的质量比数量更重要
2. 持续更新：建立知识更新机制，保持知识图谱的时效性
3. 实体消歧：实体消歧是知识融合中最关键的环节
4. GraphRAG：将知识图谱与向量检索结合，取长补短
5. 评估体系：建立知识图谱质量评估指标（完整性、一致性、准确性）
6. 可视化：知识图谱可视化有助于发现和纠正错误

总结

本教程系统讲解了AI驱动的知识图谱技术栈，从知识抽取、图数据库、GraphRAG到知识推理和智能问答。知识图谱与LLM的结合正在开启智能知识管理的新范式，企业可以通过构建领域知识图谱来显著提升AI系统的知识理解和推理能力。