时序知识图谱

时序知识图谱(Temporal Knowledge Graph,TKG)是知识图谱的时间扩展,其中事实(三元组)被标注有时间戳,捕捉知识随时间的动态演化。本文档系统介绍时序知识图谱的前沿方法。

1. 时序知识图谱基础

1.1 定义与形式化

时序知识图谱

  • :实体集合
  • :关系集合
  • :时间戳集合
  • :时序三元组

1.2 与静态KG的关键区别

特性静态KG时序KG
事实表示
关系建模单一关系时间条件关系
推理任务KG补全时序预测
核心挑战嵌入学习动态演化建模

1.3 核心任务

时序知识图谱任务:

1. 时序链接预测 (Temporal Link Prediction)
   给定 (h, r, ?, τ),预测尾实体 t

2. 时序事实预测 (Temporal Fact Prediction)
   给定 (h, r, ?, ?),预测 (t, τ)

3. 时序路径推理 (Temporal Path Reasoning)
   考虑时间约束的多跳推理

4. 未来事件预测 (Future Event Prediction)
   基于历史预测未来事件

5. 时序实体对齐 (Temporal Entity Alignment)
   跨图谱实体匹配

2. MATA:记忆增强时序锚点

2.1 核心思想

MATA (Memory-Augmented Temporal Anchors)1提出了一种处理稀疏时序知识图谱的方法,通过可学习的记忆模块实现时间插值。

2.2 稀疏时间挑战

传统方法的局限:

时序KG往往非常稀疏:
- 每年可能只有少数几个事实
- 某些时间段完全缺失数据
- 实体在不同时期可能有不同关系

例如:
2019: CEO(Apple, TimCook)
2020: (无数据)
2021: (无数据)
2022: CEO(Apple, TimCook)
2023: CEO(Apple, TimCook)

传统方法:无法处理中间缺失
MATA:学习时间锚点进行插值

2.3 记忆增强架构

┌─────────────────────────────────────────────────────────────┐
│                        MATA 架构                             │
│                                                              │
│  输入: (head, relation, query_time)                         │
│                                                              │
│  ┌─────────────────────────────────────────────────────┐    │
│  │                 记忆模块 (Memory Module)             │    │
│  │                                                       │    │
│  │    Memory Bank: M = [m_1, m_2, ..., m_k]            │    │
│  │                  ↑ 可学习的时序锚点                    │    │
│  │                                                       │    │
│  │    时间注意力:                                          │    │
│  │    α_t = softmax(q · M^T / √d)                      │    │
│  │    temporal_context = Σ_t α_t[i] · m_i               │    │
│  └─────────────────────────────────────────────────────┘    │
│                           │                                 │
│                           ▼                                 │
│  ┌─────────────────────────────────────────────────────┐    │
│  │              时序推理层 (Temporal Reasoning)           │    │
│  │                                                       │    │
│  │    score = f(head_emb, rel_emb, temporal_context)    │    │
│  └─────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────┘

2.4 时间插值机制

