关系抽取 - 全面指南

pip install semantica
# 或安装所有可选依赖：
pip install semantica[all]

# 安装 semantica 包
!pip install semantica

from semantica.semantic_extract import RelationExtractor, NERExtractor

# 初始化抽取器
ner_extractor = NERExtractor()
relation_extractor = RelationExtractor()

# 包含清晰关系的示例文本
text = """
Apple Inc. was founded by Steve Jobs, Steve Wozniak, and Ronald Wayne in 1976.
The company is headquartered in Cupertino, California. Tim Cook is the current CEO
of Apple Inc. and took over from Steve Jobs in August 2011.
"""

# 首先，抽取实体
entities = ner_extractor.extract(text)
print(f" Extracted {len(entities)} entities\n")

# 然后，抽取关系
relationships = relation_extractor.extract(text, entities)

print(f" Extracted {len(relationships)} relationships:\n")
print("=" * 80)

for i, rel in enumerate(relationships, 1):
    # 同时处理 dict 和对象两种格式
    source = rel.get('source', rel.get('subject', '')) if isinstance(rel, dict) else getattr(rel, 'subject', '')
    target = rel.get('target', rel.get('object', '')) if isinstance(rel, dict) else getattr(rel, 'object', '')
    rel_type = rel.get('type', rel.get('predicate', 'related_to')) if isinstance(rel, dict) else getattr(rel, 'predicate', 'related_to')
    confidence = rel.get('confidence', 1.0) if isinstance(rel, dict) else getattr(rel, 'confidence', 1.0)

    # 获取源和目标文本
    if isinstance(source, dict):
        source_text = source.get('text', source.get('entity', str(source)))
    else:
        source_text = getattr(source, 'text', str(source))

    if isinstance(target, dict):
        target_text = target.get('text', target.get('entity', str(target)))
    else:
        target_text = getattr(target, 'text', str(target))

    print(f"{i:2d}. {source_text:20s} --[{rel_type:15s}]--> {target_text:20s} (conf: {confidence:.2f})")

print("=" * 80)

from semantica.semantic_extract import RelationExtractor

sample_text = "Apple Inc. was founded by Steve Jobs in Cupertino, California."
sample_entities = ner_extractor.extract(sample_text)

print(" Comparing Relation Extraction Methods:\n")
print("=" * 80)

# 尝试不同的方法
methods_to_try = ["pattern", "dependency", "cooccurrence"]

for method_name in methods_to_try:
    try:
        print(f"\n Method: {method_name.upper()}")
        print("-" * 40)

        extractor = RelationExtractor(method=method_name)
        relations = extractor.extract(sample_text, sample_entities)

        print(f"Found {len(relations)} relations:")
        for rel in relations[:3]:  # 显示前 3 个
            source = rel.get('source', rel.get('subject', '')) if isinstance(rel, dict) else getattr(rel, 'subject', '')
            target = rel.get('target', rel.get('object', '')) if isinstance(rel, dict) else getattr(rel, 'object', '')
            rel_type = rel.get('type', rel.get('predicate', 'related_to')) if isinstance(rel, dict) else getattr(rel, 'predicate', 'related_to')

            # 获取文本表示
            source_text = source.get('text', str(source)) if isinstance(source, dict) else getattr(source, 'text', str(source))
            target_text = target.get('text', str(target)) if isinstance(target, dict) else getattr(target, 'text', str(target))

            print(f"  • {source_text} --[{rel_type}]--> {target_text}")

    except Exception as e:
        print(f"  ️  Method '{method_name}' not available: {str(e)[:50]}")

print("\n" + "=" * 80)

# 使用自定义配置创建抽取器
advanced_extractor = RelationExtractor(
    relation_types=["founded_by", "located_in", "works_for"],  # 只抽取这些类型
    confidence_threshold=0.7,  # 更高的阈值以保证质量
    bidirectional=False,  # 只抽取单向关系
    max_distance=50  # 实体之间最多 50 个 token
)

# 示例文本
texts = [
    "Microsoft was founded by Bill Gates and Paul Allen in Albuquerque, New Mexico.",
    "Satya Nadella works for Microsoft as the CEO.",
    "Google is located in Mountain View, California.",
    "Amazon was founded by Jeff Bezos in Seattle, Washington."
]

print(" Advanced Relation Extraction:\n")
print("=" * 80)

for i, text in enumerate(texts, 1):
    entities = ner_extractor.extract(text)
    relations = advanced_extractor.extract(text, entities)

    print(f"\n Text {i}: {text}")
    print(f"   Relations found: {len(relations)}")

    for rel in relations:
        source = rel.get('source', rel.get('subject', '')) if isinstance(rel, dict) else getattr(rel, 'subject', '')
        target = rel.get('target', rel.get('object', '')) if isinstance(rel, dict) else getattr(rel, 'object', '')
        rel_type = rel.get('type', rel.get('predicate', 'related_to')) if isinstance(rel, dict) else getattr(rel, 'predicate', 'related_to')
        confidence = rel.get('confidence', 1.0) if isinstance(rel, dict) else getattr(rel, 'confidence', 1.0)

        source_text = source.get('text', str(source)) if isinstance(source, dict) else getattr(source, 'text', str(source))
        target_text = target.get('text', str(target)) if isinstance(target, dict) else getattr(target, 'text', str(target))

        print(f"     • {source_text} --[{rel_type}]--> {target_text} (conf: {confidence:.2f})")

