word_gen = jieba.cut("我的名字叫Karry，来自计算机科学技术学院")

for word in word_gen:

    print(word)

from gensim.models import KeyedVectors

model_path = 'sgns.weibo.word.bz2'

model = KeyedVectors.load_word2vec_format(model_path)

similarity = model.similarity('公交', '地铁')

print('公交和地铁的相似度', similarity)

model.most_similar(positive=['男人', '女孩'], negative=['男孩'], topn=10)

import pandas as pd

import jieba

from gensim.models import Word2Vec

comments = pd.read_csv('./data/online_shopping_10_cats.csv', encoding='utf-8')

reviews = comments['review']

reviews = reviews.dropna()

sentences = [[token for token in jieba.__lcut(review) if token.strip() != ''] for review in reviews]

model = Word2Vec(

    sentences,            # 已分词的句子序列

    vector_size=100,      # 词向量维度

    window=5,             # 上下文窗口大小

    min_count=2,          # 最小词频（低于将被忽略）

    sg=1,                 # 1:Skip-Gram，0:CBOW

    workers=4             # 并行训练线程数

)

model.wv.save_word2vec_format('./data/word2vec.txt')

import torch

from torch import nn

from gensim.models import KeyedVectors

# 加载词向量

wv = KeyedVectors.load_word2vec_format('./data/word2vec.txt')

# 构建词向量矩阵

num_embedding = len(wv.key_to_index)

embedding_dim = wv.vector_size

embedding_matrix = torch.randn(num_embedding, embedding_dim)

for word, index in wv.key_to_index.items():

    embedding_matrix[index] = torch.from_numpy(wv[word])

print(embedding_matrix.shape)

# 创建 Embedding

embedding = nn.Embedding.from_pretrained(embedding_matrix)

text = '我喜欢乘坐地铁'

tokens= jieba.lcut(text)

input_ids = [wv.key_to_index[token] for token in tokens if token in wv.key_to_index]

input_tensor = torch.LongTensor(input_ids)

embedding(input_tensor)

import torch.nn as nn

# 双层双向 RNN

rnn = nn.RNN(input_size=3, hidden_size=4, num_layers=2, batch_first=True, bidirectional=True)

# shape: (batch, seq_len, input_size)

input = torch.randn(2, 4, 3)

output, hn = rnn(input)

print(output.shape)

print(hn.shape)

import torch

import jieba

import torch.nn as nn

def dm01():

    text = "当您登录平台即表示您接受本隐私政策的全部内容，并同意平台按本政策收集、使用和保护您的相关个人信息。"

    words = jieba.lcut(text)

    print(words)

    '''

    len(words) : 词表大小

    embedding_dim : 词向量维度

    '''

    embed = nn.Embedding(len(words), embedding_dim=4)

    for i, word in enumerate(words):

        # 词索引（张量）转变为词向量

        word2Vec = embed(torch.tensor([i]))

        print(word, word2Vec)

if __name__ == '__main__':

    dm01()

import torch

import torch.nn as nn

'''

input_size：词向量的维度

hidden_size：隐藏状态的维度

num_layers：隐藏层数

batch_first：批次优先

'''

rnn = nn.RNN(input_size=10, hidden_size=20, num_layers=1)

'''

5：有5个词

3：3个批次

10：每个词的细节是10维度

'''

x = torch.rand(5, 3, 10)

'''

1：隐藏层层数

3：句子数量

20：有20个循环维度，也就是隐藏状态的维度

'''

h0 = torch.zeros(1, 3, 20)

'''

RNN处理

x：本次的输入

h0：上一次的隐藏状态

输出

output：每个时间步的输出，包含了所有时间步骤的隐藏状态

h1：最后一次隐藏状态，大脑里最新的剧情

'''

output, h1 = rnn(x, h0)

print(f'output_shape:{output.shape}')

print(f'h1_shape:{h1.shape}')

# _*_ coding : utf-8 _*_

# @Time : 2025/10/31 10:02

# @Author : KarryLiu

# File : 歌词生成

# @Project : pytorchSTU

import math

import torch

import jieba

from torch.utils.data import DataLoader, Dataset

import torch.nn as nn

import torch.optim as optim

import time

from tqdm import tqdm

"""

实现步骤：

    1. 获取数据，进行分词

    2. 获取词表，构建数据集

    3. 搭建RNN神经网络模型

    4. 训练模型

    5. 模型预测

    6. 测试

"""

