基于动态时间规整（Dynamic Time Warping, DTW）的文本序列相似度计算算法详解

我将为您详细讲解DTW在文本序列相似度计算中的应用。DTW是一种经典的序列对齐算法，虽然最初用于语音识别中的时间序列对齐，但在自然语言处理中也有重要应用，特别是在处理长度不等的文本序列时。

1. 算法背景与问题定义

在自然语言处理中，我们经常需要比较两个序列的相似度，例如：

• 句子对匹配（语义相似度计算）
• 时间序列文本（如用户行为序列）的对比
• 长度不同的词向量序列比较

传统方法（如欧氏距离、余弦相似度）要求序列长度相等，而DTW可以处理长度不同的序列，通过动态规划找到最优对齐路径。

核心问题：给定两个长度分别为n和m的序列：

• 序列A：a₁, a₂, ..., aₙ
• 序列B：b₁, b₂, ..., bₘ
  计算它们之间的“弯曲距离”，使相似的点对齐。

2. 算法核心思想

2.1 基本直觉

想象两个人在不同速度下说同一句话。DTW通过“拉伸”或“压缩”时间轴，使两个序列在时间上对齐，从而比较它们的内容相似性。

2.2 关键特性

1. 一对一或多对一匹配：允许一个序列中的点与另一个序列中的多个点匹配
2. 保持顺序：对齐必须保持序列的顺序性
3. 边界条件：必须从起点到终点完整匹配

3. 算法详细步骤

步骤1：构建距离矩阵

首先，我们需要一个基础的距离度量来计算两个序列点之间的“局部距离”。

距离计算示例：
假设我们比较两个句子：

• 句子1：["I", "love", "NLP"]
• 句子2：["I", "really", "like", "NLP"]

使用词向量余弦距离：

d("I", "I") = 0
d("love", "really") = 0.3
d("love", "like") = 0.2
...

构建n×m的距离矩阵D，其中D[i][j] = dist(aᵢ, bⱼ)

步骤2：动态规划递推

我们定义一个累加距离矩阵DTW，大小为n×m

递推公式：

DTW[1][1] = D[1][1]
DTW[i][j] = D[i][j] + min(DTW[i-1][j],    # 插入
                         DTW[i][j-1],    # 删除  
                         DTW[i-1][j-1])  # 匹配

其中：

• DTW[i-1][j]：序列A的点aᵢ重复使用（拉伸）
• DTW[i][j-1]：序列B的点bⱼ重复使用（压缩）
• DTW[i-1][j-1]：点对点匹配

步骤3：路径搜索

从DTW[n][m]开始回溯，找到最小代价路径：

路径 = []
i, j = n, m
while i > 1 or j > 1:
    路径.append((i, j))
    找到DTW[i-1][j], DTW[i][j-1], DTW[i-1][j-1]中最小的
    更新i, j
路径.append((1, 1))
反转路径

步骤4：计算最终距离

最终距离 = DTW[n][m] / (路径长度) # 可选归一化

4. 文本处理中的具体实现

4.1 词级别DTW

def dtw_distance(seq1, seq2, distance_func):
    """
    seq1: 第一个词序列 [word1, word2, ...]
    seq2: 第二个词序列
    distance_func: 词对距离函数
    """
    n, m = len(seq1), len(seq2)
    
    # 初始化距离矩阵
    d = [[0] * m for _ in range(n)]
    for i in range(n):
        for j in range(m):
            d[i][j] = distance_func(seq1[i], seq2[j])
    
    # DTW矩阵
    dtw = [[float('inf')] * m for _ in range(n)]
    dtw[0][0] = d[0][0]
    
    # 第一行
    for j in range(1, m):
        dtw[0][j] = d[0][j] + dtw[0][j-1]
    
    # 第一列
    for i in range(1, n):
        dtw[i][0] = d[i][0] + dtw[i-1][0]
    
    # 填充剩余
    for i in range(1, n):
        for j in range(1, m):
            cost = d[i][j]
            dtw[i][j] = cost + min(dtw[i-1][j],   # 插入
                                  dtw[i][j-1],   # 删除
                                  dtw[i-1][j-1]) # 匹配
    
    return dtw[n-1][m-1] / (n + m)  # 归一化

