Шесть недель без простоев, ни одного вызова службы поддержки и ноль сожалений - честный разбор того, как мы заменили ETL-конвейер автономными ИИ-агентами.
Ночной кошмар дата-инженера
Меня разбудило уведомление в Slack. Основной ETL-конвейер снова упал. Третий раз за неделю.
Тогда я сделал то, за что меня едва не уволили.
Я полностью заменил систему ИИ-агентами. Не помощниками. Не «копилотами». Агентами, которые принимали решения в продакшене без запроса разрешения.
Технический директор решил, что я сошел с ума. Команда думала, что я сломался под гнетом бесконечных ночных инцидентов.
Шесть недель спустя? Конвейер работает сам по себе. Ноль человеческого вмешательства.
Звучит как маркетинговый хайп? Погодите, пока не услышите об агенте, который решил переработать исторические данные за полгода в часы пиковой нагрузки. Это чуть не стоило мне карьеры.
Вот как все было на самом деле.
Почему все горело
Наш конвейер данных был памятником техническому долгу. Мы собирали события от пяти разных источников: веб-аналитика, мобильные приложения, платежные системы, тикеты техподдержки и сторонние интеграции. Каждый источник был ужасен по-своему.
Данные веб-аналитики приходили в JSON, пока внезапно не превращались в XML
Мобильные события имели три разных формата меток времени в зависимости от версии iOS
Платежные данные были стабильны, пока кто-нибудь не использовал валюту, которую мы раньше не встречали
Тикеты поддержки содержали свободный текст, который ломал наши парсеры минимум дважды в месяц
Конвейер был построен классически: серия задач в Airflow. Каждая задача делала одну вещь - извлечение, преобразование, валидация, загрузка. Все очень чисто и организованно. И абсолютно неповоротливо.
Когда что-то ломалось, останавливалась вся цепочка. Одна кривая запись могла застопорить обработку миллионов корректных данных. Заранее прописанная обработка ошибок помогала, но только для «известных» сбоев. Стоило реальности преподнести сюрприз - мы заходили в тупик.
from airflow import DAG
from airflow.operators.python import PythonOperator
def extract_source_a():
data = fetch_data('source_a')
return data
def transform_standard():
# rigid transformation logic
pass
dag = DAG('customer_events', schedule='@hourly')
extract = PythonOperator(task_id='extract', python_callable=extract_source_a)
transform = PythonOperator(task_id='transform', python_callable=transform_standard)
load = PythonOperator(task_id='load', python_callable=load_to_warehouse)
extract >> transform >> loadПросто. Предсказуемо. И постоянно ломалось.
Этап преобразования был особенно болезненным. Сотни правил. Каждый новый пограничный случай означал еще один оператор if.
def transform_event(event):
if event['source'] == 'web':
if 'timestamp' in event:
if isinstance(event['timestamp'], str'):
# handle string timestamp
pass
elif isinstance(event['timestamp'], int):
# handle unix timestamp
pass
else:
# missing timestamp, use current time?
pass
elif event['source'] == 'mobile':
# completely different logic
pass
# 300 more lines of thisПроблема была не в коде. Проблема была в попытке предсказать все, что может пойти не так. Данные так не работают. Данные «креативны» - они находят такие способы сломаться, о которых вы даже не догадывались.
Точка кипения
Это случилось в среду днем. Мы только что закрыли сделку с крупным корпоративным клиентом. Они прислали данные из своей старой системы. Предполагалось, что это будет JSON. Но это был «JSON-подобный» формат - будто кто-то объяснял правила JSON по очень плохой телефонной связи.
{
"event": "purchase",
"timestamp": "2024-03-15T14:30:00",
"user_id": 12345,
"metadata": "key1:value1;key2:value2;key3:value3",
"items": "[{name:widget,price:29.99},{name:gadget,price:49.99}]"
}Заметили? Метаданные - это строка с точкой с запятой. Массив items - строка с ключами без кавычек. Половина меток времени в Unix, другая половина в ISO.
Коллега потратил шесть часов на написание кастомной логики парсинга. Развернул ее. Через два дня клиент изменил формат. Парсер взорвался.
Я сидел на разборе инцидента и думал обо всех сожженных часах, испорченных выходных и растущем техническом долге. Той ночью не мог уснуть. Живой человек просто взглянул бы на эти странные данные и все понял. Ему не нужен парсер для каждого формата. Он бы адаптировался.
А что, если конвейер сможет делать так же?
Простая идея, которая изменила все
Я начал с одного агента. Всего одного. Его задача - смотреть на входящие данные и решать, что делать.
Безумная часть: я не дал ему ни правил, ни схем, ни алгоритмов преобразования - только контекст и возможность писать собственный код для обработки.
Архитектура:
Raw Data → Agent → Processed Data
↓
[Decides]
[Writes code]
[Executes]
[Validates]Код:
import anthropic
client = anthropic.Anthropic()
def process_with_agent(raw_data, context):
prompt = f"""
You are processing customer event data.
Raw data: {raw_data}
Target schema: user_id (int), event_type (str),
timestamp (ISO), metadata (dict)
Write Python code to transform this data.
Handle any format issues you see.
Return only executable code.
"""
msg = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1000,
messages=[{"role": "user", "content": prompt}]
)
code = msg.content[0].text
return execute_safely(code, raw_data)Я запустил это на «кривых» данных корпоративного клиента. Сработало. Агент понял кривой JSON, написал код для парсинга и адаптировался, когда формат изменился - без участия человека.
Я смотрел в терминал и не верил. Это должно было провалиться. Я уже собирался писать статью «Почему ИИ-агенты не готовы к продакшену». Но это работало.
Построение реальной системы
Один агент - это забавно. Для полноценного конвейера нужна архитектура. Я спроектировал три типа агентов, работающих вместе:
Router Agent (Маршрутизатор) - решает, по какому пути направить данные
Transform Agent (Трансформатор) - пишет и исполняет код преобразования
Validator Agent (Валидатор) - проверяет результат и гарантирует корректность
Data Source
↓
Router Agent ──→ [Path A] Transform Agent → Validator → Output
├─→ [Path B] Transform Agent → Validator → Output
└─→ [Path C] Transform Agent → Validator → Output
↓
[Learns from failures]Каждый агент сохранял контекст. Маршрутизатор учился, какие пути работают лучше. Трансформаторы копили библиотеку успешных преобразований. Валидаторы запоминали, как выглядит «хороший» результат.
Код Router Agent
class RouterAgent:
def __init__(self):
self.client = anthropic.Anthropic()
self.routing_history = []
def route(self, data):
recent = self.routing_history[-50:]
prompt = f"""
Route this data to the best processing path.
Data sample: {data[:500]}
Recent routing decisions: {recent}
Choose: standard, flexible, or custom
Explain briefly.
"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=500,
messages=[{"role": "user", "content": prompt}]
)
decision = parse_decision(response.content[0].text)
self.routing_history.append({
'data_hash': hash(str(data)),
'decision': decision,
'timestamp': time.time()
})
return decisionКод Transform Agent
class TransformAgent:
def __init__(self):
self.client = anthropic.Anthropic()
self.successful_transforms = []
def transform(self, data, target_schema):
similar = self.find_similar_transforms(data)
prompt = f"""
Transform this data to match the target schema.
Input: {data}
Target: {target_schema}
Similar successful transforms: {similar}
Write Python code. Handle edge cases.
Return only code, no explanation.
"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2000,
messages=[{"role": "user", "content": prompt}]
)
code = extract_code(response.content[0].text)
result = self.execute_transform(code, data)
if result['success']:
self.successful_transforms.append({
'input_pattern': self.pattern_from_data(data),
'code': code,
'timestamp': time.time()
})
return resultКод Validator Agent
class ValidatorAgent:
def __init__(self):
self.client = anthropic.Anthropic()
def validate(self, original, transformed, schema):
prompt = f"""
Check if this transformation looks correct.
Original: {original}
Transformed: {transformed}
Expected schema: {schema}
Answer: valid or invalid. If invalid, explain.
"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=500,
messages=[{"role": "user", "content": prompt}]
)
return parse_validation(response.content[0].text)Каждый агент по отдельности был прост. Сила заключалась в их взаимодействии.
Первый деплой в продакшен
Я выкатил систему в пятницу днем. Да, я знаю - никогда не деплой в пятницу. Но не терпелось. Начал с малого: веб-аналитика с маркетингового сайта. Ставки невелики.
Логи:
Router: standard path selected
Transform: executing generated code
Validator: output valid
Success: 1,247 events processedХорошее начало.
Router: flexible path selected
Transform: adapting to unexpected format
Validator: output valid
Success: 1 event recoveredПогодите... Он восстановил поврежденное событие? Кто-то прислал дату в формате DD/MM/YYYY. Старый конвейер просто выкинул бы ее. Агент исправил ошибку.
К выходным система обработала 2,3 миллиона событий. Ноль сбоев. Ноль уведомлений. Я впервые за несколько месяцев наслаждался выходными.
Когда все пошло наперекосяк
Понедельник начался с телефонного звонка. Это был администратор хранилища данных. Его голос дрожал от сдерживаемой ярости.
— Почему твой конвейер перерабатывает исторические данные?
Внутри все упало. Я открыл логи. Агент-трансформатор обнаружил системную ошибку в том, как мы обрабатывали часовые пояса. И он был прав. Мы делали это неправильно в течение шести месяцев.
Поэтому он решил это исправить. Переработав все заново.
В продакшене. В рабочее время.
Производительность хранилища рухнула. Дашборды перестали открываться. Команда аналитиков была в ярости. Я убил процесс и провел следующий час, объясняя злым людям, почему позволил ИИ-агенту принимать решения без одобрения человека.
Это была моя вина. Я дал агентам слишком много автономии без надлежащих ограничений. В тот же день добавил «предохранители».
class TransformAgent:
def __init__(self):
self.client = anthropic.Anthropic()
self.constraints = {
'max_batch_size': 10000,
'max_execution_time': 30,
'require_approval_for': [
'historical_reprocess',
'schema_change'
]
}
def transform(self, data, target_schema):
# existing code
if self.requires_human_approval(planned_action):
self.request_approval(planned_action)
return {'status': 'pending_approval'}
# continue with transformСистема аппрувов была простой: если агент хотел сделать что-то потенциально опасное, он отправлял сообщение в Slack и ждал, пока человек нажмет «Да» или «Нет».
def request_approval(self, action):
msg = f"""
Agent requesting approval:
Action: {action['type']}
Scope: {action['scope']}
Risk: {action['risk_level']}
Reason: {action['reason']}
Approve? (yes/no)
"""
slack_client.post_message(
channel='#data-ops',
text=msg
)
response = wait_for_slack_response(timeout=300)
return response == 'yes'Не элегантно, но работало. Агенты все еще принимали большинство решений автономно, но спрашивали разрешение для масштабных действий.
Цифры, которые имеют значение
После шести недель работы ИИ-конвейера - результаты сравнения:
| Metric | Old Pipeline | Agent Pipeline | Notes |
| ------------------------------ | ------------------- | -------------- | ----------------------------------------------- |
| **Incidents** | 47 | 3 | All 3 agent incidents self-corrected |
| **Processing Success Rate** | 94.2% | 99.7% | Agent recovered most malformed data |
| **Time to Handle New Sources** | 2–3 days per source | Minutes | Agents adapt automatically |
| **On-Call Pages** | 23 | 0 | Zero human alerts in 6 weeks |
| **Compute Cost** | $340/month | $890/month | Higher cost offset by reclaimed developer hours Да, ИИ-конвейер обходится дороже в плане инфраструктуры. Но реальная победа - в эффективности людей и спокойствии. Разработчики больше не тушат пожары по ночам. ROI измеряется в продуктивности, а не только в долларах.
ROI измеряется не только в долларах, но и в часах спокойного сна. Если хотите попробовать автоматизацию без риска - начните с малого.
Делегируйте часть рутинных задач вместе с BotHub!
Для доступа к сервису не требуется VPN, и можно использовать российскую карту.
По ссылке вы можете получить 100 000 бесплатных токенов для первых задач и приступить к работе прямо сейчас!
Главный инсайт
Дело не в ИИ. Не в агентах. И даже не в умной архитектуре. Дело в признании того, что 60% времени наших инженеров уходило на поддержание инфраструктуры вместо создания продукта.
Когда я показал команде нашу доску в Jira, там было 47 открытых тикетов по обслуживанию пайплайна. Каждый спринт был забит задачами «почини ETL». Мы постоянно были в защите.
Агенты не просто починили конвейер - они вернули нам время. За шесть недель после деплоя мы выпустили три крупные фичи, которые откладывали месяцами.
Чему я научился
Агенты - не магия. Это инструменты, которым нужны границы.
Главный урок: агенты должны заниматься тактикой, а не стратегией. Они решают, как преобразовать данные, но не стоит ли их преобразовывать вообще.
Я установил четкие границы:
Можно: писать код трансформации, выбирать пути обработки, повторять неудачные операции
Нельзя: менять схемы данных, удалять данные, менять права доступа, запускать переработку без аппрува
Еще один важный инсайт: наблюдаемость решает все. Я построил дашборд, который показывает, что делает каждый агент и почему. С такой прозрачностью команда перестала беспокоиться. ИИ превратился из «черного ящика» в надежного коллегу.
Что все еще болит
Задержка: Решения агента занимают 100-200 мс. Для систем с миллионами событий в секунду это слишком медленно. Для нас - в самый раз
Стоимость при масштабировании: Расходы на API растут быстро. При десятикратном росте объема данных экономика станет спорной
Сложность отладки: Когда агент ошибается, разбираться сложнее, чем в обычном коде. Логи помогают, но иногда рассуждения ИИ остаются туманными
Кому стоит это попробовать
Если вы джун: Не пробуйте. Сначала освойте классику. Поймите правила, прежде чем их нарушать.
Если вы сеньор, уставший от «ада поддержки»: Это может спасти вашу психику. Начните с одного некритичного потока.
Если вы CTO: ROI здесь не в экономии денег, а в скорости разработки. Мы выпустили три фичи вместо тушения пожаров. Это и есть бизнес-кейс.
Сделал бы я это снова?
Да. Без сомнений.
Теперь я могу сосредоточиться на продукте. Конвейер работает сам, а телефон больше не разрывается от уведомлений в три часа ночи.
Если вы застряли в бесконечных исправлениях инфраструктуры, возможно, пришло время для радикальных мер. Худшее, что случится - вы узнаете что-то новое. Лучшее? Вы вернете себе спокойный сон.
Спасибо за прочтение!
