Refactor: Move GPT Prompt Agent to DeepSeek model via smolagents, all 4 agents now use transformers

agents/__init__.py

@@ -186,12 +186,57 @@ def run_rca_agent(anomaly_report: dict):
     validate_schema(result, rca_schema)
     return result
 
-# Экспорт для обратной совместимости
-from .gpt_prompt_agent import GPTPromptAgent
+# Агент 4: Генерация промпта для GPT
+gpt_prompt_agent = ToolCallingAgent(
+    model=model,
+    tools=[final_tool],
+    instructions="""
+Ты эксперт по созданию детальных промптов для GPT-моделей. Твоя задача - создать готовый промпт для анализа проблем на основе структурированных данных о логах, аномалиях и рекомендациях.
+
+Промпт должен быть структурированным и содержать:
+1. Контекст проблемы - общее описание ситуации
+2. Информация о системе - статистика, временные диапазоны, типы событий
+3. Обнаруженные проблемы - детальное описание аномалий с приоритетами
+4. Статистика и метрики - количественные показатели
+5. Примеры ошибок - ключевые ошибки из логов
+6. Предварительный анализ - рекомендации от предыдущих агентов (если есть)
+7. Запрос на решение - конкретные вопросы для GPT
+
+Промпт должен быть готов к использованию - его можно скопировать и вставить в ChatGPT, Claude или другую GPT-модель.
+
+Ответ верни как обычный текст (не JSON), используя final_answer. Это должен быть готовый промпт на русском языке в формате Markdown.
+""",
+    name="GPTPromptAgent",
+    max_steps=10,
+)
+
+# Pipeline шаг 4 — генерация промпта для GPT
+def run_gpt_prompt_agent(structured_data: dict, anomaly_report: dict, recommendations: str = None):
+    """Генерирует промпт для GPT на основе всех данных анализа."""
+    # Подготавливаем входные данные для агента
+    input_data = {
+        "structured_data": structured_data,
+        "anomaly_report": anomaly_report,
+        "recommendations": recommendations
+    }
+    input_json = json.dumps(input_data, ensure_ascii=False, indent=2)
+    
+    result = run_agent_safely(gpt_prompt_agent, task=input_json)
+    
+    # Результат должен быть строкой (промпт), а не JSON
+    if isinstance(result, dict) and "answer" in result:
+        return result["answer"]
+    elif isinstance(result, str):
+        return result
+    else:
+        # Если агент вернул JSON, попробуем извлечь промпт
+        if isinstance(result, dict) and "prompt" in result:
+            return result["prompt"]
+        return str(result)
 
 __all__ = [
     'run_parser_agent',
     'run_anomaly_agent',
     'run_rca_agent',
-    'GPTPromptAgent',
+    'run_gpt_prompt_agent',
 ]

agents/gpt_prompt_agent.py