def build_vocab():

    unique_words, all_words = [], []

    with open('data/jaychou_lyrics.txt', 'r', encoding='utf-8') as f:

        for line in tqdm(f, desc='正在处理数据', total=5819):

            words = jieba.lcut(line)

            all_words.append(words)

            for word in words:

                if word not in unique_words:

                    unique_words.append(word)

    word2index = {word: index for index, word in enumerate(unique_words)}

    index2word = {index: word for index, word in enumerate(unique_words)}

    # 将词表转为索引，【0，1，2，3】，这个是一句歌词

    corpus_idx = []

    for all_word in all_words:

        tmp = []

        for word in all_word:

            tmp.append(word2index[word])

        tmp.append(word2index[' '])

        corpus_idx.extend(tmp)

    word_count = len(unique_words)

    return unique_words, word2index, word_count, corpus_idx

class LyricsDataset(Dataset):

    def __init__(self, corpus_idx, num_chars):

        super().__init__()

        self.corpus_idx = corpus_idx

        self.num_chars = num_chars

        self.word_count = len(corpus_idx)

        self.sentence_len = self.word_count // num_chars

    def __len__(self):

        return self.sentence_len

    def __getitem__(self, index):

        start = min(max(index, 0), self.word_count - self.num_chars - 1)

        end = start + self.num_chars

        sentence_x = self.corpus_idx[start:end]

        sentence_y = self.corpus_idx[start + 1:end + 1]

        sentence_x = torch.tensor(sentence_x, dtype=torch.long)

        sentence_y = torch.tensor(sentence_y, dtype=torch.long)

        return sentence_x, sentence_y

class TextGenerator(nn.Module):

    def __init__(self, unique_words_count, ):

        super().__init__()

        self.embedding = nn.Embedding(unique_words_count, 128)

        self.rnn = nn.RNN(128, 256, 1)

        self.linear = nn.Linear(256, unique_words_count)

    def forward(self, input, hidden):

        embed = self.embedding(input)

        output, hidden = self.rnn(embed.transpose(0, 1), hidden)

        output = self.linear(output.reshape(shape=(-1, output.shape[-1])))

        return output, hidden

    def init_hidden(self, batch_size):

        return torch.zeros(1, batch_size, 256)

