基于 Triplet Loss 正确地将 3 个输入提交给 Keras 模型
Correctly submitting 3 inputs to a Keras model based on Triplet Loss
我正在研究一个由两部分组成的模型，正如我在 this question 中所讨论的那样：第一部分应该采用三元组的元素（由一个锚、一个正例和一个负例组成， FaceNet 采用相同的原理）并将它们转换为向量（word2vec + lstm），而第二个应该采用这些向量并使用它们来计算三元组损失。我开始编写一些代码，这是我现在拥有的：
import pandas as pd
import numpy as np
import tensorflow as tf
from nltk.tokenize import WordPunctTokenizer
from collections import Counter
from string import punctuation, ascii_lowercase
import regex as re
from tqdm import tqdm
from gensim.models import Word2Vec
from keras.preprocessing.sequence import pad_sequences
from keras.layers import Dense, Input, LSTM, Embedding, Dropout, SpatialDropout1D, Bidirectional, concatenate, Lambda
from keras.models import Model
from keras.optimizers import Adam
from keras.layers.normalization import BatchNormalization
from keras.utils import plot_model
# Constants and initial settings
path = 'Datasets/DBLP-ACM/'
tf.compat.v1.set_random_seed(1)
ALPHA = 0.2
TRIPLETS_DATA_FILE = path + 'triplets/random_triplets.csv'
MAX_SEQUENCE_LENGTH = 300
tokenizer = WordPunctTokenizer()
vocab = Counter()
# Tokenize the text
def text_to_wordlist(text, lower=False):
# Tokenize
text = tokenizer.tokenize(text)
# Optional: lower case
if lower: text = [t.lower() for t in text]
# Return a list of words
vocab.update(text)
return text
# Process data
def process_triplets(list_sentences, lower=False):
triplet_elements = []
for text in tqdm(list_sentences):
txt = text_to_wordlist(text, lower=lower)
triplet_elements.append(txt)
return triplet_elements
# Define the custom loss (Triplet Loss)
def triplet_loss(x):
anchor, positive, negative = x
pos_dist = tf.reduce_sum(tf.square(tf.subtract(anchor, positive)), 1)
neg_dist = tf.reduce_sum(tf.square(tf.subtract(anchor, negative)), 1)
basic_loss = tf.add(tf.subtract(pos_dist, neg_dist), ALPHA)
loss = tf.reduce_mean(tf.maximum(basic_loss, 0.0), 0)
return loss
# Build the embedding model
def build_embedding_model(): # How can i feed the input to the word2vec part?
# Inputs
wv_layer = Embedding(nb_words, WV_DIM, mask_zero=False, weights=[wv_matrix], input_length=MAX_SEQUENCE_LENGTH, trainable=False)
embedding_input = Input(shape=(MAX_SEQUENCE_LENGTH,), dtype='int32')
embedded_sequences = wv_layer(embedding_input)
# BiGRU (aka bidirectional gru, bidirectional LSTM)
embedded_sequences = SpatialDropout1D(0.2)(embedded_sequences)
x = Bidirectional(LSTM(64, return_sequences=False))(embedded_sequences)
x = Dropout(0.2)(x)
x = BatchNormalization()(x)
# Output
preds = Dense(1, activation='sigmoid')(x) # Just one output class (dummy)
# Build the model
model = Model(inputs=[embedding_input], outputs=preds)
model.compile(loss='mse', optimizer = "adam")
return model
# Build the entire model
def build_model():
# Inputs
anchor_input = Input(shape=(MAX_SEQUENCE_LENGTH,), name='anchor_input')
positive_input = Input(shape=(MAX_SEQUENCE_LENGTH,), name='positive_input')
negative_input = Input(shape=(MAX_SEQUENCE_LENGTH,), name='negative_input')
embedding_model = build_embedding_model()
# Outputs
anchor_embedding = embedding_model(anchor_input)
positive_embedding = embedding_model(positive_input)
negative_embedding = embedding_model(negative_input)
merged_output = concatenate([anchor_embedding, positive_embedding, negative_embedding])
loss = Lambda(triplet_loss, (1,))(merged_output)
triplet_model = Model(inputs=[anchor_input, positive_input, negative_input], outputs=loss)
triplet_model.compile(loss = 'mean_absolute_error', optimizer = Adam())
return triplet_model
triplets = pd.read_csv(TRIPLETS_DATA_FILE, error_bad_lines=False, sep="|", quotechar="\"", encoding="latin_1")
list_sentences_anchor = list((triplets["anchor"].astype(str)).fillna("").values)
list_sentences_positive = list((triplets["positive"].astype(str)).fillna("").values)
list_sentences_negative = list((triplets["negative"].astype(str)).fillna("").values)
# Fill an array for anchors, one for positives and one for negatives
anchors = process_triplets(list_sentences_anchor, lower=True)
positives = process_triplets(list_sentences_positive, lower=True)
negatives = process_triplets(list_sentences_negative, lower=True)
model_anchor = Word2Vec(anchors, size=100, window=5, min_count=5, workers=16, sg=0, negative=5)
model_positive = Word2Vec(positives, size=100, window=5, min_count=5, workers=16, sg=0, negative=5)
model_negative = Word2Vec(negatives, size=100, window=5, min_count=5, workers=16, sg=0, negative=5)
word_vectors_anchor = model_anchor.wv
word_vectors_positive = model_positive.wv
word_vectors_negative = model_negative.wv
# Use the embeddings in Keras
MAX_NB_WORDS = max(len(word_vectors_anchor.vocab), len(word_vectors_positive.vocab), len(word_vectors_negative.vocab))
word_index = {t[0]: i+1 for i,t in enumerate(vocab.most_common(MAX_NB_WORDS))}
sequences = [[word_index.get(t, 0) for t in anchor] for anchor in anchors[:len(anchors)]]
# Pad
anchor_data = pad_sequences(sequences, maxlen=MAX_SEQUENCE_LENGTH, padding="pre", truncating="post")
# Create the embedding matrix
WV_DIM = 200
nb_words = min(MAX_NB_WORDS, len(word_vectors_anchor.vocab))
# Initialize the matrix with random numbers
wv_matrix = (np.random.rand(nb_words, WV_DIM) - 0.5) / 5.0
for word, i in word_index.items():
if i >= MAX_NB_WORDS: continue
try: # Words not found in embedding index will be all-zeros
embedding_vector = word_vectors_anchor[word]
wv_matrix[i] = embedding_vector
except: pass
# Build and fit the model
triplet_model = build_model()
hist = triplet_model.fit([anchor_data, anchor_data, anchor_data], 0, validation_split=0.1, epochs=50, batch_size=256, shuffle=True)
正如您肯定会看到的那样，存在很多混乱。基本上，我将三元组分成 3 个不同的部分，我在每个部分上应用 word2vec，然后在嵌入模型中使用结果（我使用相同的结果 3 次只是为了测试它是否有效，但没有）。
嵌入模型应计算要在第二个模型、拟合过程中和三元组损失中使用的向量。我是 Keras 的新手，我肯定在这里做错了什么，因为我现在收到这个错误：
TypeError: Tensor objects are only iterable when eager execution is enabled. To iterate over this tensor use tf.map_fn.
这发生在三元组损失函数本身的第一行，可能与输入格式有关。所以问题是：给定这段代码，我如何修改它以正确接受 3 个输入，生成 3 个向量，并在拟合期间在 triplet_model 中使用这些向量？
如果我修改代码或遇到不同的错误，我会更新问题。
损失不应该是 Lambda 层。删除 Lambda 层并更新您的代码，以便：
triplet_model = Model(inputs=[anchor_input, positive_input, negative_input], outputs=merged_output)
triplet_model.compile(loss = triplet_loss, optimizer = Adam())
triplet_loss需要定义为：
def triplet_loss(y_true, y_pred):
anchor_vec = y_pred[:, :VECTOR_SIZE]
positive_vec = y_pred[:, VECTOR_SIZE:2*VECTOR_SIZE]
negative_vec = y_pred[:, 2*VECTOR_SIZE:]
... code ...
然后您应该忽略 y_true。
有效的三重损失函数的完整示例：
https://colab.research.google.com/drive/1VgOTzr_VZNHkXh2z9IiTAcEgg5qr19y0
我正在研究一个由两部分组成的模型，正如我在 this question 中所讨论的那样：第一部分应该采用三元组的元素（由一个锚、一个正例和一个负例组成， FaceNet 采用相同的原理）并将它们转换为向量（word2vec + lstm），而第二个应该采用这些向量并使用它们来计算三元组损失。我开始编写一些代码，这是我现在拥有的：
import pandas as pd
import numpy as np
import tensorflow as tf
from nltk.tokenize import WordPunctTokenizer
from collections import Counter
from string import punctuation, ascii_lowercase
import regex as re
from tqdm import tqdm
from gensim.models import Word2Vec
from keras.preprocessing.sequence import pad_sequences
from keras.layers import Dense, Input, LSTM, Embedding, Dropout, SpatialDropout1D, Bidirectional, concatenate, Lambda
from keras.models import Model
from keras.optimizers import Adam
from keras.layers.normalization import BatchNormalization
from keras.utils import plot_model
# Constants and initial settings
path = 'Datasets/DBLP-ACM/'
tf.compat.v1.set_random_seed(1)
ALPHA = 0.2
TRIPLETS_DATA_FILE = path + 'triplets/random_triplets.csv'
MAX_SEQUENCE_LENGTH = 300
tokenizer = WordPunctTokenizer()
vocab = Counter()
# Tokenize the text
def text_to_wordlist(text, lower=False):
# Tokenize
text = tokenizer.tokenize(text)
# Optional: lower case
if lower: text = [t.lower() for t in text]
# Return a list of words
vocab.update(text)
return text
# Process data
def process_triplets(list_sentences, lower=False):
triplet_elements = []
for text in tqdm(list_sentences):
txt = text_to_wordlist(text, lower=lower)
triplet_elements.append(txt)
return triplet_elements
# Define the custom loss (Triplet Loss)
def triplet_loss(x):
anchor, positive, negative = x
pos_dist = tf.reduce_sum(tf.square(tf.subtract(anchor, positive)), 1)
neg_dist = tf.reduce_sum(tf.square(tf.subtract(anchor, negative)), 1)
basic_loss = tf.add(tf.subtract(pos_dist, neg_dist), ALPHA)
loss = tf.reduce_mean(tf.maximum(basic_loss, 0.0), 0)
return loss
# Build the embedding model
def build_embedding_model(): # How can i feed the input to the word2vec part?
# Inputs
wv_layer = Embedding(nb_words, WV_DIM, mask_zero=False, weights=[wv_matrix], input_length=MAX_SEQUENCE_LENGTH, trainable=False)
embedding_input = Input(shape=(MAX_SEQUENCE_LENGTH,), dtype='int32')
embedded_sequences = wv_layer(embedding_input)
# BiGRU (aka bidirectional gru, bidirectional LSTM)
embedded_sequences = SpatialDropout1D(0.2)(embedded_sequences)
x = Bidirectional(LSTM(64, return_sequences=False))(embedded_sequences)
x = Dropout(0.2)(x)
x = BatchNormalization()(x)
# Output
preds = Dense(1, activation='sigmoid')(x) # Just one output class (dummy)
# Build the model
model = Model(inputs=[embedding_input], outputs=preds)
model.compile(loss='mse', optimizer = "adam")
return model
# Build the entire model
def build_model():
# Inputs
anchor_input = Input(shape=(MAX_SEQUENCE_LENGTH,), name='anchor_input')
positive_input = Input(shape=(MAX_SEQUENCE_LENGTH,), name='positive_input')
negative_input = Input(shape=(MAX_SEQUENCE_LENGTH,), name='negative_input')
embedding_model = build_embedding_model()
# Outputs
anchor_embedding = embedding_model(anchor_input)
positive_embedding = embedding_model(positive_input)
negative_embedding = embedding_model(negative_input)
merged_output = concatenate([anchor_embedding, positive_embedding, negative_embedding])
loss = Lambda(triplet_loss, (1,))(merged_output)
triplet_model = Model(inputs=[anchor_input, positive_input, negative_input], outputs=loss)
triplet_model.compile(loss = 'mean_absolute_error', optimizer = Adam())
return triplet_model
triplets = pd.read_csv(TRIPLETS_DATA_FILE, error_bad_lines=False, sep="|", quotechar="\"", encoding="latin_1")
list_sentences_anchor = list((triplets["anchor"].astype(str)).fillna("").values)
list_sentences_positive = list((triplets["positive"].astype(str)).fillna("").values)
list_sentences_negative = list((triplets["negative"].astype(str)).fillna("").values)
# Fill an array for anchors, one for positives and one for negatives
anchors = process_triplets(list_sentences_anchor, lower=True)
positives = process_triplets(list_sentences_positive, lower=True)
negatives = process_triplets(list_sentences_negative, lower=True)
model_anchor = Word2Vec(anchors, size=100, window=5, min_count=5, workers=16, sg=0, negative=5)
model_positive = Word2Vec(positives, size=100, window=5, min_count=5, workers=16, sg=0, negative=5)
model_negative = Word2Vec(negatives, size=100, window=5, min_count=5, workers=16, sg=0, negative=5)
word_vectors_anchor = model_anchor.wv
word_vectors_positive = model_positive.wv
word_vectors_negative = model_negative.wv
# Use the embeddings in Keras
MAX_NB_WORDS = max(len(word_vectors_anchor.vocab), len(word_vectors_positive.vocab), len(word_vectors_negative.vocab))
word_index = {t[0]: i+1 for i,t in enumerate(vocab.most_common(MAX_NB_WORDS))}
sequences = [[word_index.get(t, 0) for t in anchor] for anchor in anchors[:len(anchors)]]
# Pad
anchor_data = pad_sequences(sequences, maxlen=MAX_SEQUENCE_LENGTH, padding="pre", truncating="post")
# Create the embedding matrix
WV_DIM = 200
nb_words = min(MAX_NB_WORDS, len(word_vectors_anchor.vocab))
# Initialize the matrix with random numbers
wv_matrix = (np.random.rand(nb_words, WV_DIM) - 0.5) / 5.0
for word, i in word_index.items():
if i >= MAX_NB_WORDS: continue
try: # Words not found in embedding index will be all-zeros
embedding_vector = word_vectors_anchor[word]
wv_matrix[i] = embedding_vector
except: pass
# Build and fit the model
triplet_model = build_model()
hist = triplet_model.fit([anchor_data, anchor_data, anchor_data], 0, validation_split=0.1, epochs=50, batch_size=256, shuffle=True)
正如您肯定会看到的那样，存在很多混乱。基本上，我将三元组分成 3 个不同的部分，我在每个部分上应用 word2vec，然后在嵌入模型中使用结果（我使用相同的结果 3 次只是为了测试它是否有效，但没有）。
嵌入模型应计算要在第二个模型、拟合过程中和三元组损失中使用的向量。我是 Keras 的新手，我肯定在这里做错了什么，因为我现在收到这个错误：
TypeError: Tensor objects are only iterable when eager execution is enabled. To iterate over this tensor use tf.map_fn.
这发生在三元组损失函数本身的第一行，可能与输入格式有关。所以问题是：给定这段代码，我如何修改它以正确接受 3 个输入，生成 3 个向量，并在拟合期间在 triplet_model 中使用这些向量？
如果我修改代码或遇到不同的错误，我会更新问题。
损失不应该是 Lambda 层。删除 Lambda 层并更新您的代码，以便：
triplet_model = Model(inputs=[anchor_input, positive_input, negative_input], outputs=merged_output)
triplet_model.compile(loss = triplet_loss, optimizer = Adam())
triplet_loss需要定义为：
def triplet_loss(y_true, y_pred):
anchor_vec = y_pred[:, :VECTOR_SIZE]
positive_vec = y_pred[:, VECTOR_SIZE:2*VECTOR_SIZE]
negative_vec = y_pred[:, 2*VECTOR_SIZE:]
... code ...
然后您应该忽略 y_true。
有效的三重损失函数的完整示例： https://colab.research.google.com/drive/1VgOTzr_VZNHkXh2z9IiTAcEgg5qr19y0