AI-агент упал на 7-м шаге: как продолжить workflow без повторной оплаты и дублей в CRM

Длинный workflow ИИ-агента редко ломается в удобном месте. Модель уже потратила токены, поиск собрал данные, CRM получила запись — и тут процесс упал. Если просто запустить всё заново, вы заплатите повторно и рискуете продублировать побочные эффекты. Разберём checkpointing, idempotency и outbox на практическом примере.

У короткого LLM-запроса есть понятная граница: отправили prompt, получили ответ или ошибку. У агента граница размазывается на десятки шагов: получить данные, вызвать модель, сходить в базу, создать объект в CRM, дождаться внешнего API, отправить уведомление.

Представим workflow:

Если процесс упал перед седьмым шагом, повтор первых шести — это не восстановление. Это повторные токены, дубли в CRM и две задачи менеджеру. Поэтому production-агент должен уметь продолжать выполнение с последней безопасной точки.

Durable state. Состояние нельзя держать только в памяти процесса. После каждого значимого шага сохраняем вход, результат, статус и версию workflow в постоянное хранилище.

Idempotency. Повторный вызов шага не должен создавать второй побочный эффект. Для операций записи используем стабильный idempotency key, связанный с workflow и именем шага.

Atomic transition. Нельзя отдельно записать «шаг завершён» и отдельно поставить следующий шаг в очередь. Если процесс упадёт между этими операциями, workflow зависнет. Эту щель закрывает transactional outbox.

Для начала достаточно трёх таблиц:

CREATE TABLE workflows ( id TEXT PRIMARY KEY, type TEXT NOT NULL, version INTEGER NOT NULL, status TEXT NOT NULL, input_json TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE step_runs ( workflow_id TEXT NOT NULL, step_name TEXT NOT NULL, status TEXT NOT NULL, attempt INTEGER NOT NULL DEFAULT 0, input_hash TEXT NOT NULL, output_json TEXT, error_json TEXT, started_at TEXT, finished_at TEXT, PRIMARY KEY (workflow_id, step_name) ); CREATE TABLE outbox ( id INTEGER PRIMARY KEY AUTOINCREMENT, workflow_id TEXT NOT NULL, event_type TEXT NOT NULL, payload_json TEXT NOT NULL, published_at TEXT );

Уникальный ключ (workflow_id, step_name) не позволяет случайно создать две записи одного шага. Версия workflow нужна, чтобы новая схема процесса не пыталась продолжить старый запуск несовместимым кодом.

Ниже упрощённый пример на SQLite. В реальном сервисе ту же схему можно реализовать на PostgreSQL, а очередь — на Kafka, RabbitMQ или SQS.

import hashlib import json import sqlite3 from collections.abc import Callable def stable_hash(value: dict) -> str: raw = json.dumps(value, sort_keys=True).encode() return hashlib.sha256(raw).hexdigest() def run_step( db: sqlite3.Connection, workflow_id: str, step_name: str, step_input: dict, handler: Callable[[dict, str], dict], ) -> dict: input_hash = stable_hash(step_input) row = db.execute( """ SELECT status, input_hash, output_json FROM step_runs WHERE workflow_id = ? AND step_name = ? """, (workflow_id, step_name), ).fetchone() if row and row[0] == "completed": if row[1] != input_hash: raise RuntimeError("Step input changed during resume") return json.loads(row[2]) with db: db.execute( """ INSERT INTO step_runs( workflow_id, step_name, status, attempt, input_hash, started_at ) VALUES (?, ?, 'running', 1, ?, datetime('now')) ON CONFLICT(workflow_id, step_name) DO UPDATE SET status = 'running', attempt = attempt + 1, started_at = datetime('now') """, (workflow_id, step_name, input_hash), ) idempotency_key = f"{workflow_id}:{step_name}" try: result = handler(step_input, idempotency_key) except Exception as exc: with db: db.execute( """ UPDATE step_runs SET status = 'failed', error_json = ? WHERE workflow_id = ? AND step_name = ? """, (json.dumps({"message": str(exc)}), workflow_id, step_name), ) raise with db: db.execute( """ UPDATE step_runs SET status = 'completed', output_json = ?, finished_at = datetime('now') WHERE workflow_id = ? AND step_name = ? """, (json.dumps(result), workflow_id, step_name), ) db.execute( """ INSERT INTO outbox(workflow_id, event_type, payload_json) VALUES (?, 'step.completed', ?) """, (workflow_id, json.dumps({"step": step_name})), ) return result

При повторном запуске завершённый шаг возвращает сохранённый результат. Если вход неожиданно изменился, выполнение останавливается: молча продолжать с другими данными опаснее, чем явно потребовать миграцию или новый workflow.

Стабильный ключ полезен только тогда, когда его понимает получатель. Если CRM поддерживает idempotency header — передаём его. Если нет, сохраняем ключ в пользовательском поле или создаём локальную таблицу соответствий.

def create_crm_deal(payload: dict, key: str) -> dict: existing = find_deal_by_external_key(key) if existing: return existing return crm.create_deal({ **payload, "external_id": key, })

Проверка «найти, потом создать» сама по себе имеет race condition. Надёжный вариант — уникальное ограничение на external_id в системе назначения. Если это невозможно, нужен локальный lock или отдельный сервис дедупликации.

Допустим, шаг записал результат в базу, а затем процесс упал до отправки сообщения в очередь. В базе шаг completed, но следующий worker никогда не узнает, что пора продолжать.

Outbox решает это так:

Publisher может отправить событие дважды, поэтому потребитель тоже должен быть идемпотентным. В распределённых системах практичнее строить at-least-once delivery с дедупликацией, чем пытаться обещать магическое exactly-once.

Повторная генерация текста обычно не создаёт внешний побочный эффект, но стоит денег и может дать другой результат. Поэтому полезно хранить:

Если шаг уже completed, берём сохранённый ответ. Если он завис в статусе running, нельзя сразу считать его мёртвым: внешний запрос мог завершиться, а процесс не успел записать результат. Для таких случаев нужен lease с TTL и отдельный статус unknown. Повтор разрешаем только после reconciliation или по явной политике стоимости.

У каждого шага своя политика. Чтение из API можно повторить с backoff. Создание сделки — только с idempotency key. Ошибку валидации prompt повторять бессмысленно. Ошибку лимита провайдера лучше отправить в отложенную очередь.

Я обычно разделяю:

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

Отдельный алерт нужен не только на failed, но и на «не движется». Workflow может формально не падать, оставаясь running несколько часов.

Durable execution добавляет таблицы, статусы и немного скучной инфраструктуры. Но именно эта скучная часть превращает демонстрацию агента в production-сервис: после рестарта он продолжает работу, а не делает вид, что вчера ничего не происходило.

Как вы восстанавливаете длинные agent workflows: пишете собственный checkpointing, используете workflow engine или пока перезапускаете цепочку целиком?