自注意力机制全解析——从原理到计算细节，一文尽览！

import torch
import torch.nn as nn
import torch.nn.functional as F
from math import sqrt

class SelfAttention(nn.Module):
    def __init__(self, embed_size, heads):
        super(SelfAttention, self).__init__()
        self.embed_size = embed_size
        self.heads = heads
        self.head_dim = embed_size // heads

        # Ensure the embedding size is divisible by the number of heads
        assert (self.head_dim * heads == embed_size), "Embedding size must be divisible by heads"

        # Define linear transformations for queries, keys, and values
        self.values = nn.Linear(embed_size, embed_size, bias=False)
        self.keys = nn.Linear(embed_size, embed_size, bias=False)
        self.queries = nn.Linear(embed_size, embed_size, bias=False)
        self.fc_out = nn.Linear(embed_size, embed_size)

    def forward(self, x):
        N = x.shape[0]  # Batch size
        length = x.shape[1]  # Sequence length

        # Pass inputs through linear layers for queries, keys, and values
        values = self.values(x)
        keys = self.keys(x)
        queries = self.queries(x)

        # Split into multiple heads
        values = values.view(N, length, self.heads, self.head_dim)
        keys = keys.view(N, length, self.heads, self.head_dim)
        queries = queries.view(N, length, self.heads, self.head_dim)

        # Permute dimensions to put heads second
        values = values.permute(0, 2, 1, 3)  # (N, heads, length, head_dim)
        keys = keys.permute(0, 2, 1, 3)
        queries = queries.permute(0, 2, 1, 3)

        # Compute scaled dot-product attention
        energy = torch.einsum("nqhd,nkhd->nhqk", [queries, keys])  # (N, heads, query_len, key_len)
        attention = F.softmax(energy / (self.embed_size ** (1 / 2)), dim=3)  # Softmax on the last dimension

        # Weighted sum
        out = torch.einsum("nhql,nlhd->nqhd", [attention, values]).reshape(
            N, length, self.heads * self.head_dim
        )  # (N, length, embed_size)

        return self.fc_out(out)

# Example usage
embed_size = 256  # Embedding dimension
heads = 8  # Number of attention heads
x = torch.rand(64, 10, embed_size)  # Simulate input data

self_attention = SelfAttention(embed_size, heads)
output = self_attention(x)

print(output.shape)  # Should print: torch.Size([64, 10, 256])

import torch
import torch.nn as nn
from transformers import BertTokenizer, BertModel

# 初始化BERT模型和分词器
tokenizer = BertTokenizer.from_pretrained('bert-base-multilingual-cased')
model = BertModel.from_pretrained('bert-base-multilingual-cased')

def translate_sentence(sentence, model, tokenizer):
    # 对输入句子进行编码
    inputs = tokenizer(sentence, return_tensors="pt")
    
    # 获取BERT模型的输出
    outputs = model(**inputs)
    last_hidden_states = outputs.last_hidden_state
    
    # 这里可以加入额外的解码逻辑，例如使用一个预训练好的翻译模型
    # 简单起见，我们直接返回最后一层隐藏状态作为示例
    return last_hidden_states

# 示例翻译
sentence = "Hello, how are you?"
translated = translate_sentence(sentence, model, tokenizer)
print(translated.shape)  # 输出形状应为 (batch_size, sequence_length, hidden_dim)

import pandas as pd
from sklearn.preprocessing import LabelEncoder

# 假设我们有一个包含用户行为的数据框 df
df = pd.read_csv('user_behavior.csv')

# 对用户和物品ID进行编码
user_encoder = LabelEncoder()
item_encoder = LabelEncoder()

df['user_id'] = user_encoder.fit_transform(df['original_user_id'])
df['item_id'] = item_encoder.fit_transform(df['original_item_id'])

# 将数据按用户分组，并按照时间排序
user_sequences = df.groupby('user_id')['item_id'].apply(list).reset_index(name='sequence')

