Как отлаживать AI-агентов не сжигая бюджет: replay debugger для LangChain

Пятница, вечер. Мой агент на LangChain — цепочка из девяти шагов: парсинг документа, извлечение сущностей, классификация, обогащение из базы знаний, генерация саммари, проверка качества, форматирование, пост-обработка, отправка. Шаг 7 падает с ошибкой валидации. Я фиксю промпт, запускаю заново. Все девять шагов — с нуля. Три минуты ожидания. $1.80 на токенах. Шаг 7 опять падает — другая ошибка. Фикшу. Запускаю. Ещё $1.80. Ещё три минуты.

За вечер отладки я потратил $14 и 40 минут чистого ожидания на шаги 1-6, которые работали идеально с первого раза.

Это фундаментальная проблема отладки AI-агентов: каждый перезапуск — полный прогон с начала. Нет точек сохранения. Нет replay. Каждый LLM-вызов стоит денег и времени, и при отладке одного шага вы платите за все предыдущие — снова и снова.

В обычной разработке это как если бы при каждом баге в функции приходилось перезапускать всю программу с компиляции. Абсурд — но именно так сегодня работает отладка агентов.

Идея: Flight Recorder

Концепция простая. Бортовой самописец (flight recorder) записывает входы и выходы каждого шага агентного workflow. При перезапуске — успешные шаги не выполняются заново, а воспроизводятся из кэша. Реально вызывается только сбойный шаг и всё, что после него.

Первый запуск (полный):
  Шаг 1 ✅ → записать результат
  Шаг 2 ✅ → записать результат
  ...
  Шаг 6 ✅ → записать результат
  Шаг 7 ❌ → ошибка

Повторный запуск (replay):
  Шаг 1 ⏩ → из кэша (0 токенов, 0 мс)
  Шаг 2 ⏩ → из кэша
  ...
  Шаг 6 ⏩ → из кэша
  Шаг 7 🔄 → реальный вызов (исправленный промпт)
  Шаг 8 🔄 → реальный вызов
  Шаг 9 🔄 → реальный вызов

Второй запуск стоит не $1.80, а $0.50 — только шаги 7-9. Время — не три минуты, а 40 секунд. При десяти итерациях отладки — $5 вместо $18, 7 минут вместо 30.

Реализация: три компонента

Компонент 1: Step Recorder

import hashlib
import json
import os
from pathlib import Path

class FlightRecorder:
    def __init__(self, session_id: str, cache_dir: str = ".flight_cache"):
        self.session_id = session_id
        self.cache_dir = Path(cache_dir) / session_id
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        self.replay_from = None  # Шаг, с которого перезапускать
    
    def _cache_key(self, step_name: str, input_data: dict) -> str:
        """Ключ кэша = хэш (имя шага + вход + промпт)."""
        content = json.dumps({
            "step": step_name,
            "input": input_data,
        }, sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(content.encode()).hexdigest()[:16]
    
    def _cache_path(self, step_name: str, cache_key: str) -> Path:
        return self.cache_dir / f"{step_name}_{cache_key}.json"
    
    def get_cached(self, step_name: str, input_data: dict):
        """Вернуть результат из кэша или None."""
        key = self._cache_key(step_name, input_data)
        path = self._cache_path(step_name, key)
        
        if path.exists():
            with open(path) as f:
                record = json.load(f)
            return record["output"]
        return None
    
    def save(self, step_name: str, input_data: dict, output_data: dict):
        """Записать результат шага."""
        key = self._cache_key(step_name, input_data)
        path = self._cache_path(step_name, key)
        
        record = {
            "step": step_name,
            "input": input_data,
            "output": output_data,
            "timestamp": datetime.now().isoformat(),
        }
        
        with open(path, "w") as f:
            json.dump(record, f, ensure_ascii=False, indent=2)

Компонент 2: Smart Replay Logic

class ReplayController:
    def __init__(self, recorder: FlightRecorder):
        self.recorder = recorder
        self.invalidated_steps = set()
        self.force_rerun_from = None
    
    def invalidate(self, step_name: str):
        self.invalidated_steps.add(step_name)
    
    def invalidate_from(self, step_name: str):
        self.force_rerun_from = step_name
    
    def should_replay(self, step_name: str, input_data: dict, 
                      step_order: list) -> bool:
        if step_name in self.invalidated_steps:
            return False
        
        if self.force_rerun_from:
            idx_force = step_order.index(self.force_rerun_from)
            idx_current = step_order.index(step_name)
            if idx_current >= idx_force:
                return False
        
        cached = self.recorder.get_cached(step_name, input_data)
        return cached is not None

Компонент 3: интеграция с LangChain

from langchain_core.runnables import RunnableConfig, RunnableLambda

def recorded_step(step_name: str, fn, recorder: FlightRecorder, 
                  controller: ReplayController, step_order: list):
    def wrapper(input_data, config: RunnableConfig = None):
        if controller.should_replay(step_name, input_data, step_order):
            cached = recorder.get_cached(step_name, input_data)
            print(f"  ⏩ {step_name}: replay from cache")
            return cached
        
        print(f"  🔄 {step_name}: executing...")
        try:
            result = fn(input_data)
            recorder.save(step_name, input_data, result)
            print(f"  ✅ {step_name}: done, saved to cache")
            return result
        except Exception as e:
            print(f"  ❌ {step_name}: failed — {e}")
            raise
    
    return RunnableLambda(wrapper).with_config({"run_name": step_name})

Тонкости, которые ломают наивную реализацию

Недетерминированность LLM: в режиме debug используйте temperature: 0 — результат стабильный, кэш валидный.

Побочные эффекты: разделяйте шаги на pure (трансформация данных) и impure (побочные эффекты). Pure реплеятся, impure всегда выполняются.

Изменение промпта: включайте хэш промпта в ключ кэша — изменили промпт, хэш другой, кэш невалиден, шаг перезапускается автоматически.

Разветвлённые графы (LangGraph): кэшируйте по (node_name, input_hash), а не по порядковому номеру.

Экономика

Агент из 9 шагов, 4 LLM-вызова (GPT-4o):

Без Flight Recorder: 10 итераций = $18.00, ~30 минут
С Flight Recorder: 10 итераций = $6.30, ~12 минут

Экономия 65% бюджета и 60% времени на одну сессию отладки.

Итог

Отладка AI-агентов дорогая не потому, что LLM-вызовы дорогие, а потому что при каждой итерации вы оплачиваете шаги, которые уже работали. Flight Recorder решает именно эту проблему: записал → воспроизвёл → заплатил только за то, что реально нужно было перезапустить.

Три файла Python, один вечер на настройку. После этого каждый invalidate_from("broken_step") экономит 60-90% бюджета на итерацию.