def train():

    unique_words, word2index, unique_word_count, corpus_idx = build_vocab()

    dataset = LyricsDataset(corpus_idx, 32)

    dataloader = DataLoader(dataset, batch_size=5, shuffle=True)

    model = TextGenerator(unique_word_count)

    optimizer = optim.Adam(model.parameters(), lr=0.01)

    loss_fn = nn.CrossEntropyLoss()

    min_loss = math.inf

    loss = 0

    total_loss = 0

    iter_times = 0

    for epoch in range(50):

        for i, (x, y) in enumerate(tqdm(dataloader, desc=f'正在训练第{epoch + 1}轮', total=len(dataset) // 5)):

            hidden = model.init_hidden(5)

            output, hidden = model(x, hidden)

            y = torch.transpose(y, 0, 1).reshape(shape=(-1,))

            loss = loss_fn(output, y)

            optimizer.zero_grad()

            loss.backward()

            optimizer.step()

            iter_times += 1

            total_loss += loss.item()

        print(f'第{epoch + 1}轮，总损失：{total_loss / iter_times}')

        if loss < min_loss:

            min_loss = loss

            torch.save(model.state_dict(), 'model.pth')

def eval(start_word, seq_len):

    unique_words, word2index, unique_word_count, corpus_idx = build_vocab()

    model = TextGenerator(unique_word_count)

    model.load_state_dict(torch.load('model.pth'))

    hidden = model.init_hidden(1)

    word_index = word2index[start_word]

    generate_sentence = [word_index]

    for i in range(seq_len):

        output, hidden = model(torch.tensor([[word_index]]), hidden)

        output = output.reshape(shape=(-1,))

        word_index = torch.argmax(output).item()

        generate_sentence.append(word_index)

    for e in [unique_words[index] for index in generate_sentence]:

        print(e, end='')

if __name__ == '__main__':

    train()

    eval('分手', 100)

print(__file__)

from pathlib import Path

print(Path(__file__).parent.parent / "data/raw/synthesized_.jsonl")

RAW_DATA_DIR = Path(__file__).parent.parent / "data/raw"

def process():

    print("开始处理数据")

    # 读取原始文件

    df = pd.read_json(RAW_DATA_DIR / "synthesized_.jsonl", lines=True, orient='records',

                      encoding='utf-8')

    print(df.head())

    print("数据处理完成")

import jieba

from sklearn.model_selection import train_test_split

import pandas as pd

from tqdm import tqdm

import config

def build_dataset(sentences, word2index, desc="构建数据集"):

    indexed_sentences = [[word2index.get(token, 0) for token in jieba.lcut(sentences)] for sentences in

                         sentences]

    dataset = []

    for index, sentence in tqdm(enumerate(indexed_sentences), desc=desc):

        for i in range(len(sentence) - config.SWQ_LEN):

            input = sentence[i:i + config.SWQ_LEN]

            target = sentence[i + config.SWQ_LEN]

            dataset.append({

                'input': input,

                'target': target,

            })

    return dataset

def process():

    print("开始处理数据")

    # 读取原始文件

    df = pd.read_json(config.RAW_DATA_DIR / "synthesized_.jsonl", lines=True, orient='records',

                      encoding='utf-8').sample(frac=0.1)

    # 提取句子

    sentences = []

    for sentencesX in df['dialog']:

        for sentencesXs in sentencesX:

            sentences.append(sentencesXs.split('：')[1])

    print(f"句子的数量为：{len(sentences)}")

    # 划分数据集

    train_sentences, valid_sentences = train_test_split(sentences, test_size=0.2)

    # 构建词表

    vocab_set = set()

    for sentence in tqdm(train_sentences, desc='构建词表'):

        vocab_set.update(jieba.lcut(sentence))

    vocab_list = ['<unk>'] + list(vocab_set)

    print(f"词表的数量为：{len(vocab_list)}")

    # 保存此表

    with open(config.MODELS_DIR / "vocab.txt", 'w', encoding='utf-8') as f:

        f.write('\n'.join(vocab_list))

    # 构建训练集

    word2index = {word: index for index, word in enumerate(vocab_list)}

    train_dataset = build_dataset(train_sentences, word2index)

    # 保存训练集

    pd.DataFrame(train_dataset).to_json(config.PROCESSED_DATA_DIR / "train.jsonl", orient='records', lines=True)

    # 构建验证集

    valid_dataset = build_dataset(valid_sentences, word2index)

    # 保存验证集

    pd.DataFrame(valid_dataset).to_json(config.PROCESSED_DATA_DIR / "valid.jsonl", orient='records', lines=True)

    print("数据处理完成")

if __name__ == '__main__':

    process()

import pandas as pd

import torch

from torch.utils.data import DataLoader, Dataset

from torch.utils.data.dataset import _T_co

from src.NLP.SmartPrompt.src import config

# 定义 Dataset

class InputMethodDataset(Dataset):

    def __init__(self, path):

        super().__init__()

        self.data = pd.read_json(path, lines=True, orient='records').to_dict('records')

    def __getitem__(self, index) -> _T_co:

        input_tensor = torch.tensor(self.data[index]['input'], dtype=torch.long)

        target_tensor = torch.tensor(self.data[index]['target'], dtype=torch.long)

        return input_tensor, target_tensor

    def __len__(self) -> int:

        return len(self.data)

# 提供一个获取 DATALoader 的方法

def get_dataloader(train=True):

    path = config.PROCESSED_DATA_DIR / ("train.jsonl" if train else "valid.jsonl")

    dataset = InputMethodDataset(path)

    return DataLoader(dataset, batch_size=config.BATCH_SIZE, shuffle=True)

if __name__ == '__main__':

    train_loader = get_dataloader(train=True)

    valid_loader = get_dataloader(train=False)

    print(len(train_loader))

    print(len(valid_loader))

    for input_tensor, target_tensor in train_loader:

        print(input_tensor.shape)

        print(target_tensor.shape)

        break

import time

import torch

from torch import nn

from torch.utils.tensorboard import SummaryWriter

from tqdm import tqdm

from src.NLP.SmartPrompt.src import config

from src.NLP.SmartPrompt.src.dataset import get_dataloader

from src.NLP.SmartPrompt.src.model import InputMethodModel

def train_onr_epoch(model, dataloader, loss_fn, optimizer, device):

    """

    :param model: 模型

    :param dataloader: 数据集

    :param loss_fn: 损失函数

    :param optimizer: 优化器

    :param device: 设备

    :return: 损失值

    """

    model.train()

    total_loss = 0

    for batch in tqdm(dataloader, desc="训练中"):

        features, targets = batch

        features = features.to(device)

        targets = targets.to(device)

        outputs = model(features)

        loss = loss_fn(outputs, targets)

        total_loss += loss.item()

        optimizer.zero_grad()

        loss.backward()

        optimizer.step()

    return total_loss / len(dataloader)

def train():

    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

    dataloader = get_dataloader(train=True)

    with open(config.MODELS_DIR / 'vocab.txt', 'r', encoding='utf-8') as f:

        vocab_list = f.readlines()

        vocab_list = [vocab.strip() for vocab in vocab_list]

    model = InputMethodModel(len(vocab_list)).to(device)

    loss_fn = nn.CrossEntropyLoss()

    optimizer = torch.optim.Adam(model.parameters(), lr=config.LEARNING_RATE)

    writer = SummaryWriter(log_dir=config.LOGS_DIR/time.strftime('%Y-%m-%d_%H-%M-%S'))

    bestLoss = float('inf')

    for epoch in range(1, 1 + config.EPOCHS):

        print("-" * 5, f"Epoch {epoch}/{config.EPOCHS}", "-" * 5)

        loss = train_onr_epoch(model, dataloader, loss_fn, optimizer, device)

        print(f"loss:{loss}")

        writer.add_scalar('loss', loss, epoch)

        if loss < bestLoss:

            bestLoss = loss

            torch.save(model.state_dict(), config.MODELS_DIR / 'model.pt')

    writer.close()

if __name__ == '__main__':

    train()

from torch import nn

import config

class InputMethodModel(nn.Module):

    def __init__(self, vocab_size, *args, **kwargs) -> None:

        super().__init__(*args, **kwargs)

        self.embedding = nn.Embedding(num_embeddings=vocab_size, embedding_dim=config.EMBEDDING_DIM)

        self.rnn = nn.RNN(

            input_size=config.EMBEDDING_DIM,

            hidden_size=config.HIDDEN_SIZE,

            batch_first=True,

        )

        self.linear = nn.Linear(in_features=config.HIDDEN_SIZE, out_features=vocab_size)

    def forward(self, x, *args, **kwargs):

        embedding = self.embedding(x)

        output, hidden = self.rnn(embedding)

        last_hidden_state = output[:,-1,:]

        output = self.linear(last_hidden_state)

        return output

import jieba

import torch

from src.NLP.SmartPrompt.src import config

from src.NLP.SmartPrompt.src.model import InputMethodModel

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

with open(config.MODELS_DIR / 'vocab.txt', 'r', encoding='utf-8') as f:

    vocab_list = f.readlines()

    vocab_list = [vocab.strip() for vocab in vocab_list]

word2index = {word: index for index, word in enumerate(vocab_list)}

index2word = {index: word for index, word in enumerate(vocab_list)}

model = InputMethodModel(len(vocab_list)).to(device)

model.load_state_dict(torch.load(config.MODELS_DIR / 'model.pt'))

def predict(text):

    tokens = jieba.lcut(text)

    input = [word2index.get(token, 0) for token in tokens]

    input_tenosr = torch.tensor([input], dtype=torch.long).to(device)

    model.eval()

    with torch.no_grad():

        output = model(input_tenosr)

    topk = torch.topk(output, k=5).indices.tolist()

    return [index2word[index] for index in topk[0]]

if __name__ == '__main__':

    user_str = ''

    while user_str != 'q':

        now_str = input("请输入：")

        user_str += now_str

        if now_str.strip() == 'q':

            break

        top5_tokens = predict(user_str)

        print(top5_tokens)

        print("当前输入：", user_str)

import torch

import torch.nn as nn

import math

class SelfAttention(nn.Module):

    def __init__(self, dropout=0.1) -> None:

        super().__init__()

        self.dropout = nn.Dropout(dropout)  # 对 10% 的数据进行 dropout，防止模型过拟合

        self.softmax = nn.Softmax(dim=-1)  # 对最后一维进行 softmax，为什么是最后一个维度？

        # 因为是多头注意力机制，所以是多个头，每个头对输入进行 softmax

    def forward(self, Q, K, V, mask=None):

        # X: [batch_size, seq_len, d_model]

        # batch_size: 批次大小，一次送几个句子

        # seq_len: 序列长度，一个句子中的 Token 数量

        # d_model: 模型维度，这是个 Embedding 向量的维度

        # Q：query 向量     [batch_size, heads, seq_len_q, d_k]

        # K：key 向量       [batch_size, heads, seq_len_k, d_k]

        # V：values 向量    [batch_size, heads, seq_len) v, d_v]

        # mask：掩码向量    [batch_size, seq_len, seq_len]   mask 的意义是那些需要忽略，不要看见未来的信息

        d_k = Q.size(-1)

        # Q：                [batch_size, heads, seq_len_q, d_k]       seq_len_q, d_k

        # K.T:               [batch_size, heads, d_k, seq_len_k]      d_k, seq_len_k

        # attention_scores： [batch_size, heads, seq_len_q, seq_len_k]

        attention_scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(d_k)  # (Q·K.T)/sqrt(d_k)

        # 如果提供了 mask，masked_fill 则将 mask 中的值为 0 的元素替换为负无穷大

        if mask is not None:

            attention_scores = attention_scores.masked_fill(mask == 0, float('-inf'))

        # 获取注意力权重，对最后一位进行 softmax

        attn = self.softmax(attention_scores)

        attn = self.dropout(attn)

        # 获取注意力结果，对 V 进行矩阵乘法，[batch_size, heads, seq_len_q, d_v]

        output = torch.matmul(attn, V)

        return output, attn

class MultiHeadAttention(nn.Module):

    def __init__(self, d_model, n_heads, dropout=0.1) -> None:

        super().__init__()

        # d_model: 模型维度，这是个 Embedding 向量的维度，一般是 512

        # n_heads: 多头注意力机制的个数，一般是 8

        assert d_model % n_heads == 0

        self.d_k = d_model // n_heads

        self.n_heads = n_heads

        self.W_q = nn.Linear(d_model, d_model)

        self.W_k = nn.Linear(d_model, d_model)

        self.W_v = nn.Linear(d_model, d_model)

        self.fc = nn.Linear(d_model, d_model)

        self.attention = SelfAttention(dropout)

        self.layer_norm = nn.LayerNorm(d_model)

        self.dropout = nn.Dropout(dropout)

    def forward(self, q, k, v, mask=None):

        batch_size = q.size(0)

        # Q: [batch_size, seq_len, d_model] -> [batch_size, n_heads, seq_len, d_k]

        # 为了获取多头注意力，需要将输入进行分割，分割成多个头，每个头对输入进行计算，然后进行拼接

        Q = self.W_q(q).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)

        K = self.W_k(k).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)

        V = self.W_v(v).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)

        output, attn = self.attention(Q, K, V, mask)

        output = output.transpose(1, 2).contiguous().view(batch_size, -1, self.n_heads * self.d_k)

        output = self.fc(output)

        output = self.layer_norm(q + self.dropout(output))

        return output, attn

