app.py

"""

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()