1概览
本 notebook 使用 Semantica 对 时序知识图谱（Temporal Knowledge Graphs，TKG） 进行全面深入的探究。与静态知识图谱不同，TKG 能够捕捉事实、关系和实体随时间的演变。这一能力对于以下应用至关重要：
- 企业历史分析：追踪并购、收购和领导层更迭。
- 供应链监控：追踪产品流转和状态变化。
- 金融欺诈检测：分析交易序列。
我们将构建一个丰富的场景，对一个科技生态系统的历史进行建模，覆盖 40 年的演变。
涵盖的关键组件
GraphBuilder（时序模式）：构建具有时间感知属性的知识图谱。TemporalGraphQuery：执行时间点、区间和路径查询。TemporalPatternDetector：识别序列和循环模式。TemporalVersionManager：管理快照并比较图谱状态。TemporalVisualizer：交互式时间线和演变图。
文档：API 参考
2安装
# !pip install semantica[all]
import json
from datetime import datetime
from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, TemporalVersionManager
from semantica.visualization import TemporalVisualizer
import plotly.offline as pyo
pyo.init_notebook_mode(connected=True)
# 确保输出一致以便复现
import random
random.seed(42)
3步骤 1：场景定义与数据准备
我们定义一个数据集，表示 "TechCorp" 和 "InnovateInc" 的历史，包括它们的创始人、产品以及最终的合并。
时序属性：
- 实体具有 founded、born、released 日期。
- 关系具有 timestamp（时间点事件）或 valid_from/valid_to（区间）。
# 1. 定义带时序元数据的实体
entities = [
# 组织
{"id": "org_1", "type": "Organization", "name": "TechCorp", "properties": {"founded": "1980-01-01", "industry": "Hardware"}},
{"id": "org_2", "type": "Organization", "name": "InnovateInc", "properties": {"founded": "1995-06-15", "industry": "Software"}},
{"id": "org_3", "type": "Organization", "name": "FutureSystems", "properties": {"founded": "2010-03-10", "industry": "AI"}},
# 人物
{"id": "per_1", "type": "Person", "name": "Alice Founder", "properties": {"born": "1955-05-20"}},
{"id": "per_2", "type": "Person", "name": "Bob Coder", "properties": {"born": "1970-08-12"}},
{"id": "per_3", "type": "Person", "name": "Charlie CEO", "properties": {"born": "1980-02-28"}},
# 产品
{"id": "prod_1", "type": "Product", "name": "HomePC", "properties": {"released": "1985-11-20"}},
{"id": "prod_2", "type": "Product", "name": "SoftOS", "properties": {"released": "1998-07-25"}},
{"id": "prod_3", "type": "Product", "name": "SmartAI", "properties": {"released": "2015-01-10"}}
]
# 2. 定义时序关系
relationships = [
# 创立事件（时间点）
{"source": "per_1", "target": "org_1", "type": "founded", "timestamp": "1980-01-01", "properties": {"timestamp": "1980-01-01"}},
{"source": "per_2", "target": "org_2", "type": "founded", "timestamp": "1995-06-15", "properties": {"timestamp": "1995-06-15"}},
# 任职（区间）
{"source": "per_1", "target": "org_1", "type": "ceo_of", "valid_from": "1980-01-01", "valid_to": "2000-01-01", "properties": {"role": "CEO"}},
{"source": "per_3", "target": "org_1", "type": "ceo_of", "valid_from": "2000-01-02", "valid_to": "2023-01-01", "properties": {"role": "CEO"}},
{"source": "per_2", "target": "org_2", "type": "cto_of", "valid_from": "1995-06-15", "valid_to": "2010-05-01", "properties": {"role": "CTO"}},
# 产品发布
{"source": "org_1", "target": "prod_1", "type": "launched", "timestamp": "1985-11-20", "properties": {"timestamp": "1985-11-20"}},
{"source": "org_2", "target": "prod_2", "type": "launched", "timestamp": "1998-07-25", "properties": {"timestamp": "1998-07-25"}},
{"source": "org_3", "target": "prod_3", "type": "launched", "timestamp": "2015-01-10", "properties": {"timestamp": "2015-01-10"}},
# 企业行为
{"source": "org_1", "target": "org_2", "type": "acquired", "timestamp": "2010-05-01", "properties": {"amount": "$5B", "timestamp": "2010-05-01"}},
{"source": "org_1", "target": "org_3", "type": "invested_in", "timestamp": "2012-08-15", "properties": {"amount": "$100M", "timestamp": "2012-08-15"}}
]
print(f"Defined {len(entities)} entities and {len(relationships)} temporal relationships.")
4步骤 2：构建时序图谱
我们使用 GraphBuilder 并设置 enable_temporal=True。这会指示构建器对 timestamp、valid_from 和 valid_to 等时序属性建立索引。
builder = GraphBuilder(
enable_temporal=True,
temporal_granularity="day" # 可以是 'year'、'month'、'day'、'hour'
)
temporal_kg = builder.build(entities, relationships)
# 图谱对象现在包含时序索引
print("Graph built successfully.")
print(f"Nodes: {len(temporal_kg['entities'])}")
print(f"Edges: {len(temporal_kg['relationships'])}")
5步骤 3：高级时序查询
我们使用 TemporalGraphQuery 来提出时间敏感的问题。
query_engine = TemporalGraphQuery()
# 1. 时间点查询
# "1990 年 TechCorp 的 CEO 是谁？"
ceo_1990 = query_engine.query_at_time(
temporal_kg,
query="Find the CEO of TechCorp",
at_time="1990-06-01"
)
print("CEO in 1990:", [e['id'] for e in ceo_1990.get('entities', [])])
# "2015 年 TechCorp 的 CEO 是谁？"
ceo_2015 = query_engine.query_at_time(
temporal_kg,
query="Find the CEO of TechCorp",
at_time="2015-06-01"
)
print("CEO in 2015:", [e['id'] for e in ceo_2015.get('entities', [])])
# 2. 时序路径查找
# "Alice（创始人）是如何与 SmartAI（2015 年发布的产品）联系起来的？"
# 这需要穿越时间进行遍历：Alice -> 创立 TechCorp -> 投资 FutureSystems -> 发布 SmartAI
paths = query_engine.find_temporal_paths(
graph=temporal_kg,
source="per_1", # Alice
target="prod_3", # SmartAI
start_time="1980-01-01",
end_time="2020-01-01"
)
print(f"\nFound {len(paths)} temporal paths from Alice to SmartAI.")
for i, path in enumerate(paths):
print(f"Path {i+1}: {path}")
6步骤 4：图谱演变分析
我们可以使用 analyze_evolution 分析图谱属性随时间的变化。
# 分析图谱属性随时间的变化
evolution_stats = query_engine.analyze_evolution(
temporal_kg,
start_time="1980-01-01",
end_time="2025-01-01",
metrics=["count", "diversity", "stability"]
)
# 输出各项演变指标
print("\nEvolution Statistics (1980-2025):")
print(f"Total Relationships: {evolution_stats.get('count', 'N/A')}")
print(f"Relationship Diversity: {evolution_stats.get('diversity', 'N/A')}")
print(f"Graph Stability: {evolution_stats.get('stability', 'N/A')}")
7步骤 5：时序模式检测
我们使用 TemporalPatternDetector 自动查找重复出现的结构，例如序列（A -> B -> C）或循环。
detector = TemporalPatternDetector()
# 检测序列模式（例如，创立 -> 发布 -> 收购）
sequences = detector.detect_temporal_patterns(
temporal_kg,
pattern_type="sequence",
min_frequency=1
)
print(f"\nDetected {len(sequences)} sequential patterns.")
for seq in sequences[:3]: # 显示前 3 个
print(f"Pattern: {seq.get('pattern')}")
print(f"Support: {seq.get('support')}")
8步骤 6：版本管理与比较
在真实场景中，知识图谱是分批更新的。TemporalVersionManager 负责处理这些版本。
version_manager = TemporalVersionManager()
# 创建显式版本
v1_1990 = version_manager.create_version(temporal_kg, timestamp="1990-01-01", version_label="v1.0 (Early Days)")
v2_2010 = version_manager.create_version(temporal_kg, timestamp="2010-01-01", version_label="v2.0 (Post-Merger)")
# 比较版本
diff = version_manager.compare_versions(v1_1990, v2_2010)
print(f"\nComparing {v1_1990['label']} vs {v2_2010['label']}:")
print(f"New Entities: {diff.get('entities_added', 0)}")
print(f"New Relationships: {diff.get('relationships_added', 0)}")
9步骤 7：可视化时间线
最后，TemporalVisualizer 让数据变得生动起来。我们将创建一个交互式时间线和一个快照比较。
visualizer = TemporalVisualizer()
# 1. 交互式时间线
# 准备用于可视化的事件（从知识图谱中提取）
def extract_events(graph):
events = []
for rel in graph['relationships']:
# 时间点事件
if rel.get('timestamp'):
events.append({
'timestamp': rel['timestamp'],
'type': rel['type'],
'label': f"{rel['source']} -> {rel['target']}",
'entity': rel['source']
})
# 区间事件（开始）
if rel.get('valid_from'):
events.append({
'timestamp': rel['valid_from'],
'type': f"{rel['type']} (start)",
'label': f"{rel['source']} -> {rel['target']}",
'entity': rel['source']
})
return {'events': events}
temporal_data = extract_events(temporal_kg)
timeline_fig = visualizer.visualize_timeline(temporal_data, output="interactive")
# 在 notebook 中，这会渲染一个 Plotly 图形。
timeline_fig.show()
# 2. 版本历史可视化
history = [
{"version": "v1.0", "timestamp": "1990-01-01", "changes": "Founding Era"},
{"version": "v2.0", "timestamp": "2010-01-01", "changes": "Expansion Era"},
{"version": "v3.0", "timestamp": "2020-01-01", "changes": "AI Era"}
]
history_fig = visualizer.visualize_version_history(history, output="interactive")
history_fig.show()
print("Visualizations generated (render requires Jupyter environment).")
10小结
在本次深入探究中，我们：
1. 建模了一个带有时序元数据的复杂企业历史。
2. 使用 GraphBuilder 构建了一个时间感知的知识图谱。
3. 查询了特定的时间切片和区间以重建历史。
4. 追踪了时序路径以理解间接连接。
5. 分析了图谱的演变指标。
6. 管理了版本并可视化了时间线。
7. 使用 TemporalVisualizer 可视化了数据。
这一工作流构成了 Semantica 中时序智能应用的骨干。