DeepSeek-R1-Distill-Llama-8B与LangGraph的知识图谱构建
·
DeepSeek-R1-Distill-Llama-8B与LangGraph：构建智能知识图谱的实战指南
1. 引言：当推理模型遇上知识图谱
想象一下，你手头有一堆杂乱无章的技术文档、产品说明和客户反馈，想要从中提取有价值的信息，却发现人工整理耗时耗力。或者，你的客服系统每天处理大量相似问题，但回答质量参差不齐。这正是知识图谱能够大显身手的地方。
知识图谱不是简单的数据库，它更像是一个智能的大脑，能够理解实体之间的关系，进行逻辑推理。而DeepSeek-R1-Distill-Llama-8B这个模型，正好擅长推理和逻辑分析。当它遇上LangGraph这个专门处理复杂工作流的框架，就能构建出一个真正“智能”的知识图谱系统。
我最近在一个客户项目中尝试了这个组合，效果出乎意料的好。他们原本需要3个人花一周时间整理的行业知识，现在系统能在几小时内自动完成，而且准确率还更高。这就是技术带来的效率提升。
2. 技术选型：为什么是这两个组合？
2.1 DeepSeek-R1-Distill-Llama-8B的优势
这个模型有个很特别的地方：它是通过蒸馏技术从更大的DeepSeek-R1模型学来的推理能力。简单说，就是“大老师教小学生”，把复杂的推理模式教给了这个相对较小的模型。
在实际测试中，我发现它在几个方面表现很突出：
数学和逻辑推理能力强：处理“A是B的供应商，B是C的客户，那么A和C是什么关系？”这类问题时，它能准确推理出间接关系。
代码理解不错：对于技术文档中的代码片段，它能理解功能和作用，这在构建技术知识图谱时特别有用。
上下文长度够用：32K的上下文，意味着它能处理较长的文档，不用频繁切割。
不过要注意的是，这个模型有个小特点：它喜欢“一步一步思考”。如果你问它问题，它会在回答前先进行内部推理。这在构建知识图谱时反而是个优点，因为我们需要的就是这种严谨的推理过程。
2.2 LangGraph的工作流管理
LangGraph不是另一个大语言模型，而是一个框架，专门用来管理复杂的工作流。你可以把它想象成一个智能的流水线控制器。
在知识图谱构建中，通常需要多个步骤：文本解析、实体识别、关系抽取、图结构构建、质量验证等。LangGraph能把这些步骤串起来，还能处理分支、循环、条件判断等复杂逻辑。
我特别喜欢它的“状态管理”功能。比如在构建知识图谱时，某个文档可能解析失败，LangGraph能自动记录这个状态，然后尝试其他方法，或者标记出来让人工处理。这种容错机制在实际应用中非常重要。
3. 实战开始：搭建你的第一个知识图谱系统
3.1 环境准备
首先，你需要安装必要的库。我建议创建一个虚拟环境，避免依赖冲突：
# 创建虚拟环境
python -m venv kg_env
source kg_env/bin/activate # Linux/Mac
# 或者 kg_env\Scripts\activate # Windows
# 安装核心库
pip install langgraph langchain langchain-community
pip install transformers torch
pip install networkx pyvis # 用于图可视化的库
pip install sentence-transformers # 用于文本相似度计算
对于DeepSeek-R1-Distill-Llama-8B，你可以直接从Hugging Face加载：
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
# 加载模型和分词器
model_name = "deepseek-ai/DeepSeek-R1-Distill-Llama-8B"
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.float16,
device_map="auto",
trust_remote_code=True
)
3.2 设计知识图谱构建流程
一个完整的知识图谱构建流程通常包括以下几个步骤，我们可以用LangGraph来管理：
from typing import TypedDict, List, Dict, Any
from langgraph.graph import StateGraph, END
# 定义状态结构
class KnowledgeGraphState(TypedDict):
raw_text: str
chunks: List[str]
entities: List[Dict[str, Any]]
relations: List[Dict[str, Any]]
graph_data: Dict[str, Any]
validation_results: Dict[str, Any]
final_output: str
# 创建状态图
workflow = StateGraph(KnowledgeGraphState)
# 第一步：文本预处理
def text_preprocessing(state: KnowledgeGraphState) -> KnowledgeGraphState:
"""将长文本分割成适合处理的片段"""
text = state["raw_text"]
# 简单的按段落分割，实际中可以用更智能的方法
paragraphs = [p.strip() for p in text.split('\n\n') if p.strip()]
# 如果段落太长，进一步分割
chunks = []
for para in paragraphs:
if len(para) > 1000: # 太长的段落需要分割
sentences = para.split('. ')
current_chunk = ""
for sentence in sentences:
if len(current_chunk) + len(sentence) < 800:
current_chunk += sentence + ". "
else:
chunks.append(current_chunk.strip())
current_chunk = sentence + ". "
if current_chunk:
chunks.append(current_chunk.strip())
else:
chunks.append(para)
return {"chunks": chunks}
# 第二步：实体识别
def entity_extraction(state: KnowledgeGraphState) -> KnowledgeGraphState:
"""使用DeepSeek模型识别文本中的实体"""
chunks = state["chunks"]
entities = []
for chunk in chunks:
# 构建提示词
prompt = f"""请从以下文本中识别出所有重要的实体（如人物、组织、产品、概念等），并按JSON格式返回：
文本：{chunk}
请返回格式：
{{
"entities": [
{{
"name": "实体名称",
"type": "实体类型",
"description": "简要描述"
}}
]
}}"""
# 使用DeepSeek模型处理
inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=500,
temperature=0.6,
do_sample=True
)
response = tokenizer.decode(outputs[0], skip_special_tokens=True)
# 解析响应（这里简化处理，实际需要更健壮的解析）
try:
import json
# 提取JSON部分
json_start = response.find('{')
json_end = response.rfind('}') + 1
if json_start != -1 and json_end != -1:
json_str = response[json_start:json_end]
result = json.loads(json_str)
for entity in result.get("entities", []):
entity["source_chunk"] = chunk[:100] # 记录来源
entities.append(entity)
except:
# 如果解析失败，使用简单的规则提取
pass
return {"entities": entities}
# 第三步：关系抽取
def relation_extraction(state: KnowledgeGraphState) -> KnowledgeGraphState:
"""识别实体之间的关系"""
chunks = state["chunks"]
entities = state["entities"]
relations = []
# 为每个实体创建简化的表示
entity_names = [e["name"] for e in entities]
for chunk in chunks:
# 检查这个chunk中是否包含多个实体
chunk_entities = []
for entity in entities:
if entity["name"] in chunk:
chunk_entities.append(entity)
if len(chunk_entities) >= 2:
# 构建关系抽取提示词
entity_list = ", ".join([e["name"] for e in chunk_entities])
prompt = f"""分析以下文本，识别提到的实体之间的关系：
文本：{chunk}
涉及的实体：{entity_list}
请分析这些实体之间的关系，按JSON格式返回：
{{
"relations": [
{{
"source": "源实体",
"target": "目标实体",
"relation": "关系类型",
"description": "关系描述",
"confidence": 置信度(0-1)
}}
]
}}"""
# 使用模型处理
inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=300,
temperature=0.6
)
response = tokenizer.decode(outputs[0], skip_special_tokens=True)
# 解析关系
try:
import json
json_start = response.find('{')
json_end = response.rfind('}') + 1
if json_start != -1 and json_end != -1:
json_str = response[json_start:json_end]
result = json.loads(json_str)
for rel in result.get("relations", []):
relations.append(rel)
except:
pass
return {"relations": relations}
# 第四步：构建图结构
def graph_construction(state: KnowledgeGraphState) -> KnowledgeGraphState:
"""将实体和关系构建成图结构"""
entities = state["entities"]
relations = state["relations"]
# 创建图数据结构
graph_data = {
"nodes": [],
"edges": [],
"metadata": {
"total_entities": len(entities),
"total_relations": len(relations),
"timestamp": "2024-01-01" # 实际应该用当前时间
}
}
# 添加节点
for i, entity in enumerate(entities):
graph_data["nodes"].append({
"id": f"entity_{i}",
"label": entity["name"],
"type": entity.get("type", "unknown"),
"description": entity.get("description", ""),
"properties": entity
})
# 添加边
for i, relation in enumerate(relations):
# 查找源和目标实体的ID
source_id = None
target_id = None
for j, entity in enumerate(entities):
if entity["name"] == relation["source"]:
source_id = f"entity_{j}"
if entity["name"] == relation["target"]:
target_id = f"entity_{j}"
if source_id and target_id:
graph_data["edges"].append({
"id": f"relation_{i}",
"source": source_id,
"target": target_id,
"label": relation["relation"],
"description": relation.get("description", ""),
"confidence": relation.get("confidence", 0.5)
})
return {"graph_data": graph_data}
# 第五步：验证和优化
def validation_and_optimization(state: KnowledgeGraphState) -> KnowledgeGraphState:
"""验证知识图谱的质量并进行优化"""
graph_data = state["graph_data"]
validation_results = {
"checks_passed": [],
"issues_found": [],
"suggestions": []
}
# 检查1：是否有孤立节点（没有连接的实体）
connected_nodes = set()
for edge in graph_data["edges"]:
connected_nodes.add(edge["source"])
connected_nodes.add(edge["target"])
all_nodes = {node["id"] for node in graph_data["nodes"]}
isolated_nodes = all_nodes - connected_nodes
if isolated_nodes:
validation_results["issues_found"].append({
"type": "isolated_nodes",
"count": len(isolated_nodes),
"nodes": list(isolated_nodes)
})
validation_results["suggestions"].append(
"考虑为孤立节点添加关系，或者检查是否识别了所有相关关系"
)
else:
validation_results["checks_passed"].append("没有孤立节点")
# 检查2：关系置信度
low_confidence_edges = [
edge for edge in graph_data["edges"]
if edge.get("confidence", 1) < 0.3
]
if low_confidence_edges:
validation_results["issues_found"].append({
"type": "low_confidence_relations",
"count": len(low_confidence_edges),
"edges": low_confidence_edges[:5] # 只显示前5个
})
# 检查3：重复实体
entity_names = [node["label"] for node in graph_data["nodes"]]
duplicates = {}
for name in entity_names:
if entity_names.count(name) > 1:
duplicates[name] = entity_names.count(name)
if duplicates:
validation_results["issues_found"].append({
"type": "duplicate_entities",
"count": len(duplicates),
"examples": dict(list(duplicates.items())[:3])
})
return {"validation_results": validation_results}
# 第六步：生成最终输出
def generate_final_output(state: KnowledgeGraphState) -> KnowledgeGraphState:
"""生成最终的知识图谱报告"""
graph_data = state["graph_data"]
validation = state["validation_results"]
report = f"""# 知识图谱构建报告
## 概览
- 实体数量：{graph_data['metadata']['total_entities']}
- 关系数量：{graph_data['metadata']['total_relations']}
- 构建时间：{graph_data['metadata']['timestamp']}
## 主要实体
"""
# 添加重要实体
for node in graph_data["nodes"][:10]: # 只显示前10个
report += f"- **{node['label']}** ({node['type']}): {node['description'][:100]}...\n"
report += "\n## 质量检查结果\n"
if validation["checks_passed"]:
report += " 通过的检查：\n"
for check in validation["checks_passed"]:
report += f" - {check}\n"
if validation["issues_found"]:
report += "\n 发现的问题：\n"
for issue in validation["issues_found"]:
report += f" - {issue['type']}: {issue['count']}个\n"
if validation["suggestions"]:
report += "\n 优化建议：\n"
for suggestion in validation["suggestions"]:
report += f" - {suggestion}\n"
report += "\n## 下一步建议\n1. 人工审核低置信度的关系\n2. 为孤立实体补充关系\n3. 合并重复的实体\n4. 扩展相关领域的知识"
return {"final_output": report}
# 将各个步骤添加到工作流中
workflow.add_node("preprocess", text_preprocessing)
workflow.add_node("extract_entities", entity_extraction)
workflow.add_node("extract_relations", relation_extraction)
workflow.add_node("build_graph", graph_construction)
workflow.add_node("validate", validation_and_optimization)
workflow.add_node("generate_report", generate_final_output)
# 设置执行顺序
workflow.set_entry_point("preprocess")
workflow.add_edge("preprocess", "extract_entities")
workflow.add_edge("extract_entities", "extract_relations")
workflow.add_edge("extract_relations", "build_graph")
workflow.add_edge("build_graph", "validate")
workflow.add_edge("validate", "generate_report")
workflow.add_edge("generate_report", END)
# 编译工作流
kg_workflow = workflow.compile()
3.3 运行知识图谱构建
现在我们可以使用这个工作流来处理文本了：
# 示例文本
sample_text = """
OpenAI是一家美国人工智能研究实验室，由Elon Musk、Sam Altman等人于2015年创立。
该公司最著名的产品是ChatGPT，这是一个基于GPT架构的对话AI系统。
GPT-4是OpenAI在2023年发布的大型语言模型，在多项基准测试中表现出色。
微软是OpenAI的重要投资方和合作伙伴，双方在Azure云服务上有深度合作。
"""
# 初始化状态
initial_state = {
"raw_text": sample_text,
"chunks": [],
"entities": [],
"relations": [],
"graph_data": {},
"validation_results": {},
"final_output": ""
}
# 执行工作流
try:
final_state = kg_workflow.invoke(initial_state)
print("=" * 60)
print("知识图谱构建完成！")
print("=" * 60)
print("\n最终报告：")
print(final_state["final_output"])
# 可视化图结构（可选）
import networkx as nx
import matplotlib.pyplot as plt
G = nx.Graph()
# 添加节点
for node in final_state["graph_data"]["nodes"]:
G.add_node(node["label"], type=node["type"])
# 添加边
for edge in final_state["graph_data"]["edges"]:
source_label = None
target_label = None
for node in final_state["graph_data"]["nodes"]:
if node["id"] == edge["source"]:
source_label = node["label"]
if node["id"] == edge["target"]:
target_label = node["label"]
if source_label and target_label:
G.add_edge(source_label, target_label, label=edge["label"])
# 绘制图形
plt.figure(figsize=(12, 8))
pos = nx.spring_layout(G, seed=42)
nx.draw(G, pos, with_labels=True, node_color='lightblue',
node_size=2000, font_size=10, font_weight='bold')
# 添加边标签
edge_labels = nx.get_edge_attributes(G, 'label')
nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels)
plt.title("知识图谱可视化")
plt.tight_layout()
plt.savefig("knowledge_graph.png", dpi=300, bbox_inches='tight')
print("\n 知识图谱已保存为 knowledge_graph.png")
except Exception as e:
print(f"构建过程中出现错误：{e}")
import traceback
traceback.print_exc()
4. 进阶技巧：让知识图谱更智能
4.1 处理大规模文档
当处理大量文档时，我们需要优化流程：
def batch_processing(documents: List[str], batch_size: int = 5):
"""批量处理多个文档"""
all_entities = []
all_relations = []
for i in range(0, len(documents), batch_size):
batch = documents[i:i+batch_size]
print(f"处理批次 {i//batch_size + 1}/{(len(documents)+batch_size-1)//batch_size}")
for doc in batch:
state = {
"raw_text": doc,
"chunks": [],
"entities": [],
"relations": [],
"graph_data": {},
"validation_results": {},
"final_output": ""
}
try:
result = kg_workflow.invoke(state)
all_entities.extend(result["entities"])
all_relations.extend(result["relations"])
except Exception as e:
print(f"文档处理失败：{e}")
continue
# 去重和合并
unique_entities = merge_duplicate_entities(all_entities)
consolidated_relations = consolidate_relations(all_relations, unique_entities)
return unique_entities, consolidated_relations
def merge_duplicate_entities(entities: List[Dict]) -> List[Dict]:
"""合并重复的实体"""
entity_dict = {}
for entity in entities:
name = entity["name"]
if name not in entity_dict:
entity_dict[name] = entity
else:
# 合并描述和属性
existing = entity_dict[name]
if "descriptions" not in existing:
existing["descriptions"] = []
existing["descriptions"].append(entity.get("description", ""))
# 合并来源
if "sources" not in existing:
existing["sources"] = []
existing["sources"].append(entity.get("source_chunk", ""))
return list(entity_dict.values())
def consolidate_relations(relations: List[Dict], entities: List[Dict]) -> List[Dict]:
"""合并和验证关系"""
# 按实体对分组
relation_dict = {}
for rel in relations:
key = (rel["source"], rel["target"], rel["relation"])
if key not in relation_dict:
relation_dict[key] = {
"source": rel["source"],
"target": rel["target"],
"relation": rel["relation"],
"descriptions": [],
"confidences": [],
"sources": []
}
relation_dict[key]["descriptions"].append(rel.get("description", ""))
relation_dict[key]["confidences"].append(rel.get("confidence", 0.5))
if "source_chunk" in rel:
relation_dict[key]["sources"].append(rel["source_chunk"])
# 计算平均置信度，合并描述
consolidated = []
for key, rel_data in relation_dict.items():
avg_confidence = sum(rel_data["confidences"]) / len(rel_data["confidences"])
# 选择最常见的描述
from collections import Counter
if rel_data["descriptions"]:
desc_counter = Counter(rel_data["descriptions"])
most_common_desc = desc_counter.most_common(1)[0][0]
else:
most_common_desc = ""
consolidated.append({
"source": rel_data["source"],
"target": rel_data["target"],
"relation": rel_data["relation"],
"description": most_common_desc,
"confidence": avg_confidence,
"source_count": len(rel_data["sources"])
})
# 按置信度排序
consolidated.sort(key=lambda x: x["confidence"], reverse=True)
return consolidated
4.2 增量更新知识图谱
知识图谱需要定期更新，而不是每次都从头构建：
class KnowledgeGraphUpdater:
"""知识图谱增量更新器"""
def __init__(self, existing_graph):
self.existing_graph = existing_graph
self.new_entities = []
self.new_relations = []
self.conflicts = []
def update_with_new_document(self, document: str):
"""用新文档更新知识图谱"""
# 从文档中提取信息
state = {
"raw_text": document,
"chunks": [],
"entities": [],
"relations": [],
"graph_data": {},
"validation_results": {},
"final_output": ""
}
result = kg_workflow.invoke(state)
new_entities = result["entities"]
new_relations = result["relations"]
# 检查与现有知识的冲突
self._check_conflicts(new_entities, new_relations)
# 合并新信息
self._merge_entities(new_entities)
self._merge_relations(new_relations)
return {
"added_entities": len(new_entities),
"added_relations": len(new_relations),
"conflicts_found": len(self.conflicts)
}
def _check_conflicts(self, new_entities, new_relations):
"""检查新旧知识之间的冲突"""
existing_entity_types = {
entity["name"]: entity["type"]
for entity in self.existing_graph.get("nodes", [])
}
for entity in new_entities:
name = entity["name"]
if name in existing_entity_types:
if entity["type"] != existing_entity_types[name]:
self.conflicts.append({
"type": "entity_type_conflict",
"entity": name,
"existing_type": existing_entity_types[name],
"new_type": entity["type"]
})
def _merge_entities(self, new_entities):
"""合并实体"""
existing_names = {e["name"] for e in self.existing_graph.get("nodes", [])}
for entity in new_entities:
if entity["name"] not in existing_names:
self.new_entities.append(entity)
self.existing_graph["nodes"].append({
"id": f"entity_{len(self.existing_graph['nodes'])}",
"label": entity["name"],
"type": entity.get("type", "unknown"),
"description": entity.get("description", ""),
"properties": entity
})
def _merge_relations(self, new_relations):
"""合并关系"""
existing_relation_keys = set()
for edge in self.existing_graph.get("edges", []):
# 找到对应的节点标签
source_label = None
target_label = None
for node in self.existing_graph["nodes"]:
if node["id"] == edge["source"]:
source_label = node["label"]
if node["id"] == edge["target"]:
target_label = node["label"]
if source_label and target_label:
key = (source_label, target_label, edge["label"])
existing_relation_keys.add(key)
for relation in new_relations:
key = (relation["source"], relation["target"], relation["relation"])
if key not in existing_relation_keys:
self.new_relations.append(relation)
# 添加到图中
source_id = None
target_id = None
for node in self.existing_graph["nodes"]:
if node["label"] == relation["source"]:
source_id = node["id"]
if node["label"] == relation["target"]:
target_id = node["id"]
if source_id and target_id:
self.existing_graph["edges"].append({
"id": f"relation_{len(self.existing_graph['edges'])}",
"source": source_id,
"target": target_id,
"label": relation["relation"],
"description": relation.get("description", ""),
"confidence": relation.get("confidence", 0.5)
})
4.3 知识图谱查询接口
构建好的知识图谱需要提供查询接口：
class KnowledgeGraphQueryEngine:
"""知识图谱查询引擎"""
def __init__(self, graph_data):
self.graph_data = graph_data
self._build_index()
def _build_index(self):
"""构建查询索引"""
self.entity_index = {}
self.relation_index = {}
for node in self.graph_data["nodes"]:
name = node["label"].lower()
if name not in self.entity_index:
self.entity_index[name] = []
self.entity_index[name].append(node)
for edge in self.graph_data["edges"]:
# 找到对应的节点标签
source_label = None
target_label = None
for node in self.graph_data["nodes"]:
if node["id"] == edge["source"]:
source_label = node["label"]
if node["id"] == edge["target"]:
target_label = node["label"]
if source_label and target_label:
key = (source_label.lower(), target_label.lower())
if key not in self.relation_index:
self.relation_index[key] = []
self.relation_index[key].append(edge)
def find_entity(self, name: str):
"""查找实体"""
name_lower = name.lower()
results = []
# 精确匹配
if name_lower in self.entity_index:
results.extend(self.entity_index[name_lower])
# 模糊匹配
for entity_name, entities in self.entity_index.items():
if name_lower in entity_name or entity_name in name_lower:
for entity in entities:
if entity not in results:
results.append(entity)
return results
def find_relations(self, entity1: str, entity2: str = None):
"""查找关系"""
entity1_lower = entity1.lower()
results = []
if entity2:
entity2_lower = entity2.lower()
key = (entity1_lower, entity2_lower)
if key in self.relation_index:
results.extend(self.relation_index[key])
# 反向查找
key_reverse = (entity2_lower, entity1_lower)
if key_reverse in self.relation_index:
for rel in self.relation_index[key_reverse]:
# 标记为反向关系
rel_copy = rel.copy()
rel_copy["is_reverse"] = True
results.append(rel_copy)
else:
# 查找所有与entity1相关的关系
for (source, target), relations in self.relation_index.items():
if source == entity1_lower or target == entity1_lower:
results.extend(relations)
return results
def query_with_natural_language(self, question: str):
"""用自然语言查询"""
# 使用DeepSeek模型理解查询意图
prompt = f"""分析以下问题，理解用户想查询知识图谱中的什么信息：
问题：{question}
请将问题解析为知识图谱查询，按JSON格式返回：
{{
"intent": "查询意图描述",
"target_entities": ["实体1", "实体2", ...],
"query_type": "find_entity|find_relations|path_query|...",
"parameters": {{
// 根据查询类型不同而不同
}}
}}"""
inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=300,
temperature=0.3
)
response = tokenizer.decode(outputs[0], skip_special_tokens=True)
try:
import json
json_start = response.find('{')
json_end = response.rfind('}') + 1
if json_start != -1 and json_end != -1:
json_str = response[json_start:json_end]
query_spec = json.loads(json_str)
# 根据查询规格执行查询
return self._execute_query(query_spec)
except:
pass
return {"error": "无法解析查询"}
def _execute_query(self, query_spec):
"""执行查询"""
query_type = query_spec.get("query_type", "")
if query_type == "find_entity":
entities = query_spec.get("target_entities", [])
if entities:
results = {}
for entity in entities:
results[entity] = self.find_entity(entity)
return results
elif query_type == "find_relations":
entities = query_spec.get("target_entities", [])
if len(entities) >= 2:
return self.find_relations(entities[0], entities[1])
elif len(entities) == 1:
return self.find_relations(entities[0])
elif query_type == "path_query":
# 查找两个实体之间的路径
entities = query_spec.get("target_entities", [])
if len(entities) >= 2:
return self._find_path(entities[0], entities[1])
return {"error": f"不支持的查询类型: {query_type}"}
def _find_path(self, entity1: str, entity2: str, max_depth: int = 3):
"""查找两个实体之间的路径"""
import networkx as nx
# 构建NetworkX图
G = nx.Graph()
for node in self.graph_data["nodes"]:
G.add_node(node["label"])
for edge in self.graph_data["edges"]:
source_label = None
target_label = None
for node in self.graph_data["nodes"]:
if node["id"] == edge["source"]:
source_label = node["label"]
if node["id"] == edge["target"]:
target_label = node["label"]
if source_label and target_label:
G.add_edge(source_label, target_label,
relation=edge["label"],
confidence=edge.get("confidence", 0.5))
# 查找路径
try:
paths = list(nx.all_simple_paths(G, entity1, entity2, cutoff=max_depth))
results = []
for path in paths[:5]: # 只返回前5条路径
path_info = {
"entities": path,
"relations": [],
"total_confidence": 1.0
}
for i in range(len(path) - 1):
edge_data = G[path[i]][path[i+1]]
path_info["relations"].append({
"from": path[i],
"to": path[i+1],
"relation": edge_data["relation"],
"confidence": edge_data["confidence"]
})
path_info["total_confidence"] *= edge_data["confidence"]
results.append(path_info)
return {
"entity1": entity1,
"entity2": entity2,
"paths_found": len(paths),
"paths": results
}
except nx.NetworkXNoPath:
return {"message": f"在{max_depth}步内未找到从{entity1}到{entity2}的路径"}
except nx.NodeNotFound as e:
return {"error": f"实体不存在: {str(e)}"}
5. 实际应用案例
5.1 技术文档知识图谱
假设我们有一些技术文档，想要构建一个技术栈知识图谱：
# 技术文档示例
tech_docs = [
"""
Docker是一个开源的应用容器引擎，基于Go语言开发。
它允许开发者将应用及其依赖打包到一个可移植的容器中。
Kubernetes是Google开源的容器编排系统，用于自动化部署、扩展和管理容器化应用。
Docker容器可以在Kubernetes集群中运行。
""",
"""
React是一个用于构建用户界面的JavaScript库，由Facebook开发。
Vue.js是另一个流行的JavaScript框架，用于构建用户界面。
两者都可以用于创建单页面应用（SPA）。
""",
"""
TensorFlow是Google开发的开源机器学习框架。
PyTorch是Facebook开发的另一个流行的机器学习框架。
两者都支持深度学习模型的训练和部署。
"""
]
# 批量处理文档
entities, relations = batch_processing(tech_docs)
print(f"提取到 {len(entities)} 个实体")
print(f"提取到 {len(relations)} 个关系")
# 构建完整的知识图谱
graph_state = {
"raw_text": "",
"chunks": [],
"entities": entities,
"relations": relations,
"graph_data": {},
"validation_results": {},
"final_output": ""
}
# 执行后续步骤
graph_state = kg_workflow.invoke(graph_state)
# 创建查询引擎
query_engine = KnowledgeGraphQueryEngine(graph_state["graph_data"])
# 示例查询
print("\n查询示例：")
print("1. 查找Docker的相关信息：")
docker_info = query_engine.find_entity("Docker")
for info in docker_info:
print(f" - {info['label']} ({info['type']}): {info['description'][:100]}...")
print("\n2. 查找Docker和Kubernetes的关系：")
docker_k8s_relations = query_engine.find_relations("Docker", "Kubernetes")
for rel in docker_k8s_relations:
print(f" - {rel['source']} --[{rel['label']}]--> {rel['target']}")
print("\n3. 自然语言查询：")
nl_query_result = query_engine.query_with_natural_language(
"React和Vue.js有什么共同点？"
)
print(f" 查询结果：{nl_query_result}")
5.2 客户服务知识图谱
在客户服务场景中，知识图谱可以帮助快速找到解决方案：
class CustomerServiceKG:
"""客户服务知识图谱系统"""
def __init__(self):
self.knowledge_base = {
"常见问题": [],
"解决方案": [],
"产品信息": [],
"故障排除": []
}
self.query_engine = None
def build_from_faqs(self, faqs: List[Dict]):
"""从FAQ构建知识图谱"""
documents = []
for faq in faqs:
doc = f"""
问题：{faq['question']}
答案：{faq['answer']}
分类：{faq.get('category', '未分类')}
关键词：{', '.join(faq.get('keywords', []))}
"""
documents.append(doc)
# 提取知识
entities, relations = batch_processing(documents)
# 构建图谱
graph_state = {
"raw_text": "",
"chunks": [],
"entities": entities,
"relations": relations,
"graph_data": {},
"validation_results": {},
"final_output": ""
}
graph_state = kg_workflow.invoke(graph_state)
self.query_engine = KnowledgeGraphQueryEngine(graph_state["graph_data"])
return graph_state["final_output"]
def answer_question(self, question: str):
"""回答客户问题"""
if not self.query_engine:
return "知识图谱尚未构建，请先构建知识库"
# 首先尝试直接匹配
relevant_entities = self.query_engine.find_entity(question)
if relevant_entities:
# 找到相关实体，查找相关关系
answers = []
for entity in relevant_entities[:3]: # 最多3个相关实体
relations = self.query_engine.find_relations(entity["label"])
for rel in relations[:2]: # 每个实体最多2个关系
answer = {
"相关实体": entity["label"],
"关系": rel["label"],
"目标实体": rel.get("target", ""),
"描述": rel.get("description", ""),
"置信度": rel.get("confidence", 0)
}
answers.append(answer)
if answers:
# 按置信度排序
answers.sort(key=lambda x: x["置信度"], reverse=True)
response = f"根据知识库，找到以下相关信息：\n\n"
for i, ans in enumerate(answers[:3], 1): # 返回前3个
response += f"{i}. {ans['相关实体']} {ans['关系']} {ans['目标实体']}\n"
if ans["描述"]:
response += f" 说明：{ans['描述']}\n"
response += f" 相关度：{ans['置信度']:.2f}\n\n"
return response
# 如果没有直接匹配，使用自然语言查询
nl_result = self.query_engine.query_with_natural_language(question)
if "error" not in nl_result:
return f"智能分析结果：\n{nl_result}"
else:
return "抱歉，在知识库中没有找到相关信息。建议联系人工客服。"
# 使用示例
faqs = [
{
"question": "如何重置密码？",
"answer": "请访问登录页面，点击'忘记密码'链接，按照提示操作即可重置密码。",
"category": "账户管理",
"keywords": ["密码", "重置", "登录"]
},
{
"question": "支付失败怎么办？",
"answer": "请检查银行卡余额、网络连接，或尝试更换支付方式。如问题持续，请联系客服。",
"category": "支付问题",
"keywords": ["支付", "失败", "银行卡"]
},
{
"question": "如何查看订单状态？",
"answer": "登录账户后，进入'我的订单'页面即可查看所有订单的当前状态。",
"category": "订单管理",
"keywords": ["订单", "状态", "查看"]
}
]
cs_system = CustomerServiceKG()
report = cs_system.build_from_faqs(faqs)
print("知识图谱构建报告：")
print(report)
# 测试查询
test_questions = [
"我忘记密码了怎么办？",
"支付时显示失败",
"想看看我的订单到哪里了"
]
for q in test_questions:
print(f"\n问题：{q}")
print(f"回答：{cs_system.answer_question(q)}")
print("-" * 50)
6. 性能优化和最佳实践
6.1 模型推理优化
DeepSeek-R1-Distill-Llama-8B虽然相对较小，但在大规模应用时仍需优化：
class OptimizedKGProcessor:
"""优化的知识图谱处理器"""
def __init__(self, model_path="deepseek-ai/DeepSeek-R1-Distill-Llama-8B"):
# 使用量化模型减少内存占用
from transformers import BitsAndBytesConfig
quantization_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_compute_dtype=torch.float16,
bnb_4bit_quant_type="nf4",
bnb_4bit_use_double_quant=True,
)
self.tokenizer = AutoTokenizer.from_pretrained(
model_path,
trust_remote_code=True
)
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
quantization_config=quantization_config,
device_map="auto",
trust_remote_code=True
)
# 启用缓存提高推理速度
self.model.config.use_cache = True
# 预热模型
self._warmup_model()
def _warmup_model(self):
"""预热模型"""
warmup_text = "这是一个测试。"
inputs = self.tokenizer(warmup_text, return_tensors="pt").to(self.model.device)
with torch.no_grad():
_ = self.model.generate(**inputs, max_new_tokens=10)
def batch_extract(self, texts: List[str], batch_size: int = 4):
"""批量提取信息"""
results = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i+batch_size]
batch_results = self._process_batch(batch)
results.extend(batch_results)
# 清理缓存避免内存泄漏
if hasattr(torch.cuda, 'empty_cache'):
torch.cuda.empty_cache()
return results
def _process_batch(self, texts: List[str]):
"""处理一个批次"""
# 构建批量提示词
prompts = []
for text in texts:
prompt = f"""从以下文本中提取关键信息：
文本：{text}
请提取实体和关系，按JSON格式返回：
{{
"entities": [
{{"name": "名称", "type": "类型", "description": "描述"}}
],
"relations": [
{{"source": "源实体", "target": "目标实体", "relation": "关系类型"}}
]
}}"""
prompts.append(prompt)
# 批量编码
inputs = self.tokenizer(
prompts,
return_tensors="pt",
padding=True,
truncation=True,
max_length=2048
).to(self.model.device)
# 批量生成
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=512,
temperature=0.6,
do_sample=True,
top_p=0.95,
pad_token_id=self.tokenizer.eos_token_id
)
# 解码和解析
batch_results = []
for i, output in enumerate(outputs):
response = self.tokenizer.decode(output, skip_special_tokens=True)
try:
import json
json_start = response.find('{')
json_end = response.rfind('}') + 1
if json_start != -1 and json_end != -1:
json_str = response[json_start:json_end]
result = json.loads(json_str)
batch_results.append(result)
else:
batch_results.append({"entities": [], "relations": []})
except:
batch_results.append({"entities": [], "relations": []})
return batch_results
6.2 缓存和持久化
对于生产环境，需要添加缓存和持久化：
import hashlib
import pickle
from pathlib import Path
class PersistentKGSystem:
"""持久化的知识图谱系统"""
def __init__(self, cache_dir="./kg_cache"):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
self.processor = OptimizedKGProcessor()
self.graphs = {} # 内存中的图
def _get_cache_key(self, text: str) -> str:
"""生成缓存键"""
return hashlib.md5(text.encode()).hexdigest()
def process_document(self, text: str, use_cache: bool = True):
"""处理文档，支持缓存"""
cache_key = self._get_cache_key(text)
cache_file = self.cache_dir / f"{cache_key}.pkl"
# 检查缓存
if use_cache and cache_file.exists():
try:
with open(cache_file, 'rb') as f:
cached_result = pickle.load(f)
print(f"使用缓存结果：{cache_key}")
return cached_result
except:
pass
# 处理文档
print(f"处理新文档：{cache_key}")
result = self.processor.batch_extract([text])[0]
# 保存到缓存
try:
with open(cache_file, 'wb') as f:
pickle.dump(result, f)
except:
pass
return result
def build_graph_from_documents(self, documents: List[str], graph_name: str):
"""从多个文档构建知识图谱"""
all_entities = []
all_relations = []
for i, doc in enumerate(documents):
print(f"处理文档 {i+1}/{len(documents)}")
result = self.process_document(doc)
if "entities" in result:
for entity in result["entities"]:
entity["source_doc"] = f"doc_{i}"
all_entities.append(entity)
if "relations" in result:
for relation in result["relations"]:
relation["source_doc"] = f"doc_{i}"
all_relations.append(relation)
# 构建图
graph_state = {
"raw_text": "",
"chunks": [],
"entities": all_entities,
"relations": all_relations,
"graph_data": {},
"validation_results": {},
"final_output": ""
}
kg_workflow = workflow.compile()
final_state = kg_workflow.invoke(graph_state)
# 保存到内存
self.graphs[graph_name] = {
"data": final_state["graph_data"],
"metadata": {
"doc_count": len(documents),
"entity_count": len(all_entities),
"relation_count": len(all_relations),
"build_time": "2024-01-01" # 实际用当前时间
}
}
# 持久化到文件
graph_file = self.cache_dir / f"graph_{graph_name}.pkl"
with open(graph_file, 'wb') as f:
pickle.dump(self.graphs[graph_name], f)
return final_state["final_output"]
def load_graph(self, graph_name: str):
"""加载已保存的知识图谱"""
graph_file = self.cache_dir / f"graph_{graph_name}.pkl"
if graph_file.exists():
with open(graph_file, 'rb') as f:
graph_data = pickle.load(f)
self.graphs[graph_name] = graph_data
return True
else:
return False
def query_graph(self, graph_name: str, query: str):
"""查询知识图谱"""
if graph_name not in self.graphs:
if not self.load_graph(graph_name):
return {"error": f"知识图谱 '{graph_name}' 不存在"}
graph_data = self.graphs[graph_name]["data"]
query_engine = KnowledgeGraphQueryEngine(graph_data)
return query_engine.query_with_natural_language(query)
7. 总结与展望
通过DeepSeek-R1-Distill-Llama-8B和LangGraph的结合，我们构建了一个相对完整的智能知识图谱系统。这个系统能够从非结构化文本中自动提取知识，构建图结构，并提供智能查询功能。
在实际使用中，我发现这个组合有几个明显的优势：DeepSeek模型的推理能力让实体和关系抽取更加准确，而LangGraph的工作流管理让整个构建过程更加可控和可维护。特别是对于需要多步骤处理的复杂任务，LangGraph的状态管理机制非常实用。
不过也要注意一些挑战。比如，模型的输出格式需要仔细解析，有时候会出现格式错误。另外，对于大规模文档集，需要做好批处理和缓存优化。我在代码中提供了一些优化方案，但实际应用中可能需要根据具体场景调整。
未来可以考虑的改进方向包括：添加多语言支持、集成更多的数据源（如图片、表格）、实现实时更新机制，以及加入更复杂的推理功能。知识图谱的潜力很大，特别是在企业知识管理、智能客服、推荐系统等领域，都有广泛的应用前景。
如果你刚开始接触这个领域，建议从小规模数据开始，逐步优化流程。遇到问题时，多看看模型的中间输出，调整提示词，往往能找到解决方案。技术总是在不断进步，保持学习和实践，就能构建出越来越智能的系统。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
2
0- 0
已为社区贡献25条内容
所有评论(0)