← 返回 Semantica 专题首页 🚀 SEMANTICA · COOKBOOK · 进阶系列

图分析

Semantica 官方 Cookbook 中文翻译 · 第 23 / 37 篇

📦 semantica 🕸️ 知识图谱 🔎 GraphRAG

欢迎来到 Semantica 图分析能力的全面演练。本 notebook 超越了简单的图构建，演示了一个全生命周期的生产流水线。

我们将模拟一个混乱的真实世界场景，涉及一个创业生态系统（投资者、创业公司、创始人），并引导你完成流程的每一步：

  1. 验证：在坏数据进入图之前将其捕获。
  2. 清洗：对实体去重并消解冲突。
  3. 结构分析：理解你的网络的形态与健康状况。
  4. 深度分析：中心性、社区与路径查找。
  5. 时序分析：在图数据中进行时间旅行。
  6. 溯源：追踪你的数据来自何处。

让我们开始吧！

# 安装 Semantica
!pip install -q semantica

import logging
import json
from datetime import datetime

# 设置日志以查看底层发生的情况
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 从 Semantica 导入所有强大的工具
from semantica.kg import (
    GraphBuilder,
    GraphAnalyzer,
    GraphValidator,
    ConnectivityAnalyzer,
    CentralityCalculator,
    CommunityDetector,
    TemporalGraphQuery,
    ProvenanceTracker
)
from semantica.deduplication import DuplicateDetector
from semantica.conflicts import ConflictDetector, ConflictResolver

11. 场景：一个混乱的创业生态系统

我们拥有来自多个来源（爬虫、新闻、用户提交）的数据。这些数据很混乱： - 重复："TechFlow AI" 和 "TechFlow Inc." - 冲突：同一家公司有不同的营收数字。 - 错误：关系指向不存在的节点（悬空边）。 - 历史：投资轮次发生在不同时间。

# 我们的"原始"混乱数据
raw_entities = [
    {"id": "startup_1", "type": "Startup", "name": "TechFlow AI", "revenue": 1000000, "founded": "2021-01-01"},
    {"id": "startup_2", "type": "Startup", "name": "GreenEnergy Co", "revenue": 500000, "founded": "2020-05-15"},
    {"id": "startup_1_dup", "type": "Startup", "name": "TechFlow Inc.", "revenue": 1200000, "founded": "2021-01-01"}, # 重复！
    {"id": "investor_1", "type": "Investor", "name": "Venture Capital X"},
    {"id": "founder_1", "type": "Person", "name": "Alice Chen"},
    {"id": "founder_2", "type": "Person", "name": "Bob Smith"}
]

raw_relationships = [
    # 有效关系
    {"source": "founder_1", "target": "startup_1", "type": "FOUNDED", "valid_from": "2021-01-01"},
    {"source": "investor_1", "target": "startup_1", "type": "INVESTED_IN", "amount": 5000000, "valid_from": "2023-06-01"},

    # 悬空边（错误！）
    {"source": "founder_2", "target": "startup_999", "type": "FOUNDED", "valid_from": "2020-05-15"}, 

    # 时序数据（历史）
    {"source": "founder_1", "target": "startup_2", "type": "ADVISED", "valid_from": "2020-01-01", "valid_until": "2021-01-01"}
]

print(f"Loaded {len(raw_entities)} raw entities and {len(raw_relationships)} raw relationships.")

22. 阶段 1：验证（守门人）

在做任何事之前，我们必须验证图。垃圾数据进 = 垃圾洞察出。 我们使用 GraphValidator 来检查： - 结构完整性：所有关系的目标是否都存在？ - 模式合规性：实体是否具有必需字段？ - 一致性：ID 是否唯一？

# 初始化验证器
validator = GraphValidator()

# 创建一个临时图对象用于验证
temp_graph = {"entities": raw_entities, "relationships": raw_relationships}

# 运行验证
print("Running Validation Check...")
validation_result = validator.validate(temp_graph)

if not validation_result.is_valid:
    print("Validation Failed! Issues found:")
    for issue in validation_result.issues:
        print(f"   - [{issue.severity.name}] {issue.message} (Code: {issue.code})")

        # 自动修复：如果是悬空边，则将其移除
        if issue.code == "DANGLING_EDGE":
            print("     Auto-Fixing: Removing invalid relationship...")
            raw_relationships = [r for r in raw_relationships 
                               if r['target'] != issue.details.get('target_id')]
else:
    print("Graph is valid!")

# 重新验证以确认修复
print("\nRe-validating after fixes...")
temp_graph = {"entities": raw_entities, "relationships": raw_relationships}
if validator.validate(temp_graph).is_valid:
    print("Graph is now clean and valid!")

33. 阶段 2：去重与冲突消解

我们有 "TechFlow AI" 和 "TechFlow Inc."。它们很可能是同一家公司。 我们还有相互冲突的营收数据。

# 1. 检测重复
print("Scanning for duplicates...")
deduper = DuplicateDetector(similarity_threshold=0.7) # 70% 相似度阈值
duplicates = deduper.detect_duplicates(raw_entities)

for candidate in duplicates:
    print(f"Found potential duplicate pair (Score: {candidate.similarity_score:.2f}):")
    print(f"   - {candidate.entity1['name']} (ID: {candidate.entity1['id']})")
    print(f"   - {candidate.entity2['name']} (ID: {candidate.entity2['id']})")

    # 合并策略：保留 entity1，合并 entity2 的数据
    print("   Merging entities...")
    # （在真实应用中，你会使用 EntityMerger，但这里是逻辑：）
    # 我们保留 startup_1 并丢弃 startup_1_dup，但记录冲突