@@ -1,325 +0,0 @@
-"""
-Agent 4: GPT Prompt Generator Agent
-Формирует детальный промпт для GPT-модели с полным контекстом проблемы и системы.
-"""
-
-from typing import Dict, List, Any, Optional
-from datetime import datetime
-
-
-class GPTPromptAgent:
-    """Генерирует промпт для GPT на основе анализа логов."""
-    
-    def __init__(self):
-        """Инициализация агента."""
-        pass
-    
-    def generate_prompt(
-        self, 
-        structured_data: Dict[str, Any],
-        anomaly_report: Dict[str, Any],
-        recommendations: Optional[str] = None
-    ) -> str:
-        """
-        Генерирует детальный промпт для GPT-модели.
-        
-        Args:
-            structured_data: Структурированные данные от LogParserAgent
-            anomaly_report: Отчёт об аномалиях от AnomalyDetectionAgent
-            recommendations: Рекомендации от RootCauseAgent (опционально)
-            
-        Returns:
-            Готовый промпт в виде строки для вставки в GPT
-        """
-        prompt_parts = []
-        
-        # Заголовок промпта
-        prompt_parts.append("# Контекст проблемы из системных логов\n")
-        prompt_parts.append("\nЯ анализирую логи системы и обнаружил несколько проблем. ")
-        prompt_parts.append("Мне нужна помощь в диагностике и решении этих проблем.\n")
-        
-        # Раздел 1: Общая информация о системе
-        prompt_parts.append("\n## 1. Информация о системе\n\n")
-        system_info = self._extract_system_info(structured_data)
-        prompt_parts.append(system_info)
-        
-        # Раздел 2: Обнаруженные проблемы
-        prompt_parts.append("\n## 2. Обнаруженные проблемы и аномалии\n\n")
-        problems_info = self._format_anomalies(anomaly_report)
-        prompt_parts.append(problems_info)
-        
-        # Раздел 3: Статистика и метрики
-        prompt_parts.append("\n## 3. Статистика событий\n\n")
-        stats_info = self._format_statistics(structured_data, anomaly_report)
-        prompt_parts.append(stats_info)
-        
-        # Раздел 4: Примеры ошибок и логов
-        prompt_parts.append("\n## 4. Примеры ошибок и ключевые логи\n\n")
-        examples_info = self._format_error_examples(structured_data, anomaly_report)
-        prompt_parts.append(examples_info)
-        
-        # Раздел 5: Анализ первопричин (если есть рекомендации)
-        if recommendations:
-            prompt_parts.append("\n## 5. Предварительный анализ и рекомендации\n\n")
-            prompt_parts.append(recommendations)
-            prompt_parts.append("\n")
-        
-        # Раздел 6: Запрос к GPT
-        prompt_parts.append("\n## 6. Запрос на решение\n\n")
-        prompt_parts.append("Пожалуйста, помогите мне:\n")
-        prompt_parts.append("1. Определить точную первопричину проблемы\n")
-        prompt_parts.append("2. Предложить конкретные шаги для решения проблемы\n")
-        prompt_parts.append("3. Предложить меры профилактики, чтобы избежать подобных проблем в будущем\n")
-        prompt_parts.append("4. Если это программная ошибка, предложите патч или исправление кода (если возможно)\n")
-        prompt_parts.append("5. Оцените критичность проблемы и приоритет решения\n\n")
-        
-        prompt_parts.append("Проанализируйте предоставленную информацию и дайте развёрнутый ответ с конкретными рекомендациями.\n")
-        
-        return "".join(prompt_parts)
-    
-    def _extract_system_info(self, structured_data: Dict[str, Any]) -> str:
-        """Извлекает информацию о системе из структурированных данных."""
-        info_parts = []
-        
-        statistics = structured_data.get('statistics', {})
-        events = structured_data.get('events', [])
-        
-        # Временной диапазон
-        time_range = statistics.get('time_range')
-        if time_range:
-            start_time = time_range.get('start') if isinstance(time_range, dict) else None
-            end_time = time_range.get('end') if isinstance(time_range, dict) else None
-            if start_time and end_time:
-                info_parts.append(f"- **Временной диапазон логов:** с {start_time} по {end_time}\n")
-            elif start_time:
-                info_parts.append(f"- **Начало периода:** {start_time}\n")
-        
-        # Общая статистика
-        total_lines = statistics.get('total_lines', 0)
-        parsed_events = statistics.get('parsed_events', 0)
-        errors_count = statistics.get('errors', 0)
-        warnings_count = statistics.get('warnings', 0)
-        
-        info_parts.append(f"- **Всего строк в логах:** {total_lines}\n")
-        info_parts.append(f"- **Распарсенных событий:** {parsed_events}\n")
-        info_parts.append(f"- **Ошибок:** {errors_count}\n")
-        info_parts.append(f"- **Предупреждений:** {warnings_count}\n")
-        
-        # Типы событий
-        event_types = statistics.get('event_types', {})
-        if event_types:
-            info_parts.append(f"- **Типы событий в системе:**\n")
-            for event_type, count in sorted(event_types.items(), key=lambda x: x[1], reverse=True)[:10]:
-                info_parts.append(f"  - {event_type}: {count} событий\n")
-        
-        # Попытка извлечь информацию о системе из логов
-        system_info = self._extract_technical_details(events)
-        if system_info:
-            info_parts.append(f"\n### Технические детали системы:\n\n")
-            info_parts.append(system_info)
-        
-        return "".join(info_parts) if info_parts else "Информация о системе не обнаружена в логах.\n"
-    
-    def _extract_technical_details(self, events: List[Dict]) -> str:
-        """Извлекает технические детали из событий (версии, компоненты и т.д.)."""
-        details = []
-        seen_details = set()
-        
-        # Ключевые слова для поиска системной информации
-        keywords = {
-            'version': ['version', 'версия', 'v.', 'ver '],
-            'component': ['component', 'компонент', 'module', 'модуль', 'service', 'сервис'],
-            'framework': ['framework', 'framework version', 'django', 'flask', 'spring', 'express'],
-            'database': ['database', 'db', 'mysql', 'postgresql', 'mongodb', 'redis'],
-            'os': ['linux', 'windows', 'ubuntu', 'centos', 'debian', 'os version'],
-            'language': ['python', 'java', 'node', 'javascript', 'typescript', 'go', 'rust'],
-            'protocol': ['http', 'https', 'tcp', 'udp', 'grpc']
-        }
-        
-        for event in events[:200]:  # Проверяем первые 200 событий
-            message = event.get('message', '').lower()
-            level = event.get('level', '').upper()
-            
-            # Извлекаем информацию о версиях и компонентах
-            for detail_type, search_terms in keywords.items():
-                for term in search_terms:
-                    if term in message and detail_type not in seen_details:
-                        # Пытаемся извлечь более конкретную информацию
-                        context = message[max(0, message.find(term) - 50):message.find(term) + 100]
-                        if context and len(context) > 10:
-                            details.append(f"- **{detail_type.upper()}:** обнаружен в логах (контекст: {context[:80]}...)\n")
-                            seen_details.add(detail_type)
-                            break
-                if detail_type in seen_details:
-                    break
-        
-        return "".join(details) if details else "Специфические технические детали не обнаружены в логах.\n"
-    
-    def _format_anomalies(self, anomaly_report: Dict[str, Any]) -> str:
-        """Форматирует информацию об аномалиях для промпта."""
-        if not anomaly_report or not anomaly_report.get('anomalies'):
-            return "Критических аномалий не обнаружено. Система работает в штатном режиме.\n"
-        
-        anomalies = anomaly_report.get('anomalies', [])
-        statistics = anomaly_report.get('statistics', {})
-        severity_summary = anomaly_report.get('severity_summary', {})
-        
-        info_parts = []
-        
-        # Общая сводка
-        total_anomalies = statistics.get('total', 0)
-        info_parts.append(f"**Всего обнаружено аномалий:** {total_anomalies}\n\n")
-        
-        # Сводка по серьёзности
-        if severity_summary:
-            info_parts.append("**Распределение по серьёзности:**\n")
-            severity_names = {
-                'CRITICAL': '🔴 КРИТИЧЕСКИЕ',
-                'HIGH': '🟠 ВЫСОКИЕ',
-                'MEDIUM': '🟡 СРЕДНИЕ',
-                'LOW': '🟢 НИЗКИЕ'
-            }
-            for severity, count in sorted(severity_summary.items(), key=lambda x: ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW'].index(x[0]) if x[0] in ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW'] else 999):
-                name = severity_names.get(severity, severity)
-                info_parts.append(f"- {name}: {count} проблем(ы)\n")
-            info_parts.append("\n")
-        
-        # Детальное описание каждой аномалии
-        info_parts.append("### Детальное описание проблем:\n\n")
-        
-        # Сортируем по серьёзности
-        severity_order = {'CRITICAL': 0, 'HIGH': 1, 'MEDIUM': 2, 'LOW': 3}
-        sorted_anomalies = sorted(
-            anomalies, 
-            key=lambda x: severity_order.get(x.get('severity', 'LOW'), 999)
-        )
-        
-        for i, anomaly in enumerate(sorted_anomalies, 1):
-            anomaly_type = anomaly.get('type', 'UNKNOWN')
-            severity = anomaly.get('severity', 'UNKNOWN')
-            description = anomaly.get('description', 'Нет описания')
-            
-            info_parts.append(f"#### Проблема #{i}: {self._get_anomaly_type_name(anomaly_type)} ({severity})\n\n")
-            info_parts.append(f"{description}\n\n")
-            
-            # Дополнительные детали
-            if anomaly.get('count'):
-                info_parts.append(f"- **Количество вхождений:** {anomaly.get('count')}\n")
-            
-            if anomaly.get('error_message'):
-                error_msg = anomaly.get('error_message', '')[:300]
-                info_parts.append(f"- **Сообщение об ошибке:** `{error_msg}`\n")
-            
-            if anomaly.get('sample_messages'):
-                info_parts.append(f"- **Примеры сообщений:**\n")
-                for msg in anomaly.get('sample_messages', [])[:3]:
-                    info_parts.append(f"  - `{msg[:200]}`\n")
-            
-            if anomaly.get('time_window'):
-                info_parts.append(f"- **Временное окно:** {anomaly.get('time_window')}\n")
-            
-            if anomaly.get('first_occurrence'):
-                info_parts.append(f"- **Первое появление:** {anomaly.get('first_occurrence')}\n")
-            
-            if anomaly.get('last_occurrence'):
-                info_parts.append(f"- **Последнее появление:** {anomaly.get('last_occurrence')}\n")
-            
-            if anomaly.get('metadata', {}).get('affected_lines'):
-                lines = anomaly.get('metadata', {}).get('affected_lines', [])[:10]
-                info_parts.append(f"- **Затронутые строки логов:** {', '.join(map(str, lines))}\n")
-            
-            info_parts.append("\n")
-        
-        return "".join(info_parts)
-    
-    def _format_statistics(self, structured_data: Dict[str, Any], anomaly_report: Dict[str, Any]) -> str:
-        """Форматирует статистику для промпта."""
-        stats_parts = []
-        
-        statistics = structured_data.get('statistics', {})
-        anomaly_stats = anomaly_report.get('statistics', {})
-        
-        stats_parts.append("### Общая статистика событий:\n\n")
-        
-        total_lines = statistics.get('total_lines', 0)
-        parsed_events = statistics.get('parsed_events', 0)
-        errors = statistics.get('errors', 0)
-        warnings = statistics.get('warnings', 0)
-        info_count = statistics.get('info_messages', 0)
-        
-        if total_lines > 0:
-            error_rate = (errors / parsed_events * 100) if parsed_events > 0 else 0
-            warning_rate = (warnings / parsed_events * 100) if parsed_events > 0 else 0
-            
-            stats_parts.append(f"- **Всего строк логов:** {total_lines}\n")
-            stats_parts.append(f"- **Распарсенных событий:** {parsed_events}\n")
-            stats_parts.append(f"- **Ошибок:** {errors} ({error_rate:.2f}% от всех событий)\n")
-            stats_parts.append(f"- **Предупреждений:** {warnings} ({warning_rate:.2f}% от всех событий)\n")
-            stats_parts.append(f"- **Информационных сообщений:** {info_count}\n")
-        
-        if anomaly_stats:
-            stats_parts.append(f"\n### Статистика аномалий:\n\n")
-            by_type = anomaly_stats.get('by_type', {})
-            if by_type:
-                stats_parts.append("**По типам:**\n")
-                for anomaly_type, count in sorted(by_type.items(), key=lambda x: x[1], reverse=True):
-                    type_name = self._get_anomaly_type_name(anomaly_type)
-                    stats_parts.append(f"- {type_name}: {count}\n")
-        
-        return "".join(stats_parts)
-    
-    def _format_error_examples(self, structured_data: Dict[str, Any], anomaly_report: Dict[str, Any]) -> str:
-        """Форматирует примеры ошибок для промпта."""
-        examples_parts = []
-        
-        errors = structured_data.get('errors', [])
-        anomalies = anomaly_report.get('anomalies', [])
-        
-        if not errors:
-            return "Явных ошибок в логах не обнаружено.\n"
-        
-        examples_parts.append("### Ключевые ошибки из логов:\n\n")
-        
-        # Показываем уникальные ошибки (первые 10)
-        unique_errors = []
-        seen_messages = set()
-        
-        for error in errors[:50]:
-            message = error.get('message', '').strip()
-            if message and message not in seen_messages:
-                unique_errors.append(error)
-                seen_messages.add(message)
-                if len(unique_errors) >= 10:
-                    break
-        
-        for i, error in enumerate(unique_errors, 1):
-            level = error.get('level', 'UNKNOWN')
-            message = error.get('message', '')[:400]
-            timestamp = error.get('timestamp', 'N/A')
-            line_num = error.get('line_number', 'N/A')
-            
-            examples_parts.append(f"**Ошибка #{i}** (уровень: {level}, строка: {line_num}, время: {timestamp}):\n")
-            examples_parts.append(f"```\n{message}\n```\n\n")
-        
-        # Если есть примеры из аномалий, добавляем их
-        if anomalies:
-            examples_parts.append("### Примеры проблемных паттернов из аномалий:\n\n")
-            for anomaly in anomalies[:5]:
-                if anomaly.get('sample_messages'):
-                    examples_parts.append(f"**{self._get_anomaly_type_name(anomaly.get('type'))}:**\n")
-                    for msg in anomaly.get('sample_messages', [])[:2]:
-                        examples_parts.append(f"- `{msg[:200]}`\n")
-                    examples_parts.append("\n")
-        
-        return "".join(examples_parts)
-    
-    def _get_anomaly_type_name(self, anomaly_type: str) -> str:
-        """Возвращает читаемое имя типа аномалии."""
-        names = {
-            'BURST_ERRORS': 'Всплеск ошибок (Burst Errors)',
-            'REPEATED_ERRORS': 'Повторяющиеся ошибки (Repeated Errors)',
-            'ERROR_BEFORE_CRASH': 'Ошибка перед крашем (Error Before Crash)',
-            'TEMPORAL_SPIKE': 'Временной всплеск событий (Temporal Spike)',
-            'REPEATED_STACK_TRACES': 'Повторяющиеся stack traces (Repeated Stack Traces)',
-        }
-        return names.get(anomaly_type, anomaly_type)

