KK100000000000000/FEMA
1
0
Fork 0
Code Issues Pull Requests Actions Projects Releases Wiki Activity
Files
9365dd22663ef05870935f3f10c330e07349d098
FEMA/пайплайн_обучения_модели__mt5_small_.py

1294 lines
49 KiB
Python
Raw Normal View History

初始化项目，由ModelHub XC社区提供模型 Model: KK100000000000000/FEMA Source: Original Platform
2026-07-10 16:22:17 +08:00
# -*- coding: utf-8 -*-
"""Пайплайн обучения модели "mT5-small"
Automatically generated by Colab.
Original file is located at
https://colab.research.google.com/drive/1oUJ0Dw-EcH91-4S1kUV_n33RKJwXj2oL
# **Тестирование модели с помощью zero-shot и few-shots**
"""
# Установка необходимых библиотек
!pip install transformers torch pandas sentencepiece
import pandas as pd
import torch
from transformers import MT5ForConditionalGeneration, MT5Tokenizer
import random
# Загрузка модели и токенизатора MT5-small
model_name = "google/mt5-small"
tokenizer = MT5Tokenizer.from_pretrained(model_name)
model = MT5ForConditionalGeneration.from_pretrained(model_name)
# Если доступен GPU
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device)
# Функция для загрузки данных из файлов
def load_announcements_data():
# Данные для английского языка
eng_data = [
{
"query": "Announcement of the exhibition",
"title": "The Rossettis",
"date": "6 april – 24 September 2023",
"text": "A major exhibition devoted to the radical Rossetti generation."
},
{
"query": "Announcement photography",
"title": "A World in Common: Contemporary African Photography",
"date": "6 July – 14 January 2024",
"text": "A celebration of the varied landscape of contemporary African photography today."
},
{
"query": "Announcement installation",
"title": "JMW Turner with Lamin Fofana: Dark Waters",
"date": "27 September 2022 – 24 September 2023",
"text": "Experience the power of the sea through paintings, sketches and an immersive sound environment."
},
{
"query": "write an announcement for the exhibition",
"title": "Hilma af Klint & Piet Mondrian: Forms of Life",
"date": "18 April 2023",
"text": "Explore the powerful work of two groundbreaking modern artists."
}
]
# Данные для финского языка
fin_data = [
{
"query": "Ilmoitus näyttelystä",
"title": "KOKOELMA: AJAN KYSYMYS.",
"date": "Pysyvä näyttely",
"text": "Kokoelmanäyttelymme Ajan kysymys valottaa nykyhetken suuria kysymyksiä kuvataiteen tarjoaman peilin kautta."
},
{
"query": "Ilmoitus näyttelystä",
"title": "RAJOJEN RIKKOJAT. 1800-luvun matkustavat naistaiteilijat.",
"date": "07.3.–24.8.2025",
"text": "Näyttely tuo ensimmäistä kertaa yhteen 1800-luvulla Saksassa opiskelleiden ja työskennelleiden naistaiteilijoiden töitä."
},
{
"query": "Ilmoitus taidepajasta",
"title": "Lauantaipajat.",
"date": "26.4.2025",
"text": "Tutustu taiteeseen itse tekemällä. Tule Ateneumin työpajaan kokeilemaan ja tekemään itse!"
}
]
return eng_data, fin_data
# Функция для промптов
def create_prompts(eng_data, fin_data, few_shot_k=3):
# Zero-shot промпты
zero_shot_prompts = {
"en_zero_shot": "[EN] Generate museum announcement: Query: \"Announcement of the exhibition\"",
"fi_zero_shot": "[FI] Generate museum announcement: Query: \"Ilmoitus näyttelystä\""
}
# Few-shot промпты для английского
eng_examples = eng_data[:few_shot_k]
en_few_shot_prompt = ""
for example in eng_examples:
en_few_shot_prompt += f"[EN] Query: \"{example['query']}\"\n"
en_few_shot_prompt += f"Title: \"{example['title']}\"\n"
en_few_shot_prompt += f"Date: {example['date']}\n"
en_few_shot_prompt += f"Text: {example['text']}\n\n"
en_few_shot_prompt += "[EN] Generate museum announcement: Query: \"Announcement of the exhibition\""
# Few-shot промпты для финского
fin_examples = fin_data[:few_shot_k]
fi_few_shot_prompt = ""
for example in fin_examples:
fi_few_shot_prompt += f"[FI] Query: \"{example['query']}\"\n"
fi_few_shot_prompt += f"Title: \"{example['title']}\"\n"
fi_few_shot_prompt += f"Date: {example['date']}\n"
fi_few_shot_prompt += f"Text: {example['text']}\n\n"
fi_few_shot_prompt += "[FI] Generate museum announcement: Query: \"Ilmoitus näyttelystä\""
few_shot_prompts = {
"en_few_shot": en_few_shot_prompt,
"fi_few_shot": fi_few_shot_prompt
}
return zero_shot_prompts, few_shot_prompts
# Функция для генерации текста
def generate_announcement(prompt, max_length=150, num_beams=5, temperature=0.9):
# Токенизация промпта
inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512).to(device)
# Генерация текста
with torch.no_grad():
outputs = model.generate(
inputs.input_ids,
max_length=max_length,
num_beams=num_beams,
temperature=temperature,
early_stopping=True,
no_repeat_ngram_size=3,
do_sample=True
)
# Декодирование сгенерированного текста
generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
return generated_text
# Функция для запуска тестов
def run_tests():
print("=" * 80)
print("Загрузка данных и подготовка промптов...")
print("=" * 80)
# Загрузка данных
eng_data, fin_data = load_announcements_data()
# Создание промптов
zero_shot_prompts, few_shot_prompts = create_prompts(eng_data, fin_data, few_shot_k=2)
results = {}
# Zero-shot тесты
print("\n" + "=" * 80)
print("ZERO-SHOT ТЕСТЫ (без примеров)")
print("=" * 80)
for lang, prompt in zero_shot_prompts.items():
print(f"\n{lang.upper()} ZERO-SHOT:")
print(f"Промпт: {prompt[:150]}...")
print("-" * 40)
generated = generate_announcement(prompt)
results[lang] = generated
print(f"Сгенерированный анонс:\n{generated}")
print("=" * 80)
# Few-shot тесты
print("\n" + "=" * 80)
print("FEW-SHOT ТЕСТЫ (с 2 примерами)")
print("=" * 80)
for lang, prompt in few_shot_prompts.items():
print(f"\n{lang.upper()} FEW-SHOT:")
print(f"Промпт (первые 200 символов): {prompt[:200]}...")
print("-" * 40)
generated = generate_announcement(prompt, max_length=200)
results[f"{lang}_few_shot"] = generated
print(f"Сгенерированный анонс:\n{generated}")
print("=" * 80)
# Тесты с разными параметрами
print("\n" + "=" * 80)
print("ТЕСТЫ С РАЗНЫМИ ПАРАМЕТРАМИ ГЕНЕРАЦИИ")
print("=" * 80)
# Английский с более свободными параметрами (температура)
print("\nАнглийский с более креативными параметрами (temperature=1.2):")
creative_prompt = "[EN] Generate museum announcement: Query: \"Announcement of a contemporary art exhibition\""
creative_output = generate_announcement(creative_prompt, temperature=1.2, num_beams=3)
print(f"Промпт: {creative_prompt}")
print(f"Сгенерированный анонс:\n{creative_output}")
print("\n" + "=" * 80)
# Финский с более структурированными параметрами
print("\nФинский с более структурированными параметрами (num_beams=7, temperature=0.7):")
structured_prompt = "[FI] Generate museum announcement: Query: \"Ilmoitus taidepajasta\""
structured_output = generate_announcement(structured_prompt, num_beams=7, temperature=0.7)
print(f"Промпт: {structured_prompt}")
print(f"Сгенерированный анонс:\n{structured_output}")
print("\n" + "=" * 80)
print("ВСЕ ТЕСТЫ ЗАВЕРШЕНЫ")
print("=" * 80)
return results
# Функция для оценки качества сгенерированных текстов
def evaluate_results(results):
print("\n" + "=" * 80)
print("КАЧЕСТВЕННАЯ ОЦЕНКА СГЕНЕРИРОВАННЫХ ТЕКСТОВ")
print("=" * 80)
for test_name, text in results.items():
print(f"\n{test_name.upper()}:")
print(f"Длина текста: {len(text)} символов")
# Простая проверка на наличие ключевых элементов
checks = {
"Есть заголовок (Title)": "Title" in text or "title" in text or "nimi" in text.lower(),
"Есть дата/время": any(word in text.lower() for word in ["date", "202", "time", "klo", "aika"]),
"Есть описание": len(text.split()) > 10,
"Языковая корректность": True # Здесь можно добавить более сложные проверки
}
for check_name, check_result in checks.items():
status = "✓" if check_result else "✗"
print(f" {status} {check_name}")
print(f"Текст: {text[:150]}...")
# Запуск тестов
if __name__ == "__main__":
print("Инициализация модели MT5-small...")
print(f"Используется устройство: {device}")
# Запуск основных тестов
test_results = run_tests()
# Оценка результатов
evaluate_results(test_results)
# Пример дополнительного теста с пользовательским промптом
print("\n" + "=" * 80)
print("ДОПОЛНИТЕЛЬНЫЙ ТЕСТ: ГЕНЕРАЦИЯ ПОЛЬЗОВАТЕЛЬСКОГО ПРОМПТА")
print("=" * 80)
# Пользователь может ввести свой промпт
custom_prompt = input("Введите промпт для генерации (или нажмите Enter для примера): ")
if not custom_prompt:
custom_prompt = "[EN] Generate museum announcement: Query: \"Announcement of a photography exhibition about urban landscapes\""
print(f"\nПромпт: {custom_prompt}")
custom_output = generate_announcement(custom_prompt, max_length=200)
print(f"Сгенерированный анонс:\n{custom_output}")
"""# **1 ЭТАП ОБУЧЕНИЯ МОДЕЛИ "mT5-SMALL". АДАПТАЦИЯ.**"""
pip install transformers datasets torch pandas
import pandas as pd
import os
from transformers import (
MT5ForConditionalGeneration,
MT5Tokenizer,
Seq2SeqTrainer,
Seq2SeqTrainingArguments,
DataCollatorForSeq2Seq
)
from datasets import Dataset, concatenate_datasets
import torch
import re
# Параметры
MODEL_NAME = "google/mt5-small"
BATCH_SIZE = 4
MAX_LENGTH = 256
SOURCE_MAX_LENGTH = 128
EPOCHS = 5
# Загрузка и предобработка данных
def load_data(file_path, lang):
df = pd.read_csv(file_path, sep='\t', header=None, names=['input', 'target'])
df = df.dropna()
# Добавляем языковые префиксы
df['input'] = f"{lang}: " + df['input']
# Очистка и форматирование целевых текстов
def format_target(text):
# Удаляем лишние пробелы и переносы строк
text = re.sub(r'\s+', ' ', str(text)).strip()
return text
df['target'] = df['target'].apply(format_target)
return df
# Загрузка данных
df_en = load_data("ENG.tsv", "en")
df_fi = load_data("FI.tsv", "fi")
print(f"Загружено {len(df_en)} английских примеров")
print(f"Загружено {len(df_fi)} финских примеров")
# Создание datasets
dataset_en = Dataset.from_pandas(df_en)
dataset_fi = Dataset.from_pandas(df_fi)
# Объединение датасетов
combined_dataset = concatenate_datasets([dataset_en, dataset_fi])
dataset = combined_dataset.train_test_split(test_size=0.2, seed=42)
print(f"Размер тренировочного набора: {len(dataset['train'])}")
print(f"Размер тестового набора: {len(dataset['test'])}")
# Загрузка модели и токенизатора
print("Загрузка модели и токенизатора...")
tokenizer = MT5Tokenizer.from_pretrained(MODEL_NAME)
model = MT5ForConditionalGeneration.from_pretrained(MODEL_NAME)
# Функция токенизации
def tokenize_function(examples):
model_inputs = tokenizer(
examples['input'],
max_length=SOURCE_MAX_LENGTH,
padding='max_length',
truncation=True
)
with tokenizer.as_target_tokenizer():
labels = tokenizer(
examples['target'],
max_length=MAX_LENGTH,
padding='max_length',
truncation=True
)
# Для mT5 нужно заменить padding токены в labels на -100
labels['input_ids'] = [
[(label if label != tokenizer.pad_token_id else -100) for label in labels_example]
for labels_example in labels['input_ids']
]
model_inputs['labels'] = labels['input_ids']
return model_inputs
# Токенизация данных
print("Токенизация данных...")
tokenized_datasets = dataset.map(
tokenize_function,
batched=True,
remove_columns=dataset['train'].column_names
)
# Data collator
data_collator = DataCollatorForSeq2Seq(
tokenizer=tokenizer,
model=model,
padding=True
)
# Аргументы обучения
training_args = Seq2SeqTrainingArguments(
output_dir="./mt5-museum-announcements",
eval_strategy="epoch",
learning_rate=3e-4,
per_device_train_batch_size=BATCH_SIZE,
per_device_eval_batch_size=BATCH_SIZE,
weight_decay=0.01,
save_total_limit=3,
num_train_epochs=EPOCHS,
predict_with_generate=True,
logging_dir='./logs',
logging_steps=10,
save_steps=100,
warmup_steps=100,
report_to="none"
)
# Trainer
trainer = Seq2SeqTrainer(
model=model,
args=training_args,
train_dataset=tokenized_datasets["train"],
eval_dataset=tokenized_datasets["test"],
data_collator=data_collator,
tokenizer=tokenizer,
)
# Запуск обучения
print("Начало обучения...")
trainer.train()
# Сохранение модели
trainer.save_model("./mt5-museum-final")
tokenizer.save_pretrained("./mt5-museum-final")
print("Модель сохранена в ./mt5-museum-final")
# Загружаем сохраненную модель для инференса
model = MT5ForConditionalGeneration.from_pretrained("./mt5-museum-final")
tokenizer = MT5Tokenizer.from_pretrained("./mt5-museum-final")
# Перемещаем модель на CPU для инференса
model = model.to('cpu')
# Пример генерации - исправленная версия
def generate_announcement(text, language="en"):
input_text = f"{language}: {text}"
# Токенизируем входной текст
inputs = tokenizer(
input_text,
return_tensors="pt",
max_length=SOURCE_MAX_LENGTH,
truncation=True,
padding=True
)
# Получаем device модели
device = next(model.parameters()).device
# Перемещаем входные данные на тот же device, что и модель
input_ids = inputs['input_ids'].to(device)
attention_mask = inputs['attention_mask'].to(device)
# Генерируем выходной текст
outputs = model.generate(
input_ids=input_ids,
attention_mask=attention_mask,
max_length=MAX_LENGTH,
num_beams=5,
early_stopping=True,
no_repeat_ngram_size=2
)
return tokenizer.decode(outputs[0], skip_special_tokens=True)
# Тестирование
print("\nТестирование модели:")
test_examples = [
("Announcement of the exhibition", "en"),
("kirjoita ilmoitus taidepajasta", "fi"),
("Music event announcement", "en"),
("Näyttelyilmoitus", "fi")
]
for text, lang in test_examples:
result = generate_announcement(text, lang)
print(f"Input ({lang}): {text}")
print(f"Output: {result}\n")
import torch
from transformers import MT5ForConditionalGeneration, MT5Tokenizer
# Загрузка обученной модели и токенизатора
model_path = "./mt5-museum-final"
tokenizer = MT5Tokenizer.from_pretrained(model_path)
model = MT5ForConditionalGeneration.from_pretrained(model_path)
# Перемещаем модель на CPU (или GPU, если доступно)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = model.to(device)
print(f"Модель загружена на устройство: {device}")
# Функция для генерации анонсов
def generate_museum_announcement(prompt, language="en", max_length=256):
"""
Генерирует музейный анонс на основе промпта и языка
Args:
prompt (str): Запрос для генерации
language (str): Язык ('en' для английского, 'fi' для финского)
max_length (int): Максимальная длина генерируемого текста
Returns:
str: Сгенерированный анонс
"""
# Добавляем языковой префикс
input_text = f"{language}: {prompt}"
# Токенизируем входной текст
inputs = tokenizer(
input_text,
return_tensors="pt",
max_length=128,
truncation=True,
padding=True
).to(device)
# Генерируем текст
with torch.no_grad():
outputs = model.generate(
inputs.input_ids,
attention_mask=inputs.attention_mask,
max_length=max_length,
num_beams=5,
early_stopping=True,
no_repeat_ngram_size=2,
temperature=0.9,
do_sample=True
)
# Декодируем сгенерированный текст
generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
return generated_text
# Теперь можно использовать функцию
# Генерация нескольких вариантов
def generate_multiple_announcements(prompt, language="en", num_variants=3):
"""Генерирует несколько вариантов анонса"""
print(f"Генерация {num_variants} вариантов для: '{prompt}' ({language})")
print("-" * 50)
for i in range(num_variants):
result = generate_museum_announcement(
prompt,
language=language,
max_length=200
)
print(f"Вариант {i+1}: {result}\n")
# Пример использования
print("=== ГЕНЕРАЦИЯ МУЗЕЙНЫХ АНОНСОВ ===\n")
# Английские примеры
english_prompts = [
"Announcement of the exhibition",
"Workshop announcement for children",
"Music concert in the museum",
"Dance performance announcement",
"Photography exhibition opening"
]
print("--- АНГЛИЙСКИЕ АНОНСЫ ---")
for i, prompt in enumerate(english_prompts, 1):
result = generate_museum_announcement(prompt, language="en")
print(f"{i}. Запрос: {prompt}")
print(f" Анонс: {result}\n")
# Финские примеры
finnish_prompts = [
"Ilmoitus taidepajasta",
"Näyttelyilmoitus",
"Musiikkitapahtuman ilmoitus",
"Tanssiesitys",
"Valokuvausnäyttely"
]
print("--- ФИНСКИЕ АНОНСЫ ---")
for i, prompt in enumerate(finnish_prompts, 1):
result = generate_museum_announcement(prompt, language="fi")
print(f"{i}. Запрос: {prompt}")
print(f" Анонс: {result}\n")
# Генерация нескольких вариантов
print("=== МНОЖЕСТВЕННАЯ ГЕНЕРАЦИЯ ===")
generate_multiple_announcements("Art exhibition announcement", "en", 3)
generate_multiple_announcements("Taidenäyttely", "fi", 2)
# Интерактивный режим
def interactive_mode():
"""Интерактивный режим для генерации анонсов"""
print("=== ИНТЕРАКТИВНЫЙ РЕЖИМ ===")
print("Доступные языки: 'en' (английский), 'fi' (финский)")
print("Для выхода введите 'quit'\n")
while True:
language = input("Выберите язык (en/fi): ").strip().lower()
if language == 'quit':
break
if language not in ['en', 'fi']:
print("Неверный язык. Используйте 'en' или 'fi'")
continue
prompt = input("Введите запрос: ").strip()
if prompt == 'quit':
break
result = generate_museum_announcement(prompt, language=language)
print(f"\nСгенерированный анонс:")
print(f"📝 {result}\n")
# Запуск интерактивного режима (раскомментируйте если нужно)
# interactive_mode()
import matplotlib.pyplot as plt
import numpy as np
# Данные экспериментов
experiments = [
{
'name': 'Эксперимент 1',
'epochs': 10,
'learning_rate': 3e-5,
'batch_size': 2,
'temperature': 1.0,
'train_loss': 30.782700,
'val_loss': 33.190113
},
{
'name': 'Эксперимент 2',
'epochs': 20,
'learning_rate': 2e-5,
'batch_size': 2,
'temperature': 0.5,
'train_loss': 19.559500,
'val_loss': 17.091799
},
{
'name': 'Эксперимент 3',
'epochs': 50,
'learning_rate': 1.5e-5,
'batch_size': 2,
'temperature': 0.4,
'train_loss': 10.598900,
'val_loss': 8.103792
},
{
'name': 'Эксперимент 4',
'epochs': 100,
'learning_rate': 1e-5,
'batch_size': 2,
'temperature': 0.3,
'train_loss': 1.276500,
'val_loss': 0.911331
}
]
# Создаем фигуру и оси
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(16, 7))
fig.suptitle('Результаты экспериментов по обучению модели на мультиязычном датасете',
fontsize=16, fontweight='bold', color='#1a3c8c')
# Оттенки синего для экспериментов
blue_palette = ['#1a3c8c', '#2a5caa', '#3a7dc8', '#4a9de6']
# График 1: Loss по экспериментам
experiment_numbers = np.arange(len(experiments))
width = 0.35
bars1 = ax1.bar(experiment_numbers - width/2,
[exp['train_loss'] for exp in experiments],
width, label='Training Loss', color='#1a3c8c', alpha=0.8)
bars2 = ax1.bar(experiment_numbers + width/2,
[exp['val_loss'] for exp in experiments],
width, label='Validation Loss', color='#4a9de6', alpha=0.8)
ax1.set_xlabel('Эксперимент', fontsize=12, color='#1a3c8c')
ax1.set_ylabel('Loss', fontsize=12, color='#1a3c8c')
ax1.set_title('Сравнение Training и Validation Loss', fontsize=14, color='#1a3c8c')
ax1.set_xticks(experiment_numbers)
ax1.set_xticklabels([exp['name'] for exp in experiments], rotation=45, ha='right')
ax1.legend()
ax1.grid(True, alpha=0.3, linestyle='--')
# Добавляем значения на столбцы
for bars in [bars1, bars2]:
for bar in bars:
height = bar.get_height()
ax1.annotate(f'{height:.2f}',
xy=(bar.get_x() + bar.get_width() / 2, height),
xytext=(0, 3), # 3 points vertical offset
textcoords="offset points",
ha='center', va='bottom',
fontsize=9, color='#1a3c8c')
# График 2: Параметры и метрики
x = np.arange(len(experiments))
# Создаем вторую ось Y
ax2_twin = ax2.twinx()
# Линия для Loss
line1 = ax2.plot(x, [exp['train_loss'] for exp in experiments],
'o-', linewidth=2, markersize=8, color='#1a3c8c',
label='Training Loss')
# Линия для Learning Rate (используем логарифмическую шкалу)
line2 = ax2_twin.plot(x, [exp['learning_rate'] for exp in experiments],
's--', linewidth=2, markersize=8, color='#3a7dc8',
label='Learning Rate')
# Столбцы для Epochs
bars_epochs = ax2.bar(x, [exp['epochs'] for exp in experiments],
alpha=0.3, color='#2a5caa', label='Epochs')
# Настройки основной оси Y
ax2.set_xlabel('Эксперимент', fontsize=12, color='#1a3c8c')
ax2.set_ylabel('Loss / Epochs', fontsize=12, color='#1a3c8c')
ax2.set_title('Параметры обучения и результаты', fontsize=14, color='#1a3c8c')
ax2.set_xticks(x)
ax2.set_xticklabels([f"Exp {i+1}" for i in range(len(experiments))])
# Настройки второй оси Y
ax2_twin.set_ylabel('Learning Rate', fontsize=12, color='#3a7dc8')
ax2_twin.tick_params(axis='y', labelcolor='#3a7dc8')
ax2_twin.set_yscale('log') # Логарифмическая шкала для learning rate
# Добавляем аннотации для температуры
for i, exp in enumerate(experiments):
ax2.annotate(f"T={exp['temperature']}",
xy=(i, exp['train_loss']),
xytext=(0, 10),
textcoords='offset points',
ha='center', fontsize=9,
bbox=dict(boxstyle="round,pad=0.3", facecolor='white', alpha=0.8, edgecolor='#4a9de6'))
# Объединяем легенды
lines1, labels1 = ax2.get_legend_handles_labels()
lines2, labels2 = ax2_twin.get_legend_handles_labels()
ax2.legend(lines1 + lines2, labels1 + labels2, loc='upper right')
ax2.grid(True, alpha=0.3, linestyle='--')
# Добавляем таблицу с параметрами
table_data = []
for exp in experiments:
table_data.append([
exp['epochs'],
f"{exp['learning_rate']:.1e}",
exp['temperature'],
f"{exp['train_loss']:.2f}",
f"{exp['val_loss']:.2f}"
])
# Создаем таблицу под графиками
col_labels = ['Эпохи', 'LR', 'Temp', 'Train Loss', 'Val Loss']
row_labels = [exp['name'] for exp in experiments]
# Добавляем таблицу как отдельный subplot
fig, ax_table = plt.subplots(figsize=(12, 3))
ax_table.axis('tight')
ax_table.axis('off')
table = ax_table.table(cellText=table_data,
colLabels=col_labels,
rowLabels=row_labels,
cellLoc='center',
loc='center',
colColours=['#e6f2ff']*5,
rowColours=['#f0f8ff']*4)
table.auto_set_font_size(False)
table.set_fontsize(10)
table.scale(1.2, 1.5)
# Настраиваем цвета заголовков
for i in range(len(col_labels)):
table[(0, i)].set_facecolor('#1a3c8c')
table[(0, i)].set_text_props(color='white', weight='bold')
# Объединяем все графики
plt.figure(fig.number)
plt.tight_layout()
plt.subplots_adjust(top=0.9)
# Сохраняем график
plt.savefig('experiments_results_blue_theme.png', dpi=300, bbox_inches='tight', facecolor='white')
plt.show()
# Дополнительно: создаем компактный линейный график
fig2, ax3 = plt.subplots(figsize=(12, 6))
# Преобразуем эпохи в массив для smooth линии
epochs_array = np.array([exp['epochs'] for exp in experiments])
train_loss_array = np.array([exp['train_loss'] for exp in experiments])
val_loss_array = np.array([exp['val_loss'] for exp in experiments])
# Создаем интерполированные данные для плавных линий
epochs_smooth = np.linspace(epochs_array.min(), epochs_array.max(), 100)
train_loss_smooth = np.interp(epochs_smooth, epochs_array, train_loss_array)
val_loss_smooth = np.interp(epochs_smooth, epochs_array, val_loss_array)
# График с плавными линиями
ax3.plot(epochs_smooth, train_loss_smooth, '-', linewidth=3,
color='#1a3c8c', alpha=0.7, label='Training Loss (сглаженный)')
ax3.plot(epochs_smooth, val_loss_smooth, '-', linewidth=3,
color='#4a9de6', alpha=0.7, label='Validation Loss (сглаженный)')
# Точки фактических измерений
ax3.scatter(epochs_array, train_loss_array, s=100, color='#1a3c8c',
edgecolors='white', linewidth=2, zorder=5, label='Training Loss')
ax3.scatter(epochs_array, val_loss_array, s=100, color='#4a9de6',
edgecolors='white', linewidth=2, zorder=5, label='Validation Loss')
# Добавляем аннотации с параметрами
for i, exp in enumerate(experiments):
text = f"Exp {i+1}\nLR={exp['learning_rate']:.1e}\nT={exp['temperature']}\nBS={exp['batch_size']}"
ax3.annotate(text,
xy=(exp['epochs'], exp['train_loss']),
xytext=(10, 20 if i%2==0 else -30),
textcoords='offset points',
ha='left',
fontsize=9,
bbox=dict(boxstyle="round,pad=0.5", facecolor='white',
alpha=0.9, edgecolor=blue_palette[i]),
arrowprops=dict(arrowstyle='->', color=blue_palette[i], alpha=0.7))
ax3.set_xlabel('Количество эпох', fontsize=12, color='#1a3c8c')
ax3.set_ylabel('Loss', fontsize=12, color='#1a3c8c')
ax3.set_title('Динамика обучения модели: зависимость Loss от количества эпох и гиперпараметров',
fontsize=14, fontweight='bold', color='#1a3c8c')
ax3.set_yscale('log') # Логарифмическая шкала для лучшей визуализации
ax3.grid(True, alpha=0.3, linestyle='--')
ax3.legend(loc='upper right')
plt.tight_layout()
plt.savefig('training_dynamics_blue_theme.png', dpi=300, bbox_inches='tight', facecolor='white')
plt.show()
"""# **2 ЭТАП ОБУЧЕНИЯ МОДЕЛИ "mT5-SMALL". ОПТИМИЗАЦИЯ.**"""
# # Часть 2: Генерация синтетических данных и дообучение
import pandas as pd
import os
from transformers import MT5ForConditionalGeneration, MT5Tokenizer
from datasets import Dataset
import torch
import re
import numpy as np
# =============================================================================
# КОНФИГУРАЦИЯ СИНТЕТИЧЕСКИХ ДАННЫХ
# =============================================================================
class SyntheticDataConfig:
"""Конфигурация для генерации синтетических данных"""
NUM_SYNTHETIC_SAMPLES = 1000
MIN_REWARD_SCORE = 7 # Минимальный балл для сохранения синтетических данных
LEARNING_EPOCHS = 3 # Эпохи дообучения на синтетических данных
CATEGORIES = {
'en': ['Exhibition', 'Workshop', 'Other events', 'Cinema', 'Photography',
'Dance', 'Video art', 'Music', 'Performance', 'Installation'],
'fi': ['Näyttely', 'Taidepaja', 'Esitys', 'Muut tapahtumat', 'Elokuvataide',
'Valokuvataide', 'Tanssitaide', 'Videotaide', 'Musiikki', 'Performanssi', 'Installaatio']
}
# =============================================================================
# ФУНКЦИИ ДЛЯ РАБОТЫ С СИНТЕТИЧЕСКИМИ ДАННЫМИ
# =============================================================================
def load_best_model():
"""Загружает лучшую модель из экспериментов"""
# Находим лучшую модель по результатам экспериментов
results_df = pd.read_csv("./experiments/results_summary.csv")
best_exp = results_df.loc[results_df['rouge1_f1'].idxmax()]['experiment']
model_path = f"./models/{best_exp}"
print(f"Загрузка лучшей модели: {best_exp}")
# Загружаем только веса модели
model = MT5ForConditionalGeneration.from_pretrained(model_path)
tokenizer = MT5Tokenizer.from_pretrained(model_path)
# Используем CUDA если доступно
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = model.to(device)
print(f"Модель загружена на устройство: {device}")
return model, tokenizer, best_exp
def generate_synthetic_announcement(model, tokenizer, query, language="en", max_length=128):
"""Генерирует синтетический анонс"""
input_text = f"[{language.upper()}] {query}"
inputs = tokenizer(
input_text,
return_tensors="pt",
max_length=64,
truncation=True,
padding=True
)
# Перемещаем входные данные на то же устройство, что и модель
device = next(model.parameters()).device
inputs = {key: value.to(device) for key, value in inputs.items()}
outputs = model.generate(
input_ids=inputs['input_ids'],
attention_mask=inputs['attention_mask'],
max_length=max_length,
num_beams=4,
early_stopping=True,
temperature=0.8,
do_sample=True
)
generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
generated_text = re.sub(r'\s+', ' ', generated_text).strip()
return generated_text
def validate_synthetic_text(text):
"""Проверяет валидность синтетического текста"""
required_fields = ['Title:', 'Date:', 'Text:']
return all(field in text for field in required_fields)
def calculate_reward_score(text):
"""Упрощенная reward-система для синтетических данных"""
score = 0
# Проверка структуры
if validate_synthetic_text(text):
score += 3
# Проверка длины
sentences = re.split(r'[.!?]+', text)
sentence_count = len([s for s in sentences if len(s.strip()) > 5])
if 1 <= sentence_count <= 3: # 1-3 предложения
score += 2
# Проверка уникальности слов
words = re.findall(r'\b\w+\b', text.lower())
unique_ratio = len(set(words)) / len(words) if words else 0
if unique_ratio > 0.7:
score += 2
return min(7, score) # Максимум 7 баллов
def generate_synthetic_dataset(model, tokenizer, num_samples=1000):
"""Генерирует датасет синтетических данных"""
print("Генерация синтетических данных...")
synthetic_data = []
queries = {
'en': [
"Announcement of the exhibition",
"Write an announcement for the exhibition",
"Announcement exhibition",
"Announcement of the workshop",
"Write an announcement for the workshop",
"Announcement workshop",
"Announcement of other events",
"Write an announcement for other events",
"Announcement other events",
"Announcement of the cinema",
"Write an announcement for the cinema",
"Announcement cinema",
"Announcement of photography",
"Write an announcement for photography",
"Announcement photography",
"Announcement of the dance",
"Write an announcement for the dance",
"Announcement dance",
"Announcement of video art",
"Write an announcement for video art",
"Announcement video art",
"Announcement of the music",
"Write an announcement for the music",
"Announcement music",
"Announcement of the performance",
"Write an announcement for the performance",
"Announcement performance",
"Announcement of the installation",
"Write an announcement for the installation",
"Announcement installation"
],
'fi': [
"Ilmoitus näyttelystä",
"Kirjoita ilmoitus näyttelystä",
"Ilmoitus taidepajasta",
"Kirjoita ilmoitus taidepajasta",
"Ilmoitus esityksestä",
"Kirjoita ilmoitus esityksestä",
"Ilmoitus muista tapahtumista",
"Kirjoita ilmoitus muista tapahtumista",
"Ilmoitus elokuvataiteesta",
"Kirjoita ilmoitus elokuvataiteesta",
"Ilmoitus valokuvataiteesta",
"Kirjoita ilmoitus valokuvataiteesta",
"Ilmoitus tanssitaiteesta",
"Kirjoita ilmoitus tanssitaiteesta",
"Ilmoitus videotaiteesta",
"Kirjoita ilmoitus videotaiteesta",
"Ilmoitus musiikista",
"Kirjoita ilmoitus musiikista",
"Ilmoitus performanssista",
"Kirjoita ilmoitus performanssista",
"Ilmoitus installaatiosta",
"Kirjoita ilmoitus installaatiosta"
]
}
generated_count = 0
while len(synthetic_data) < num_samples and generated_count < num_samples * 2:
lang = 'en' if generated_count % 2 == 0 else 'fi'
query_list = queries[lang]
query = query_list[generated_count % len(query_list)]
try:
synthetic_text = generate_synthetic_announcement(model, tokenizer, query, lang)
reward_score = calculate_reward_score(synthetic_text)
if reward_score >= SyntheticDataConfig.MIN_REWARD_SCORE:
# Определяем категорию на основе запроса
category = determine_category_from_query(query, lang)
synthetic_data.append({
'input': f"[{lang.upper()}] {query}",
'target': synthetic_text,
'language': lang,
'category': category,
'reward_score': reward_score
})
if len(synthetic_data) % 100 == 0:
print(f"Сгенерировано {len(synthetic_data)} синтетических примеров")
except Exception as e:
print(f"Ошибка генерации: {e}")
generated_count += 1
print(f"Генерация завершена: {len(synthetic_data)} качественных примеров")
return pd.DataFrame(synthetic_data)
def determine_category_from_query(query, language):
"""Определяет категорию на основе запроса"""
query_lower = query.lower()
if language == 'en':
if 'exhibition' in query_lower:
return 'Exhibition'
elif 'workshop' in query_lower:
return 'Workshop'
elif 'cinema' in query_lower:
return 'Cinema'
elif 'photography' in query_lower:
return 'Photography'
elif 'dance' in query_lower:
return 'Dance'
elif 'video art' in query_lower:
return 'Video art'
elif 'music' in query_lower:
return 'Music'
elif 'performance' in query_lower:
return 'Performance'
elif 'installation' in query_lower:
return 'Installation'
else:
return 'Other events'
else: # fi
if 'näyttely' in query_lower:
return 'Näyttely'
elif 'taidepaja' in query_lower:
return 'Taidepaja'
elif 'esitys' in query_lower:
return 'Esitys'
elif 'elokuvataide' in query_lower:
return 'Elokuvataide'
elif 'valokuvataide' in query_lower:
return 'Valokuvataide'
elif 'tanssitaide' in query_lower:
return 'Tanssitaide'
elif 'videotaide' in query_lower:
return 'Videotaide'
elif 'musiikki' in query_lower:
return 'Musiikki'
elif 'performanssi' in query_lower:
return 'Performanssi'
elif 'installaatio' in query_lower:
return 'Installaatio'
else:
return 'Muut tapahtumat'
# =============================================================================
# ДООБУЧЕНИЕ НА СИНТЕТИЧЕСКИХ ДАННЫХ
# =============================================================================
def fine_tune_on_synthetic_data(model, tokenizer, synthetic_df, original_train_data, experiment_name):
"""Дообучает модель на синтетических данных"""
print("Начало дообучения на синтетических данных...")
# Подготовка данных
synthetic_dataset = Dataset.from_pandas(synthetic_df)
combined_dataset = concatenate_datasets([original_train_data, synthetic_dataset])
# Токенизация
def tokenize_function(examples):
inputs = [f"{inp}" for inp in examples['input']]
targets = [f"{tgt}" for tgt in examples['target']]
model_inputs = tokenizer(
inputs,
max_length=64,
padding='max_length',
truncation=True,
return_tensors="pt"
)
with tokenizer.as_target_tokenizer():
labels = tokenizer(
targets,
max_length=128,
padding='max_length',
truncation=True,
return_tensors="pt"
)
labels = labels["input_ids"]
labels[labels == tokenizer.pad_token_id] = -100
model_inputs["labels"] = labels
return model_inputs
tokenized_combined = combined_dataset.map(tokenize_function, batched=True)
# Data collator
from transformers import DataCollatorForSeq2Seq
data_collator = DataCollatorForSeq2Seq(
tokenizer=tokenizer,
model=model,
padding=True
)
# Аргументы дообучения
from transformers import Seq2SeqTrainingArguments
training_args = Seq2SeqTrainingArguments(
output_dir=f"./synthetic_finetune/{experiment_name}",
evaluation_strategy="epoch",
save_strategy="epoch",
learning_rate=1e-5, # Меньший LR для тонкой настройки
per_device_train_batch_size=8,
per_device_eval_batch_size=8,
num_train_epochs=SyntheticDataConfig.LEARNING_EPOCHS,
predict_with_generate=True,
logging_dir=f'./logs/synthetic_{experiment_name}',
logging_steps=20,
save_total_limit=2,
load_best_model_at_end=True,
metric_for_best_model="eval_loss",
greater_is_better=False,
)
# Trainer
from Part1 import EnhancedContrastiveTrainer # Импортируем из первой части
trainer = EnhancedContrastiveTrainer(
model=model,
args=training_args,
train_dataset=tokenized_combined,
data_collator=data_collator,
tokenizer=tokenizer,
)
# Дообучение
print("Запуск дообучения...")
trainer.train()
# Сохранение модели
trainer.save_model(f"./models/synthetic_finetuned_{experiment_name}")
tokenizer.save_pretrained(f"./models/synthetic_finetuned_{experiment_name}")
print("Дообучение завершено!")
return trainer
# =============================================================================
# ТЕСТИРОВАНИЕ ДООБУЧЕННОЙ МОДЕЛИ
# =============================================================================
def test_finetuned_model(model, tokenizer, test_dataset):
"""Тестирует дообученную модель"""
from Part1 import MetricsCalculator # Импортируем из первой части
metrics_calculator = MetricsCalculator()
print("Тестирование дообученной модели...")
# Генерация предсказаний для тестового набора
predictions = []
references = []
for example in test_dataset:
input_text = example['input']
# Генерация
generated = generate_synthetic_announcement(model, tokenizer,
input_text.replace(f"[{example['language'].upper()}] ", ""),
example['language'])
predictions.append(generated)
references.append(example['target'])
# Вычисление метрик
metrics = metrics_calculator.calculate_custom_metrics(predictions, references)
print("Результаты тестирования дообученной модели:")
for metric, value in metrics.items():
print(f" {metric}: {value:.4f}")
return metrics
# =============================================================================
# ОСНОВНОЙ ПРОЦЕСС ЧАСТИ 2
# =============================================================================
def run_synthetic_data_pipeline():
"""Запускает полный пайплайн синтетических данных"""
# Загрузка лучшей модели
model, tokenizer, best_exp = load_best_model()
# Загрузка оригинальных данных для тестирования
from Part1 import load_excel_data
df_en = load_excel_data("ENG1.xlsx", lang='en')
df_fi = load_excel_data("FI1.xlsx", lang='fi')
from datasets import Dataset, concatenate_datasets
dataset_en = Dataset.from_pandas(df_en)
dataset_fi = Dataset.from_pandas(df_fi)
combined_dataset = concatenate_datasets([dataset_en, dataset_fi])
train_test_split = combined_dataset.train_test_split(test_size=0.15, seed=42)
# Генерация синтетических данных
synthetic_df = generate_synthetic_dataset(
model, tokenizer,
SyntheticDataConfig.NUM_SYNTHETIC_SAMPLES
)
# Сохранение синтетических данных
synthetic_df.to_csv("./synthetic_data/synthetic_announcements.csv", index=False)
print("Синтетические данные сохранены")
# Анализ распределения категорий
print("\nРаспределение категорий в синтетических данных:")
print("Английские данные:")
en_categories = synthetic_df[synthetic_df['language'] == 'en']['category'].value_counts()
for category, count in en_categories.items():
print(f" {category}: {count} примеров")
print("\nФинские данные:")
fi_categories = synthetic_df[synthetic_df['language'] == 'fi']['category'].value_counts()
for category, count in fi_categories.items():
print(f" {category}: {count} примеров")
# Дообучение на синтетических данных
finetuned_trainer = fine_tune_on_synthetic_data(
model, tokenizer, synthetic_df,
train_test_split['train'], best_exp
)
# Тестирование
test_metrics = test_finetuned_model(
finetuned_trainer.model, tokenizer, train_test_split['test']
)
# Сравнение с оригинальными результатами
original_results = pd.read_csv("./experiments/results_summary.csv")
best_original = original_results.loc[original_results['rouge1_f1'].idxmax()]
print(f"Сравнение результатов:")
print(f" Оригинальная модель - ROUGE-1 F1: {best_original['rouge1_f1']:.4f}")
print(f" Дообученная модель - ROUGE-1 F1: {test_metrics['rouge1_f1']:.4f}")
print(f" Улучшение: {test_metrics['rouge1_f1'] - best_original['rouge1_f1']:.4f}")
return {
'synthetic_data': synthetic_df,
'finetuned_trainer': finetuned_trainer,
'test_metrics': test_metrics,
'improvement': test_metrics['rouge1_f1'] - best_original['rouge1_f1']
}
# =============================================================================
# ЗАПУСК
# =============================================================================
if __name__ == "__main__":
# Создаем директории
os.makedirs("./synthetic_data", exist_ok=True)
os.makedirs("./synthetic_finetune", exist_ok=True)
# Запускаем пайплайн
results = run_synthetic_data_pipeline()
if results['improvement'] > 0:
print(f"Дообучение улучшило модель на {results['improvement']:.4f}!")
else:
print(f"Дообучение не улучшило модель")
Reference in New Issue Copy Permalink