Как создать самовосстанавливающийся ETL-пайплайн на Python с помощью LLM: полное руководство с кодом

Почему ETL-пайплайны ломаются и как заставить их чинить себя самим

Представьте: 3 часа ночи, ваш ETL-пайплайн падает из-за того, что в CSV-файле появилась новая колонка с кириллицей в названии. Или потому что даты теперь в формате DD/MM/YYYY вместо YYYY-MM-DD. Или потому что в числовом поле внезапно оказалась строка "N/A".

Вы просыпаетесь от алерта, лезете в логи, ищете проблему, пишете хотфикс, деплоите. На это уходит час. А завтра история повторится с другой ошибкой.

Знакомо? Тогда забудьте. Сегодня мы построим пайплайн, который сам находит ошибки, сам их диагностирует и сам исправляет. Без вашего участия.

Архитектура Try-Heal-Retry Loop: как это работает

Классический ETL: Extract → Transform → Load. Если на любом этапе ошибка — всё падает.

Наш подход: Try → Heal → Retry → Loop.

Ключевая идея: вместо того чтобы писать код на все возможные ошибки (невозможно), мы учим систему понимать ошибки и генерировать исправления на лету.

1 Устанавливаем зависимости (актуально на январь 2026)

Сначала подготовим окружение. Мы будем использовать:

• Python 3.11+ (3.12 уже стабильна, но проверьте совместимость библиотек)
• OpenAI API для GPT-4o (или локальную модель через Ollama)
• Pandas 2.2+ с поддержкой новых типов данных
• Pydantic 2.6+ для валидации

# requirements.txt
pandas>=2.2.0
openai>=1.12.0
pydantic>=2.6.0
python-dotenv>=1.0.0
langchain>=0.2.0  # опционально, для более сложных агентов

2 Создаем ядро: класс SelfHealingETL

Этот класс будет основой нашего пайплайна. Он умеет:

1. Ловить любые исключения
2. Анализировать их с помощью LLM
3. Генерировать и применять исправления
4. Повторять попытку

import pandas as pd
import json
import traceback
from typing import Any, Callable, Dict, Optional
from pydantic import BaseModel, Field
from openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()

class HealingInstruction(BaseModel):
    """Инструкция по исправлению ошибки от LLM"""
    error_type: str = Field(description="Тип ошибки (Pandas, IO, Validation и т.д.)")
    root_cause: str = Field(description="Корневая причина ошибки")
    fix_code: str = Field(description="Код на Python для исправления")
    confidence: float = Field(description="Уверенность в исправлении (0-1)", ge=0, le=1)
    risks: list[str] = Field(description="Потенциальные риски этого исправления")

class SelfHealingETL:
    def __init__(self, llm_client=None, max_retries=3):
        self.llm_client = llm_client or OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.max_retries = max_retries
        self.healing_history = []
        
    def try_heal_retry(self, operation: Callable, operation_name: str, **kwargs) -> Any:
        """Основной цикл Try-Heal-Retry"""
        
        for attempt in range(self.max_retries):
            try:
                result = operation(**kwargs)
                print(f"✅ {operation_name} успешно выполнена")
                return result
                
            except Exception as e:
                print(f"⚠️ Попытка {attempt + 1}/{self.max_retries}: {operation_name} упала")
                print(f"Ошибка: {str(e)}")
                
                if attempt == self.max_retries - 1:
                    print("❌ Достигнут лимит попыток")
                    raise
                    
                # Фаза Heal: анализируем и исправляем
                healing_plan = self._analyze_error(e, traceback.format_exc(), operation_name, kwargs)
                
                if healing_plan.confidence > 0.7:  # Порог уверности
                    print(f"🔧 Применяю исправление: {healing_plan.root_cause}")
                    self._apply_healing(healing_plan, kwargs)
                    self.healing_history.append({
                        'attempt': attempt,
                        'operation': operation_name,
                        'error': str(e),
                        'healing': healing_plan.dict()
                    })
                else:
                    print(f"⚠️ LLM не уверена в исправлении ({healing_plan.confidence:.2f})")
                    raise
    
    def _analyze_error(self, error: Exception, traceback_str: str, operation_name: str, context: Dict) -> HealingInstruction:
        """Анализ ошибки с помощью LLM"""
        
        prompt = f"""
Ты — эксперт по ETL и обработке данных. Произошла ошибка в пайплайне.

Операция: {operation_name}
Контекст: {json.dumps(context, indent=2)}
Ошибка: {str(error)}
Traceback:
{traceback_str}

Проанализируй ошибку и предложи исправление в виде кода Python.
Верни ответ в JSON формате:
{{
    "error_type": "тип ошибки",
    "root_cause": "корневая причина",
    "fix_code": "код для исправления",
    "confidence": 0.95,
    "risks": ["риск1", "риск2"]
}}

Критические правила:
1. Код fix_code должен быть ВЫПОЛНИМЫМ Python кодом
2. Код должен модифицировать переменную `context`
3. Не предлагай изменить логику операции, только подготовку данных
4. Будь конкретен: какие колонки, какие типы данных, какие значения
"""
        
        try:
            # Используем GPT-4o как самую продвинутую модель на январь 2026
            response = self.llm_client.chat.completions.create(
                model="gpt-4o",
                messages=[
                    {"role": "system", "content": "Ты эксперт по ETL и обработке данных."},
                    {"role": "user", "content": prompt}
                ],
                response_format={"type": "json_object"},
                temperature=0.1  # Низкая температура для детерминированности
            )
            
            result = json.loads(response.choices[0].message.content)
            return HealingInstruction(**result)
            
        except Exception as llm_error:
            print(f"❌ Ошибка при анализе LLM: {llm_error}")
            # Возвращаем запасной вариант
            return HealingInstruction(
                error_type="Unknown",
                root_cause="LLM analysis failed",
                fix_code="# No fix available",
                confidence=0.0,
                risks=["Cannot analyze error"]
            )
    
    def _apply_healing(self, healing: HealingInstruction, context: Dict):
        """Применяем исправление, сгенерированное LLM"""
        try:
            # Создаем безопасное окружение для выполнения кода
            safe_globals = {
                'context': context,
                'pd': pd,
                'json': json
            }
            
            # Выполняем код исправления
            exec(healing.fix_code, safe_globals)
            
            # Обновляем контекст
            context.update(safe_globals.get('context', {}))
            
            print(f"✅ Исправление применено: {healing.root_cause}")
            
        except Exception as e:
            print(f"❌ Не удалось применить исправление: {e}")
            raise