4.2 使用词向量的距离函数

def word_vector_distance(word1, word2, word_vectors):
    """基于预训练词向量的距离计算"""
    vec1 = word_vectors[word1]
    vec2 = word_vectors[word2]
    
    # 使用余弦距离
    cosine_sim = np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
    return 1 - cosine_sim

5. 优化与变体

5.1 窗口约束（Sakoe-Chiba Band）

限制对齐路径的偏离程度，避免不合理的匹配：

def dtw_with_window(seq1, seq2, window_size):
    window = max(window_size, abs(len(seq1) - len(seq2)))
    # 在动态规划时，限制|i-j| <= window

5.2 斜率约束

控制路径的斜率，避免过度拉伸：

• 类型I：最小斜率约束
• 类型II：最大斜率约束

5.3 快速DTW算法

通过多级分辨率加速计算：

1. 在粗粒度上计算DTW
2. 在细粒度上细化路径
3. 时间复杂度从O(nm)降低到O(n+m)

6. 文本序列的特定处理

6.1 处理不等长序列

对于句子对匹配：

句子A: "The cat sat on the mat" (6词)
句子B: "A cat is sitting on a mat" (7词)

对齐可能为：
The - A
cat - cat
sat - is
on - sitting
the - on
mat - a
     - mat

6.2 结合语义信息

使用预训练语言模型的表示：

def sentence_dtw(sent1, sent2, model):
    # 获取句子中每个词的上下文表示
    tokens1 = model.encode(sent1, output_hidden_states=True)
    tokens2 = model.encode(sent2, output_hidden_states=True)
    
    # 使用最后一层隐藏状态作为词表示
    seq1 = tokens1.last_hidden_state[0]  # 形状: [len1, dim]
    seq2 = tokens2.last_hidden_state[0]  # 形状: [len2, dim]
    
    # 计算DTW距离
    return dtw_distance(seq1, seq2, euclidean_distance)

7. 实际应用示例

示例1：短文本相似度计算

# 两个不等长句子
sentence1 = ["I", "like", "machine", "learning"]
sentence2 = ["I", "enjoy", "AI", "and", "deep", "learning"]

# 使用简单的词形距离
def simple_distance(word1, word2):
    if word1 == word2:
        return 0
    elif word1 in synonyms.get(word2, []):
        return 0.3
    else:
        return 1.0

distance = dtw_distance(sentence1, sentence2, simple_distance)
print(f"DTW距离: {distance}")

示例2：时间序列文本对齐

用户行为序列：

用户A: [登录, 搜索, 点击, 购买]
用户B: [登录, 浏览, 浏览, 搜索, 收藏, 购买]

DTW对齐：
登录-登录
搜索-浏览
点击-浏览
购买-搜索
     -收藏
     -购买

8. 算法复杂度与优化

时间复杂度

• 基本DTW：O(nm)
• 带窗口约束：O(nw)，w为窗口大小
• 快速DTW：O(n+m)

空间优化

使用滚动数组：

def dtw_space_optimized(seq1, seq2):
    prev_row = [0] * len(seq2)
    curr_row = [0] * len(seq2)
    
    for i in range(len(seq1)):
        for j in range(len(seq2)):
            # 计算并更新
            pass
        prev_row, curr_row = curr_row, prev_row

9. 与其他方法的对比

10. 在NLP中的典型应用

1. 文本对齐：平行语料对齐，翻译记忆库检索
2. 语音文本对齐：ASR输出与参考文本的对齐
3. 行为序列分析：用户行为模式的相似性比较
4. 时间序列分类：情感变化序列的分类
5. 异常检测：检测异常的行为或文本模式

11. 总结

DTW为文本序列相似度计算提供了一个强大的工具，特别是当：

1. 序列长度不同但语义相关
2. 需要考虑时间/顺序扭曲
3. 需要对齐点对点的对应关系

虽然计算成本较高，但通过窗口约束、快速算法等优化，DTW在实际应用中仍然具有重要价值。在深度学习时代，DTW的思想也被融入到注意力机制等现代架构中，继续发挥着重要作用。