""" Gradio приложение для мультиагентной системы анализа логов. Оркестрирует работу трёх агентов и предоставляет веб-интерфейс. """ import gradio as gr import json from typing import Tuple, Dict, Any from agents import LogParserAgent, AnomalyDetectionAgent, RootCauseAgent class LogAnalysisOrchestrator: """Оркестратор для последовательного выполнения агентов.""" def __init__(self): """Инициализация оркестратора и агентов.""" self.parser_agent = LogParserAgent() self.anomaly_agent = AnomalyDetectionAgent() self.rca_agent = RootCauseAgent() def analyze(self, raw_logs: str) -> Tuple[str, str, str]: """ Выполняет полный цикл анализа логов через всех агентов. Args: raw_logs: Сырые логи в виде строки Returns: Кортеж из трёх строк: - JSON с распарсенными логами - JSON с обнаруженными аномалиями - Markdown с рекомендациями """ try: # Валидация входных данных if not raw_logs or not raw_logs.strip(): empty_json = json.dumps({"error": "Логи не предоставлены"}, ensure_ascii=False, indent=2) return empty_json, empty_json, "# Ошибка\n\nЛоги не предоставлены для анализа." # Agent 1: Парсинг логов try: structured_data = self.parser_agent.parse(raw_logs) parsed_json = json.dumps(structured_data, ensure_ascii=False, indent=2) except Exception as e: error_msg = {"error": f"Ошибка парсинга логов: {str(e)}"} parsed_json = json.dumps(error_msg, ensure_ascii=False, indent=2) return parsed_json, parsed_json, f"# Ошибка\n\nОшибка на этапе парсинга: {str(e)}" # Agent 2: Обнаружение аномалий try: anomaly_report = self.anomaly_agent.detect(structured_data) anomalies_json = json.dumps(anomaly_report, ensure_ascii=False, indent=2) except Exception as e: error_msg = {"error": f"Ошибка обнаружения аномалий: {str(e)}"} anomalies_json = json.dumps(error_msg, ensure_ascii=False, indent=2) return parsed_json, anomalies_json, f"# Ошибка\n\nОшибка на этапе обнаружения аномалий: {str(e)}" # Agent 3: Анализ первопричин и рекомендации try: recommendations_md = self.rca_agent.analyze(anomaly_report) except Exception as e: recommendations_md = f"# Ошибка\n\nОшибка на этапе анализа первопричин: {str(e)}" return parsed_json, anomalies_json, recommendations_md except Exception as e: error_json = json.dumps({"error": f"Критическая ошибка: {str(e)}"}, ensure_ascii=False, indent=2) return error_json, error_json, f"# Критическая ошибка\n\n{str(e)}" # Глобальный экземпляр оркестратора orchestrator = LogAnalysisOrchestrator() def analyze_logs(raw_logs: str) -> Tuple[str, str, str]: """ Обёртка для Gradio интерфейса. Args: raw_logs: Сырые логи из интерфейса Returns: Кортеж результатов для отображения """ return orchestrator.analyze(raw_logs) def create_interface(): """Создаёт и настраивает Gradio интерфейс.""" # Описание интерфейса description = """ # 🔍 Мультиагентная система анализа логов Система использует трёх независимых агентов для анализа логов: 1. **Log Parser Agent** - парсит и структурирует логи 2. **Anomaly Detection Agent** - обнаруживает аномалии и паттерны 3. **Root Cause Agent** - анализирует первопричины и генерирует рекомендации Вставьте логи в поле ниже или загрузите лог-файл, затем нажмите "Анализировать". """ # Создание интерфейса with gr.Blocks(title="Multi-Agent Log Analysis") as app: gr.Markdown(description) with gr.Row(): with gr.Column(scale=1): log_input = gr.Textbox( label="Логи для анализа", placeholder="Вставьте логи здесь или используйте кнопку загрузки ниже...", lines=15, max_lines=30 ) upload_btn = gr.UploadButton( "📁 Загрузить лог-файл", file_types=[".log", ".txt"], file_count="single" ) analyze_btn = gr.Button("🔍 Анализировать", variant="primary", size="lg") # Обработчик загрузки файла def load_file(file): if file is None: return "" try: with open(file.name, 'r', encoding='utf-8') as f: content = f.read() return content except Exception as e: return f"Ошибка чтения файла: {str(e)}" upload_btn.upload(load_file, inputs=upload_btn, outputs=log_input) with gr.Row(): with gr.Tabs(): with gr.Tab("📊 Распарсенные логи (JSON)"): parsed_output = gr.JSON( label="Структурированные данные" ) with gr.Tab("⚠️ Обнаруженные аномалии (JSON)"): anomalies_output = gr.JSON( label="Отчёт об аномалиях" ) with gr.Tab("💡 Анализ и рекомендации (Markdown)"): recommendations_output = gr.Markdown( label="Рекомендации по устранению проблем" ) # Примеры логов для тестирования gr.Markdown("### 📝 Примеры логов для тестирования") with gr.Row(): example_logs = [ """2024-01-15 10:00:00 INFO Application started 2024-01-15 10:00:05 INFO Database connection established 2024-01-15 10:01:00 ERROR Connection timeout to external API 2024-01-15 10:01:05 ERROR Connection timeout to external API 2024-01-15 10:01:10 ERROR Connection timeout to external API 2024-01-15 10:01:15 WARNING High memory usage detected: 85% 2024-01-15 10:02:00 CRITICAL System crash detected 2024-01-15 10:02:01 INFO Application shutdown""", """[2024-01-15 14:30:00] INFO User authentication successful [2024-01-15 14:30:01] DEBUG Request received: GET /api/users [2024-01-15 14:30:02] ERROR Database query failed: connection lost [2024-01-15 14:30:03] ERROR Database query failed: connection lost [2024-01-15 14:30:04] ERROR Database query failed: connection lost [2024-01-15 14:30:05] ERROR Database query failed: connection lost [2024-01-15 14:30:06] WARNING Retrying database connection [2024-01-15 14:30:10] INFO Database connection restored""" ] example_btn1 = gr.Button("Загрузить пример 1", size="sm") example_btn2 = gr.Button("Загрузить пример 2", size="sm") example_btn1.click( lambda: example_logs[0], outputs=log_input ) example_btn2.click( lambda: example_logs[1], outputs=log_input ) # Связывание кнопки анализа с обработчиком analyze_btn.click( fn=analyze_logs, inputs=log_input, outputs=[parsed_output, anomalies_output, recommendations_output] ) # Информация о системе gr.Markdown(""" --- ### ℹ️ Информация о системе - **Архитектура:** Мультиагентная система (3 независимых агента) - **Платформа:** Hugging Face Spaces - **Интерфейс:** Gradio - **Поддержка:** До 10,000 строк логов """) return app # Создание и запуск приложения if __name__ == "__main__": app = create_interface() app.launch(server_name="0.0.0.0", server_port=7860) else: # Для Hugging Face Spaces app = create_interface()