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>
- Перед началом работы
- Зависимости
- Запуск приложения