Реальные кейсы: как это работает на практике

Кейс 1: CSV с проблемными данными

Представьте, что вы загружаете CSV, а там:

• Даты в формате "01/15/2026" вместо "2026-01-15"
• Числа с разделителем тысяч: "1,234.56"
• Пропущенные значения как "NULL", "N/A", "-"
• Новая колонка с непредвиденным названием

def load_problematic_csv(filepath):
    """Наивная загрузка CSV — упадет на первой же проблеме"""
    df = pd.read_csv(filepath)
    df['date'] = pd.to_datetime(df['date'])  # Упадет если формат другой
    df['amount'] = df['amount'].astype(float)  # Упадет на "1,234.56"
    return df

# Создаем самовосстанавливающуюся версию
healing_etl = SelfHealingETL()

# Оборачиваем нашу функцию в try-heal-retry
result = healing_etl.try_heal_retry(
    operation=load_problematic_csv,
    operation_name="load_csv",
    filepath="problematic_data.csv"
)

Что произойдет:

1. Первая попытка упадет на pd.to_datetime()
2. LLM проанализирует ошибку: "Cannot parse date '01/15/2026'"
3. Сгенерирует fix_code: context['df']['date'] = pd.to_datetime(context['df']['date'], format='%m/%d/%Y')
4. Применит исправление и повторит
5. Если упадет на amount — повторит цикл

Кейс 2: API с меняющейся структурой

Вы тянете данные из внешнего API. Вчера ответ был:

{
  "data": {
    "users": [
      {"id": 1, "name": "John", "age": 30}
    ]
  }
}

Сегодня API обновился и возвращает:

{
  "response": {
    "items": [
      {"user_id": 1, "full_name": "John Doe", "user_age": 30}
    ]
  }
}

Обычный пайплайн сломается. Наш — адаптируется.

def extract_from_api(api_response):
    """Ожидаем старую структуру"""
    users = api_response['data']['users']
    df = pd.DataFrame(users)
    df.rename(columns={'id': 'user_id'}, inplace=True)
    return df

# Когда API изменится, LLM предложит:
# fix_code = """
# # API изменил структуру
# if 'response' in context['api_response']:
#     users_data = context['api_response']['response']['items']
#     # Приводим к ожидаемому формату
#     formatted_users = []
#     for item in users_data:
#         formatted_users.append({
#             'id': item['user_id'],
#             'name': item['full_name'],
#             'age': item['user_age']
#         })
#     context['api_response'] = {'data': {'users': formatted_users}}
# """

Продвинутые техники: выходим за рамки базового сценария

Техника 1: Контекстное обучение

LLM работает лучше, когда понимает ваш домен. Добавьте контекст в промпт:

class DomainAwareHealer(SelfHealingETL):
    def __init__(self, domain_knowledge=None, **kwargs):
        super().__init__(**kwargs)
        self.domain_knowledge = domain_knowledge or ""
    
    def _analyze_error(self, error: Exception, traceback_str: str, operation_name: str, context: Dict) -> HealingInstruction:
        prompt = f"""
{self.domain_knowledge}

Ошибка в {operation_name}:
{str(error)}

Доменные правила:
1. Даты всегда в формате YYYY-MM-DD
2. Валюта: USD, разделитель тысяч - запятая
3. Обязательные поля: id, timestamp, value
"""
        # ... остальной код анализа