# 2. 检测冲突
print("\nChecking for data conflicts...")
conflict_detector = ConflictDetector()

# 模拟 TechFlow 两个版本之间的冲突检查
# 为了检查冲突，我们将它们视为同一实体（相同 ID）
entity_a = raw_entities[0].copy()
entity_b = raw_entities[2].copy()
entity_b['id'] = entity_a['id'] # 强制使用相同 ID 以进行冲突检测

conflicts = conflict_detector.detect_conflicts([entity_a, entity_b])

for conflict in conflicts:
    print(f"   Conflict detected in field '{conflict.property_name}':")
    print(f"      Values: {conflict.conflicting_values}")

    # 消解：信任更高的数字（乐观！）
    if conflict.property_name == "revenue":
        # 值可能是字符串或整数，需要处理类型
        vals = [float(v) for v in conflict.conflicting_values if v is not None]
        resolved_val = max(vals)
        print(f"      Resolved to: {resolved_val}")
        raw_entities[0]['revenue'] = resolved_val

# 最终清理：从列表中移除重复实体
clean_entities = [e for e in raw_entities if e['id'] != 'startup_1_dup']
clean_relationships = raw_relationships # （我们通常也会重新链接关系）

print(f"\nCleaned Data: {len(clean_entities)} entities remaining.")

44. 阶段 3：构建知识图谱

现在我们的数据已经干净，我们构建正式的图对象。

# 手动图构建（因为我们已经清洗过它）
kg = {
    "entities": clean_entities,
    "relationships": clean_relationships,
    "metadata": {
        "created_at": datetime.now().isoformat(),
        "source": "Manual Advanced Pipeline"
    }
}
print("Knowledge Graph Assembled Successfully!")

55. 阶段 4：高级分析

这就是魔法发生的地方。我们将使用多个分析器来提取洞察。

# 初始化主分析器
analyzer = GraphAnalyzer(enable_temporal=True)

# 1. 结构分析（连通性）
print("\n--- Connectivity Analysis ---")
connectivity = analyzer.analyze_connectivity(kg)
print(f"   • Graph Connected? {'Yes' if connectivity['is_connected'] else 'No'}")
print(f"   • Connected Components: {connectivity['num_components']}")

# 2. 中心性（谁重要？）
print("\n--- Centrality Analysis ---")
centrality_result = analyzer.calculate_centrality(kg, centrality_type="degree")
degree_data = centrality_result["centrality_measures"]["degree"]

# 获取预先计算的排名
top_nodes = degree_data["rankings"][:3]

print("   • Top Influencers (Degree Centrality):")
for item in top_nodes:
    print(f"     - {item['node']}: {item['score']:.2f}")

# 3. 社区检测（聚类）
print("\n--- Community Detection ---")
community_result = analyzer.detect_communities(kg, algorithm="louvain")
communities = community_result["communities"]

print(f"   • Detected {len(communities)} communities.")
for i, comm in enumerate(communities):
    # comm 是一组节点 ID
    members = list(comm)
    print(f"     Community {i+1}: {', '.join(members)}")

66. 阶段 5：时序分析（时间旅行）

静态图很无聊。真实世界是变化的。让我们分析我们生态系统的演化。

temporal_engine = TemporalGraphQuery(temporal_granularity="year")

# 1. 时间旅行查询：2020 年的世界是什么样子？
print("\n--- Time Travel: 2020 ---")
snapshot_2020 = temporal_engine.query_at_time(kg, query="*", at_time="2020-06-01")
print(f"   Active Relationships in 2020: {len(snapshot_2020['relationships'])}")
for rel in snapshot_2020['relationships']:
    print(f"   - {rel['source']} --[{rel['type']}]--> {rel['target']}")

# 2. 时间旅行查询：2023 年呢？
print("\n--- Time Travel: 2023 ---")
snapshot_2023 = temporal_engine.query_at_time(kg, query="*", at_time="2023-07-01")
print(f"   Active Relationships in 2023: {len(snapshot_2023['relationships'])}")
for rel in snapshot_2023['relationships']:
    print(f"   - {rel['source']} --[{rel['type']}]--> {rel['target']}")

# 注意 'ADVISED' 如果已结束可能会消失，而 'INVESTED_IN' 会出现！

77. 阶段 6：溯源（数据血缘）

最后，在生产系统中，你需要知道一个事实来自何处。这对建立信任至关重要。

tracker = ProvenanceTracker()

# 让我们假装正在追踪数据的来源
tracker.track_entity("startup_1", source="Crunchbase_API_v2", metadata={"confidence": 0.95})
tracker.track_entity("startup_1", source="Manual_Entry_User_Bob", metadata={"confidence": 1.0})

print("\n--- Provenance Report: TechFlow AI ---")
lineage = tracker.get_lineage("startup_1")
print(f"   Entity: startup_1")
print(f"   First Seen: {lineage['first_seen']}")
print(f"   Sources:")
for src in lineage['sources']:
    print(f"     - {src['source']} (at {src['timestamp']})")

8结论

你刚刚走完了一个完整的、高级的知识图谱流水线：

  1. 验证了混乱的输入数据。
  2. 清洗了重复和冲突。
  3. 分析了结构和社区动态。
  4. 跨时间维度查询。
  5. 追踪了数据血缘。

这代表了使用 Semantica 进行现代 KG 工程的最先进水平。