pengxiao/KG_ICH
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init: KG_ICH 项目初始化

Browse Source 
- data/: 非遗地理编码数据（GIS shapefile + CSV）
- dofile/kg_project/: 知识图谱构建代码（纳入主仓库）
- dofile/visulization/: 可视化数据与路线图
- officefile/: 文献、草稿、bib 文档
- officefile/latex/: Overleaf 同步目录（独立管理，不纳入）
- output/: 输出目录
- logs/: 日志目录
This commit is contained in:
xiaopeng
2026-05-30 00:52:36 +08:00
commit 834cad729f
44 changed files with 8586 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dofile/kg_project/src/__init__.py
View File
dofile/kg_project/src/data_processing/__init__.py
View File
dofile/kg_project/src/data_processing/entity_normalizer.py
+344
View File
@@ -0,0 +1,344 @@
# -*- coding: utf-8 -*-
"""
实体规范化器
负责实体的去重、ID生成和规范化
"""
import json
import hashlib
import logging
from typing import Dict, List, Any, Optional
from pathlib import Path
import yaml
class EntityNormalizer:
"""实体规范化器"""
def __init__(self, ontology_file: str):
"""
初始化规范化器
Args:
ontology_file: 本体配置文件路径
"""
# 加载本体配置
with open(ontology_file, 'r', encoding='utf-8') as f:
self.ontology = yaml.safe_load(f)
self.logger = logging.getLogger(__name__)
# 实体注册表：{entity_id: entity_data}
self.entity_registry = {}
# 文本到ID的映射：{normalized_text: entity_id}
self.text_to_id_map = {}
# 类型计数器：{entity_type: count}
self.type_counters = {}
self.logger.info("EntityNormalizer初始化完成")
def normalize_text(self, text: str) -> str:
"""
标准化文本
Args:
text: 原始文本
Returns:
标准化后的文本
"""
if not text:
return ""
# 去除首尾空格
text = text.strip()
# 统一全角/半角字符
text = text.replace('　', ' ').replace('，', ',').replace('、', ',')
# 转换为小写进行比较（保持原文用于显示）
return text
def calculate_similarity(self, text1: str, text2: str) -> float:
"""
计算两个文本的相似度（基于编辑距离）
Args:
text1: 文本1
text2: 文本2
Returns:
相似度（0-1之间）
"""
import Levenshtein
norm1 = self.normalize_text(text1)
norm2 = self.normalize_text(text2)
if not norm1 or not norm2:
return 0.0
max_len = max(len(norm1), len(norm2))
if max_len == 0:
return 1.0
distance = Levenshtein.distance(norm1, norm2)
similarity = 1.0 - (distance / max_len)
return similarity
def generate_entity_id(self, entity_type: str, entity_text: str) -> str:
"""
生成实体ID
Args:
entity_type: 实体类型
entity_text: 实体文本
Returns:
实体ID
"""
# 获取类型配置
type_config = self.ontology['entity_types'].get(entity_type, {})
prefix = type_config.get('prefix', 'UNK')
# 生成哈希值
text_hash = abs(hash(entity_text)) % 100000
# 检查是否需要计数器（某些类型可能需要）
if entity_type not in self.type_counters:
self.type_counters[entity_type] = 0
# 根据类型生成ID
if entity_type in ['Ethnic_Group', 'Geographic_Location', 'Geographic_Environment', 'Time_Period']:
# 这些类型使用唯一哈希，不需要计数器
entity_id = f"{prefix}-{text_hash:05d}"
else:
# 其他类型使用计数器
self.type_counters[entity_type] += 1
counter = self.type_counters[entity_type]
entity_id = f"{prefix}-{text_hash:05d}-{counter}"
return entity_id
def find_similar_entity(
self,
entity_text: str,
entity_type: str,
similarity_threshold: float = 0.85
) -> Optional[str]:
"""
查找相似实体
Args:
entity_text: 实体文本
entity_type: 实体类型
similarity_threshold: 相似度阈值
Returns:
相似实体的ID，如果不存在返回None
"""
normalized_text = self.normalize_text(entity_text)
# 只在相同类型的实体中查找
for entity_id, entity_data in self.entity_registry.items():
if entity_data['type'] != entity_type:
continue
# 检查文本相似度
similarity = self.calculate_similarity(normalized_text, entity_data['canonical_name'])
if similarity >= similarity_threshold:
self.logger.info(f"发现相似实体: '{entity_text}' ~ '{entity_data['canonical_name']}' (相似度: {similarity:.2f})")
return entity_id
return None
def is_generic_concept(self, entity_text: str, entity_type: str) -> bool:
"""
检查是否为泛指概念（应该被过滤）
Args:
entity_text: 实体文本
entity_type: 实体类型
Returns:
True表示是泛指概念，应该过滤
"""
generic_patterns = [
'东北少数民族',
'本地土著民族',
'当地民族',
'少数民族',
'土著',
'东北民族',
'本地民族',
'地区民族',
]
normalized_text = self.normalize_text(entity_text)
# 检查是否匹配泛指模式
for pattern in generic_patterns:
if pattern in normalized_text:
self.logger.info(f"过滤泛指概念: '{entity_text}' (类型: {entity_type})")
return True
# 对于Ethnic_Group类型，必须是具体的民族名称
if entity_type == 'Ethnic_Group':
specific_ethnic_groups = [
'满族', '赫哲族', '鄂伦春族', '鄂温克族',
'达斡尔族', '朝鲜族', '蒙古族', '回族',
'汉族', '锡伯族', '柯尔克孜族'
]
# 如果不在具体民族列表中，可能是泛指概念
is_specific = any(group in normalized_text for group in specific_ethnic_groups)
if not is_specific:
self.logger.info(f"过滤非具体民族: '{entity_text}'")
return True
return False
def normalize_entity(
self,
raw_entity: Dict[str, Any],
similarity_threshold: float = 0.85,
filter_generic: bool = True
) -> str:
"""
规范化单个实体
Args:
raw_entity: 原始实体数据（从LLM返回）
similarity_threshold: 相似度阈值
filter_generic: 是否过滤泛指概念
Returns:
实体ID
"""
entity_text = raw_entity.get('text', '')
entity_type = raw_entity.get('type', '')
attributes = raw_entity.get('attributes', {})
if not entity_text or not entity_type:
self.logger.error(f"实体缺少text或type: {raw_entity}")
return None
# 验证实体类型是否在本体中定义
if entity_type not in self.ontology['entity_types']:
self.logger.warning(f"未知实体类型 '{entity_type}'，跳过: '{entity_text}'")
return None
# 过滤泛指概念
if filter_generic and self.is_generic_concept(entity_text, entity_type):
return None
# 标准化文本
normalized_text = self.normalize_text(entity_text)
# 检查是否已存在完全相同的实体
if normalized_text in self.text_to_id_map:
existing_id = self.text_to_id_map[normalized_text]
self.logger.info(f"实体已存在: '{entity_text}' -> {existing_id}")
return existing_id
# 查找相似实体
similar_id = self.find_similar_entity(entity_text, entity_type, similarity_threshold)
if similar_id:
# 合并到相似实体
self.logger.info(f"合并实体: '{entity_text}' -> {similar_id}")
self.text_to_id_map[normalized_text] = similar_id
return similar_id
# 生成新实体ID
entity_id = self.generate_entity_id(entity_type, entity_text)
# 获取标准名称
standard_name = attributes.get('name') or entity_text
# 保存到注册表
self.entity_registry[entity_id] = {
'canonical_name': normalized_text,
'display_name': standard_name,
'type': entity_type,
'attributes': attributes,
'source_projects': []
}
# 保存文本映射
self.text_to_id_map[normalized_text] = entity_id
self.logger.info(f"新建实体: {entity_id} - '{standard_name}' ({entity_type})")
return entity_id
def normalize_batch(
self,
extraction_results: List[Dict[str, Any]],
similarity_threshold: float = 0.85
) -> Dict[str, str]:
"""
批量规范化实体
Args:
extraction_results: LLM抽取结果列表
similarity_threshold: 相似度阈值
Returns:
文本到ID的映射字典（包含所有已处理的实体）
"""
self.logger.info(f"开始批量规范化，共{len(extraction_results)}个抽取结果")
for idx, result in enumerate(extraction_results):
if not result or 'entities' not in result:
continue
for entity in result['entities']:
self.normalize_entity(entity, similarity_threshold)
self.logger.info(f"批量规范化完成，生成{len(self.entity_registry)}个唯一实体")
# 返回完整的文本到ID映射（包含所有已处理的实体，不仅仅是本批次）
return self.text_to_id_map.copy()
def get_entity_nodes(self) -> List[Dict[str, Any]]:
"""
获取所有实体节点（用于生成CSV）
Returns:
节点列表
"""
nodes = []
for entity_id, entity_data in self.entity_registry.items():
node = {
'id': entity_id,
'label': entity_data['display_name'],
'type': entity_data['type'],
'properties': json.dumps(entity_data['attributes'], ensure_ascii=False)
}
nodes.append(node)
return nodes
def get_statistics(self) -> Dict[str, Any]:
"""
获取统计信息
Returns:
统计信息字典
"""
stats = {
'total_entities': len(self.entity_registry),
'entities_by_type': {},
'type_counters': self.type_counters.copy()
}
# 按类型统计
for entity_data in self.entity_registry.values():
entity_type = entity_data['type']
stats['entities_by_type'][entity_type] = stats['entities_by_type'].get(entity_type, 0) + 1
return stats
dofile/kg_project/src/data_processing/relationship_builder.py
+280
View File
@@ -0,0 +1,280 @@
# -*- coding: utf-8 -*-
"""
关系构建器
负责构建非遗项目与深层实体之间的关系
"""
import json
import logging
from typing import Dict, List, Any, Optional
from pathlib import Path
import yaml
class RelationshipBuilder:
"""关系构建器"""
def __init__(self, ontology_file: str):
"""
初始化关系构建器
Args:
ontology_file: 本体配置文件路径
"""
# 加载本体配置
with open(ontology_file, 'r', encoding='utf-8') as f:
self.ontology = yaml.safe_load(f)
self.logger = logging.getLogger(__name__)
# 关系注册表：用于去重
self.relationship_registry = set()
self.logger.info("RelationshipBuilder初始化完成")
def build_relationship(
self,
source_id: str,
target_id: str,
rel_type: str,
properties: Dict[str, Any] = None
) -> Optional[Dict[str, Any]]:
"""
构建单个关系
Args:
source_id: 源实体ID
target_id: 目标实体ID
rel_type: 关系类型
properties: 关系属性
Returns:
关系字典，如果验证失败返回None
"""
# 验证关系类型
if rel_type not in self.ontology['relationship_types']:
self.logger.warning(f"未知关系类型: {rel_type}")
return None
# 验证ID不为空
if not source_id or not target_id:
self.logger.error(f"关系ID不能为空: source={source_id}, target={target_id}")
return None
# 生成关系唯一键（用于去重）
rel_key = f"{source_id}-{target_id}-{rel_type}"
# 检查是否重复
if rel_key in self.relationship_registry:
self.logger.debug(f"关系已存在，跳过: {rel_key}")
return None
# 添加到注册表
self.relationship_registry.add(rel_key)
# 构建关系
relationship = {
'source': source_id,
'target': target_id,
'type': rel_type,
'properties': json.dumps(properties or {}, ensure_ascii=False)
}
return relationship
def build_relationships_from_extraction(
self,
project_id: str,
extraction_result: Dict[str, Any],
entity_id_map: Dict[str, str]
) -> List[Dict[str, Any]]:
"""
从抽取结果构建关系
Args:
project_id: 项目ID
extraction_result: LLM抽取结果
entity_id_map: 实体文本到ID的映射
Returns:
关系列表
"""
relationships = []
if not extraction_result or 'relationships' not in extraction_result:
return relationships
for rel in extraction_result['relationships']:
# 获取源实体（应该是项目ID）
source_entity = rel.get('source_entity', '')
# 验证源实体是否匹配当前项目
if source_entity != project_id:
self.logger.warning(f"关系源实体不匹配: 期望{project_id}, 实际{source_entity}")
# 如果不匹配，尝试修正
source_entity = project_id
# 获取目标实体文本
target_text = rel.get('target_entity', '')
# 查找目标实体ID
target_id = entity_id_map.get(target_text)
if not target_id:
self.logger.warning(f"未找到目标实体ID: {target_text}")
continue
# 获取关系类型
rel_type = rel.get('type', '')
rel_properties = rel.get('properties', {})
# 构建关系
relationship = self.build_relationship(
source_entity,
target_id,
rel_type,
rel_properties
)
if relationship:
relationships.append(relationship)
return relationships
def build_batch_relationships(
self,
projects_data: List[Dict[str, Any]],
extraction_results: List[Dict[str, Any]],
entity_id_map: Dict[str, str]
) -> List[Dict[str, Any]]:
"""
批量构建关系
Args:
projects_data: 项目数据列表
extraction_results: 抽取结果列表
entity_id_map: 实体文本到ID的映射
Returns:
关系列表
"""
self.logger.info(f"开始批量构建关系，共{len(projects_data)}个项目")
all_relationships = []
for idx, (project_data, extraction_result) in enumerate(zip(projects_data, extraction_results)):
if not extraction_result:
continue
project_id = project_data.get('project_id', '')
if not project_id:
self.logger.warning(f"项目{idx}缺少project_id")
continue
# 构建该项目的关系
relationships = self.build_relationships_from_extraction(
project_id,
extraction_result,
entity_id_map
)
all_relationships.extend(relationships)
self.logger.info(f"项目 {project_id} 构建了{len(relationships)}个关系")
self.logger.info(f"批量构建完成，共{len(all_relationships)}个关系")
return all_relationships
def validate_relationships(
self,
relationships: List[Dict[str, Any]],
node_ids: set
) -> Dict[str, Any]:
"""
验证关系完整性
Args:
relationships: 关系列表
node_ids: 节点ID集合
Returns:
验证报告
"""
report = {
'total_relationships': len(relationships),
'valid_relationships': 0,
'invalid_relationships': 0,
'broken_links': [],
'invalid_types': [],
'relationships_by_type': {}
}
for rel in relationships:
source_id = rel.get('source', '')
target_id = rel.get('target', '')
rel_type = rel.get('type', '')
# 统计关系类型
report['relationships_by_type'][rel_type] = \
report['relationships_by_type'].get(rel_type, 0) + 1
is_valid = True
# 验证关系类型
if rel_type not in self.ontology['relationship_types']:
report['invalid_types'].append({
'source': source_id,
'target': target_id,
'type': rel_type
})
is_valid = False
# 验证链接完整性
if source_id not in node_ids:
report['broken_links'].append({
'source': source_id,
'target': target_id,
'type': rel_type,
'issue': 'source_not_found'
})
is_valid = False
if target_id not in node_ids:
report['broken_links'].append({
'source': source_id,
'target': target_id,
'type': rel_type,
'issue': 'target_not_found'
})
is_valid = False
if is_valid:
report['valid_relationships'] += 1
else:
report['invalid_relationships'] += 1
return report
def get_statistics(self, relationships: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
获取关系统计信息
Args:
relationships: 关系列表
Returns:
统计信息
"""
stats = {
'total_relationships': len(relationships),
'relationships_by_type': {}
}
for rel in relationships:
rel_type = rel.get('type', 'Unknown')
stats['relationships_by_type'][rel_type] = \
stats['relationships_by_type'].get(rel_type, 0) + 1
return stats
dofile/kg_project/src/deep_extraction_pipeline.py
+406
View File
@@ -0,0 +1,406 @@
# -*- coding: utf-8 -*-
"""
深度文化实体抽取主流程
从Excel备注字段抽取深层文化实体并生成知识图谱CSV文件
"""
import asyncio
import pandas as pd
import json
import logging
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Any
# 添加模块路径
import sys
sys.path.append(str(Path(__file__).parent))
from knowledge_extraction.deep_entity_extractor import DeepEntityExtractor
from data_processing.entity_normalizer import EntityNormalizer
from data_processing.relationship_builder import RelationshipBuilder
class DeepExtractionPipeline:
"""深度文化实体抽取流程"""
def __init__(self, config_file: str):
"""
初始化流程
Args:
config_file: 配置文件路径
"""
self.config_file = config_file
# 设置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
# 初始化组件
self.extractor = DeepEntityExtractor(config_file)
self.normalizer = EntityNormalizer(
str(Path(config_file).parent / 'entity_ontology.yaml')
)
self.builder = RelationshipBuilder(
str(Path(config_file).parent / 'entity_ontology.yaml')
)
self.logger.info("DeepExtractionPipeline初始化完成")
def load_excel_data(self, excel_file: str, max_rows: int = None) -> List[Dict[str, str]]:
"""
加载Excel数据
Args:
excel_file: Excel文件路径
max_rows: 最大读取行数（用于测试）
Returns:
项目数据列表
"""
self.logger.info(f"读取Excel文件: {excel_file}")
df = pd.read_excel(excel_file)
# 限制行数
if max_rows:
df = df.head(max_rows)
self.logger.info(f"限制读取行数: {max_rows}")
# 检查必需列
required_columns = ['总序号', '项目名称', '备注']
missing_columns = [col for col in required_columns if col not in df.columns]
if missing_columns:
self.logger.error(f"Excel缺少必需列: {missing_columns}")
raise ValueError(f"缺少必需列: {missing_columns}")
# 构建项目数据列表
projects_data = []
for idx, row in df.iterrows():
seq_num = row.get('总序号', idx + 1)
project_name = row.get('项目名称', '')
remark_text = row.get('备注', '')
# 过滤空备注
if not remark_text or len(str(remark_text).strip()) < 10:
continue
projects_data.append({
'project_id': f"ICH-{int(seq_num)}",
'project_name': str(project_name),
'remark_text': str(remark_text),
'row_number': idx + 2 # Excel行号（1-based + header）
})
self.logger.info(f"加载了{len(projects_data)}个项目的数据")
return projects_data
async def run_extraction(
self,
excel_file: str,
output_prefix: str = "",
max_rows: int = None
) -> Dict[str, Any]:
"""
执行完整的抽取流程（支持增量保存）
Args:
excel_file: Excel文件路径
output_prefix: 输出文件前缀（用于测试）
max_rows: 最大读取行数（用于测试）
Returns:
处理结果统计
"""
self.logger.info("="*60)
self.logger.info("开始深度文化实体抽取流程（增量保存模式）")
self.logger.info("="*60)
start_time = datetime.now()
# 1. 加载数据
projects_data = self.load_excel_data(excel_file, max_rows)
if not projects_data:
self.logger.error("没有可处理的数据")
return {}
# 准备输出目录
output_dir = Path('output')
output_dir.mkdir(parents=True, exist_ok=True)
# 确定输出文件名
if output_prefix:
nodes_file = output_dir / f"{output_prefix}_nodes.csv"
rels_file = output_dir / f"{output_prefix}_rels.csv"
report_file = output_dir / f"{output_prefix}_report.md"
else:
nodes_file = output_dir / Path(self.extractor.config['output']['nodes_file']).name
rels_file = output_dir / Path(self.extractor.config['output']['relationships_file']).name
report_file = output_dir / Path(self.extractor.config['output']['report_file']).name
# 2. 定义增量保存回调函数
async def save_progress(batch_num, total_batches, extraction_results, processed_projects):
"""每批次完成后保存进度（仅保存节点，不构建关系）"""
try:
self.logger.info(f"\n[增量保存] 批次 {batch_num}/{total_batches} 开始保存...")
# 实体规范化（累积到实体注册表）
self.normalizer.normalize_batch(extraction_results)
# 生成节点（累积所有已处理的实体）
entity_nodes = self.normalizer.get_entity_nodes()
# 添加项目节点（仅当前批次）
project_nodes = [
{
'id': item['project_id'],
'label': item['project_name'],
'type': 'ICH_Project',
'properties': '{}'
}
for item in processed_projects
]
all_nodes = entity_nodes + project_nodes
# 保存节点
nodes_df = pd.DataFrame(all_nodes)
nodes_df.to_csv(nodes_file, index=False, encoding='utf-8-sig')
self.logger.info(f"[增量保存] 节点已保存: {nodes_file} ({len(all_nodes)}行)")
except Exception as e:
self.logger.error(f"[增量保存] 批次 {batch_num} 保存失败: {str(e)}", exc_info=True)
# 3. LLM批量抽取（带增量保存回调）
self.logger.info("\n步骤1: LLM批量抽取（增量保存模式）")
extraction_results = await self.extractor.batch_extract(
projects_data,
progress_callback=save_progress
)
# 4. 最终数据验证和报告
self.logger.info("\n步骤2: 最终数据验证")
# 重新规范化所有实体（确保一致性）
entity_id_map = self.normalizer.normalize_batch(extraction_results)
# 重新构建所有关系
relationships = self.builder.build_batch_relationships(
projects_data,
extraction_results,
entity_id_map
)
# 生成最终节点
entity_nodes = self.normalizer.get_entity_nodes()
project_nodes = [
{
'id': item['project_id'],
'label': item['project_name'],
'type': 'ICH_Project',
'properties': '{}'
}
for item in projects_data
]
all_nodes = entity_nodes + project_nodes
# 数据验证
node_ids = set(node['id'] for node in all_nodes)
validation_report = self.builder.validate_relationships(relationships, node_ids)
# 保存最终结果
self.logger.info("\n步骤3: 保存最终结果")
nodes_df = pd.DataFrame(all_nodes)
nodes_df.to_csv(nodes_file, index=False, encoding='utf-8-sig')
self.logger.info(f"最终节点已保存: {nodes_file} ({len(all_nodes)}行)")
rels_df = pd.DataFrame(relationships)
rels_df.to_csv(rels_file, index=False, encoding='utf-8-sig')
self.logger.info(f"最终关系已保存: {rels_file} ({len(relationships)}行)")
# 生成报告
self.logger.info("\n步骤4: 生成报告")
self.generate_report(
report_file,
projects_data,
all_nodes,
relationships,
validation_report,
start_time
)
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
# 返回统计信息
result_stats = {
'total_projects': len(projects_data),
'total_nodes': len(all_nodes),
'total_relationships': len(relationships),
'entity_nodes': len(entity_nodes),
'project_nodes': len(project_nodes),
'duration_seconds': duration,
'validation_report': validation_report
}
self.logger.info("\n"+"="*60)
self.logger.info(f"抽取完成！耗时: {duration:.2f}秒")
self.logger.info(f"项目数: {result_stats['total_projects']}")
self.logger.info(f"节点数: {result_stats['total_nodes']}")
self.logger.info(f"关系数: {result_stats['total_relationships']}")
self.logger.info("="*60)
return result_stats
def generate_report(
self,
report_file: Path,
projects_data: List[Dict[str, str]],
nodes: List[Dict[str, Any]],
relationships: List[Dict[str, Any]],
validation_report: Dict[str, Any],
start_time: datetime
):
"""
生成抽取报告
Args:
report_file: 报告文件路径
projects_data: 项目数据
nodes: 节点列表
relationships: 关系列表
validation_report: 验证报告
start_time: 开始时间
"""
report_lines = []
# 标题
report_lines.append("# 深度文化实体抽取报告\n")
report_lines.append(f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
report_lines.append(f"**耗时**: {(datetime.now() - start_time).total_seconds():.2f}秒\n")
report_lines.append("---\n\n")
# 1. 处理统计
report_lines.append("## 1. 处理统计\n\n")
report_lines.append(f"- **处理项目数**: {len(projects_data)}\n")
report_lines.append(f"- **总节点数**: {len(nodes)}\n")
report_lines.append(f"- **总关系数**: {len(relationships)}\n")
report_lines.append(f"- **实体节点数**: {len([n for n in nodes if n['type'] != 'ICH_Project'])}\n")
report_lines.append(f"- **项目节点数**: {len([n for n in nodes if n['type'] == 'ICH_Project'])}\n")
# 2. 节点类型分布
report_lines.append("\n## 2. 节点类型分布\n\n")
node_types = {}
for node in nodes:
node_type = node['type']
node_types[node_type] = node_types.get(node_type, 0) + 1
report_lines.append("| 节点类型 | 数量 | 占比 |\n")
report_lines.append("|---------|------|------|\n")
for node_type, count in sorted(node_types.items()):
percentage = (count / len(nodes) * 100) if len(nodes) > 0 else 0
report_lines.append(f"| {node_type} | {count} | {percentage:.1f}% |\n")
# 3. 关系类型分布
report_lines.append("\n## 3. 关系类型分布\n\n")
rel_types = {}
for rel in relationships:
rel_type = rel['type']
rel_types[rel_type] = rel_types.get(rel_type, 0) + 1
report_lines.append("| 关系类型 | 数量 | 占比 |\n")
report_lines.append("|---------|------|------|\n")
for rel_type, count in sorted(rel_types.items()):
percentage = (count / len(relationships) * 100) if len(relationships) > 0 else 0
report_lines.append(f"| {rel_type} | {count} | {percentage:.1f}% |\n")
# 4. 数据质量
report_lines.append("\n## 4. 数据质量\n\n")
report_lines.append(f"- **有效关系**: {validation_report['valid_relationships']}\n")
report_lines.append(f"- **无效关系**: {validation_report['invalid_relationships']}\n")
if validation_report['broken_links']:
report_lines.append(f"\n**断链警告**: {len(validation_report['broken_links'])}个\n")
report_lines.append("```json\n")
report_lines.append(json.dumps(validation_report['broken_links'][:10], ensure_ascii=False, indent=2))
if len(validation_report['broken_links']) > 10:
report_lines.append(f"\n... (还有{len(validation_report['broken_links'])-10}个)")
report_lines.append("\n```\n")
# 5. 项目详情（前5个）
report_lines.append("\n## 5. 项目抽取详情（前5个）\n\n")
for idx, (project_data, node) in enumerate(zip(projects_data[:5], nodes[:5])):
if node['type'] != 'ICH_Project':
continue
report_lines.append(f"### {idx+1}. {project_data['project_name']}\n\n")
report_lines.append(f"**项目ID**: {project_data['project_id']}\n\n")
report_lines.append(f"**备注**: {project_data['remark_text'][:200]}...\n\n")
# 查找相关关系
related_rels = [r for r in relationships if r['source'] == project_data['project_id']]
if related_rels:
report_lines.append(f"**关系数**: {len(related_rels)}\n\n")
report_lines.append("| 关系类型 | 目标实体 |\n")
report_lines.append("|---------|---------|\n")
for rel in related_rels[:10]:
target_node = next((n for n in nodes if n['id'] == rel['target']), None)
if target_node:
report_lines.append(f"| {rel['type']} | {target_node['label']} |\n")
if len(related_rels) > 10:
report_lines.append(f"| ... | 还有{len(related_rels)-10}个关系 |\n")
report_lines.append("\n")
# 6. 输出文件
report_lines.append("## 6. 输出文件\n\n")
report_lines.append(f"- **节点文件**: `{report_file.parent / (report_file.stem.replace('_report', '') + '_nodes.csv')}`\n")
report_lines.append(f"- **关系文件**: `{report_file.parent / (report_file.stem.replace('_report', '') + '_rels.csv')}`\n")
# 保存报告
with open(report_file, 'w', encoding='utf-8') as f:
f.writelines(report_lines)
self.logger.info(f"报告已保存: {report_file}")
async def main():
"""主函数"""
import sys
# 配置文件
config_file = r"E:\Project\2026_KG_ICH\dofile\kg_project\config\deep_extraction_config.yaml"
# 数据文件：从命令行参数获取，或使用默认完整数据文件
if len(sys.argv) > 1:
excel_file = sys.argv[1]
else:
excel_file = r"E:\Project\2026_KG_ICH\data\黑龙江国家级和省级非遗名单.xlsx"
# 创建流程
pipeline = DeepExtractionPipeline(config_file)
# 执行抽取
results = await pipeline.run_extraction(
excel_file=excel_file,
output_prefix="", # 不使用前缀（完整抽取）
max_rows=None # 读取所有行
)
print("\n" + "="*60)
print("抽取完成！")
print("="*60)
print(f"\n结果统计:")
print(json.dumps(results, indent=2, ensure_ascii=False))
if __name__ == '__main__':
asyncio.run(main())
dofile/kg_project/src/knowledge_extraction/__init__.py
View File
dofile/kg_project/src/knowledge_extraction/deep_entity_extractor.py
+420
View File
@@ -0,0 +1,420 @@
# -*- coding: utf-8 -*-
"""
深度文化实体抽取器
利用LLM从非遗项目备注中抽取深层文化实体和关系
"""
import json
import asyncio
import logging
from typing import Dict, List, Any, Optional
from pathlib import Path
import yaml
from langchain_deepseek import ChatDeepSeek
from langchain_core.messages import HumanMessage, SystemMessage
class DeepEntityExtractor:
"""深度文化实体抽取器"""
def __init__(self, config_file: str):
"""
初始化抽取器
Args:
config_file: 配置文件路径
"""
# 加载配置
with open(config_file, 'r', encoding='utf-8') as f:
self.config = yaml.safe_load(f)
# 设置日志
log_config = self.config.get('logging', {})
log_file = log_config.get('log_file', 'logs/deep_extraction.log')
Path(log_file).parent.mkdir(parents=True, exist_ok=True)
# 创建logger
self.logger = logging.getLogger(__name__)
self.logger.setLevel(getattr(logging, log_config.get('level', 'INFO')))
# 清除已有的handlers
self.logger.handlers.clear()
# 文件handler（实时刷新）
file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_handler.setLevel(getattr(logging, log_config.get('level', 'INFO')))
file_formatter = logging.Formatter(log_config.get('format', '%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
file_handler.setFormatter(file_formatter)
# 强制实时刷新
file_handler.flush = lambda: file_handler.stream.flush()
self.logger.addHandler(file_handler)
# 控制台handler
console_handler = logging.StreamHandler()
console_handler.setLevel(getattr(logging, log_config.get('level', 'INFO')))
console_formatter = logging.Formatter(log_config.get('format', '%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
console_handler.setFormatter(console_formatter)
self.logger.addHandler(console_handler)
# 加载实体本体配置
ontology_file = Path(config_file).parent / 'entity_ontology.yaml'
with open(ontology_file, 'r', encoding='utf-8') as f:
self.ontology = yaml.safe_load(f)
# 初始化LLM
self._init_llm()
# 构建提示词
self.system_prompt = self._build_system_prompt()
self.logger.info("DeepEntityExtractor初始化完成")
def _init_llm(self):
"""初始化LLM模型"""
llm_config = self.config['llm']
# 读取API密钥（从现有的api_keys.yaml）
api_keys_file = Path(__file__).parent.parent.parent / 'config' / 'api_keys.yaml'
with open(api_keys_file, 'r', encoding='utf-8') as f:
api_keys = yaml.safe_load(f)
# 适配API密钥格式
api_key = api_keys.get('deepseek_api_key', '')
if not api_key:
# 尝试嵌套格式
api_key = api_keys.get('deepseek', {}).get('api_key', '')
self.llm = ChatDeepSeek(
model=llm_config['model'],
temperature=llm_config['temperature'],
max_tokens=llm_config['max_tokens'],
api_key=api_key
)
self.logger.info(f"LLM初始化完成: {llm_config['model']}")
def _build_system_prompt(self) -> str:
"""构建系统提示词"""
entity_prompts = self.config.get('entity_type_prompts', {})
relationship_prompts = self.config.get('relationship_type_prompts', {})
prompt = """你是一位非物质文化遗产领域的专家，擅长从项目描述中识别深层文化实体和关系。
【任务】
请从非遗项目描述中识别以下9类深层文化实体和9种深层关系：
=== 实体类型 ===
"""
# 添加实体类型说明
for entity_type, description in entity_prompts.items():
prompt += f"{entity_type}：{description}\n"
prompt += "\n=== 关系类型 ===\n"
# 添加关系类型说明
for rel_type, description in relationship_prompts.items():
prompt += f"{rel_type}：{description}\n"
prompt += """
【输出要求】
1. 严格按JSON格式输出，不要包含任何其他文本
2. 只输出明确的、文本中提到的实体和关系，不要臆测
3. 实体text必须从原文中提取，不要自行改写
4. 同一个实体只识别一次，避免重复
5. 关系必须基于文本中的明确描述
6. 如果某类实体或关系不存在，相应数组为空
7. 确保JSON格式正确，可以被Python解析
【JSON输出格式】
```json
{
"entities": [
{
"text": "实体文本（如：鱼皮）",
"type": "实体类型（如：Material）",
"attributes": {
"name": "标准名称",
"category": "类别（如：动物材料）",
"description": "详细描述"
}
}
],
"relationships": [
{
"source_entity": "ICH-{项目ID}",
"target_entity": "目标实体文本",
"type": "关系类型",
"properties": {
"description": "关系描述",
"context": "上下文信息"
}
}
],
"summary": {
"total_entities": 0,
"total_relationships": 0
}
}
```
"""
return prompt
async def extract_from_remark(
self,
project_id: str,
project_name: str,
remark_text: str
) -> Optional[Dict[str, Any]]:
"""
从单条备注抽取实体和关系
Args:
project_id: 项目ID（如ICH-1）
project_name: 项目名称
remark_text: 备注文本
Returns:
抽取结果（包含entities, relationships, summary）
"""
if not remark_text or len(remark_text.strip()) < 10:
self.logger.warning(f"项目 {project_id} 备注文本过短，跳过抽取")
return None
# 构建用户提示词
user_prompt = f"""【项目信息】
项目名称：{project_name}
项目ID：{project_id}
【描述文本】
{remark_text}
请从上述描述中识别深层文化实体和关系，严格按JSON格式输出。"""
try:
# 调用LLM
messages = [
SystemMessage(content=self.system_prompt),
HumanMessage(content=user_prompt)
]
max_retries = self.config['llm']['max_retries']
retry_delay = self.config['llm']['retry_delay']
request_timeout = self.config['llm'].get('request_timeout', 120)
for attempt in range(max_retries):
try:
self.logger.info(f"项目 {project_id} 开始LLM调用（尝试{attempt+1}/{max_retries}）")
# 添加超时控制
response = await asyncio.wait_for(
self.llm.ainvoke(messages),
timeout=request_timeout
)
result_text = response.content
self.logger.info(f"项目 {project_id} LLM调用成功，开始提取JSON")
# 提取JSON
extraction_result = self._extract_json(result_text)
if extraction_result:
# 验证结果
if self._validate_extraction(extraction_result):
self.logger.info(f"项目 {project_id} 抽取成功：{extraction_result['summary']['total_entities']}个实体，{extraction_result['summary']['total_relationships']}个关系")
return extraction_result
else:
self.logger.warning(f"项目 {project_id} 抽取结果验证失败")
else:
self.logger.warning(f"项目 {project_id} JSON提取失败")
except asyncio.TimeoutError:
self.logger.error(f"项目 {project_id} LLM调用超时（{request_timeout}秒）（尝试{attempt+1}/{max_retries}）")
if attempt < max_retries - 1:
await asyncio.sleep(retry_delay)
else:
self.logger.error(f"项目 {project_id} 达到最大重试次数，放弃抽取")
raise
except Exception as e:
self.logger.error(f"项目 {project_id} LLM调用失败（尝试{attempt+1}/{max_retries}）: {type(e).__name__}: {str(e)}")
if attempt < max_retries - 1:
await asyncio.sleep(retry_delay)
else:
raise
except Exception as e:
self.logger.error(f"项目 {project_id} 抽取失败: {str(e)}", exc_info=True)
return None
def _extract_json(self, text: str) -> Optional[Dict[str, Any]]:
"""
从文本中提取JSON
Args:
text: LLM返回的文本
Returns:
解析后的JSON对象，失败返回None
"""
# 尝试直接解析
try:
return json.loads(text)
except json.JSONDecodeError:
pass
# 尝试提取JSON块
import re
json_pattern = r'```json\s*(.*?)\s*```'
match = re.search(json_pattern, text, re.DOTALL)
if match:
try:
return json.loads(match.group(1))
except json.JSONDecodeError:
pass
# 尝试提取花括号内容
brace_pattern = r'\{.*\}'
match = re.search(brace_pattern, text, re.DOTALL)
if match:
try:
return json.loads(match.group(0))
except json.JSONDecodeError:
pass
self.logger.error(f"无法从文本中提取有效JSON: {text[:200]}...")
return None
def _validate_extraction(self, result: Dict[str, Any]) -> bool:
"""
验证抽取结果
Args:
result: 抽取结果
Returns:
是否有效
"""
# 检查必需字段
required_fields = ['entities', 'relationships', 'summary']
for field in required_fields:
if field not in result:
self.logger.error(f"缺少必需字段: {field}")
return False
# 检查entities格式
if not isinstance(result['entities'], list):
self.logger.error("entities必须是列表")
return False
for entity in result['entities']:
if not all(k in entity for k in ['text', 'type', 'attributes']):
self.logger.error(f"实体缺少必需字段: {entity}")
return False
# 检查实体类型是否合法
entity_type = entity['type']
if entity_type not in self.ontology['entity_types']:
self.logger.warning(f"未知实体类型: {entity_type}")
# 检查relationships格式
if not isinstance(result['relationships'], list):
self.logger.error("relationships必须是列表")
return False
for rel in result['relationships']:
if not all(k in rel for k in ['source_entity', 'target_entity', 'type', 'properties']):
self.logger.error(f"关系缺少必需字段: {rel}")
return False
# 检查关系类型是否合法
rel_type = rel['type']
if rel_type not in self.ontology['relationship_types']:
self.logger.warning(f"未知关系类型: {rel_type}")
return True
async def batch_extract(
self,
projects_data: List[Dict[str, str]],
batch_size: int = None,
progress_callback=None
) -> List[Optional[Dict[str, Any]]]:
"""
批量抽取实体和关系
Args:
projects_data: 项目数据列表，每项包含project_id, project_name, remark_text
batch_size: 批次大小（从配置文件读取）
progress_callback: 进度回调函数，每批次完成后调用
Returns:
抽取结果列表
"""
if batch_size is None:
batch_size = self.config['batch_processing']['batch_size']
batch_timeout = self.config['llm'].get('batch_timeout', 300)
incremental_save = self.config['batch_processing'].get('incremental_save', False)
results = []
total = len(projects_data)
self.logger.info(f"="*60)
self.logger.info(f"开始批量抽取，共{total}个项目，批次大小{batch_size}")
self.logger.info(f"增量保存: {'启用' if incremental_save else '禁用'}")
self.logger.info(f"请求超时: {self.config['llm'].get('request_timeout', 120)}秒")
self.logger.info(f"批次超时: {batch_timeout}秒")
self.logger.info(f"="*60)
for i in range(0, total, batch_size):
batch = projects_data[i:i+batch_size]
batch_num = i // batch_size + 1
total_batches = (total + batch_size - 1) // batch_size
self.logger.info(f"-"*60)
self.logger.info(f"开始处理批次 {batch_num}/{total_batches}，包含{len(batch)}个项目")
for item in batch:
self.logger.info(f" - {item['project_id']}: {item['project_name']}")
# 并发处理当前批次
batch_tasks = [
self.extract_from_remark(
item['project_id'],
item['project_name'],
item['remark_text']
)
for item in batch
]
try:
# 添加批次级别的超时控制
batch_results = await asyncio.wait_for(
asyncio.gather(*batch_tasks, return_exceptions=True),
timeout=batch_timeout
)
results.extend(batch_results)
# 统计本批次结果
success_count = sum(1 for r in batch_results if r is not None and not isinstance(r, Exception))
error_count = len(batch_results) - success_count
self.logger.info(f"批次 {batch_num}/{total_batches} 完成 - 成功: {success_count}, 失败: {error_count}")
# 调用进度回调（用于增量保存）
if progress_callback:
await progress_callback(batch_num, total_batches, results, projects_data[:i+len(batch)])
except asyncio.TimeoutError:
self.logger.error(f"批次 {batch_num}/{total_batches} 超时（{batch_timeout}秒），部分请求失败")
# 将未完成的结果标记为None
for item in batch[len(results):]:
results.append(None)
self.logger.info(f"="*60)
self.logger.info(f"批量抽取完成，共处理{len(results)}个项目")
self.logger.info(f"="*60)
return results
dofile/kg_project/src/knowledge_extraction/llm_extractor.py
+486
View File
@@ -0,0 +1,486 @@
"""
DeepSeek实体识别模块
使用DeepSeek API进行非遗知识抽取
"""
import yaml
import asyncio
import json
from pathlib import Path
from typing import Dict, List, Optional, Any
import logging
from datetime import datetime
try:
from langchain_deepseek import ChatDeepSeek
from langchain_core.messages import HumanMessage, SystemMessage
except ImportError:
print("错误: 请先安装依赖包")
print("运行: pip install langchain-deepseek langchain-core")
raise
class DeepSeekExtractor:
"""基于DeepSeek的知识抽取器"""
def __init__(self, config_file: str = 'config/ich_config.yaml'):
"""
初始化抽取器
Args:
config_file: 配置文件路径
"""
# 加载配置
self.config = self._load_config(config_file)
self.api_key = self._load_api_key()
# 初始化模型
self.model = ChatDeepSeek(
model=self.config['api']['model'],
api_key=self.api_key,
temperature=self.config['api']['temperature'],
max_tokens=self.config['api']['max_tokens']
)
# 设置日志
self.logger = self._setup_logger()
# 加载本体配置
self.entity_types = self.config['extraction']['entity_types']
self.relation_types = self.config['extraction']['relation_types']
def _setup_logger(self):
"""设置日志"""
log_dir = Path(self.config['paths']['logs_dir'])
log_dir.mkdir(parents=True, exist_ok=True)
log_file = log_dir / f"extraction_{datetime.now().strftime('%Y%m%d')}.log"
logging.basicConfig(
level=getattr(logging, self.config['logging']['level']),
format=self.config['logging']['format'],
handlers=[
logging.FileHandler(log_file, encoding='utf-8'),
logging.StreamHandler()
]
)
return logging.getLogger(__name__)
def _load_config(self, config_file: str) -> Dict:
"""加载配置"""
config_path = Path(config_file)
if not config_path.exists():
raise FileNotFoundError(f"配置文件不存在: {config_file}")
with open(config_path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
def _load_api_key(self) -> str:
"""加载API密钥"""
api_config_file = Path(self.config['api']['config_file'])
if not api_config_file.exists():
raise FileNotFoundError(f"API配置文件不存在: {api_config_file}")
with open(api_config_file, 'r', encoding='utf-8') as f:
api_config = yaml.safe_load(f)
return api_config['deepseek_api_key']
def extract_json_from_response(self, response_text: str) -> str:
"""
从API响应中提取JSON内容
Args:
response_text: API响应文本
Returns:
str: 提取的JSON字符串
"""
# 如果响应包含markdown代码块，提取其中的JSON
if '```json' in response_text:
start = response_text.find('```json') + 7
end = response_text.find('```', start)
if start > 6 and end > start:
return response_text[start:end].strip()
elif '```' in response_text:
start = response_text.find('```') + 3
end = response_text.find('```', start)
if start > 2 and end > start:
content = response_text[start:end].strip()
if not content.startswith('json'):
return content
return content[5:].strip() if content.startswith('json') else content.strip()
# 否则直接返回原始文本
return response_text.strip()
async def extract_entities_from_text(
self,
text: str,
project_name: str = "",
max_retries: int = 3
) -> Optional[Dict[str, Any]]:
"""
从文本中抽取实体
Args:
text: 输入文本
project_name: 项目名称（可选）
max_retries: 最大重试次数
Returns:
Dict: 抽取结果
"""
# 构建提示词
entity_types_str = "、".join(self.entity_types)
prompt = f"""
你是一位非物质文化遗产领域的专家。请从以下文本中识别非遗相关实体，并提取属性。
项目名称：{project_name}
文本内容：
{text}
请识别以下类型的实体：
{entity_types_str}
对于每个实体，请提取以下信息：
1. 实体文本
2. 实体类型
3. 相关属性（如：民族、地点、时间、技艺特点等）
输出格式（JSON）：
{{
"entities": [
{{
"text": "实体文本",
"type": "实体类型",
"attributes": {{
"ethnic_group": "民族（如果适用）",
"location": "地点（如果适用）",
"time_period": "时期（如果适用）",
"skill_feature": "技艺特点（如果适用）",
"cultural_value": "文化价值（如果适用）"
}}
}}
],
"relationships": [
{{
"from": "实体1",
"to": "实体2",
"type": "关系类型",
"description": "关系描述"
}}
]
}}
请确保输出是有效的JSON格式。
"""
# 调用API
for attempt in range(max_retries):
try:
self.logger.info(f"开始抽取实体 (尝试 {attempt + 1}/{max_retries})")
messages = [HumanMessage(content=prompt)]
response = await self.model.ainvoke(messages)
result_text = response.content
# 提取JSON
json_text = self.extract_json_from_response(result_text)
# 解析JSON
try:
result = json.loads(json_text)
# 添加元数据
result['metadata'] = {
'project_name': project_name,
'extraction_time': datetime.now().isoformat(),
'text_length': len(text),
'model': self.config['api']['model']
}
self.logger.info(f"成功抽取 {len(result.get('entities', []))} 个实体")
return result
except json.JSONDecodeError as e:
self.logger.warning(f"JSON解析失败: {str(e)}")
self.logger.debug(f"响应文本: {result_text[:500]}")
if attempt == max_retries - 1:
raise
except Exception as e:
self.logger.error(f"API调用失败 (尝试 {attempt + 1}/{max_retries}): {str(e)}")
if attempt == max_retries - 1:
return None
# 等待后重试
await asyncio.sleep(self.config['processing']['retry_delay'])
return None
async def extract_relationships(
self,
entities: List[Dict],
context: str = "",
max_retries: int = 3
) -> Optional[List[Dict[str, Any]]]:
"""
抽取实体间的关系
Args:
entities: 实体列表
context: 上下文文本
max_retries: 最大重试次数
Returns:
List[Dict]: 关系列表
"""
if not entities:
return []
# 构建实体描述
entity_descriptions = []
for entity in entities:
desc = f"- {entity.get('text', '')} ({entity.get('type', '')})"
if entity.get('attributes'):
attrs = ", ".join([f"{k}={v}" for k, v in entity['attributes'].items() if v])
if attrs:
desc += f" [{attrs}]"
entity_descriptions.append(desc)
entities_str = "\n".join(entity_descriptions)
relations_str = "、".join(self.relation_types)
prompt = f"""
基于以下实体和上下文，识别实体间的关系：
实体列表：
{entities_str}
上下文：
{context}
可能的关系类型：
{relations_str}
对于每个关系，请提供：
1. 头实体（from）
2. 尾实体（to）
3. 关系类型
4. 关系描述
输出格式（JSON）：
{{
"relationships": [
{{
"from": "实体1文本",
"to": "实体2文本",
"type": "关系类型",
"description": "关系描述",
"confidence": 0.9
}}
]
}}
请确保输出是有效的JSON格式。
"""
# 调用API
for attempt in range(max_retries):
try:
messages = [HumanMessage(content=prompt)]
response = await self.model.ainvoke(messages)
result_text = response.content
# 提取JSON
json_text = self.extract_json_from_response(result_text)
# 解析JSON
try:
result = json.loads(json_text)
relationships = result.get('relationships', [])
self.logger.info(f"成功抽取 {len(relationships)} 个关系")
return relationships
except json.JSONDecodeError as e:
self.logger.warning(f"JSON解析失败: {str(e)}")
if attempt == max_retries - 1:
raise
except Exception as e:
self.logger.error(f"API调用失败 (尝试 {attempt + 1}/{max_retries}): {str(e)}")
if attempt == max_retries - 1:
return None
await asyncio.sleep(self.config['processing']['retry_delay'])
return None
async def enrich_project_entity(
self,
project_data: Dict[str, Any],
max_retries: int = 3
) -> Optional[Dict[str, Any]]:
"""
丰富非遗项目实体信息
Args:
project_data: 项目数据
max_retries: 最大重试次数
Returns:
Dict: 丰富后的实体信息
"""
project_name = project_data.get('properties', {}).get('name', '')
description = project_data.get('properties', {}).get('description', '')
prompt = f"""
你是一位非物质文化遗产领域的专家。请分析以下非遗项目，提取和丰富实体信息。
项目名称：{project_name}
项目描述：
{description}
请提取以下信息：
1. **民族特色**：是否与特定少数民族相关？（满族、赫哲族、鄂伦春族等）
2. **地域特征**：体现哪些黑龙江地域特征？（寒地、冰雪、森林、江河等）
3. **技艺特点**：核心技艺特点是什么？
4. **文化价值**：有哪些重要的文化价值？
5. **传承方式**：如何传承？（家族传承、师徒制度、口传心授等）
6. **濒危状况**：是否濒危？原因是什么？
7. **保护措施**：有哪些保护措施？
输出格式（JSON）：
{{
"ethnic_features": ["民族1", "民族2"],
"regional_characteristics": ["特征1", "特征2"],
"skill_features": ["技艺特点1", "技艺特点2"],
"cultural_values": ["价值1", "价值2"],
"transmission_methods": ["方式1", "方式2"],
"endangerment_status": "濒危状况描述",
"protection_measures": ["措施1", "措施2"],
"summary": "项目总结（50字以内）"
}}
请确保输出是有效的JSON格式。
"""
# 调用API
for attempt in range(max_retries):
try:
messages = [HumanMessage(content=prompt)]
response = await self.model.ainvoke(messages)
result_text = response.content
# 提取JSON
json_text = self.extract_json_from_response(result_text)
# 解析JSON
try:
result = json.loads(json_text)
self.logger.info(f"成功丰富项目实体信息: {project_name}")
return result
except json.JSONDecodeError as e:
self.logger.warning(f"JSON解析失败: {str(e)}")
if attempt == max_retries - 1:
raise
except Exception as e:
self.logger.error(f"API调用失败 (尝试 {attempt + 1}/{max_retries}): {str(e)}")
if attempt == max_retries - 1:
return None
await asyncio.sleep(self.config['processing']['retry_delay'])
return None
async def batch_extract(
self,
items: List[Dict[str, Any]],
concurrent: int = 5
) -> List[Optional[Dict[str, Any]]]:
"""
批量抽取
Args:
items: 待处理项目列表
concurrent: 并发数
Returns:
List[Dict]: 抽取结果列表
"""
results = []
batch_size = concurrent
for i in range(0, len(items), batch_size):
batch = items[i:i + batch_size]
self.logger.info(f"处理批次 {i//batch_size + 1}: {len(batch)} 个项目")
# 并发处理
tasks = []
for item in batch:
project_name = item.get('properties', {}).get('name', '')
description = item.get('properties', {}).get('description', '')
if description:
task = self.enrich_project_entity(item)
tasks.append(task)
else:
tasks.append(asyncio.sleep(0)) # 占位任务
batch_results = await asyncio.gather(*tasks, return_exceptions=True)
results.extend(batch_results)
# 显示进度
completed = min(i + batch_size, len(items))
self.logger.info(f"进度: {completed}/{len(items)}")
return results
def main():
"""主函数 - 测试实体抽取"""
import sys
# 添加项目根目录到路径
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
# 创建抽取器
extractor = DeepSeekExtractor('config/ich_config.yaml')
# 测试文本
test_text = """
桦树皮制作技艺是鄂伦春族的传统手工艺，利用桦树皮制作各种生活用品。
传承人莫桂树2008年被评为国家级传承人。这项技艺体现了鄂伦春族对自然资源的
巧妙利用，具有鲜明的渔猎文化特色。
"""
# 测试实体抽取
print("="*60)
print("测试实体抽取")
print("="*60)
async def test():
result = await extractor.extract_entities_from_text(
text=test_text,
project_name="桦树皮制作技艺"
)
if result:
print("\n抽取结果:")
print(json.dumps(result, indent=2, ensure_ascii=False))
else:
print("抽取失败")
asyncio.run(test())
if __name__ == '__main__':
main()
dofile/kg_project/src/main.py
+276
View File
@@ -0,0 +1,276 @@
"""
黑龙江省非物质文化遗产知识图谱构建 - 主处理脚本
"""
import json
import asyncio
import yaml
from pathlib import Path
from typing import Dict, List, Any
import logging
from datetime import datetime
import sys
# 添加项目路径
sys.path.insert(0, str(Path(__file__).parent))
from knowledge_extraction.llm_extractor import DeepSeekExtractor
class ICKnowledgeGraphBuilder:
"""非遗知识图谱构建器"""
def __init__(self, config_file: str = 'config/ich_config.yaml'):
"""初始化构建器"""
# 加载配置
with open(config_file, 'r', encoding='utf-8') as f:
self.config = yaml.safe_load(f)
# 设置路径
self.project_root = Path(__file__).parent.parent
self.data_dir = self.project_root / self.config['paths']['data_dir']
self.output_dir = self.project_root / self.config['paths']['output_dir']
self.logs_dir = self.project_root / self.config['paths']['logs_dir']
# 创建目录
self.output_dir.mkdir(parents=True, exist_ok=True)
self.logs_dir.mkdir(parents=True, exist_ok=True)
# 设置日志
self.logger = self._setup_logger()
# 初始化DeepSeek抽取器
self.extractor = DeepSeekExtractor(config_file)
# 加载进度
self.progress_file = self.output_dir / 'progress.json'
self.progress = self._load_progress()
def _setup_logger(self):
"""设置日志"""
log_file = self.logs_dir / f"build_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
logging.basicConfig(
level=getattr(logging, self.config['logging']['level']),
format=self.config['logging']['format'],
handlers=[
logging.FileHandler(log_file, encoding='utf-8'),
logging.StreamHandler()
]
)
return logging.getLogger(__name__)
def _load_progress(self) -> Dict:
"""加载进度"""
if self.progress_file.exists():
with open(self.progress_file, 'r', encoding='utf-8') as f:
return json.load(f)
return {
'total': 0,
'processed': [],
'successful': [],
'failed': [],
'last_update': None
}
def _save_progress(self):
"""保存进度"""
self.progress['last_update'] = datetime.now().isoformat()
with open(self.progress_file, 'w', encoding='utf-8') as f:
json.dump(self.progress, f, indent=2, ensure_ascii=False)
def load_nodes(self) -> List[Dict[str, Any]]:
"""加载节点数据"""
nodes_file = self.output_dir / 'nodes.json'
if not nodes_file.exists():
self.logger.error(f"节点文件不存在: {nodes_file}")
return []
with open(nodes_file, 'r', encoding='utf-8') as f:
nodes = json.load(f)
self.logger.info(f"加载了 {len(nodes)} 个节点")
self.progress['total'] = len(nodes)
return nodes
async def enrich_single_node(self, node: Dict[str, Any]) -> Dict[str, Any]:
"""丰富单个节点"""
node_id = node.get('id')
project_name = node.get('properties', {}).get('name', '')
self.logger.info(f"开始处理节点: {node_id} - {project_name}")
try:
# 调用DeepSeek丰富信息
enriched_data = await self.extractor.enrich_project_entity(node)
if enriched_data:
# 合并到原节点
node['properties']['enriched_data'] = enriched_data
node['properties']['enriched_at'] = datetime.now().isoformat()
self.logger.info(f"成功丰富节点: {node_id}")
return {
'node_id': node_id,
'status': 'success',
'enriched_data': enriched_data
}
else:
self.logger.warning(f"丰富节点失败（无返回数据）: {node_id}")
return {
'node_id': node_id,
'status': 'failed',
'error': 'No data returned'
}
except Exception as e:
self.logger.error(f"丰富节点失败: {node_id} - {str(e)}")
return {
'node_id': node_id,
'status': 'failed',
'error': str(e)
}
async def batch_enrich_nodes(
self,
nodes: List[Dict[str, Any]],
max_nodes: int = None,
concurrent: int = 5
):
"""批量丰富节点"""
# 过滤已处理的节点
pending_nodes = [
node for node in nodes
if node.get('id') not in self.progress['processed']
]
# 限制处理数量（用于测试）
if max_nodes:
pending_nodes = pending_nodes[:max_nodes]
self.logger.info(f"开始批量处理: {len(pending_nodes)} 个节点待处理")
# 分批处理
batch_size = concurrent
for i in range(0, len(pending_nodes), batch_size):
batch = pending_nodes[i:i + batch_size]
batch_num = i // batch_size + 1
total_batches = (len(pending_nodes) + batch_size - 1) // batch_size
self.logger.info(f"处理批次 {batch_num}/{total_batches}: {len(batch)} 个节点")
# 并发处理
tasks = [self.enrich_single_node(node) for node in batch]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 更新进度
for result in results:
if isinstance(result, Exception):
self.logger.error(f"处理异常: {str(result)}")
continue
node_id = result.get('node_id')
status = result.get('status')
if status == 'success':
self.progress['successful'].append(node_id)
else:
self.progress['failed'].append(node_id)
self.progress['processed'].append(node_id)
# 保存进度
self._save_progress()
# 显示进度
self._print_progress()
def _print_progress(self):
"""打印进度"""
total = self.progress['total']
processed = len(self.progress['processed'])
successful = len(self.progress['successful'])
failed = len(self.progress['failed'])
print("\n" + "="*60)
print("处理进度")
print("="*60)
print(f"总节点数: {total}")
print(f"已处理: {processed} ({processed/total*100:.1f}%)")
print(f"成功: {successful}")
print(f"失败: {failed}")
print(f"成功率: {successful/processed*100:.1f}%" if processed > 0 else "成功率: N/A")
print(f"最后更新: {self.progress['last_update']}")
print("="*60 + "\n")
def save_enriched_nodes(self, nodes: List[Dict[str, Any]]):
"""保存丰富后的节点"""
output_file = self.output_dir / 'nodes_enriched.json'
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(nodes, f, indent=2, ensure_ascii=False)
self.logger.info(f"丰富后的节点已保存到: {output_file}")
# 统计
enriched_count = sum(
1 for node in nodes
if 'enriched_data' in node.get('properties', {})
)
print(f"\n统计信息:")
print(f"总节点数: {len(nodes)}")
print(f"已丰富节点数: {enriched_count}")
print(f"丰富率: {enriched_count/len(nodes)*100:.1f}%")
def main():
"""主函数"""
import argparse
parser = argparse.ArgumentParser(description='黑龙江非遗知识图谱构建')
parser.add_argument('--config', default='config/ich_config.yaml', help='配置文件路径')
parser.add_argument('--max-nodes', type=int, help='最大处理节点数（用于测试）')
parser.add_argument('--status', action='store_true', help='查看进度')
parser.add_argument('--resume', action='store_true', help='断点续传')
args = parser.parse_args()
# 创建构建器
builder = ICKnowledgeGraphBuilder(args.config)
if args.status:
# 显示进度
builder._print_progress()
return
# 加载节点数据
print("加载节点数据...")
nodes = builder.load_nodes()
if not nodes:
print("没有找到节点数据")
return
# 批量处理
print("开始批量处理节点...")
asyncio.run(builder.batch_enrich_nodes(
nodes,
max_nodes=args.max_nodes,
concurrent=builder.config['processing']['concurrent_requests']
))
# 加载并保存丰富后的节点
print("保存处理结果...")
nodes_file = builder.output_dir / 'nodes.json'
with open(nodes_file, 'r', encoding='utf-8') as f:
all_nodes = json.load(f)
builder.save_enriched_nodes(all_nodes)
print("\n处理完成！")
builder._print_progress()
if __name__ == '__main__':
main()
dofile/kg_project/src/validation/__init__.py
View File