class TemporalInterpolation:
    """时间插值机制"""
    
    def __init__(self, memory_dim, num_anchors):
        # 可学习的时序锚点
        self.memory = nn.Parameter(torch.randn(num_anchors, memory_dim))
    
    def interpolate(self, query_time, entity_history):
        """
        基于记忆模块进行时间插值
        
        Args:
            query_time: 查询时间戳
            entity_history: 实体的历史表示
        """
        # 1. 时间编码
        time_encoding = self.get_time_encoding(query_time)
        
        # 2. 与记忆模块的注意力交互
        attention_scores = torch.matmul(
            time_encoding, 
            self.memory.T
        ) / np.sqrt(self.memory.shape[-1])
        
        attention_weights = F.softmax(attention_scores, dim=-1)
        
        # 3. 加权聚合记忆
        temporal_context = torch.matmul(
            attention_weights,
            self.memory
        )
        
        # 4. 结合实体历史
        combined = entity_history + temporal_context
        
        return combined
    
    def get_time_encoding(self, t):
        """
        时间编码:sinusoidal encoding
        """
        d_model = self.memory.shape[-1]
        position = torch.arange(d_model // 2)
        
        # Sinusoidal encoding
        pe = torch.zeros(d_model)
        pe[0::2] = torch.sin(t / (10000 ** (position / (d_model // 2))))
        pe[1::2] = torch.cos(t / (10000 ** (position / (d_model // 2))))
        
        return pe

2.5 实验结果

方法GDELT MRRICEWS14 MRRICEWS18 MRR
MATA0.420.580.55
RE-NET0.380.510.48
CyGNet0.350.490.45
提升+10.5%+13.7%+15.6%

3. DynaGen:动态子图与扩散正则化

3.1 核心思想

DynaGen2统一了时序KG的插值和外推任务,通过动态子图构建和扩散正则化学习底层演化规律。

3.2 统一框架

传统方法的分裂:
┌──────────────┐     ┌──────────────┐
│    插值      │     │    外推      │
│  预测中间时间 │     │  预测未来时间 │
│  2019 → 2020 │     │  2022 → 2023 │
└──────────────┘     └──────────────┘
       │                   │
       └─────────┬─────────┘
                 ▼
         ┌──────────────┐
         │   DynaGen    │
         │   统一框架    │
         └──────────────┘

3.3 动态子图构建

class DynamicSubgraphConstructor:
    """动态子图构建器"""
    
    def __init__(self, kg_history, time_range):
        self.kg_history = kg_history
        self.time_range = time_range
    
    def construct_subgraph(self, center_entity, time_window):
        """
        为实体构建时序子图
        """
        # 1. 时间窗口内的事实
        in_window = [
            (h, r, t, τ) 
            for h, r, t, τ in self.kg_history
            if τ ∈ time_window and (h == center_entity or t == center_entity)
        ]
        
        # 2. 构建实体为中心的子图
        subgraph_nodes = set()
        subgraph_edges = []
        
        for h, r, t, τ in in_window:
            subgraph_nodes.add(h)
            subgraph_nodes.add(t)
            subgraph_edges.append({
                'source': h,
                'relation': r,
                'target': t,
                'time': τ
            })
        
        # 3. 添加邻居实体的子结构
        for entity in subgraph_nodes:
            neighbors = self.get_temporal_neighbors(entity, time_window)
            subgraph_nodes.update(neighbors)
        
        return Subgraph(nodes=subgraph_nodes, edges=subgraph_edges)
    
    def get_temporal_neighbors(self, entity, time_window):
        """获取时间感知的邻居"""
        neighbors = []
        for h, r, t, τ in self.kg_history:
            if τ ∈ time_window:
                if h == entity:
                    neighbors.append(t)
                elif t == entity:
                    neighbors.append(h)
        return neighbors

3.4 扩散正则化

核心思想:使用扩散模型学习KG的演化规律

扩散过程:
t=0 (观测数据): [观测事实分布]
    ↓ 前向扩散
t=T (噪声):     [均匀随机分布]

反向过程(学习):
t=T (噪声):     [随机初始化]
    ↓ 反向扩散
t=0 (预测):     [预测事实分布]
class DiffusionRegularizer:
    """扩散正则化器"""
    
    def __init__(self, dim, num_timesteps=100):
        self.num_timesteps = num_timesteps
        self.noise_net = NoiseNetwork(dim)
        self.score_net = ScoreNetwork(dim)
    
    def forward_diffusion(self, x, t):
        """
        前向扩散:添加噪声
        """
        # x_t = √(ᾱ_t) * x_0 + √(1 - ᾱ_t) * ε
        alpha_bar = self.get_alpha_bar(t)
        noise = torch.randn_like(x)
        
        x_t = torch.sqrt(alpha_bar) * x + torch.sqrt(1 - alpha_bar) * noise
        return x_t, noise
    
    def training_loss(self, model, kg_embeddings):
        """
        训练损失:score matching
        """
        # 随机采样时间步
        t = torch.randint(0, self.num_timesteps, (kg_embeddings.shape[0],))
        
        # 前向扩散
        x_t, noise = self.forward_diffusion(kg_embeddings, t)
        
        # 预测噪声
        predicted_noise = model(x_t, t)
        
        # Score matching loss
        loss = F.mse_loss(predicted_noise, noise)
        
        return loss
    
    def regularization_loss(self, embeddings, temporal_graph):
        """
        时序图正则化:鼓励相似时间的实体接近
        """
        # 1. 获取时序邻居
        neighbors = self.get_temporal_neighbors(embeddings, temporal_graph)
        
        # 2. 同一时间的邻居应该接近
        same_time_loss = 0
        for entity, (nbrs, times) in neighbors.items():
            for nbr, tau in zip(nbrs, times):
                # 拉近同一时间的实体
                same_time_loss += F.mse_loss(
                    embeddings[entity],
                    embeddings[nbr]
                )
        
        # 3. 扩散正则化
        diffusion_reg = self.training_loss(self.score_net, embeddings)
        
        return same_time_loss + λ * diffusion_reg

4. TeRDy:频率分解方法

4.1 核心思想

TeRDy (Temporal Regularity Decomposition)3通过频率分解同时捕捉时序KG的长期趋势和短期波动。

4.2 频率视角的时序KG

时序知识图谱 = 低频成分 + 高频成分

低频成分:
- 实体/关系的长期特性
- 缓慢变化的知识模式
- 例如:公司CEO可能在10年内不变

高频成分:
- 短期事件和交互
- 快速变化的动态
- 例如:社交媒体互动可能每秒都在变

4.3 数学框架

class TeRDyDecomposition:
    """TeRDy频率分解"""
    
    def __init__(self, num_freq_components):
        self.num_low = num_freq_components // 2
        self.num_high = num_freq_components - self.num_low
    
    def decompose(self, entity_sequence):
        """
        将实体表示序列分解为频率成分
        
        Args:
            entity_sequence: [T, d] 时间序列
            
        Returns:
            low_freq: 低频成分(长期趋势)
            high_freq: 高频成分(短期波动)
        """
        T, d = entity_sequence.shape
        
        # 1. 傅里叶变换
        fft = torch.fft.fft(entity_sequence, dim=0)
        freqs = torch.fft.fftfreq(T)
        
        # 2. 低频提取(保留低频分量)
        low_mask = torch.abs(freqs) < self.freq_threshold
        fft_low = fft * low_mask.unsqueeze(-1).float()
        low_freq = torch.fft.ifft(fft_low, dim=0).real
        
        # 3. 高频提取(保留高频分量)
        high_mask = torch.abs(freqs) >= self.freq_threshold
        fft_high = fft * high_mask.unsqueeze(-1).float()
        high_freq = torch.fft.ifft(fft_high, dim=0).real
        
        return low_freq, high_freq
    
    def frequency_attention(self, query, low_freq, high_freq):
        """
        频率感知注意力
        """
        # 低频注意力
        attn_low = torch.sigmoid(
            query @ low_freq.T / np.sqrt(self.dim)
        )
        
        # 高频注意力
        attn_high = torch.sigmoid(
            query @ high_freq.T / np.sqrt(self.dim)
        )
        
        # 组合
        combined = attn_low * low_freq + attn_high * high_freq
        
        return combined

4.4 关系动态建模

class TemporalRelationModel:
    """时序关系建模"""
    
    def __init__(self, entity_dim, relation_dim):
        self.entity_proj = nn.Linear(entity_dim, relation_dim)
        
        # 低频和高频的关系嵌入
        self.low_freq_rel = nn.Parameter(torch.randn(relation_dim))
        self.high_freq_rel = nn.Parameter(torch.randn(relation_dim))
    
    def get_relation_embedding(self, tau, relation_base):
        """
        获取时间依赖的关系嵌入
        """
        # 基础关系 + 低频偏移 + 高频偏移
        low_offset = self.low_freq_rel * torch.sin(2 * np.pi * tau / 365)
        high_offset = self.high_freq_rel * torch.sin(2 * np.pi * tau / 7)
        
        return relation_base + low_offset + high_offset
    
    def score_triplet(self, head_emb, rel_emb, tail_emb, time):
        """
        计算时序三元组分数
        """
        # 获取时间依赖的关系嵌入
        rel_t = self.get_relation_embedding(time, rel_emb)
        
        # 翻译评分
        score = -torch.norm(head_emb + rel_t - tail_emb, p=2, dim=-1)
        
        return score

5. SALoM:结构感知时序GNN

5.1 核心思想

SALoM (Structure-Aware Long and Short Memory)4结合了结构感知和时间感知,通过长期记忆捕捉全局模式,短期记忆捕捉局部动态。

5.2 双记忆架构

┌─────────────────────────────────────────────────────────────┐
│                      SALoM 架构                               │
│                                                              │
│  输入: 时序KG快照序列 G_1, G_2, ..., G_T                      │
│                                                              │
│  ┌──────────────────────────────────────────────────────┐   │
│  │                    长期记忆 (Long-term Memory)          │   │
│  │                                                       │   │
│  │  作用: 捕捉跨时间的全局结构模式                        │   │
│  │  实现: 累积历史嵌入的指数移动平均                       │   │
│  │                                                       │   │
│  │  h_long^t = α * h_long^{t-1} + (1-α) * h_t           │   │
│  │  ↑ α = 0.9 (高衰减慢,保留长期信息)                   │   │
│  └──────────────────────────────────────────────────────┘   │
│                           │                                   │
│                           ▼                                   │
│  ┌──────────────────────────────────────────────────────┐   │
│  │                    短期记忆 (Short-term Memory)          │   │
│  │                                                       │   │
│  │  作用: 捕捉最近邻时间步的局部动态                      │   │
│  │  实现: 滑动窗口内的GNN消息传递                         │   │
│  │                                                       │   │
│  │  Window: [G_{t-k}, ..., G_{t-1}, G_t]                │   │
│  │  Message: m_{ij} = f(h_i, h_j, r_{ij}, τ_{ij})       │   │
│  └──────────────────────────────────────────────────────┘   │
│                           │                                   │
│                           ▼                                   │
│  ┌──────────────────────────────────────────────────────┐   │
│  │                    记忆融合 (Memory Fusion)             │   │
│  │                                                       │   │
│  │  h_fused = σ(W_long · h_long ⊕ W_short · h_short)     │   │
│  └──────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────┘

5.3 实现细节

class SALoMLayer(nn.Module):
    """SALoM层"""
    
    def __init__(self, dim, num_heads=4, window_size=5):
        super().__init__()
        self.dim = dim
        self.window_size = window_size
        
        # 长期记忆更新
        self.long_memory_alpha = 0.9
        self.long_proj = nn.Linear(dim, dim)
        
        # 短期记忆更新 (GNN)
        self.short_gnn = TemporalGNN(dim, num_heads)
        
        # 记忆融合
        self.fusion_gate = nn.Linear(dim * 2, dim)
        
        # 时间编码
        self.time_encoder = SinusoidalTimeEncoder(dim)
    
    def forward(self, snapshots, prev_long_memory=None):
        """
        Args:
            snapshots: [window_size, num_entities, dim] 时间窗口内的快照
            prev_long_memory: [num_entities, dim] 之前的长期记忆
        """
        T, N, D = snapshots.shape
        
        # === 短期记忆:时序GNN ===
        # 在滑动窗口上执行消息传递
        short_features = []
        for t in range(T):
            gnn_out = self.short_gnn(snapshots[t])
            short_features.append(gnn_out)
        
        # 时序注意力聚合
        short_memory = self.temporal_attention(short_features)
        
        # === 长期记忆:EMA更新 ===
        if prev_long_memory is not None:
            # 指数移动平均
            long_memory = (
                self.long_memory_alpha * prev_long_memory + 
                (1 - self.long_memory_alpha) * snapshots[-1]
            )
        else:
            long_memory = snapshots[-1]
        
        # === 记忆融合 ===
        combined = torch.cat([long_memory, short_memory], dim=-1)
        fused = torch.sigmoid(self.fusion_gate(combined))
        
        return fused, long_memory
    
    def temporal_attention(self, features):
        """
        时序注意力聚合
        """
        features = torch.stack(features, dim=1)  # [N, T, D]
        
        # 时间注意力
        time_emb = self.time_encoder(torch.arange(T))
        attn_scores = (features * time_emb.unsqueeze(0)).sum(dim=-1)
        attn_weights = F.softmax(attn_scores, dim=1)
        
        # 加权聚合
        aggregated = (features * attn_weights.unsqueeze(-1)).sum(dim=1)
        
        return aggregated
 
 
class TemporalGNN(nn.Module):
    """时序GNN消息传递"""
    
    def __init__(self, dim, num_heads):
        super().__init__()
        self.num_heads = num_heads
        self.head_dim = dim // num_heads
        
        self.query_proj = nn.Linear(dim, dim)
        self.key_proj = nn.Linear(dim, dim)
        self.value_proj = nn.Linear(dim, dim)
        self.rel_proj = nn.Linear(dim, dim)
        
        self.out_proj = nn.Linear(dim, dim)
    
    def forward(self, graph_snapshot):
        """
        单个时间步的GNN前向传播
        """
        h = graph_snapshot.node_features  # [N, D]
        edges = graph_snapshot.edges       # [E, 3] (src, dst, rel_idx)
        
        # 多头注意力
        Q = self.query_proj(h).view(-1, self.num_heads, self.head_dim)
        K = self.key_proj(h).view(-1, self.num_heads, self.head_dim)
        V = self.value_proj(h).view(-1, self.num_heads, self.head_dim)
        
        # 边注意力(包含关系信息)
        rel_emb = self.rel_proj(graph_snapshot.rel_embeddings)
        edge_K = K[edges[:, 0]] + rel_emb
        edge_V = V[edges[:, 0]]
        
        # 聚合邻居信息
        # (这里简化了,实际需要更复杂的聚合逻辑)
        aggregated = self.aggregate(edge_V, edge_K, Q)
        
        return self.out_proj(aggregated)

6. 因果增强时序图表示学习(CEGRL-TKGR)

6.1 核心思想

CEGRL-TKGR5将因果推理引入时序知识图谱推理,区分相关性和因果性,实现更鲁棒的预测。

6.2 因果视角的时序KG

相关 ≠ 因果

示例:
- 冰淇淋销量 ↑ 和 溺水人数 ↑ 相关
- 但两者都是由"夏天"这个共同原因导致的
- 冰淇淋 → 溺水 是虚假相关

因果建模的优势:
- 更鲁棒的预测
- 可干预分析
- 回答"如果...会怎样"的问题

6.3 因果图结构

时序KG的因果图:

          ┌─────────────┐
          │  历史事件    │
          │  H_t        │
          └──────┬──────┘
                 │
     ┌───────────┼───────────┐
     │           │           │
     ▼           ▼           ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ 实体状态 │ │ 关系动态 │ │ 时间效应 │
│ S_t     │ │ R_t     │ │ T_t     │
└────┬────┘ └────┬────┘ └────┬────┘
     │           │           │
     └───────────┼───────────┘
                 │
                 ▼
          ┌─────────────┐
          │  未来预测    │
          │  Y_{t+1}   │
          └─────────────┘

6.4 因果干预建模

class CausalTemporalReasoner:
    """因果时序推理器"""
    
    def __init__(self, dim):
        self.dim = dim
        
        # 因果效应估计器
        self.causal_effect_net = CausalEffectNetwork(dim)
        
        # 时序预测器
        self.temporal_predictor = TemporalPredictor(dim)
    
    def estimate_causal_effect(self, historical_graphs, target_entity):
        """
        估计因果效应
        """
        # 1. 区分直接原因和混杂因素
        direct_causes = self.identify_direct_causes(
            historical_graphs, 
            target_entity
        )
        confounders = self.identify_confounders(
            historical_graphs
        )
        
        # 2. 计算因果效应
        # E[Y | do(X=x)] vs E[Y | X=x]
        causal_effect = self.causal_effect_net(
            direct_causes,
            confounders
        )
        
        return causal_effect
    
    def predict_with_intervention(self, graph, intervention):
        """
        干预预测:回答"如果...会怎样"
        """
        # 1. 移除混杂影响
        adjusted_graph = self.remove_confounding(graph)
        
        # 2. 应用干预
        intervened_graph = self.apply_intervention(
            adjusted_graph,
            intervention
        )
        
        # 3. 基于调整后的图进行预测
        prediction = self.temporal_predictor(intervened_graph)
        
        return prediction

7. 实践指南

7.1 数据集

数据集规模特点下载
ICEWS1423K事实年度事件数据Link
ICEWS1823K事实每日事件数据Link
GDELT100M+事实全球事件Link
Wikidata500K事实时序事实子集Link
YAGO100K事实含时间注释Link

7.2 方法选择指南

场景推荐方法理由
稀疏数据MATA记忆增强插值
统一插值外推DynaGen动态子图+扩散
分离长短期TeRDy频率分解
结构+时间SALoM双记忆架构
因果分析CEGRL-TKGR因果推断能力

7.3 实现注意事项

  1. 时间编码:选择合适的时间编码方法(sinusoidal vs learned)
  2. 时间粒度:根据任务选择合适的时间粒度
  3. 计算效率:大规模时序KG需要高效的批处理

8. 总结与展望

8.1 方法对比

方法核心创新优势局限
MATA记忆增强锚点处理稀疏数据记忆管理复杂
DynaGen动态子图+扩散统一插值外推扩散训练不稳定
TeRDy频率分解清晰的长短期分离需要足够长序列
SALoM双记忆架构综合结构时间信息记忆融合困难
CEGRL-TKGR因果增强可干预分析因果识别困难

8.2 未来方向

  1. 大规模时序KG:处理百万级实体和事件
  2. 实时更新:在线学习和增量更新
  3. 多模态时序KG:整合文本、图像等
  4. 跨语言时序KG:多语言知识图谱的对齐

8.3 开放问题

  • 如何有效捕捉长期依赖而不遗忘短期信息?
  • 时序KG中的因果关系如何识别?
  • 如何处理时间的不确定性?

参考文献


相关主题

Footnotes

  1. [ICLR 2026] MATA: Memory-Augmented Temporal Anchors for Sparse Temporal Knowledge Graphs

  2. [arXiv 2026] DynaGen: Dynamic Subgraph Generation with Diffusion Regularization for Temporal Knowledge Graphs

  3. [ACL 2025] TeRDy: Temporal Regularity Decomposition for Event Forecasting

  4. [NeurIPS 2025] SALoM: Structure-Aware Long and Short Memory for Temporal Knowledge Graphs

  5. [NeusymBridge 2025] CEGRL-TKGR: Causal Enhanced Graph Representation Learning for Temporal Knowledge Graph Reasoning