Files
jerryandClaude Fable 5 2f82c4a8dc P4/M5: agent/legacy-rag 源码迁移(449 模块 191K 行)+ 影子 7911/7912 冒烟
- 按生产 7900 容器黄金清单迁移(compile.manifest 449 条目,0 缺失)
- 入口双态:api.app(生产同源 3 路由)/ api.main(裁剪 11 路由族 91 路径)
- 修复:advanced/__init__ 空文件(包标记误覆盖,已从主项目恢复)
- 修复:api/rag_routes.py 跨环境 site-packages 注入改 LEGACY_GF_SITE_PACKAGES 门控
- 修复:scripts/utils 缺失(动态加载 reasoning_engine)
- 影子验证:kg/stats 数据连通(6.2s,与 7900 同源 normalized 后端)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-31 14:20:41 +08:00

265 lines
9.8 KiB
Python
Executable File

#!/usr/bin/env python3
"""
规则验证层 - 验证LLM提取结果的准确性
原则:实事求是,过滤编造,标注不确定
"""
import json
import re
from typing import Dict, List, Set, Tuple
class ExtractionValidator:
"""LLM提取结果验证器"""
def __init__(self, gene_list_path: str = None):
"""
初始化验证器
:param gene_list_path: 附表基因列表路径,如果有的话
"""
self.known_genes: Set[str] = set()
self.known_groups: Set[str] = {
'Group_I', 'Group_II', 'Group_III', 'Group_IV',
'Group_V', 'Group_VI', 'Group_VII', 'Group_VIII'
}
self.known_domains: Set[str] = {'SBP', 'ANK'}
if gene_list_path:
self._load_gene_list(gene_list_path)
def _load_gene_list(self, path: str):
"""加载附表中的基因列表作为ground truth"""
try:
with open(path, 'r') as f:
data = json.load(f)
for gene in data:
symbol = gene.get('symbol')
if symbol:
self.known_genes.add(symbol)
print(f"✓ 加载附表基因列表: {len(self.known_genes)} 个基因")
except Exception as e:
print(f"⚠ 无法加载基因列表: {e}")
def extract_genes_from_text(self, text: str) -> Set[str]:
"""从原文中提取所有基因名(正则)"""
# 匹配 FtSPL1, FtSPL2, ... FtSPL99
pattern = r'FtSPL\d+'
return set(re.findall(pattern, text))
def validate_entity(self, entity: Dict, source_text: str = None) -> Tuple[bool, str]:
"""
验证单个实体
:return: (是否有效, 原因)
"""
entity_id = entity.get('id', '')
entity_type = entity.get('type', '')
# 1. 验证基因实体
if entity_type == 'gene':
# 检查格式
if not re.match(r'^FtSPL\d+$', entity_id):
return False, f"基因ID格式错误: {entity_id}"
# 如果有附表,检查是否在附表中
if self.known_genes and entity_id not in self.known_genes:
return False, f"基因不在附表中: {entity_id}"
# 如果有原文,检查是否在原文中
if source_text:
text_genes = self.extract_genes_from_text(source_text)
if entity_id not in text_genes:
return False, f"基因不在原文中: {entity_id}"
return True, "有效"
# 2. 验证组实体
elif entity_type == 'gene_subfamily':
if entity_id not in self.known_groups:
return False, f"未知的组: {entity_id}"
return True, "有效"
# 3. 验证domain实体
elif entity_type == 'protein_domain':
if entity_id not in self.known_domains:
return False, f"未知的domain: {entity_id}"
return True, "有效"
# 其他类型暂不验证
return True, "未验证类型"
def validate_relation(self, relation: Dict, valid_entities: Set[str]) -> Tuple[bool, str]:
"""
验证关系
:return: (是否有效, 原因)
"""
source = relation.get('source', '')
target = relation.get('target', '')
rel_type = relation.get('type', '')
# 1. 检查源实体是否存在
if source not in valid_entities:
return False, f"源实体无效或不存在: {source}"
# 2. 检查目标实体是否存在
if target not in valid_entities:
return False, f"目标实体无效或不存在: {target}"
# 3. 检查关系类型是否合理
valid_relations = {
'classified_into', 'member_of', 'contains', 'expressed_in',
'upregulated_in', 'downregulated_in', 'involved_in', 'belongs_to'
}
if rel_type not in valid_relations:
return False, f"未知的关系类型: {rel_type}"
return True, "有效"
def validate_extraction(self, extraction_result: Dict, source_text: str = None) -> Dict:
"""
验证完整的提取结果
"""
validated = {
"metadata": {
"original_entities": len(extraction_result.get('entities', [])),
"original_relations": len(extraction_result.get('relations', [])),
"validation_time": "2026-03-30"
},
"entities": {
"valid": [],
"invalid": []
},
"relations": {
"valid": [],
"invalid": []
},
"statistics": {}
}
# 验证实体
valid_entity_ids = set()
for entity in extraction_result.get('entities', []):
is_valid, reason = self.validate_entity(entity, source_text)
entity['_validation'] = {
"is_valid": is_valid,
"reason": reason
}
if is_valid:
validated["entities"]["valid"].append(entity)
valid_entity_ids.add(entity.get('id'))
else:
validated["entities"]["invalid"].append(entity)
# 验证关系(只验证指向有效实体的关系)
for relation in extraction_result.get('relations', []):
is_valid, reason = self.validate_relation(relation, valid_entity_ids)
relation['_validation'] = {
"is_valid": is_valid,
"reason": reason
}
if is_valid:
validated["relations"]["valid"].append(relation)
else:
validated["relations"]["invalid"].append(relation)
# 统计
validated["statistics"] = {
"entity_precision": len(validated["entities"]["valid"]) / validated["metadata"]["original_entities"]
if validated["metadata"]["original_entities"] > 0 else 0,
"relation_precision": len(validated["relations"]["valid"]) / validated["metadata"]["original_relations"]
if validated["metadata"]["original_relations"] > 0 else 0,
"invalid_entities": len(validated["entities"]["invalid"]),
"invalid_relations": len(validated["relations"]["invalid"])
}
return validated
def generate_clean_output(self, validated: Dict) -> Dict:
"""
生成清理后的输出(只包含验证通过的内容)
"""
return {
"metadata": {
"extraction_method": "LLM + Rule Validation",
"validation_applied": True,
"original_entities": validated["metadata"]["original_entities"],
"original_relations": validated["metadata"]["original_relations"],
"filtered_entities": len(validated["entities"]["invalid"]),
"filtered_relations": len(validated["relations"]["invalid"])
},
"entities": validated["entities"]["valid"],
"relations": validated["relations"]["valid"],
"filtered": {
"entities": validated["entities"]["invalid"],
"relations": validated["relations"]["invalid"]
}
}
def main():
"""测试验证器"""
import sys
# 初始化验证器
validator = ExtractionValidator(
gene_list_path="ftspl_gene_list.json"
)
# 加载LLM提取结果
input_path = "evaluation/auto_results/pmid31286919_sections.json"
try:
with open(input_path, 'r') as f:
extraction_result = json.load(f)
except Exception as e:
print(f"✗ 无法加载提取结果: {e}")
sys.exit(1)
print(f"\n{'='*70}")
print("LLM提取结果验证")
print(f"{'='*70}")
# 执行验证
validated = validator.validate_extraction(extraction_result)
# 显示统计
print(f"\n📊 验证统计:")
print(f" 原始实体数: {validated['metadata']['original_entities']}")
print(f" 有效实体: {len(validated['entities']['valid'])} ({validated['statistics']['entity_precision']*100:.1f}%)")
print(f" 无效实体: {validated['statistics']['invalid_entities']}")
print(f" 原始关系数: {validated['metadata']['original_relations']}")
print(f" 有效关系: {len(validated['relations']['valid'])} ({validated['statistics']['relation_precision']*100:.1f}%)")
print(f" 无效关系: {validated['statistics']['invalid_relations']}")
# 显示被过滤的内容
if validated['entities']['invalid']:
print(f"\n⚠️ 被过滤的实体:")
for e in validated['entities']['invalid']:
print(f" - {e.get('id')} ({e.get('type')}): {e['_validation']['reason']}")
if validated['relations']['invalid']:
print(f"\n⚠️ 被过滤的关系:")
for r in validated['relations']['invalid']:
print(f" - {r.get('source')} --{r.get('type')}--> {r.get('target')}: {r['_validation']['reason']}")
# 生成清理后的输出
clean_output = validator.generate_clean_output(validated)
# 保存结果
output_path = "evaluation/auto_results/pmid31286919_sections_validated.json"
with open(output_path, 'w') as f:
json.dump(clean_output, f, indent=2, ensure_ascii=False)
print(f"\n✓ 验证结果已保存到: {output_path}")
print(f"\n{'='*70}")
print("清理后统计:")
print(f" 实体: {len(clean_output['entities'])} (过滤了 {clean_output['metadata']['filtered_entities']} 个)")
print(f" 关系: {len(clean_output['relations'])} (过滤了 {clean_output['metadata']['filtered_relations']} 个)")
print(f"{'='*70}")
if __name__ == "__main__":
main()