Техника 2: Валидация исправлений

Прежде чем применять исправление, проверьте его на безопасность:

def _validate_healing(self, healing: HealingInstruction) -> bool:
    """Проверяем, что исправление безопасно"""
    
    red_flags = [
        'os.system', 'subprocess', 'eval', 'exec',  # Опасные операции
        '__',  # Магические методы
        'delete', 'drop', 'remove'  # Деструктивные операции
    ]
    
    for flag in red_flags:
        if flag in healing.fix_code.lower():
            print(f"🚨 Обнаружена опасная операция: {flag}")
            return False
    
    # Проверяем, что код хотя бы синтаксически корректен
    try:
        ast.parse(healing.fix_code)
        return True
    except SyntaxError:
        return False

Техника 3: Коллективное обучение

Сохраняйте историю исправлений и используйте ее для улучшения:

class LearningHealer(SelfHealingETL):
    def __init__(self, knowledge_base_path="healing_kb.json", **kwargs):
        super().__init__(**kwargs)
        self.knowledge_base_path = knowledge_base_path
        self.load_knowledge_base()
    
    def load_knowledge_base(self):
        """Загружаем прошлые успешные исправления"""
        try:
            with open(self.knowledge_base_path, 'r') as f:
                self.knowledge_base = json.load(f)
        except FileNotFoundError:
            self.knowledge_base = []
    
    def _analyze_error(self, error: Exception, traceback_str: str, operation_name: str, context: Dict) -> HealingInstruction:
        # Сначала ищем в базе знаний
        for kb_entry in self.knowledge_base:
            if kb_entry['error_pattern'] in str(error):
                print(f"📚 Найдено в базе знаний: {kb_entry['description']}")
                return HealingInstruction(**kb_entry['healing'])
        
        # Если не нашли — используем LLM
        healing = super()._analyze_error(error, traceback_str, operation_name, context)
        
        # Сохраняем успешное исправление
        if healing.confidence > 0.8:
            self._save_to_knowledge_base(error, healing, operation_name)
        
        return healing

Интеграция с существующими пайплайнами

Не нужно переписывать весь ETL с нуля. Оберните критические части:

# Ваш существующий ETL
class LegacyETL:
    def extract(self):
        # ... ваш код
    
    def transform(self, data):
        # ... ваш код
    
    def load(self, transformed_data):
        # ... ваш код

# Модернизируем с самовосстановлением
class ModernizedETL(LegacyETL):
    def __init__(self):
        super().__init__()
        self.healer = SelfHealingETL()
    
    def safe_transform(self, data):
        """Обертка с самовосстановлением"""
        return self.healer.try_heal_retry(
            operation=self.transform,
            operation_name="transform",
            data=data
        )

Ограничения и подводные камни

• Критические данные: Не позволяйте ИИ "исправлять" финансовые транзакции или медицинские записи без человеческой проверки
• Бесконечные циклы: Установите жесткий лимит попыток (3-5)
• Стоимость: Каждый вызов LLM стоит денег. Для массовой обработки используйте локальные модели
• Детерминизм: LLM могут предлагать разные исправления для одной ошибки. Это проблема для воспроизводимости
• Безопасность: Всегда валидируйте сгенерированный код перед выполнением

Что дальше? Эволюция подхода

Try-Heal-Retry Loop — только начало. Следующий шаг — полностью автономные ИИ-агенты для ETL, как в нашем кейсе "Я заменил ETL-конвейер на ИИ-агентов".

Представьте агентов, которые:

1. Самостоятельно мониторят источники данных
2. Обнаруживают дрейф схемы (schema drift)
3. Тестируют гипотезы исправлений на снэпшотах данных
4. Ведет журнал принятых решений для аудита

Такой подход уже работает в продакшене. Компании экономят сотни человеко-часов на поддержке ETL.

Стартовый чеклист

Прежде чем внедрять в продакшен:

1. Начните с не-критических пайплайнов (отчетность, аналитика)
2. Установите жесткие лимиты на стоимость LLM-вызовов
3. Реализуйте обязательную валидацию сгенерированного кода
4. Ведите детальный лог всех исправлений
5. Периодически ревьюируйте решения, принятые ИИ
6. Имейте ручной "аварийный выключатель"

Самый важный совет: не пытайтесь автоматизировать всё сразу. Начните с одной, самой болезненной ошибки. Добейтесь, чтобы система надежно с ней справлялась. Затем добавляйте следующую.

Через месяц у вас будет пайплайн, который сам чинит 80% типовых ошибок. Вы будете спать по ночам. Ваши коллеги будут спрашивать, как вы это сделали. Покажете им эту статью.