Add: Streaming processing toggle and speed optimization guide - Add streaming processing option in Gradio UI - Auto-enable for files >500 lines - Add chunk size slider - Create SPEED_OPTIMIZATION.md with model suggestions

SPEED_OPTIMIZATION.md

@@ -0,0 +1,173 @@
+# Оптимизация скорости обработки
+
+## Текущая ситуация
+
+- **Модель:** DeepSeek-V3.1-Terminus (через smolagents InferenceClientModel)
+- **Время обработки:** ~2.5 минуты для 13 строк логов
+- **Проблема:** Модель медленная из-за размера и API задержек
+
+## Варианты ускорения
+
+### 1. Использование более быстрых моделей DeepSeek
+
+#### Вариант A: DeepSeek-R1 (более быстрая версия)
+```python
+model = InferenceClientModel(
+    model_id="deepseek-ai/DeepSeek-R1",
+    token=hf_token,
+    max_tokens=4096
+)
+```
+- **Плюсы:** Быстрее, меньше задержка
+- **Минусы:** Может быть менее точной для сложных задач
+
+#### Вариант B: DeepSeek-Coder (специализированная модель для кода/структурированных данных)
+```python
+model = InferenceClientModel(
+    model_id="deepseek-ai/deepseek-coder-6.7b-instruct",
+    token=hf_token,
+    max_tokens=2048
+)
+```
+- **Плюсы:** Оптимизирована для структурированных данных (JSON, логи)
+- **Минусы:** Меньший контекст, может быть менее точной для анализа
+
+### 2. Использование других быстрых моделей через HF Inference API
+
+#### Qwen/Qwen2.5 (быстрая и качественная)
+```python
+model = InferenceClientModel(
+    model_id="Qwen/Qwen2.5-7B-Instruct",
+    token=hf_token,
+    max_tokens=2048
+)
+```
+
+#### Mistral (баланс скорости и качества)
+```python
+model = InferenceClientModel(
+    model_id="mistralai/Mistral-7B-Instruct-v0.2",
+    token=hf_token,
+    max_tokens=2048
+)
+```
+
+### 3. Оптимизация промптов и параметров
+
+#### Уменьшение max_tokens для более коротких ответов
+```python
+model = InferenceClientModel(
+    model_id="deepseek-ai/DeepSeek-V3.1-Terminus",
+    token=hf_token,
+    max_tokens=2048  # Вместо 4096 - быстрее генерация
+)
+```
+
+#### Уменьшение max_steps для агентов
+```python
+parser_agent = ToolCallingAgent(
+    model=model,
+    tools=[final_tool],
+    instructions="...",
+    name="LogParserAgent",
+    max_steps=5,  # Вместо 10 - меньше итераций
+)
+```
+
+### 4. Параллельная обработка чанков
+
+В `streaming_processor.py` можно добавить:
+```python
+from concurrent.futures import ThreadPoolExecutor
+
+def process_chunks_parallel(chunks, max_workers=3):
+    with ThreadPoolExecutor(max_workers=max_workers) as executor:
+        results = executor.map(process_single_chunk, chunks)
+    return list(results)
+```
+
+### 5. Кэширование результатов
+
+Кэшировать результаты парсинга одинаковых строк:
+```python
+from functools import lru_cache
+import hashlib
+
+@lru_cache(maxsize=1000)
+def cached_parse(log_hash: str, logs: str):
+    return run_parser_agent(logs)
+```
+
+### 6. Использование локальной модели (если доступно)
+
+Если есть GPU, можно использовать локальную модель вместо API:
+```python
+from transformers import AutoModelForCausalLM, AutoTokenizer
+from smolagents import LocalModel
+
+model = LocalModel(
+    model_id="deepseek-ai/DeepSeek-V3.1-Terminus",
+    device="cuda"  # или "cpu"
+)
+```
+
+## Рекомендации
+
+### Для максимальной скорости (уровень 1):
+1. Использовать более быструю модель (Qwen2.5 или DeepSeek-R1)
+2. Уменьшить max_tokens до 2048
+3. Уменьшить max_steps до 5
+4. Включить потоковую обработку для больших файлов
+
+### Для баланса скорости и качества (уровень 2):
+1. Оставить DeepSeek-V3.1-Terminus
+2. Уменьшить max_tokens до 2048
+3. Оптимизировать промпты (делать их короче и конкретнее)
+4. Использовать потоковую обработку
+
+### Для максимального качества (текущий уровень):
+1. Оставить текущую настройку
+2. Использовать retry механизм
+3. Использовать потоковую обработку для больших файлов
+
+## Пример реализации быстрой модели
+
+Создать `agents/fast_agents.py`:
+```python
+import os
+from smolagents import ToolCallingAgent, InferenceClientModel, FinalAnswerTool
+
+hf_token = os.getenv("HF_TOKEN")
+
+# Быстрая модель для быстрой обр��ботки
+fast_model = InferenceClientModel(
+    model_id="Qwen/Qwen2.5-7B-Instruct",  # или "deepseek-ai/DeepSeek-R1"
+    token=hf_token,
+    max_tokens=2048,
+    temperature=0.3  # Меньше творчества = быстрее
+)
+
+# Использовать fast_model вместо model для быстрых задач
+```
+
+## Тестирование скорости
+
+Создать бенчмарк:
+```python
+import time
+from test_logs.good_example_1_web_server import test_logs
+
+start = time.time()
+result = analyze_logs(test_logs)
+end = time.time()
+
+print(f"Время обработки: {end - start:.2f} секунд")
+```
+
+## Вывод
+
+**Рекомендуемый подход:**
+1. Добавить выбор модели в интерфейс (Fast/Standard)
+2. Использовать Qwen2.5 для быстрого режима
+3. Оставить DeepSeek-V3.1 для режима максимального качества
+4. Включить потоковую обработку по умолчанию для файлов >500 строк