app.py

@@ -8,13 +8,7 @@ import json
 import os
 from typing import Tuple
 
-from agents import run_parser_agent, run_anomaly_agent, run_rca_agent
-
-# Импорт GPT Prompt Agent (остаётся rule-based, не использует трансформеры)
-from agents.gpt_prompt_agent import GPTPromptAgent
-
-# Инициализация GPT Prompt Agent (остаётся rule-based)
-gpt_prompt_agent = GPTPromptAgent()
+from agents import run_parser_agent, run_anomaly_agent, run_rca_agent, run_gpt_prompt_agent
 
 
 def format_rca_as_markdown(rca_result: dict) -> str:
@@ -131,7 +125,7 @@ def analyze_logs(raw_logs: str) -> Tuple[str, str, str, str]:
         
         # Agent 4: Генерация промпта для GPT
         try:
-            gpt_prompt = gpt_prompt_agent.generate_prompt(
+            gpt_prompt = run_gpt_prompt_agent(
                 structured_data, 
                 anomaly_report, 
                 recommendations_md
@@ -281,7 +275,7 @@ def create_interface():
         ---
         ### ℹ️ Информация о системе
         
-        - **Архитектура:** Мультиагентная система (4 независимых агента на основе трансформеров)
+        - **Архитектура:** Мультиагентная система (4 независимых агента на основе DeepSeek через smolagents)
         - **Платформа:** Hugging Face Spaces
         - **Интерфейс:** Gradio
         - **Поддержка:** До 10,000 строк логов