AI 知识图谱与 GraphRAG 应用教程
前言
知识图谱（Knowledge Graph）是一种以图结构组织和表示知识的技术，它将现实世界中的实体及其关系以节点和边的形式存储，为智能问答、推荐系统、数据分析等应用提供强大的知识支撑。近年来，随着大语言模型（LLM）的发展，知识图谱与 RAG（检索增强生成）技术的结合催生了 GraphRAG 这一新兴范式，它利用图结构的拓扑信息增强 LLM 的检索和推理能力。
本教程将从知识图谱基础概念出发，系统讲解 LLM 驱动的图谱自动构建、图数据库选型与操作、GraphRAG 架构与实现，最终带你构建一个行业知识问答系统。
第一章：知识图谱基础
1.1 什么是知识图谱
知识图谱是一种语义网络，它以三元组（Triple）的形式表示知识：
(主语 Subject, 谓语 Predicate, 宾语 Object)
例如：
- （爱因斯坦，出生于，乌尔姆）
- （爱因斯坦，提出了，相对论）
- （相对论，属于，物理学）
这些三元组相互连接，形成一个庞大的知识网络。
1.2 核心概念
实体（Entity）：现实世界中的对象，如人名、地名、组织、概念等。每个实体有唯一标识符。
关系（Relation）：实体之间的联系，如"出生于"、"工作于"、"属于"等。关系是有方向和类型的。
属性（Attribute）：实体的特征描述，如"出生日期=1879年3月14日"。属性是实体与值之间的关系。
本体（Ontology）：对领域知识的形式化描述，定义了概念、关系类型、属性和约束规则。本体是知识图谱的"模式层"（Schema），而具体的知识实例是"数据层"（Instance）。
本体层（Schema）:
Person --出生日期--> Date
Person --出生于--> Location
Person --提出了--> Theory
Theory --属于--> Field
数据层（Instance）:
爱因斯坦 --出生于--> 乌尔姆
爱因斯坦 --提出了--> 相对论
相对论 --属于--> 物理学
1.3 知识图谱的表示方法
RDF（Resource Description Framework）：W3C 标准，使用三元组表示，支持 SPARQL 查询语言。
# RDF Turtle 格式
@prefix ex: <http://example.org/> .
@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
ex:Einstein rdf:type ex:Person ;
ex:birthPlace ex:Ulm ;
ex:proposed ex:Relativity .
ex:Relativity rdf:type ex:Theory ;
ex:belongsTo ex:Physics .
属性图（Property Graph）：工业界主流，节点和边都可以有属性，更适合工程实现。
// Neo4j Cypher 查询语言
CREATE (einstein:Person {name: "爱因斯坦", birth: "1879-03-14"})
CREATE (ulm:Location {name: "乌尔姆", country: "德国"})
CREATE (relativity:Theory {name: "相对论", year: 1905})
CREATE (physics:Field {name: "物理学"})
CREATE (einstein)-[:BORN_IN]->(ulm)
CREATE (einstein)-[:PROPOSED {year: 1905}]->(relativity)
CREATE (relativity)-[:BELONGS_TO]->(physics)
1.4 知识图谱的应用场景
| 应用场景 | 典型案例 | 知识图谱的作用 |
|---|---|---|
| 智能问答 | 百度知道、智能客服 | 基于图谱的语义理解和精确回答 |
| 推荐系统 | 淘宝商品推荐 | 利用关系发现潜在关联 |
| 搜索引擎 | Google Knowledge Panel | 增强搜索结果的知识展示 |
| 风控反欺诈 | 金融风控系统 | 关系链路分析发现异常模式 |
| 药物研发 | 生物医学知识图谱 | 药物-靶点-疾病关系发现 |
第二章：LLM 驱动的知识图谱自动构建
2.1 传统方法 vs LLM 方法
传统知识图谱构建依赖 NLP Pipeline（命名实体识别→关系抽取→实体链接→知识融合），流程复杂、领域迁移困难。LLM 的出现彻底改变了这一格局：
| 对比维度 | 传统NLP方法 | LLM方法 |
|---|---|---|
| 实体识别 | BiLSTM-CRF 微调 | Prompt + LLM 零样本 |
| 关系抽取 | CNN/RNN 监督学习 | Prompt + LLM 零样本/少样本 |
| 领域适配 | 重新标注数据、重新训练 | 调整 Prompt 即可 |
| 多语言 | 需要各语言模型 | LLM 原生支持 |
| 成本 | 高（标注成本+训练成本） | 低（API调用成本） |
2.2 使用 LLM 抽取知识三元组
import json
from openai import OpenAI
client = OpenAI()
def extract_triples_llm(text, domain="通用"):
"""使用LLM从文本中抽取知识三元组"""
prompt = f"""你是一个专业的知识图谱构建专家。请从以下文本中抽取所有有意义的知识三元组。
领域：{domain}
要求：
1. 每个三元组格式为 (主语, 关系, 宾语)
2. 主语和宾语应为具体的实体或概念
3. 关系应为动词或动词短语，简洁明确
4. 保留原文中的关键信息，不要过度概括
5. 输出JSON数组格式
文本：
{text}
请输出三元组JSON数组，格式如：
[{{"subject": "...", "relation": "...", "object": "..."}}, ...]"""
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "你是一个专业的知识图谱构建专家。"},
{"role": "user", "content": prompt}
],
temperature=0.1,
response_format={"type": "json_object"}
)
result = json.loads(response.choices[0].message.content)
return result.get("triples", result)
# 使用示例
text = """
爱因斯坦于1879年出生于德国乌尔姆。他在苏黎世联邦理工学院学习物理学。
1905年，他发表了关于光电效应的论文，这为量子力学的发展奠定了基础。
同年，他提出了狭义相对论，其中包含了著名的质能方程E=mc²。
1915年，他完成了广义相对论，这一理论描述了引力如何弯曲时空。
"""
triples = extract_triples_llm(text, domain="物理学历史")
for t in triples:
print(f"({t['subject']}, {t['relation']}, {t['object']})")
# 输出示例：
# (爱因斯坦, 出生于, 乌尔姆)
# (爱因斯坦, 出生年份, 1879年)
# (乌尔姆, 位于, 德国)
# (爱因斯坦, 学习于, 苏黎世联邦理工学院)
# (爱因斯坦, 学习专业, 物理学)
# (爱因斯坦, 发表, 关于光电效应的论文)
# (爱因斯坦, 提出, 狭义相对论)
# (狭义相对论, 包含, 质能方程E=mc²)
# (爱因斯坦, 完成, 广义相对论)
2.3 实体类型识别与标准化
def extract_entities_with_types(text):
"""抽取实体并识别类型"""
prompt = f"""从以下文本中抽取所有命名实体，并标注其类型。
实体类型包括：
- PERSON（人物）
- ORGANIZATION（组织/机构）
- LOCATION（地点）
- DATE（日期）
- EVENT（事件）
- CONCEPT（概念/理论）
- PRODUCT（产品）
- TECHNOLOGY（技术）
文本：
{text}
输出JSON格式：
[{{"entity": "...", "type": "...", "aliases": ["别名1", "别名2"]}}]"""
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.1,
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
def normalize_entities(entities):
"""实体标准化：合并同义实体"""
from collections import defaultdict
# 构建标准化映射
prompt = f"""以下是一些实体列表，请识别其中指代同一事物的实体，返回标准化映射。
实体列表：
{json.dumps(entities, ensure_ascii=False)}
输出格式：
{{"标准名": ["别名1", "别名2", ...], ...}}"""
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.1,
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
2.4 文档级知识图谱构建 Pipeline
import re
from typing import List, Dict
class DocumentKGBuilder:
"""文档级知识图谱自动构建"""
def __init__(self, llm_model="gpt-4"):
self.model = llm_model
self.all_triples = []
self.all_entities = {}
def split_text(self, text: str, chunk_size: int = 1000, overlap: int = 200) -> List[str]:
"""将长文档分块"""
sentences = re.split(r'[。！？\n]', text)
chunks = []
current_chunk = ""
for sentence in sentences:
sentence = sentence.strip()
if not sentence:
continue
if len(current_chunk) + len(sentence) > chunk_size:
chunks.append(current_chunk)
# 保留最后overlap个字符作为上下文
current_chunk = current_chunk[-overlap:] + sentence
else:
current_chunk += sentence + "。"
if current_chunk:
chunks.append(current_chunk)
return chunks
def process_document(self, text: str, domain: str = "通用") -> Dict:
"""处理整篇文档，构建知识图谱"""
chunks = self.split_text(text)
all_triples = []
all_entities = set()
for i, chunk in enumerate(chunks):
print(f"处理第 {i+1}/{len(chunks)} 个文本块...")
# 抽取三元组
triples = extract_triples_llm(chunk, domain)
all_triples.extend(triples)
# 收集实体
for t in triples:
all_entities.add(t['subject'])
all_entities.add(t['object'])
# 去重
unique_triples = self._deduplicate_triples(all_triples)
# 实体标准化
entity_list = list(all_entities)
normalization = normalize_entities(entity_list)
# 应用标准化
normalized_triples = self._apply_normalization(unique_triples, normalization)
return {
"triples": normalized_triples,
"entities": entity_list,
"normalization": normalization,
"stats": {
"total_triples": len(normalized_triples),
"unique_entities": len(set(
[t['subject'] for t in normalized_triples] +
[t['object'] for t in normalized_triples]
))
}
}
def _deduplicate_triples(self, triples: List[Dict]) -> List[Dict]:
"""三元组去重"""
seen = set()
unique = []
for t in triples:
key = (t['subject'], t['relation'], t['object'])
if key not in seen:
seen.add(key)
unique.append(t)
return unique
def _apply_normalization(self, triples, normalization):
"""应用实体标准化"""
# 构建映射表
name_map = {}
for standard_name, aliases in normalization.items():
for alias in aliases:
name_map[alias] = standard_name
normalized = []
for t in triples:
new_t = {
"subject": name_map.get(t['subject'], t['subject']),
"relation": t['relation'],
"object": name_map.get(t['object'], t['object'])
}
normalized.append(new_t)
return normalized
# 使用示例
builder = DocumentKGBuilder()
document = """
（此处放置长文档文本）
"""
result = builder.process_document(document, domain="人工智能")
print(f"抽取了 {result['stats']['total_triples']} 个三元组")
print(f"涉及 {result['stats']['unique_entities']} 个实体")
第三章：图数据库选型
3.1 主流图数据库对比
| 特性 | Neo4j | NebulaGraph | JanusGraph | ArangoDB |
|---|---|---|---|---|
| 查询语言 | Cypher | nGQL | Gremlin | AQL |
| 分布式 | 企业版 | 原生分布式 | 原生分布式 | 支持 |
| 开源 | 社区版开源 | 完全开源 | 完全开源 | 完全开源 |
| 适用规模 | 中小型 | 大型 | 大型 | 中型 |
| 学习曲线 | 低 | 中 | 高 | 中 |
| 生态成熟度 | ★★★★★ | ★★★★ | ★★★ | ★★★★ |
| AI集成 | 丰富 | 较好 | 一般 | 较好 |
3.2 Neo4j 快速入门
from neo4j import GraphDatabase
class Neo4jManager:
"""Neo4j 图数据库管理器"""
def __init__(self, uri="bolt://localhost:7687", user="neo4j", password="password"):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def close(self):
self.driver.close()
def create_entity(self, name: str, entity_type: str, properties: dict = None):
"""创建实体节点"""
props = properties or {}
props["name"] = name
props["type"] = entity_type
query = """
MERGE (e:Entity {name: $name})
SET e += $props
SET e:$entity_type
RETURN e
"""
with self.driver.session() as session:
result = session.run(query, name=name, props=props, entity_type=entity_type)
return result.single()
def create_relation(self, subject: str, relation: str, obj: str, properties: dict = None):
"""创建关系"""
props = properties or {}
query = f"""
MATCH (a:Entity {{name: $subject}})
MATCH (b:Entity {{name: $object}})
MERGE (a)-[r:`{relation}`]->(b)
SET r += $props
RETURN a, r, b
"""
with self.driver.session() as session:
result = session.run(query, subject=subject, object=obj, props=props)
return result.single()
def import_triples(self, triples: list):
"""批量导入三元组"""
for t in triples:
self.create_entity(t['subject'], "Entity")
self.create_entity(t['object'], "Entity")
self.create_relation(t['subject'], t['relation'], t['object'])
print(f"成功导入 {len(triples)} 个三元组")
def query_entity(self, name: str, depth: int = 2):
"""查询实体及其关系"""
query = """
MATCH (e:Entity {name: $name})-[r*1..""" + str(depth) + """]->(related)
RETURN e, r, related
LIMIT 50
"""
with self.driver.session() as session:
result = session.run(query, name=name)
return [record for record in result]
def search_by_relation(self, relation: str, limit: int = 20):
"""按关系类型搜索"""
query = """
MATCH (a)-[r:`""" + relation + """`]->(b)
RETURN a.name AS subject, type(r) AS relation, b.name AS object
LIMIT $limit
"""
with self.driver.session() as session:
result = session.run(query, limit=limit)
return [dict(record) for record in result]
def get_entity_neighbors(self, name: str):
"""获取实体的所有邻居"""
query = """
MATCH (e:Entity {name: $name})-[r]-(neighbor)
RETURN type(r) AS relation, neighbor.name AS neighbor_name,
startNode(r).name AS direction
"""
with self.driver.session() as session:
result = session.run(query, name=name)
return [dict(record) for record in result]
# 使用示例
neo4j = Neo4jManager("bolt://localhost:7687", "neo4j", "password")
# 导入三元组
triples = [
{"subject": "爱因斯坦", "relation": "出生于", "object": "乌尔姆"},
{"subject": "爱因斯坦", "relation": "提出了", "object": "相对论"},
{"subject": "相对论", "relation": "属于", "object": "物理学"},
{"subject": "爱因斯坦", "relation": "工作于", "object": "普林斯顿大学"},
]
neo4j.import_triples(triples)
# 查询
results = neo4j.query_entity("爱因斯坦", depth=2)
for r in results:
print(r)
3.3 NebulaGraph 使用示例
from nebula3.gclient.net import ConnectionPool
from nebula3.Config import Config
class NebulaGraphManager:
"""NebulaGraph 图数据库管理器"""
def __init__(self, host="localhost", port=9669):
config = Config()
config.max_connection_pool_size = 10
self.connection_pool = ConnectionPool()
self.connection_pool.init([(host, port)], config)
def execute(self, nGQL: str):
"""执行 nGQL 查询"""
session = self.connection_pool.get_session("root", "nebula")
try:
result = session.execute(nGQL)
if not result.is_succeeded():
print(f"查询失败: {result.error_msg()}")
return None
return result
finally:
session.release()
def setup_schema(self):
"""创建图模式"""
# 创建空间
self.execute("CREATE SPACE IF NOT EXISTS knowledge_graph(vid_type=FIXED_STRING(256));")
import time
time.sleep(10) # 等待空间创建完成
# 创建标签（实体类型）
self.execute("USE knowledge_graph;")
self.execute("CREATE TAG IF NOT EXISTS entity(name string, type string);")
# 创建边类型（关系类型）
self.execute("CREATE EDGE IF NOT EXISTS relation(rel_type string);")
def import_triples(self, triples):
"""导入三元组"""
self.execute("USE knowledge_graph;")
for t in triples:
# 插入实体
self.execute(f"""
INSERT VERTEX entity(name, type) VALUES
"{t['subject']}":("{t['subject']}", "Entity");
""")
self.execute(f"""
INSERT VERTEX entity(name, type) VALUES
"{t['object']}":("{t['object']}", "Entity");
""")
# 插入关系
self.execute(f"""
INSERT EDGE relation(rel_type) VALUES
"{t['subject']}"->"{t['object']}":("{t['relation']}");
""")
第四章：从非结构化文本抽取知识三元组
4.1 基于 Prompt 的三元组抽取
TRIPLE_EXTRACTION_PROMPT = """你是一个专业的知识图谱构建专家。请从以下文本中精确抽取所有知识三元组。
## 抽取规则：
1. 每个三元组格式：(主语, 关系, 宾语)
2. 主语和宾语必须是文本中明确提到的实体或概念
3. 关系应使用简洁的动词或动词短语
4. 如果存在多跳关系，拆分为多个直接关系的三元组
5. 保留数值、日期等属性信息
## 输出格式：
请以JSON数组输出，每个元素包含 subject, relation, object 三个字段。
## 文本：
{text}
## 三元组："""
def extract_triples_batch(texts: list, batch_size: int = 5):
"""批量抽取三元组"""
all_results = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i+batch_size]
# 并发处理
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=batch_size) as executor:
futures = [
executor.submit(extract_triples_llm, text)
for text in batch
]
for future in concurrent.futures.as_completed(futures):
try:
result = future.result()
all_results.extend(result)
except Exception as e:
print(f"抽取失败: {e}")
return all_results
4.2 三元组质量评估
class TripleQualityEvaluator:
"""三元组质量评估器"""
def __init__(self):
self.quality_scores = []
def evaluate_triple(self, triple: dict, source_text: str) -> dict:
"""评估单个三元组的质量"""
scores = {
"completeness": self._check_completeness(triple),
"relevance": self._check_relevance(triple, source_text),
"specificity": self._check_specificity(triple),
"consistency": self._check_consistency(triple),
}
scores["overall"] = sum(scores.values()) / len(scores)
return scores
def _check_completeness(self, triple: dict) -> float:
"""检查完整性：三个字段都不为空"""
if triple.get("subject") and triple.get("relation") and triple.get("object"):
return 1.0
return 0.0
def _check_relevance(self, triple: dict, text: str) -> float:
"""检查相关性：实体是否在原文中出现"""
subject_in = triple["subject"] in text
object_in = triple["object"] in text
if subject_in and object_in:
return 1.0
elif subject_in or object_in:
return 0.5
return 0.0
def _check_specificity(self, triple: dict) -> float:
"""检查具体性：避免过于泛化的三元组"""
# 关系不应太长（可能是整句提取）
if len(triple["relation"]) > 20:
return 0.3
# 主语和宾语不应相同
if triple["subject"] == triple["object"]:
return 0.0
return 1.0
def _check_consistency(self, triple: dict) -> float:
"""检查一致性：实体类型是否合理"""
# 简单的类型一致性检查
# 例如：日期不应作为关系的主语
date_patterns = r'\d{4}年|\d{4}-\d{2}-\d{2}'
import re
if re.match(date_patterns, triple["subject"]):
return 0.3
return 1.0
def filter_triples(self, triples: list, source_text: str,
threshold: float = 0.6) -> list:
"""过滤低质量三元组"""
quality_triples = []
for t in triples:
scores = self.evaluate_triple(t, source_text)
if scores["overall"] >= threshold:
t["quality_score"] = scores["overall"]
quality_triples.append(t)
print(f"过滤前: {len(triples)} 个三元组, 过滤后: {len(quality_triples)} 个")
return quality_triples
4.3 人工校验工具
class TripleReviewTool:
"""三元组人工校验工具"""
def __init__(self, triples: list, source_text: str):
self.triples = triples
self.source_text = source_text
self.approved = []
self.rejected = []
self.modified = []
def interactive_review(self):
"""交互式校验"""
for i, triple in enumerate(self.triples):
print(f"\n{'='*60}")
print(f"三元组 [{i+1}/{len(self.triples)}]")
print(f"主语: {triple['subject']}")
print(f"关系: {triple['relation']}")
print(f"宾语: {triple['object']}")
print(f"质量分: {triple.get('quality_score', 'N/A')}")
print(f"\n来源文本: {self.source_text[:200]}...")
print(f"{'='*60}")
action = input("[a]批准 / [r]拒绝 / [m]修改 / [s]跳过: ").strip().lower()
if action == 'a':
self.approved.append(triple)
print("✓ 已批准")
elif action == 'r':
self.rejected.append(triple)
print("✗ 已拒绝")
elif action == 'm':
new_subject = input(f" 主语 [{triple['subject']}]: ").strip()
new_relation = input(f" 关系 [{triple['relation']}]: ").strip()
new_object = input(f" 宾语 [{triple['object']}]: ").strip()
modified = {
"subject": new_subject or triple['subject'],
"relation": new_relation or triple['relation'],
"object": new_object or triple['object']
}
self.modified.append((triple, modified))
self.approved.append(modified)
print("✓ 已修改并批准")
return {
"approved": self.approved,
"rejected": self.rejected,
"modified": self.modified
}
def batch_review_by_confidence(self, low_threshold=0.4, high_threshold=0.8):
"""按置信度批量分类"""
high_conf = [] # 自动批准
medium_conf = [] # 需要人工审核
low_conf = [] # 自动拒绝
for t in self.triples:
score = t.get('quality_score', 0.5)
if score >= high_threshold:
high_conf.append(t)
elif score >= low_threshold:
medium_conf.append(t)
else:
low_conf.append(t)
print(f"高置信度（自动批准）: {len(high_conf)}")
print(f"中置信度（需人工审核）: {len(medium_conf)}")
print(f"低置信度（自动拒绝）: {len(low_conf)}")
return high_conf, medium_conf, low_conf
第五章：GraphRAG 架构
5.1 传统 RAG vs GraphRAG
传统 RAG 将文档切分为文本块，通过向量相似度检索相关片段，然后送入 LLM 生成回答。这种方法的局限在于：
- 缺乏全局视角：无法回答需要综合多个文档段落的全局性问题
- 丢失结构信息：切块破坏了实体间的关联关系
- 推理能力弱：无法进行多跳推理（如"A的导师的学生在哪里工作？"）
GraphRAG 通过知识图谱的拓扑结构解决了这些问题：
传统 RAG:
用户问题 → 向量检索 → Top-K 文本块 → LLM 生成回答
GraphRAG:
用户问题 → 实体识别 → 图谱遍历/社区搜索 → 结构化上下文 → LLM 生成回答
5.2 GraphRAG 的两种检索模式
Local Search（局部搜索）：从问题中识别出实体，在图谱中找到该实体及其邻居，获取局部上下文。适合事实性、具体性问题。
Global Search（全局搜索）：利用社区检测算法将图谱划分为多个社区，对每个社区生成摘要，然后综合所有社区信息回答问题。适合概括性、全局性问题。
class GraphRAG:
"""GraphRAG 检索系统"""
def __init__(self, neo4j_manager, llm_client):
self.graph = neo4j_manager
self.llm = llm_client
self.community_summaries = {}
def local_search(self, query: str, top_k: int = 5, max_depth: int = 2) -> str:
"""局部搜索：从实体出发，遍历局部子图"""
# Step 1: 从问题中提取关键实体
entities = self._extract_entities_from_query(query)
print(f"识别到实体: {entities}")
# Step 2: 从图谱中检索相关子图
subgraph = []
for entity in entities:
neighbors = self.graph.get_entity_neighbors(entity)
subgraph.extend(neighbors)
# 二跳邻居
for neighbor in neighbors:
second_hop = self.graph.get_entity_neighbors(neighbor['neighbor_name'])
subgraph.extend(second_hop)
# Step 3: 构建上下文
context = self._build_context_from_subgraph(subgraph)
# Step 4: LLM 生成回答
answer = self._generate_answer(query, context, mode="local")
return answer
def global_search(self, query: str) -> str:
"""全局搜索：利用社区摘要回答全局性问题"""
# Step 1: 获取所有社区摘要
if not self.community_summaries:
self.community_summaries = self._compute_community_summaries()
# Step 2: 对每个社区摘要进行局部推理
community_answers = []
for comm_id, summary in self.community_summaries.items():
partial = self._generate_answer(
f"基于以下社区知识回答问题：{query}",
summary,
mode="global_partial"
)
community_answers.append({
"community": comm_id,
"answer": partial
})
# Step 3: 综合所有社区答案
final_answer = self._aggregate_answers(query, community_answers)
return final_answer
def _extract_entities_from_query(self, query: str) -> list:
"""从用户问题中提取实体"""
prompt = f"""从以下问题中提取所有命名实体（人名、地名、组织、概念等）。
问题：{query}
输出JSON数组格式：["实体1", "实体2", ...]"""
response = self.llm.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.1,
response_format={"type": "json_object"}
)
result = json.loads(response.choices[0].message.content)
return result.get("entities", result) if isinstance(result, dict) else result
def _build_context_from_subgraph(self, subgraph: list) -> str:
"""从子图构建上下文文本"""
relations = []
for record in subgraph:
relations.append(
f"{record.get('direction', '')} --[{record['relation']}]--> {record['neighbor_name']}"
)
return "\n".join(relations[:50]) # 限制上下文长度
def _generate_answer(self, query: str, context: str, mode: str = "local") -> str:
"""使用LLM生成回答"""
if mode == "local":
system_prompt = "你是一个知识问答助手。请基于提供的图谱关系信息，准确回答用户的问题。如果信息不足，请明确说明。"
else:
system_prompt = "你是一个知识分析专家。请基于提供的社区知识摘要，分析并回答问题。"
prompt = f"""## 知识上下文：
{context}
## 用户问题：
{query}
请基于上述知识回答问题。"""
response = self.llm.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
temperature=0.3
)
return response.choices[0].message.content
def _aggregate_answers(self, query: str, community_answers: list) -> str:
"""综合多个社区答案"""
answers_text = "\n\n".join([
f"社区 {a['community']}: {a['answer']}"
for a in community_answers
])
prompt = f"""以下是从不同知识社区得到的部分回答，请综合这些信息，给出一个完整、准确的最终回答。
## 各社区回答：
{answers_text}
## 原始问题：
{query}
## 最终回答："""
response = self.llm.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
return response.choices[0].message.content
5.3 社区检测与摘要生成
import networkx as nx
from community import community_louvain
class CommunityDetector:
"""图谱社区检测"""
def __init__(self, neo4j_manager):
self.graph = neo4j_manager
self.communities = {}
def detect_communities(self) -> dict:
"""使用 Louvain 算法检测社区"""
# 从 Neo4j 导出图结构到 NetworkX
G = self._export_to_networkx()
# Louvain 社区检测
partition = community_louvain.best_partition(G)
# 按社区分组
self.communities = {}
for node, comm_id in partition.items():
if comm_id not in self.communities:
self.communities[comm_id] = []
self.communities[comm_id].append(node)
print(f"检测到 {len(self.communities)} 个社区")
for comm_id, members in self.communities.items():
print(f" 社区 {comm_id}: {len(members)} 个节点")
return self.communities
def _export_to_networkx(self) -> nx.Graph:
"""从 Neo4j 导出图到 NetworkX"""
G = nx.Graph()
query = """
MATCH (a:Entity)-[r]-(b:Entity)
RETURN a.name AS source, b.name AS target, type(r) AS rel_type
"""
with self.graph.driver.session() as session:
result = session.run(query)
for record in result:
G.add_edge(record["source"], record["target"],
relation=record["rel_type"])
return G
def generate_community_summaries(self, llm_client) -> dict:
"""为每个社区生成摘要"""
summaries = {}
for comm_id, members in self.communities.items():
# 获取社区内的所有关系
relations = []
for member in members:
neighbors = self.graph.get_entity_neighbors(member)
for n in neighbors:
if n['neighbor_name'] in members:
relations.append(
f"{member} --[{n['relation']}]--> {n['neighbor_name']}"
)
# 使用 LLM 生成摘要
context = "\n".join(relations[:100])
prompt = f"""请为以下知识社区生成一段简洁的摘要，概括该社区的核心主题和关键信息。
社区成员：{', '.join(members[:20])}
社区内的关系：
{context}
请用2-3句话概括这个社区的核心内容。"""
response = llm_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
summaries[comm_id] = {
"members": members,
"summary": response.choices[0].message.content
}
return summaries
第六章：Neo4j + LangChain 集成实战
6.1 LangChain Graph Store 集成
from langchain_community.graphs import Neo4jGraph
from langchain.chains import GraphCypherQAChain
from langchain_openai import ChatOpenAI
# 初始化 Neo4j 图存储
graph = Neo4jGraph(
url="bolt://localhost:7687",
username="neo4j",
password="password"
)
# 刷新图模式（获取最新的 schema 信息）
graph.refresh_schema()
print(graph.schema)
# 创建 Cypher QA Chain
llm = ChatOpenAI(model="gpt-4", temperature=0)
chain = GraphCypherQAChain.from_llm(
llm=llm,
graph=graph,
verbose=True,
allow_dangerous_requests=True,
return_intermediate_steps=True
)
# 查询示例
result = chain.invoke({"query": "爱因斯坦提出了哪些理论？"})
print(result["result"])
# 查看生成的 Cypher 查询
if "intermediate_steps" in result:
for step in result["intermediate_steps"]:
if "query" in step:
print(f"生成的 Cypher: {step['query']}")
6.2 自定义图检索器
from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document
from typing import List
class KnowledgeGraphRetriever(BaseRetriever):
"""知识图谱检索器"""
neo4j_manager: object
llm: object
search_depth: int = 2
max_results: int = 10
def _get_relevant_documents(self, query: str) -> List[Document]:
"""从知识图谱中检索相关文档"""
# 提取查询中的实体
entities = self._extract_entities(query)
documents = []
for entity in entities:
# 获取实体的局部子图
subgraph = self._get_subgraph(entity, self.search_depth)
# 将子图转换为文本描述
for relation in subgraph:
text = f"{relation.get('direction', '')} {relation['neighbor_name']}。关系类型：{relation['relation']}"
doc = Document(
page_content=text,
metadata={
"source": "knowledge_graph",
"entity": entity,
"relation_type": relation['relation']
}
)
documents.append(doc)
return documents[:self.max_results]
def _extract_entities(self, query: str) -> List[str]:
"""从查询中提取实体"""
prompt = f"从以下问题中提取人名、地名、组织等命名实体，输出JSON数组：\n{query}"
response = self.llm.invoke(prompt)
try:
return json.loads(response.content)
except:
return [query]
def _get_subgraph(self, entity: str, depth: int) -> list:
"""获取实体的局部子图"""
return self.neo4j_manager.get_entity_neighbors(entity)
6.3 GraphRAG with LangChain 完整集成
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
class LangChainGraphRAG:
"""基于 LangChain 的 GraphRAG 系统"""
def __init__(self, neo4j_url, neo4j_user, neo4j_password):
self.graph = Neo4jGraph(url=neo4j_url, username=neo4j_user, password=neo4j_password)
self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
self._build_chain()
def _build_chain(self):
"""构建检索-生成链"""
self.graph.refresh_schema()
# Cypher 生成 Chain
self.cypher_chain = GraphCypherQAChain.from_llm(
llm=self.llm,
graph=self.graph,
verbose=True,
allow_dangerous_requests=True
)
# 自定义 RAG Prompt
self.rag_prompt = ChatPromptTemplate.from_template("""
你是一个专业的知识问答助手。请基于以下图谱查询结果回答用户的问题。
## 图谱查询结果：
{graph_context}
## 用户问题：
{question}
请给出准确、详细的回答。如果图谱中没有相关信息，请明确说明。
""")
def query(self, question: str, mode: str = "auto") -> str:
"""查询知识图谱"""
if mode == "auto":
mode = self._detect_query_type(question)
if mode == "local":
return self._local_search(question)
elif mode == "global":
return self._global_search(question)
else:
return self.cypher_chain.invoke({"query": question})["result"]
def _detect_query_type(self, question: str) -> str:
"""自动判断查询类型"""
prompt = f"""判断以下问题是"局部性问题"还是"全局性问题"。
局部性问题：关于具体实体的事实性问题（如：谁提出了相对论？）
全局性问题：需要综合多个信息的概括性问题（如：20世纪物理学有哪些重大突破？）
问题：{question}
只回答 "local" 或 "global"。"""
response = self.llm.invoke(prompt)
return "local" if "local" in response.content.lower() else "global"
def _local_search(self, question: str) -> str:
"""局部搜索"""
result = self.cypher_chain.invoke({"query": question})
return result["result"]
def _global_search(self, question: str) -> str:
"""全局搜索（基于社区摘要）"""
# 获取所有社区摘要
query = """
MATCH (n:Entity)-[r]-(m:Entity)
RETURN n.name AS entity, collect(DISTINCT m.name) AS neighbors,
collect(DISTINCT type(r)) AS relations
LIMIT 100
"""
with self.graph._driver.session() as session:
result = session.run(query)
context_parts = []
for record in result:
context_parts.append(
f"{record['entity']} 关联: {', '.join(record['neighbors'][:5])}"
)
context = "\n".join(context_parts)
chain = self.rag_prompt | self.llm | StrOutputParser()
return chain.invoke({"graph_context": context, "question": question})
第七章：图谱增强的 RAG 检索策略
7.1 实体链接与消歧
class EntityLinker:
"""实体链接器：将查询中的提及链接到图谱中的实体"""
def __init__(self, neo4j_manager, llm_client):
self.graph = neo4j_manager
self.llm = llm_client
self.entity_cache = {}
def link_entities(self, mention: str, context: str = "") -> dict:
"""将提及链接到知识图谱中的实体"""
# Step 1: 候选实体检索
candidates = self._search_candidates(mention)
if not candidates:
return {"entity": None, "confidence": 0}
if len(candidates) == 1:
return {"entity": candidates[0], "confidence": 1.0}
# Step 2: 使用LLM消歧
return self._disambiguate(mention, candidates, context)
def _search_candidates(self, mention: str) -> list:
"""从图谱中搜索候选实体"""
query = """
MATCH (e:Entity)
WHERE e.name CONTAINS $mention OR $mention CONTAINS e.name
RETURN e.name AS name, e.type AS type
LIMIT 10
"""
with self.graph.driver.session() as session:
result = session.run(query, mention=mention)
return [dict(record) for record in result]
def _disambiguate(self, mention: str, candidates: list, context: str) -> dict:
"""使用LLM进行实体消歧"""
candidates_text = "\n".join([
f"- {c['name']} (类型: {c.get('type', '未知')})"
for c in candidates
])
prompt = f"""从以下候选实体中，选择与给定提及最匹配的实体。
提及：{mention}
上下文：{context}
候选实体：
{candidates_text}
请只返回最匹配的实体名称。如果都不匹配，返回"无匹配"。"""
response = self.llm.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)
best_match = response.choices[0].message.content.strip()
for c in candidates:
if c['name'] in best_match:
return {"entity": c, "confidence": 0.9}
return {"entity": candidates[0], "confidence": 0.5}
7.2 图谱路径推理
class PathReasoner:
"""图谱路径推理器：通过多跳关系进行推理"""
def __init__(self, neo4j_manager, llm_client):
self.graph = neo4j_manager
self.llm = llm_client
def find_path(self, source: str, target: str, max_depth: int = 4) -> list:
"""查找两个实体之间的路径"""
query = """
MATCH path = shortestPath(
(a:Entity {name: $source})-[*1..""" + str(max_depth) + """]->(b:Entity {name: $target})
)
RETURN [n IN nodes(path) | n.name] AS nodes,
[r IN relationships(path) | type(r)] AS relations
LIMIT 5
"""
with self.graph.driver.session() as session:
result = session.run(query, source=source, target=target)
paths = []
for record in result:
paths.append({
"nodes": record["nodes"],
"relations": record["relations"]
})
return paths
def reason_over_path(self, question: str, paths: list) -> str:
"""基于路径进行推理"""
paths_text = ""
for i, path in enumerate(paths):
chain = " → ".join([
f"{node}[{rel}]" if rel else node
for node, rel in zip(path["nodes"], path["relations"] + [""])
])
paths_text += f"路径{i+1}: {chain}\n"
prompt = f"""请基于以下图谱路径信息回答问题。
## 图谱路径：
{paths_text}
## 问题：
{question}
请基于路径中的关系链进行推理，给出准确的答案。"""
response = self.llm.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
return response.choices[0].message.content
7.3 混合检索策略
class HybridRetriever:
"""混合检索器：结合向量检索和图谱检索"""
def __init__(self, vector_store, graph_rag, llm_client):
self.vector_store = vector_store
self.graph_rag = graph_rag
self.llm = llm_client
def retrieve(self, query: str, top_k: int = 5) -> dict:
"""混合检索"""
# 并行执行向量检索和图谱检索
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
vector_future = executor.submit(self._vector_search, query, top_k)
graph_future = executor.submit(self._graph_search, query)
vector_results = vector_future.result()
graph_results = graph_future.result()
# 融合结果
combined = self._merge_results(vector_results, graph_results)
return {
"vector_results": vector_results,
"graph_results": graph_results,
"combined_context": combined
}
def _vector_search(self, query: str, top_k: int) -> list:
"""向量相似度检索"""
return self.vector_store.similarity_search(query, k=top_k)
def _graph_search(self, query: str) -> str:
"""图谱检索"""
return self.graph_rag.query(query)
def _merge_results(self, vector_results: list, graph_results: str) -> str:
"""融合检索结果"""
vector_context = "\n".join([
f"[文档片段 {i+1}] {doc.page_content}"
for i, doc in enumerate(vector_results)
])
return f"""## 文档检索结果：
{vector_context}
## 知识图谱结果：
{graph_results}"""
第八章：多源知识图谱融合
8.1 图谱对齐
class GraphAligner:
"""知识图谱对齐器"""
def __init__(self, llm_client):
self.llm = llm_client
def align_entities(self, entities_a: list, entities_b: list) -> dict:
"""对齐两个图谱中的实体"""
# 使用 LLM 进行实体对齐
prompt = f"""请从以下两个实体列表中，找出指代同一事物的实体对。
列表A：
{json.dumps(entities_a, ensure_ascii=False)}
列表B：
{json.dumps(entities_b, ensure_ascii=False)}
输出JSON数组，每对格式：
[{{"entity_a": "...", "entity_b": "...", "confidence": 0.95}}]"""
response = self.llm.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.1,
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
def merge_graphs(self, graph_a_triples: list, graph_b_triples: list,
alignments: dict) -> list:
"""合并两个知识图谱"""
# 构建实体映射
entity_map = {}
for align in alignments.get("alignments", []):
entity_map[align["entity_b"]] = align["entity_a"]
# 合并三元组
merged = list(graph_a_triples)
for triple in graph_b_triples:
new_triple = {
"subject": entity_map.get(triple["subject"], triple["subject"]),
"relation": triple["relation"],
"object": entity_map.get(triple["object"], triple["object"])
}
# 去重检查
if new_triple not in merged:
merged.append(new_triple)
return merged
8.2 冲突检测与解决
class ConflictResolver:
"""知识冲突检测与解决"""
def __init__(self, llm_client):
self.llm = llm_client
def detect_conflicts(self, triples: list) -> list:
"""检测三元组之间的冲突"""
conflicts = []
# 按主语分组
subject_groups = {}
for t in triples:
subj = t["subject"]
if subj not in subject_groups:
subject_groups[subj] = []
subject_groups[subj].append(t)
# 检测同一主语下的矛盾关系
for subject, group in subject_groups.items():
relation_groups = {}
for t in group:
rel = t["relation"]
if rel not in relation_groups:
relation_groups[rel] = []
relation_groups[rel].append(t)
# 同一关系有多个不同宾语 → 可能冲突
for rel, rel_triples in relation_groups.items():
if len(rel_triples) > 1:
objects = [t["object"] for t in rel_triples]
if len(set(objects)) > 1:
conflicts.append({
"subject": subject,
"relation": rel,
"conflicting_values": objects,
"triples": rel_triples
})
return conflicts
def resolve_conflict(self, conflict: dict, source_texts: dict = None) -> dict:
"""使用LLM解决冲突"""
values_text = "\n".join([
f"- {v}" for v in conflict["conflicting_values"]
])
prompt = f"""以下是对同一实体的同一关系存在矛盾的知识：
实体：{conflict['subject']}
关系：{conflict['relation']}
矛盾的值：
{values_text}
请判断哪个值更可能正确，并说明理由。如果有多个值都正确（不矛盾），请说明。
输出JSON格式：{{"correct_value": "...", "reason": "...", "all_valid": false}}"""
response = self.llm.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
第九章：实战项目——构建行业知识问答系统
9.1 项目概述
我们将构建一个完整的行业知识问答系统，具备以下能力：
- 自动从行业文档中抽取知识图谱
- 存储到 Neo4j 图数据库
- 提供 Local/Global 两种查询模式
- 支持自然语言问答
- 包含质量评估和人工校验流程
9.2 完整系统实现
import os
import json
import yaml
from typing import List, Dict, Optional
from dataclasses import dataclass, field
from neo4j import GraphDatabase
from openai import OpenAI
@dataclass
class SystemConfig:
"""系统配置"""
neo4j_uri: str = "bolt://localhost:7687"
neo4j_user: str = "neo4j"
neo4j_password: str = "password"
llm_model: str = "gpt-4"
chunk_size: int = 1000
chunk_overlap: int = 200
quality_threshold: float = 0.6
max_search_depth: int = 3
class IndustryQA:
"""行业知识问答系统"""
def __init__(self, config_path: str = "config.yaml"):
# 加载配置
if os.path.exists(config_path):
with open(config_path, 'r', encoding='utf-8') as f:
config_dict = yaml.safe_load(f)
self.config = SystemConfig(**config_dict)
else:
self.config = SystemConfig()
# 初始化组件
self.llm = OpenAI()
self.graph = GraphDatabase.driver(
self.config.neo4j_uri,
auth=(self.config.neo4j_user, self.config.neo4j_password)
)
# 初始化子模块
self.kg_builder = DocumentKGBuilder(self.config.llm_model)
self.quality_eval = TripleQualityReviewer()
print("行业知识问答系统初始化完成")
# ========== 知识图谱构建 ==========
def build_knowledge_graph(self, documents: List[str], domain: str = "通用"):
"""从文档列表构建知识图谱"""
all_triples = []
for i, doc in enumerate(documents):
print(f"\n处理文档 [{i+1}/{len(documents)}]...")
# 文本分块
chunks = self.kg_builder.split_text(
doc,
self.config.chunk_size,
self.config.chunk_overlap
)
# 逐块抽取三元组
for j, chunk in enumerate(chunks):
print(f" 处理文本块 [{j+1}/{len(chunks)}]...")
triples = self.kg_builder.extract_triples_llm(chunk, domain)
all_triples.extend(triples)
# 质量评估
print(f"\n抽取到 {len(all_triples)} 个原始三元组")
quality_triples = self.quality_eval.filter_triples(
all_triples, threshold=self.config.quality_threshold
)
# 导入图数据库
self._import_to_neo4j(quality_triples)
print(f"\n知识图谱构建完成！共导入 {len(quality_triples)} 个三元组")
def _import_to_neo4j(self, triples: List[Dict]):
"""导入三元组到 Neo4j"""
with self.graph.session() as session:
# 创建索引
session.run("CREATE INDEX IF NOT EXISTS FOR (e:Entity) ON (e.name)")
# 批量导入
for t in triples:
session.run("""
MERGE (a:Entity {name: $subject})
MERGE (b:Entity {name: $object})
MERGE (a)-[r:`""" + t['relation'] + """`]->(b)
""", subject=t['subject'], object=t['object'])
# ========== 知识问答 ==========
def ask(self, question: str, mode: str = "auto") -> str:
"""问答入口"""
if mode == "auto":
mode = self._detect_query_type(question)
print(f"查询模式: {mode}")
if mode == "local":
return self._local_search(question)
elif mode == "global":
return self._global_search(question)
else:
return "无法确定查询类型，请明确您的问题。"
def _detect_query_type(self, question: str) -> str:
"""自动检测查询类型"""
prompt = f"""判断以下问题需要"局部查询"还是"全局查询"。
- 局部查询：针对具体实体的事实性问题
- 全局查询：需要综合多个信息的概括性问题
问题：{question}
只回答 "local" 或 "global"。"""
response = self.llm.chat.completions.create(
model=self.config.llm_model,
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)
result = response.choices[0].message.content.strip().lower()
return "local" if "local" in result else "global"
def _local_search(self, question: str) -> str:
"""局部搜索"""
# 提取实体
entities = self._extract_entities(question)
# 检索子图
context_parts = []
with self.graph.session() as session:
for entity in entities:
result = session.run("""
MATCH (e:Entity {name: $name})-[r]-(neighbor)
RETURN e.name AS entity, type(r) AS relation,
neighbor.name AS neighbor
LIMIT 20
""", name=entity)
for record in result:
context_parts.append(
f"{record['entity']} --[{record['relation']}]--> {record['neighbor']}"
)
context = "\n".join(context_parts) if context_parts else "未找到相关信息"
return self._generate_answer(question, context)
def _global_search(self, question: str) -> str:
"""全局搜索"""
# 获取图谱统计信息和关键关系
with self.graph.session() as session:
result = session.run("""
MATCH (e:Entity)-[r]-(neighbor)
WITH e, collect(DISTINCT type(r)) AS rels,
collect(DISTINCT neighbor.name) AS neighbors
RETURN e.name AS entity, rels, neighbors
ORDER BY size(rels) DESC
LIMIT 50
""")
context_parts = []
for record in result:
context_parts.append(
f"{record['entity']}: 关系类型={record['rels']}, "
f"关联实体={record['neighbors'][:5]}"
)
context = "\n".join(context_parts)
return self._generate_answer(question, context, mode="global")
def _extract_entities(self, question: str) -> List[str]:
"""从问题中提取实体"""
prompt = f"""从以下问题中提取所有命名实体，输出JSON数组。
问题：{question}
输出：["实体1", "实体2"]"""
response = self.llm.chat.completions.create(
model=self.config.llm_model,
messages=[{"role": "user", "content": prompt}],
temperature=0.1,
response_format={"type": "json_object"}
)
result = json.loads(response.choices[0].message.content)
return result if isinstance(result, list) else result.get("entities", [])
def _generate_answer(self, question: str, context: str, mode: str = "local") -> str:
"""生成回答"""
system_prompt = """你是一个专业的行业知识问答助手。
请基于提供的知识图谱信息准确回答用户问题。
如果信息不足，请明确说明并建议用户如何获取更多信息。"""
prompt = f"""## 知识图谱信息：
{context}
## 用户问题：
{question}
请基于上述知识给出准确、详细的回答。"""
response = self.llm.chat.completions.create(
model=self.config.llm_model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
temperature=0.3
)
return response.choices[0].message.content
# ========== 系统管理 ==========
def get_stats(self) -> Dict:
"""获取图谱统计信息"""
with self.graph.session() as session:
entity_count = session.run("MATCH (e:Entity) RETURN count(e) AS count").single()["count"]
relation_count = session.run("MATCH ()-[r]->() RETURN count(r) AS count").single()["count"]
return {
"entities": entity_count,
"relations": relation_count,
}
def export_graph(self, output_path: str):
"""导出图谱数据"""
with self.graph.session() as session:
result = session.run("""
MATCH (a:Entity)-[r]->(b:Entity)
RETURN a.name AS subject, type(r) AS relation, b.name AS object
""")
triples = [dict(record) for record in result]
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(triples, f, ensure_ascii=False, indent=2)
print(f"导出 {len(triples)} 个三元组到 {output_path}")
def close(self):
"""关闭连接"""
self.graph.close()
# ========== 主程序入口 ==========
if __name__ == "__main__":
# 初始化系统
qa_system = IndustryQA("config.yaml")
# 示例文档
documents = [
"""
人工智能（AI）是计算机科学的一个分支，致力于创建能够执行通常需要人类智能的任务的系统。
机器学习是AI的核心技术之一，它使计算机能够从数据中学习而无需显式编程。
深度学习是机器学习的一个子领域，使用多层神经网络来处理复杂的数据模式。
自然语言处理（NLP）是AI的另一个重要分支，专注于计算机与人类语言之间的交互。
""",
"""
大语言模型（LLM）是近年来NLP领域的重大突破。
GPT系列模型由OpenAI开发，GPT-4是目前最先进的模型之一。
Transformer架构是LLM的基础，由Google在2017年提出。
注意力机制是Transformer的核心创新，它允许模型关注输入序列中最相关的部分。
预训练+微调是LLM的标准训练范式。
"""
]
# 构建知识图谱
qa_system.build_knowledge_graph(documents, domain="人工智能")
# 问答测试
print("\n" + "="*60)
print("问答测试")
print("="*60)
questions = [
"什么是深度学习？",
"GPT-4是由谁开发的？",
"人工智能有哪些核心技术？", # 全局问题
]
for q in questions:
print(f"\n问题: {q}")
answer = qa_system.ask(q)
print(f"回答: {answer}")
# 系统统计
stats = qa_system.get_stats()
print(f"\n图谱统计: {stats}")
# 导出图谱
qa_system.export_graph("knowledge_graph_export.json")
qa_system.close()
9.3 配置文件
# config.yaml
neo4j_uri: "bolt://localhost:7687"
neo4j_user: "neo4j"
neo4j_password: "your_password_here"
llm_model: "gpt-4"
chunk_size: 1000
chunk_overlap: 200
quality_threshold: 0.6
max_search_depth: 3
9.4 系统运行
# 安装依赖
pip install neo4j openai pyyaml langchain langchain-community langchain-openai
# 启动 Neo4j（Docker方式）
docker run -d \
--name neo4j \
-p 7474:7474 -p 7687:7687 \
-e NEO4J_AUTH=neo4j/password \
neo4j:5
# 运行系统
python industry_qa.py
# Neo4j Web界面：http://localhost:7474
总结与展望
本教程系统讲解了知识图谱与 GraphRAG 的核心技术和实践方法：
- 知识图谱基础：理解实体、关系、属性、本体等核心概念
- LLM驱动构建：利用大语言模型自动从文本中抽取知识三元组
- 图数据库：掌握 Neo4j、NebulaGraph 等主流图数据库的使用
- 质量控制：三元组质量评估与人工校验流程
- GraphRAG 架构：Local Search 与 Global Search 两种检索模式
- LangChain 集成：将知识图谱与 LangChain 生态无缝结合
- 增强检索：实体链接、路径推理、混合检索等高级策略
- 多源融合：图谱对齐、冲突检测与解决
- 实战系统：完整的行业知识问答系统
随着 LLM 技术的持续进步，知识图谱与 RAG 的结合将更加紧密。GraphRAG 不仅能提升 LLM 的事实准确性，还能赋予其结构化推理能力，是构建可信赖 AI 系统的重要技术路径。
推荐学习资源
- Neo4j 官方文档：https://neo4j.com/docs/
- Microsoft GraphRAG：https://github.com/microsoft/graphrag
- LangChain 文档：https://python.langchain.com/
- Knowledge Graph Embedding：https://github.com/thunlp/OpenKE
- NebulaGraph 文档：https://docs.nebula-graph.com.cn/
本教程内容为原创撰写，基于作者对知识图谱和 GraphRAG 技术的实际项目经验整理而成。如有技术细节更新，请参考官方文档获取最新信息。