app.py

@@ -9,6 +9,7 @@ import os
 from typing import Tuple
 
 from agents import run_parser_agent, run_anomaly_agent, run_rca_agent, run_gpt_prompt_agent
+from utils.streaming_processor import process_logs_streaming
 
 
 def format_rca_as_markdown(rca_result: dict) -> str:
@@ -76,12 +77,14 @@ def format_rca_as_markdown(rca_result: dict) -> str:
     return "".join(markdown_parts)
 
 
-def analyze_logs(raw_logs: str) -> Tuple[str, str, str, str]:
+def analyze_logs(raw_logs: str, use_streaming: bool = False, chunk_size: int = 100) -> Tuple[str, str, str, str]:
     """
     Обёртка для Gradio интерфейса.
     
     Args:
         raw_logs: Сырые логи из интерфейса
+        use_streaming: Использовать потоковую обработку для больших логов
+        chunk_size: Размер чанка для потоковой обработки (в строках)
         
     Returns:
         Кортеж результатов для отображения:
@@ -97,6 +100,17 @@ def analyze_logs(raw_logs: str) -> Tuple[str, str, str, str]:
             error_prompt = "# Ошибка\n\nЛоги не предоставлены для анализа."
             return error_json, error_json, "# Ошибка\n\nЛоги не предоставлены для анализа.", error_prompt
         
+        # Определяем количество строк
+        log_lines = len(raw_logs.strip().split('\n'))
+        
+        # Автоматически включаем потоковую обработку для больших файлов
+        # или если пользователь явно включил её
+        should_stream = use_streaming or log_lines > 500
+        
+        if should_stream:
+            print(f"[Streaming Mode] Processing {log_lines} lines in chunks of {chunk_size}...")
+            return analyze_logs_streaming(raw_logs, chunk_size)
+        
         # Agent 1: Парсинг логов
         try:
             structured_data = run_parser_agent(raw_logs)
@@ -134,13 +148,48 @@ def analyze_logs(raw_logs: str) -> Tuple[str, str, str, str]:
             gpt_prompt = f"# Ошибка генерации промпта\n\nПроизошла ошибка при генерации промпта для GPT: {str(e)}\n\nПопробуйте использовать информацию из других вкладок."
         
         return parsed_json, anomalies_json, recommendations_md, gpt_prompt
-        
+    
     except Exception as e:
         error_json = json.dumps({"error": f"Критическая ошибка: {str(e)}"}, ensure_ascii=False, indent=2)
         error_prompt = f"# Критическая ошибка\n\n{str(e)}"
         return error_json, error_json, f"# Критическая ошибка\n\n{str(e)}", error_prompt
 
 
+def analyze_logs_streaming(raw_logs: str, chunk_size: int = 100) -> Tuple[str, str, str, str]:
+    """
+    Потоковая обработка логов для больших файлов.
+    
+    Args:
+        raw_logs: Сырые логи из интерфейса
+        chunk_size: Размер чанка в строках
+        
+    Returns:
+        Кортеж результатов для отображения
+    """
+    try:
+        result = process_logs_streaming(raw_logs, chunk_size=chunk_size)
+        
+        structured_data = result['structured_data']
+        anomaly_report = result['anomaly_report']
+        rca_result = result['rca_result']
+        gpt_prompt = result['gpt_prompt']
+        
+        parsed_json = json.dumps(structured_data, ensure_ascii=False, indent=2)
+        anomalies_json = json.dumps(anomaly_report, ensure_ascii=False, indent=2)
+        recommendations_md = format_rca_as_markdown(rca_result)
+        
+        # Добавляем информацию о потоковой обработке
+        info_note = f"\n\n*Обработано потоковым способом: {result['chunks_processed']} чанков по {result['chunk_size']} строк*\n"
+        recommendations_md = recommendations_md + info_note
+        
+        return parsed_json, anomalies_json, recommendations_md, gpt_prompt
+        
+    except Exception as e:
+        error_json = json.dumps({"error": f"Ошибка потоковой обработки: {str(e)}"}, ensure_ascii=False, indent=2)
+        error_prompt = f"# Ошибка\n\nОшибка потоковой обработки: {str(e)}"
+        return error_json, error_json, error_prompt, error_prompt
+
+
 def create_interface():
     """Создаёт и настраивает Gradio интерфейс."""
     
@@ -176,6 +225,22 @@ def create_interface():
                     file_count="single"
                 )
                 
+                # Настройки обработки
+                with gr.Accordion("⚙️ Настройки обработки", open=False):
+                    use_streaming = gr.Checkbox(
+                        label="Потоковая обработка (для больших файлов >500 строк)",
+                        value=False,
+                        info="Автоматически включается для файлов >500 строк. Разбивает логи на части для более быстрой обработки."
+                    )
+                    chunk_size = gr.Slider(
+                        label="Размер чанка (строк)",
+                        minimum=50,
+                        maximum=200,
+                        value=100,
+                        step=50,
+                        info="Количество строк в одном чанке при потоковой обработке"
+                    )
+                
                 analyze_btn = gr.Button("🔍 Анализировать", variant="primary", size="lg")
                 
                 # Обработчик загрузки файла
@@ -266,7 +331,7 @@ def create_interface():
         # Связывание кнопки анализа с обработчиком
         analyze_btn.click(
             fn=analyze_logs,
-            inputs=log_input,
+            inputs=[log_input, use_streaming, chunk_size],
             outputs=[parsed_output, anomalies_output, recommendations_output, gpt_prompt_output]
         )
         

utils/streaming_processor.py

@@ -0,0 +1,201 @@
+"""
+Потоковая обработка больших логов.
+Разбивает логи на чанки и обрабатывает их параллельно или последовательно.
+"""
+
+from typing import List, Dict, Any, Callable, Generator
+from agents import run_parser_agent, run_anomaly_agent, run_rca_agent, run_gpt_prompt_agent
+
+
+def split_logs_into_chunks(logs: str, chunk_size: int = 100) -> Generator[str, None, None]:
+    """
+    Разбивает логи на чанки по количеству строк.
+    
+    Args:
+        logs: Полный текст логов
+        chunk_size: Количество строк в одном чанке
+        
+    Yields:
+        str: Чанк логов
+    """
+    lines = logs.strip().split('\n')
+    total_lines = len(lines)
+    
+    for i in range(0, total_lines, chunk_size):
+        chunk_lines = lines[i:i + chunk_size]
+        yield '\n'.join(chunk_lines)
+
+
+def process_logs_streaming(
+    raw_logs: str,
+    chunk_size: int = 100,
+    parallel: bool = False
+) -> Dict[str, Any]:
+    """
+    Обрабатывает логи потоковым способом - разбивает на чанки и обрабатывает каждый.
+    
+    Args:
+        raw_logs: Сырые логи для обработки
+        chunk_size: Размер чанка в строках
+        parallel: Если True, обрабатывает чанки параллельно (требует threading/multiprocessing)
+        
+    Returns:
+        dict: Объединенные результаты всех чанков
+    """
+    chunks = list(split_logs_into_chunks(raw_logs, chunk_size))
+    total_chunks = len(chunks)
+    
+    all_events = []
+    all_errors = []
+    all_warnings = []
+    all_anomalies = []
+    
+    statistics_accumulator = {
+        'total_lines': 0,
+        'parsed_events': 0,
+        'errors': 0,
+        'warnings': 0,
+        'info_messages': 0,
+        'event_types': {},
+        'time_range': None
+    }
+    
+    # Обрабатываем каждый чанк
+    for chunk_idx, chunk in enumerate(chunks, 1):
+        print(f"[Streaming] Processing chunk {chunk_idx}/{total_chunks} ({len(chunk.split(chr(10)))} lines)...")
+        
+        # Шаг 1: Парсинг чанка
+        try:
+            structured_data = run_parser_agent(chunk)
+            
+            # Объединяем результаты
+            all_events.extend(structured_data.get('events', []))
+            all_errors.extend(structured_data.get('errors', []))
+            all_warnings.extend(structured_data.get('warnings', []))
+            
+            # Обновляем статистику
+            stats = structured_data.get('statistics', {})
+            statistics_accumulator['total_lines'] += stats.get('total_lines', 0)
+            statistics_accumulator['parsed_events'] += stats.get('parsed_events', 0)
+            statistics_accumulator['errors'] += stats.get('errors', 0)
+            statistics_accumulator['warnings'] += stats.get('warnings', 0)
+            statistics_accumulator['info_messages'] += stats.get('info_messages', 0)
+            
+            # Объединяем типы событий
+            for event_type, count in stats.get('event_types', {}).items():
+                statistics_accumulator['event_types'][event_type] = \
+                    statistics_accumulator['event_types'].get(event_type, 0) + count
+            
+            # Обновляем временной диапазон (берем самый ранний start и самый поздний end)
+            chunk_time_range = stats.get('time_range')
+            if chunk_time_range:
+                if statistics_accumulator['time_range'] is None:
+                    statistics_accumulator['time_range'] = chunk_time_range.copy()
+                else:
+                    if chunk_time_range.get('start'):
+                        if (statistics_accumulator['time_range'].get('start') is None or
+                            chunk_time_range['start'] < statistics_accumulator['time_range']['start']):
+                            statistics_accumulator['time_range']['start'] = chunk_time_range['start']
+                    
+                    if chunk_time_range.get('end'):
+                        if (statistics_accumulator['time_range'].get('end') is None or
+                            chunk_time_range['end'] > statistics_accumulator['time_range']['end']):
+                            statistics_accumulator['time_range']['end'] = chunk_time_range['end']
+        
+        except Exception as e:
+            print(f"[Streaming] Error processing chunk {chunk_idx}: {e}")
+            continue
+    
+    # Обновляем номера строк в объединенных событиях
+    for idx, event in enumerate(all_events, 1):
+        event['line_number'] = idx
+    
+    # Шаг 2: Обнаружение аномалий на объединенных данных
+    merged_structured_data = {
+        'events': all_events,
+        'errors': all_errors,
+        'warnings': all_warnings,
+        'statistics': statistics_accumulator
+    }
+    
+    print(f"[Streaming] Analyzing {len(all_events)} total events for anomalies...")
+    anomaly_report = run_anomaly_agent(merged_structured_data)
+    all_anomalies = anomaly_report.get('anomalies', [])
+    
+    # Шаг 3: Анализ первопричин
+    print(f"[Streaming] Running root cause analysis...")
+    rca_result = run_rca_agent(anomaly_report)
+    
+    # Шаг 4: Генерация промпта для GPT
+    print(f"[Streaming] Generating GPT prompt...")
+    recommendations_md = format_rca_as_markdown_streaming(rca_result)
+    gpt_prompt = run_gpt_prompt_agent(merged_structured_data, anomaly_report, recommendations_md)
+    
+    return {
+        'structured_data': merged_structured_data,
+        'anomaly_report': anomaly_report,
+        'rca_result': rca_result,
+        'gpt_prompt': gpt_prompt,
+        'chunks_processed': total_chunks,
+        'chunk_size': chunk_size
+    }
+
+
+def format_rca_as_markdown_streaming(rca_result: dict) -> str:
+    """Форматирует результат RCA в Markdown (копия из app.py для избежания циклических зависимостей)."""
+    markdown_parts = []
+    
+    # Возможные первопричины
+    analysis = rca_result.get("analysis", {})
+    root_causes = analysis.get("root_causes", [])
+    if root_causes:
+        markdown_parts.append("## Возможные первопричины\n\n")
+        for i, cause in enumerate(root_causes, 1):
+            markdown_parts.append(f"{i}. {cause}\n")
+        markdown_parts.append("\n")
+    
+    # Детальный анализ аномалий
+    details = analysis.get("details", [])
+    if details:
+        markdown_parts.append("## Детальный анализ аномалий\n\n")
+        for detail in details:
+            anomaly_type = detail.get("anomaly_type", "UNKNOWN")
+            severity = detail.get("severity", "MEDIUM")
+            description = detail.get("description", "")
+            possible_causes = detail.get("possible_causes", [])
+            
+            emoji = "🔴" if severity == "CRITICAL" else "🟡" if severity == "HIGH" else "🟢" if severity == "MEDIUM" else "⚪"
+            markdown_parts.append(f"### {emoji} {anomaly_type} ({severity})\n\n{description}\n\n")
+            
+            if possible_causes:
+                markdown_parts.append("#### Возможные первопричины:\n\n")
+                for i, cause in enumerate(possible_causes, 1):
+                    markdown_parts.append(f"{i}. {cause}\n")
+                markdown_parts.append("\n")
+    
+    # Рекомендации по устранению
+    recommendations = rca_result.get("recommendations", [])
+    if recommendations:
+        markdown_parts.append("## Рекомендации по устранению\n\n")
+        for rec in recommendations:
+            priority = rec.get("priority", "MEDIUM")
+            text = rec.get("text", "")
+            actions = rec.get("actions", [])
+            
+            emoji = "🔴" if priority == "CRITICAL" else "🟡" if priority == "HIGH" else "🟢" if priority == "MEDIUM" else "⚪"
+            markdown_parts.append(f"### {emoji} Рекомендация (Приоритет: {priority})\n\n{text}\n\n")
+            
+            if actions:
+                markdown_parts.append("**Конкретные действия:**\n\n")
+                for action in actions:
+                    markdown_parts.append(f"- {action}\n")
+                markdown_parts.append("\n")
+    
+    # Общие рекомендации
+    general = rca_result.get("general_recommendations", [])
+    if general:
+        markdown_parts.append("## Общие рекомендации\n\n")
+        for rec in general:
+            markdown_parts.append(f"- {rec}\n")
+    
+    return "".join(markdown_parts)