import tensorflow as tf
from tensorflow.keras.layers import Embedding, LayerNormalization, Dense
from tensorflow.keras.models import Model
import numpy as np

class SASRec(Model):
    def __init__(self, num_users, num_items, embedding_dim, max_len, num_heads=1, num_blocks=2):
        super(SASRec, self).__init__()
        self.user_embedding = Embedding(input_dim=num_users + 1, output_dim=embedding_dim)
        self.item_embedding = Embedding(input_dim=num_items + 1, output_dim=embedding_dim)
        self.positional_encoding = self.get_positional_encoding(max_len, embedding_dim)
        self.attention_layers = [tf.keras.layers.MultiHeadAttention(num_heads=num_heads, key_dim=embedding_dim) for _ in range(num_blocks)]
        self.layer_norms = [LayerNormalization() for _ in range(num_blocks)]
        self.dense = Dense(num_items, activation='softmax')

    def get_positional_encoding(self, seq_len, d_model):
        # 创建位置编码矩阵
        position = np.arange(seq_len)[:, np.newaxis]
        div_term = np.exp(np.arange(0, d_model, 2) * -(np.log(10000.0) / d_model))
        pe = np.zeros((seq_len, d_model))
        pe[:, 0::2] = np.sin(position * div_term)
        pe[:, 1::2] = np.cos(position * div_term)
        return pe[np.newaxis, ...]

    def call(self, inputs):
        user_ids, sequences = inputs
        seq_emb = self.item_embedding(sequences) + self.positional_encoding[:, :tf.shape(sequences)[1], :]
        mask = tf.cast(tf.not_equal(sequences, 0), dtype=tf.float32)
        
        for i in range(len(self.attention_layers)):
            seq_emb = self.layer_norms[i](seq_emb + self.attention_layers[i](seq_emb, seq_emb, attention_mask=mask))
        
        logits = self.dense(seq_emb)
        return logits

# 初始化SASRec模型
num_users = len(user_encoder.classes_)
num_items = len(item_encoder.classes_)
embedding_dim = 128
max_len = 50  # 用户行为序列的最大长度

model = SASRec(num_users, num_items, embedding_dim, max_len)

from tensorflow.keras.optimizers import Adam

# 编写简单的负采样函数以生成负样本
def negative_sampling(pos_seq, num_items, num_neg_samples=5):
    neg_samples = []
    for pos_item in pos_seq:
        neg_items = np.random.choice(num_items, size=num_neg_samples, replace=False)
        neg_samples.append(neg_items)
    return np.array(neg_samples)

# 定义损失函数（交叉熵）
loss_fn = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)

# 选择优化器
optimizer = Adam(learning_rate=0.001)

# 训练模型
for epoch in range(epochs):
    for batch in data_generator():
        with tf.GradientTape() as tape:
            user_ids, sequences, labels = batch
            predictions = model([user_ids, sequences])
            loss = loss_fn(labels, predictions)
        
        gradients = tape.gradient(loss, model.trainable_variables)
        optimizer.apply_gradients(zip(gradients, model.trainable_variables))
    
    print(f'Epoch {epoch + 1}/{epochs}, Loss: {loss.numpy()}')

def recommend_items(model, user_id, user_history, top_k=10):
    # 对用户行为序列进行填充以匹配最大长度
    sequence = np.pad(user_history[-max_len:], (max_len - len(user_history[-max_len:]), 0), 'constant')
    
    # 获取预测结果
    predictions = model.predict([[user_id], [sequence]])
    sorted_indices = np.argsort(predictions[0])[-top_k:]
    
    # 返回解码后的商品ID
    recommended_items = item_encoder.inverse_transform(sorted_indices)
    return recommended_items.tolist()

# 示例推荐
user_id = 1234
user_history = [234, 567, 890]  # 用户的历史行为记录
recommendations = recommend_items(model, user_id, user_history)
print(recommendations)  # 输出推荐商品ID