Python Flask App для управления задачами обучения
Это приложение на Flask предоставляет базовый интерфейс для создания, мониторинга и получения логов задач обучения, используя API Cloud.ru.
Перед началом работы
Необходимо настроить окружение. Для этого создайте файл конфигурации .env_demo в корневой директории проекта с следующим содержимым:
envMLS_APIKEY_ID=MLS_APIKEY_SECRET=WORKSPACE_ID=X_API_KEY=REGION=OUTPUT=
Зависимости
-
Flask
-
Pydantic
-
Requests
-
urllib3
Запуск приложения
-
Подготовьте переменные окружения.
-
Инициализируйте TrainingJobApi.
-
Выполните flask run.
Код приложения на flask
import requestsimport urllib3from flask import Flaskfrom flask import jsonifyfrom flask import render_templatefrom flask import requestfrom flask import Responsefrom pydantic.v1 import BaseSettingsfrom mls_core import TrainingJobApiclass Settings(BaseSettings):"""Конфигурационные настройки для приложения, загружаемые из файла окружения.Атрибуты:- MLS_APIKEY_ID (str): Идентификатор API ключа для доступа к MLS (MLS - Machine Learning Space).- MLS_APIKEY_SECRET (str): Секретный ключ API для доступа к MLS.- WORKSPACE_ID (str): Идентификатор рабочего пространства в MLS.- X_API_KEY (str): API ключ для дополнительного доступа к другим сервисам.- REGION (str): Регион для запуска и выполнения задач машинного обучения.- OUTPUT (str): Способ вывода результатов выполнения задач."""MLS_APIKEY_ID: strMLS_APIKEY_SECRET: strWORKSPACE_ID: strX_API_KEY: strREGION: strOUTPUT: strclass Config:"""Конфигурация.- env_file (str): Путь к файлу окружения, откуда загружаются настройки. По умолчанию '.env_demo'.- env_file_encoding (str): Кодировка файла окружения. По умолчанию 'utf-8'."""env_file = '.env_demo'env_file_encoding = 'utf-8'settings = Settings()app = Flask(__name__)client = TrainingJobApi('https://.../public/v2',settings.MLS_APIKEY_ID,settings.MLS_APIKEY_SECRET,settings.WORKSPACE_ID,settings.X_API_KEY,backoff_factor=2,connect_timeout=300,read_timeout=300,)@app.route('/', methods=['GET'])def home():"""Домашняя страница приложения."""return render_template('example.html')@app.route('/create', methods=['POST'])def create_task():"""Метод постановки задачи обучения в очередь на исполнение."""task_name = client.run_job(payload={'script': 'sleep 30; echo abc; sleep 30; echo abc; sleep 30; echo abc; sleep 30; echo abc; sleep 30; echo abc;','base_image': 'cr.ai.cloud.ru/aicloud-base-images/py3.10-torch2.1.2:0.0.40','instance_type': 'a100.1gpu','region': settings.REGION,'type': 'pytorch2','n_workers': 1,'job_desc': 'Все хорошо закончилось',},).get('job_name')if task_name:return jsonify({'task_name': task_name})return {}@app.route('/status', methods=['POST'])def status():"""Метод получения статуса задачи."""task_name = request.json.get('task_name')if task_name:res = client.get_job_status(name=task_name, region=settings.REGION)return jsonify({'status': res.get('status'), 'task_name': res.get('job_name')})return {}@app.route('/logs', methods=['POST'])def logs():"""Метод получения журнала событий в потоке."""task_name = request.json.get('task_name')try:return Response(client.stream_logs(task_name, settings.REGION))except (urllib3.exceptions.ReadTimeoutError, requests.exceptions.ConnectionError):return {}if __name__ == '__main__':app.run(debug=True)
Код html-страницы для приложения
<!DOCTYPE html><html><head><title>Запуск задачи</title><link href="https://maxcdn.bootstrapcdn.com/bootstrap/3.3.7/css/bootstrap.min.css" rel="stylesheet"><script src="https://ajax.googleapis.com/ajax/libs/jquery/3.5.1/jquery.min.js"></script></head><body><div class="container"><!-- Верхняя часть: Форма --><div class="row"><div class="col-md-12"><form id="myForm"><div class="form-group"><label>Тип инстанса:</label><select class="form-control" name="instanceType"><option value="small">Small</option><option value="large">Large</option><option value="xlarge">XLarge</option></select></div><button type="submit" class="btn btn-primary">Отправить</button></form></div></div><!-- Нижняя часть: Текстовое поле для AJAX ответа --><div class="row" style="margin-top: 20px;"><div class="col-md-12"><div id="result2" class="well"><!-- AJAX ответ будет здесь --></div><div id="result" class="well"><!-- AJAX ответ будет здесь --></div></div></div></div><script>function checkTaskStatus(data) {$.ajax({url: "/status",type: "POST",data: data,success: function(response) {$('#result2').html(response.status + " " + response.task_name);if (["failed", "completed"].includes(response.status)) {$('#result').append('<br>' + 'Задача завершилась');} else if (response.status === "running") {// Если задача в статусе "Running", убедимся, что не начинаем загрузку логов повторно// Для этого можно использовать флаг или проверять текущее состояние контента,// например, запущена ли функция streamLogs для этой задачи.// Пример с флагом:if (!window.isStreamingLogs) {window.isStreamingLogs = true; // Устанавливаем флагstreamLogs(data.task_name);} else {window.isStreamingLogs = true;checkTaskStatus(data);}} else {setTimeout(function() {checkTaskStatus(data)}, 4000);}}});}async function streamLogs (task_name){const response = await fetch("/logs",{headers: {Accept: 'text/plain','Content-Type': 'application/json',},method: 'POST',body: JSON.stringify({"task_name": task_name})});for await (const chunk of response.body) {const utf8Decoder = new TextDecoder();$('#result').append('<br>' + utf8Decoder.decode(chunk.buffer));}// После завершения чтения логов сбрасываем флаг, чтобы можно было начать поток снова// при необходимости (например, если поток прервался и задача всё ещё в статусе "Running").window.isStreamingLogs = true;checkTaskStatus({"task_name": task_name})}$(document).on("submit", "#myForm", function(e) {e.preventDefault();var formData = $(this).serialize();// Убедимся, что сбросили флаг перед началом работыwindow.isStreamingLogs = false;$.ajax({url: "/create",type: "POST",data: formData,success: function(response) {checkTaskStatus({"task_name": response.task_name});}});});</script></body></html>
- Перед началом работы
- Зависимости
- Запуск приложения