-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathutils.py
179 lines (143 loc) · 6.53 KB
/
utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
import random
import numpy as np
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.autograd import Variable
from torch.nn.utils.rnn import pad_sequence
def seed_everything(seed):
random.seed(seed)
np.random.seed(seed)
torch.manual_seed(seed)
torch.cuda.manual_seed(seed)
torch.cuda.manual_seed_all(seed)
torch.backends.cudnn.benchmark = False
torch.backends.cudnn.deterministic = True
def pad(tensor, length, no_cuda):
if isinstance(tensor, Variable):
var = tensor
if length > var.size(0):
if not no_cuda:
return torch.cat([var, torch.zeros(length - var.size(0), *var.size()[1:]).cuda()])
else:
return torch.cat([var, torch.zeros(length - var.size(0), *var.size()[1:])])
else:
return var
else:
if length > tensor.size(0):
if not no_cuda:
return torch.cat([tensor, torch.zeros(length - tensor.size(0), *tensor.size()[1:]).cuda()])
else:
return torch.cat([tensor, torch.zeros(length - tensor.size(0), *tensor.size()[1:])])
else:
return tensor
def edge_perms(l, window_past, window_future):
"""
Method to construct the edges considering the past and future window.
"""
all_perms = set()
array = np.arange(l)
for j in range(l):
perms = set()
if window_past == -1 and window_future == -1:
eff_array = array
elif window_past == -1:
eff_array = array[:min(l, j + window_future + 1)]
elif window_future == -1:
eff_array = array[max(0, j - window_past):]
else:
eff_array = array[max(0, j - window_past):min(l, j + window_future + 1)]
for item in eff_array:
perms.add((j, item))
all_perms = all_perms.union(perms)
return list(all_perms)
def batch_graphify(features, qmask, lengths, window_past, window_future, edge_type_mapping, att_model, no_cuda):
"""
Method to prepare the data format required for the GCN network. Pytorch geometric puts all nodes for classification
in one single graph. Following this, we create a single graph for a mini-batch of dialogue instances. This method
ensures that the various graph indexing is properly carried out so as to make sure that, utterances (nodes) from
each dialogue instance will have edges with utterances in that same dialogue instance, but not with utternaces
from any other dialogue instances in that mini-batch.
"""
edge_index, edge_norm, edge_type, node_features = [], [], [], []
batch_size = features.size(1)
length_sum = 0
edge_ind = []
edge_index_lengths = []
for j in range(batch_size):
edge_ind.append(edge_perms(lengths[j], window_past, window_future))
# scores are the edge weights!!!
scores = att_model(features, lengths, edge_ind)
for j in range(batch_size):
node_features.append(features[:lengths[j], j, :])
perms1 = edge_perms(lengths[j], window_past, window_future)
perms2 = [(item[0] + length_sum, item[1] + length_sum) for item in perms1] # 对当前dialogue在batch图中的真实索引
length_sum += lengths[j]
edge_index_lengths.append(len(perms1))
for item1, item2 in zip(perms1, perms2):
edge_index.append(torch.tensor([item2[0], item2[1]]))
edge_norm.append(scores[j, item1[0], item1[1]])
# qmask: speaker mask
# umask: utterance mask
speaker0 = (qmask[item1[0], j, :] == 1).nonzero()[0][0].tolist()
speaker1 = (qmask[item1[1], j, :] == 1).nonzero()[0][0].tolist()
if item1[0] < item1[1]:
edge_type.append(edge_type_mapping[str(speaker0) + str(speaker1) + '0'])
else:
edge_type.append(edge_type_mapping[str(speaker0) + str(speaker1) + '1']) # 确定边类型
node_features = torch.cat(node_features, dim=0)
edge_index = torch.stack(edge_index).transpose(0, 1)
edge_norm = torch.stack(edge_norm)
edge_type = torch.tensor(edge_type)
if not no_cuda:
node_features = node_features.cuda()
edge_index = edge_index.cuda()
edge_norm = edge_norm.cuda()
edge_type = edge_type.cuda()
return node_features, edge_index, edge_norm, edge_type, edge_index_lengths
def attentive_node_features(emotions, seq_lengths, umask, matchatt_layer, no_cuda):
"""
Method to obtain attentive node features over the graph convoluted features, as in Equation 4, 5, 6. in the paper.
"""
input_conversation_length = torch.tensor(seq_lengths)
start_zero = input_conversation_length.data.new(1).zero_()
if not no_cuda:
input_conversation_length = input_conversation_length.cuda()
start_zero = start_zero.cuda()
max_len = max(seq_lengths)
start = torch.cumsum(torch.cat((start_zero, input_conversation_length[:-1])), 0)
emotions = torch.stack([pad(emotions.narrow(0, s, l), max_len, no_cuda)
for s, l in zip(start.data.tolist(),
input_conversation_length.data.tolist())], 0).transpose(0, 1) # ???
alpha, alpha_f, alpha_b = [], [], []
att_emotions = []
for t in emotions:
att_em, alpha_ = matchatt_layer(emotions, t, mask=umask)
att_emotions.append(att_em.unsqueeze(0))
alpha.append(alpha_[:, 0, :])
att_emotions = torch.cat(att_emotions, dim=0)
return att_emotions
def classify_node_features(emotions, seq_lengths, umask, matchatt_layer, linear_layer, dropout_layer,
smax_fc_layer, nodal_attn, avec, no_cuda):
"""
Function for the final classification, as in Equation 7, 8, 9. in the paper.
"""
if nodal_attn:
emotions = attentive_node_features(emotions, seq_lengths, umask, matchatt_layer, no_cuda)
# seq_lengths: 真实的 utterances 数量
hidden = F.relu(linear_layer(emotions))
hidden = dropout_layer(hidden)
hidden = smax_fc_layer(hidden)
if avec:
return torch.cat([hidden[:, j, :][:seq_lengths[j]] for j in range(len(seq_lengths))])
log_prob = F.log_softmax(hidden, 2)
log_prob = torch.cat([log_prob[:, j, :][:seq_lengths[j]] for j in range(len(seq_lengths))])
return log_prob
else:
hidden = F.relu(linear_layer(emotions))
hidden = dropout_layer(hidden)
hidden = smax_fc_layer(hidden)
if avec:
return hidden
log_prob = F.log_softmax(hidden, 1)
return log_prob