print("\n" + "=" * 80)

# 从所有文本中抽取关系
all_relations = []
for text in texts:
    entities = ner_extractor.extract(text)
    relations = advanced_extractor.extract(text, entities)
    all_relations.extend(relations)

# 对关系进行分类
classified_relations = advanced_extractor.classify_relations(all_relations)

print("️  Relation Classification:\n")
print("=" * 80)

for rel_type, rel_list in sorted(classified_relations.items()):
    print(f"\n{rel_type.upper()} ({len(rel_list)} relations):")
    print("-" * 40)

    for rel in rel_list:
        source = rel.get('source', rel.get('subject', '')) if isinstance(rel, dict) else getattr(rel, 'subject', '')
        target = rel.get('target', rel.get('object', '')) if isinstance(rel, dict) else getattr(rel, 'object', '')

        source_text = source.get('text', str(source)) if isinstance(source, dict) else getattr(source, 'text', str(source))
        target_text = target.get('text', str(target)) if isinstance(target, dict) else getattr(target, 'text', str(target))

        print(f"  • {source_text} → {target_text}")

print("\n" + "=" * 80)

from semantica.semantic_extract import TripletExtractor

# 初始化三元组抽取器
triplet_extractor = TripletExtractor(
    include_temporal=True,  # 包含时序信息
    include_provenance=True  # 追踪来源句子
)

# 示例文本
triplet_text = """
Apple Inc. was founded by Steve Jobs in 1976. The company is based in Cupertino, California.
Tim Cook became CEO in 2011. Apple develops the iPhone and MacBook products.
"""

# 抽取三元组
triplets = triplet_extractor.extract_triplets(triplet_text)

print(f" Extracted {len(triplets)} RDF Triplets:\n")
print("=" * 80)

for i, triplet in enumerate(triplets, 1):
    subject = triplet.get('subject', '') if isinstance(triplet, dict) else triplet.subject
    predicate = triplet.get('predicate', '') if isinstance(triplet, dict) else triplet.predicate
    obj = triplet.get('object', '') if isinstance(triplet, dict) else triplet.object
    confidence = triplet.get('confidence', 1.0) if isinstance(triplet, dict) else getattr(triplet, 'confidence', 1.0)

    print(f"{i:2d}. ({subject}, {predicate}, {obj})")
    print(f"    Confidence: {confidence:.2f}")

    # 如果有时序信息则显示
    metadata = triplet.get('metadata', {}) if isinstance(triplet, dict) else getattr(triplet, 'metadata', {})
    if metadata.get('temporal'):
        print(f"    Temporal: {metadata['temporal']}")
    print()

print("=" * 80)

from semantica.semantic_extract import RDFSerializer

# 初始化序列化器
serializer = RDFSerializer()

print(" RDF Serialization Examples:\n")
print("=" * 80)

# 序列化为不同格式
formats = ["turtle", "ntriples", "jsonld"]

for fmt in formats:
    print(f"\n {fmt.upper()} Format:")
    print("-" * 40)

    try:
        serialized = serializer.serialize_to_rdf(triplets[:3], format=fmt)  # 显示前 3 个

        # 显示预览（前 300 个字符）
        preview = serialized[:300] + "..." if len(serialized) > 300 else serialized
        print(preview)

    except Exception as e:
        print(f"Error: {str(e)[:50]}")

print("\n" + "=" * 80)

def extract_knowledge(text):
    """
    完整知识抽取流水线。

    Args:
        text: 输入文本

    Returns:
        dict: 抽取出的实体、关系和三元组
    """
    # 步骤 1：抽取实体
    entities = ner_extractor.extract(text)

    # 步骤 2：抽取关系
    relations = relation_extractor.extract(text, entities)

    # 步骤 3：抽取三元组
    triplets = triplet_extractor.extract_triplets(text, entities=entities, relationships=relations)

    return {
        'entities': entities,
        'relations': relations,
        'triplets': triplets
    }

# 富含知识的示例文本
knowledge_text = """
Tesla Inc. was founded by Elon Musk, JB Straubel, Martin Eberhard, Marc Tarpenning, 
and Ian Wright in 2003. The company is headquartered in Austin, Texas. Tesla produces 
electric vehicles including the Model S, Model 3, Model X, and Model Y. Elon Musk serves 
as CEO and has been instrumental in the company's growth.
"""

print(" Complete Extraction Pipeline:\n")
print("=" * 80)

# 运行流水线
result = extract_knowledge(knowledge_text)

print(f"\n Extraction Results:")
print("-" * 40)
print(f"Entities extracted: {len(result['entities'])}")
print(f"Relations extracted: {len(result['relations'])}")
print(f"Triplets extracted: {len(result['triplets'])}")

print(f"\n Sample Triplets:")
for i, triplet in enumerate(result['triplets'][:5], 1):
    subject = triplet.get('subject', '') if isinstance(triplet, dict) else triplet.subject
    predicate = triplet.get('predicate', '') if isinstance(triplet, dict) else triplet.predicate
    obj = triplet.get('object', '') if isinstance(triplet, dict) else triplet.object
    print(f"  {i}. ({subject}, {predicate}, {obj})")

print("\n" + "=" * 80)