torch.nn.Transformer(d_model=512, 

                     nhead=8, 

                     num_encoder_layers=6, 

                     num_decoder_layers=6, 

                     dim_feedforward=2048, 

                     dropout=0.1, 

                     activation='relu', 

                     custom_encoder=None, 

                     custom_decoder=None, 

                     layer_norm_eps=1e-05, 

                     batch_first=False, 

                     norm_first=False, 

                     bias=True, 

                     device=None, 

                     dtype=None)

from torch import nn

# 初始化 Transformer

transformer = nn.Transformer(

  d_model=512, 

  nhead=8, 

  num_encoder_layers=6, 

  num_decoder_layers=6, 

  batch_first=True

)

output = transformer(

    src=src_emb,

    tgt=tgt_emb,

    src_key_padding_mask=src_pad_mask,

    tgt_key_padding_mask=tgt_pad_mask,

    tgt_mask=tgt_mask,

    memory_key_padding_mask=src_pad_mask

)

from torch import nn

# 初始化 Transformer

transformer = nn.Transformer(

    d_model=512, nhead=8,

    num_encoder_layers=6, num_decoder_layers=6,

    batch_first=True

)

# 调用编码器

memory = transformer.encoder(

    src=src_emb, 

    src_key_padding_mask=src_pad_mask

)

from torch import nn

# 初始化 Transformer

transformer = nn.Transformer(

    d_model=512, nhead=8,

    num_encoder_layers=6, num_decoder_layers=6,

    batch_first=True

)

# 调用编码器

memory = transformer.encoder(

    src=src_emb, 

    src_key_padding_mask=src_pad_mask

)

# 调用解码器（逐步生成）

output = transformer.decoder(

    tgt=tgt_emb,

    memory=memory,

    tgt_mask=tgt_mask,

    tgt_key_padding_mask=tgt_pad_mask,

    memory_key_padding_mask=src_pad_mask

)