Agent Lab Journal
Самое важное из мира AI — в канале MAX AgentLabОтдельные разборы, инструменты и практические схемы Читайте Agent Lab в Telegram Разборы, кейсы и новости о практических AI-агентах без лишней воды.

Перезапускаемый LLM-бэкфилл модерации на Node.js

Перезапускаемый LLM-бэкфилл модерации на Node.js
Temporary fallback cover; replace in editorial pass.

Уровень: продвинутый · Чтение: до 12 минут · Обновлено: 18 октября 2026

Когда в продукт добавляют LLM-модерацию, новые публикации проверяются на лету, а старые остаются непроверенными. Их прогоняют разовым бэкфиллом — пакетной классификацией архива. На практике такой скрипт падает на середине из-за таймаута, лимита запросов или перезапуска контейнера. После повторного запуска он снова отправляет уже обработанные тексты и снова за них платит. Если в одной базе лежат данные нескольких арендаторов, к этому добавляется вопрос, чьи токены были потрачены и не попал ли результат одного клиента в выдачу другого.

В статье соберём небольшой, но честный стенд: Node.js-скрипт читает JSONL с публикациями, вызывает классификатор, пишет результаты в журнал только на добавление и переживает падение в любой момент. Затем проверим четыре свойства: повторный запуск не делает лишних вызовов, оборванная запись не ломает журнал, токены считаются по арендаторам, а подмена арендатора в результатах обнаруживается автоматически.

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

Архитектура в трёх файлах

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

Шаг 1. Подготовка стенда

Нужен Node.js 18 или новее; внешние пакеты не используются. Все команды работают только внутри новой папки.

mkdir llm-backfill-lab && cd llm-backfill-lab
mkdir -p data
node --version

Тестовые данные вымышлены: два условных арендатора, совпадающие post_id у разных арендаторов и одна дублирующаяся строка.

cat > data/posts.jsonl <<'EOF'
{"tenant_id":"tenant-a","post_id":"p1","text":"Отличная статья, спасибо"}
{"tenant_id":"tenant-a","post_id":"p2","text":"Заходите в наше казино, бонус каждый день"}
{"tenant_id":"tenant-a","post_id":"p3","text":"Ты идиот и ничего не понимаешь"}
{"tenant_id":"tenant-b","post_id":"p1","text":"Где купить запчасти для велосипеда?"}
{"tenant_id":"tenant-b","post_id":"p2","text":"Спам-рассылка: заработок без вложений"}
{"tenant_id":"tenant-b","post_id":"p3","text":"Хорошая подборка, добавлю в закладки"}
{"tenant_id":"tenant-a","post_id":"p1","text":"Отличная статья, спасибо"}
{"tenant_id":"tenant-b","post_id":"p4","text":"Ненавижу понедельники"}
EOF

Во входе восемь строк, но уникальных ключей семь: седьмая строка повторяет первую.

Шаг 2. Устойчивый ключ

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

cat > key.mjs <<'EOF'
import { createHash } from 'node:crypto';

export const PROMPT_VERSION = 'moderation-v1';

export const sha256 = (s) => createHash('sha256').update(s, 'utf8').digest('hex');

// Разделитель \u0000 не встречается в идентификаторах, поэтому
// пары вида ("a", "bc") и ("ab", "c") не дадут одинаковый ключ.
export function stableKey(post, model) {
  return sha256(
    [post.tenant_id, post.post_id, sha256(post.text), PROMPT_VERSION, model].join('\u0000')
  );
}
EOF

tenant_id стоит в ключе первым не для красоты. У обоих арендаторов есть p1, и без арендатора в ключе результат одного из них затёр бы результат другого.

Шаг 3. Конвейер с журналом и контрольной точкой

cat > backfill.mjs <<'EOF'
import {
  createReadStream, existsSync, readFileSync, writeFileSync, renameSync,
  openSync, writeSync, fsyncSync, closeSync, truncateSync, mkdirSync,
} from 'node:fs';
import { createInterface } from 'node:readline';
import { stableKey, PROMPT_VERSION } from './key.mjs';

const INPUT = process.env.INPUT ?? 'data/posts.jsonl';
const OUT_DIR = process.env.OUT_DIR ?? 'out';
const RESULTS = `${OUT_DIR}/results.jsonl`;
const CHECKPOINT = `${OUT_DIR}/checkpoint.json`;
const MODEL = process.env.MODEL ?? 'mock';
const CRASH_AFTER = Number(process.env.CRASH_AFTER ?? 0);
const MAX_TOKENS_PER_TENANT = Number(process.env.MAX_TOKENS_PER_TENANT ?? Infinity);

mkdirSync(OUT_DIR, { recursive: true });

function addUsage(usage, tenant, u) {
  const t = usage.get(tenant) ?? { input_tokens: 0, output_tokens: 0 };
  t.input_tokens += u.input_tokens;
  t.output_tokens += u.output_tokens;
  usage.set(tenant, t);
}

const spent = (usage, tenant) => {
  const t = usage.get(tenant);
  return t ? t.input_tokens + t.output_tokens : 0;
};

// Журнал — источник истины. Запись без завершающего \n считается
// оборванной при падении и отрезается до последней целой строки.
function loadLedger() {
  const done = new Set();
  const usage = new Map();
  if (!existsSync(RESULTS)) return { done, usage };
  const buf = readFileSync(RESULTS);
  const complete = buf.lastIndexOf(0x0a) + 1;
  if (complete < buf.length) {
    console.warn(`ledger: отрезаю неполный хвост ${buf.length - complete} байт`);
    truncateSync(RESULTS, complete);
  }
  for (const line of buf.subarray(0, complete).toString('utf8').split('\n')) {
    if (!line) continue;
    const r = JSON.parse(line);
    done.add(r.key);
    addUsage(usage, r.tenant_id, r.usage);
  }
  return { done, usage };
}

function appendDurable(path, obj) {
  const fd = openSync(path, 'a');
  try {
    writeSync(fd, JSON.stringify(obj) + '\n');
    fsyncSync(fd);
  } finally {
    closeSync(fd);
  }
}

// Атомарная замена: читатель видит либо старый, либо новый файл целиком.
function writeCheckpoint(state) {
  const tmp = `${CHECKPOINT}.tmp`;
  writeFileSync(tmp, JSON.stringify({ ...state, at: new Date().toISOString() }));
  renameSync(tmp, CHECKPOINT);
}

// Заглушка. Для реального провайдера верните сюда ярлык и счётчики
// токенов из ответа API — не оценку по длине текста.
async function classify(post) {
  if (MODEL !== 'mock') throw new Error('Подключите реальный клиент в classify()');
  const text = post.text.toLowerCase();
  const label = /спам|казино/.test(text) ? 'spam'
    : /идиот|ненавиж/.test(text) ? 'abuse'
    : 'ok';
  return {
    label,
    usage: { input_tokens: Math.ceil(post.text.length / 4) + 50, output_tokens: 5 },
  };
}

const { done, usage } = loadLedger();
let calls = 0, skipped = 0, deferred = 0, lineNo = 0;

const rl = createInterface({ input: createReadStream(INPUT), crlfDelay: Infinity });
for await (const line of rl) {
  lineNo++;
  if (!line.trim()) continue;
  const post = JSON.parse(line);
  if (!post.tenant_id || !post.post_id || typeof post.text !== 'string') {
    throw new Error(`строка ${lineNo}: нужны tenant_id, post_id и text`);
  }
  const key = stableKey(post, MODEL);
  if (done.has(key)) { skipped++; continue; }
  if (spent(usage, post.tenant_id) >= MAX_TOKENS_PER_TENANT) { deferred++; continue; }

  const { label, usage: u } = await classify(post);
  appendDurable(RESULTS, {
    key, tenant_id: post.tenant_id, post_id: post.post_id, label,
    prompt_version: PROMPT_VERSION, model: MODEL, usage: u, at: new Date().toISOString(),
  });
  done.add(key);
  addUsage(usage, post.tenant_id, u);
  calls++;
  writeCheckpoint({ input_line: lineNo, calls, skipped, deferred });

  if (CRASH_AFTER && calls >= CRASH_AFTER) {
    console.error(`имитация падения после ${calls} вызовов`);
    process.exit(137);
  }
}

writeCheckpoint({ input_line: lineNo, calls, skipped, deferred, finished: true });
console.log(JSON.stringify({ calls, skipped, deferred, usage: Object.fromEntries(usage) }));
EOF

Порядок операций выбран так, чтобы падение в любой точке было безопасным:

  1. Падение до записи в журнал. Вызов модели потерян, ключа в журнале нет, при перезапуске публикация будет классифицирована снова. Это единственный сценарий повторной оплаты, и он ограничен одной записью.
  2. Падение посреди записи. В журнале остаётся строка без \n. loadLedger() отрезает её, и публикация обрабатывается заново.
  3. Падение между журналом и контрольной точкой. checkpoint.json отстаёт на одну запись, но пропуск решается по ключу, поэтому лишнего вызова не будет.

Бюджет на арендатора — мягкий. Проверка стоит перед вызовом, поэтому последний вызов может превысить лимит на размер одного ответа. Отложенные публикации не пишутся в журнал и подхватываются следующим запуском с бо́льшим бюджетом.

Шаг 4. Независимая проверка

Проверка не использует состояние конвейера: она заново читает вход, заново вычисляет ключи и сверяет с ними каждую запись журнала.

cat > check.mjs <<'EOF'
import { readFileSync } from 'node:fs';
import { stableKey } from './key.mjs';

const INPUT = process.env.INPUT ?? 'data/posts.jsonl';
const RESULTS = process.env.RESULTS ?? 'out/results.jsonl';
const MODEL = process.env.MODEL ?? 'mock';

const readJsonl = (p) =>
  readFileSync(p, 'utf8').split('\n').filter(Boolean).map((l) => JSON.parse(l));

const byKey = new Map(readJsonl(INPUT).map((p) => [stableKey(p, MODEL), p]));
const errors = [];
const seen = new Set();
const tokens = {};

for (const r of readJsonl(RESULTS)) {
  const short = r.key.slice(0, 12);
  if (seen.has(r.key)) errors.push(`дубликат ключа ${short}`);
  seen.add(r.key);

  const src = byKey.get(r.key);
  if (!src) { errors.push(`ключ ${short} не найден во входе`); continue; }
  if (src.tenant_id !== r.tenant_id || src.post_id !== r.post_id) {
    errors.push(`изоляция: ${short} записан как ${r.tenant_id}/${r.post_id}, во входе ${src.tenant_id}/${src.post_id}`);
  }

  const t = (tokens[r.tenant_id] ??= { calls: 0, input_tokens: 0, output_tokens: 0 });
  t.calls++;
  t.input_tokens += r.usage.input_tokens;
  t.output_tokens += r.usage.output_tokens;
}

const missing = [...byKey.keys()].filter((k) => !seen.has(k)).length;
console.log(JSON.stringify({ results: seen.size, expected: byKey.size, missing, tokens }, null, 2));
if (errors.length) {
  console.error(errors.join('\n'));
  process.exit(1);
}
EOF

Шаг 5. Проверка результата

5.1. Падение и возобновление

Каталог out создаёт сам скрипт. При повторных прогонах удаляйте только его.

CRASH_AFTER=3 node backfill.mjs; echo "exit=$?"
cat out/checkpoint.json; echo
node backfill.mjs

Ожидаемое поведение на этих данных: первый запуск делает 3 вызова и завершается с кодом 137. Второй запуск пропускает 4 строки (три уже обработанные и дубль) и делает ещё 4 вызова: "calls":4,"skipped":4.

5.2. Идемпотентность

node backfill.mjs
node check.mjs; echo "exit=$?"

Третий запуск обязан показать "calls":0,"skipped":8. Проверка должна вернуть "results": 7, "expected": 7, "missing": 0, разбивку токенов по tenant-a и tenant-b и exit=0.

5.3. Оборванная запись

printf '{"key":"broken' >> out/results.jsonl
node backfill.mjs
node check.mjs; echo "exit=$?"

Скрипт предупредит, что отрезает неполный хвост, сделает 0 вызовов, и проверка снова пройдёт.

5.4. Негативный тест изоляции

Создаём отдельную копию журнала, где первая запись приписана чужому арендатору. Исходный журнал при этом не меняется.

node -e "const fs=require('fs');const l=fs.readFileSync('out/results.jsonl','utf8').trim().split('\n');const r=JSON.parse(l[0]);r.tenant_id='tenant-b';l[0]=JSON.stringify(r);fs.writeFileSync('out/tampered.jsonl',l.join('\n')+'\n')"
RESULTS=out/tampered.jsonl node check.mjs; echo "exit=$?"

Ожидается строка изоляция: … записан как tenant-b/p1, во входе tenant-a/p1 и exit=1. Если проверка прошла, значит, она не защищает от главного риска и ей нельзя доверять.

5.5. Бюджет по арендатору

OUT_DIR=out-budget MAX_TOKENS_PER_TENANT=100 node backfill.mjs
RESULTS=out-budget/results.jsonl node check.mjs

При оценке токенов из заглушки каждый арендатор получает по два вызова, после чего лимит исчерпан. Ожидаемый итог: "calls":4,"deferred":3,"skipped":1, а в проверке "missing": 3. Это не ошибка: отложенные публикации доберёт запуск с бо́льшим лимитом в тот же OUT_DIR.

Подключение реальной модели

Замените тело classify() вызовом своего провайдера и запускайте с MODEL=<идентификатор модели>. Модель входит в ключ, поэтому результаты заглушки не смешаются с настоящими. Соблюдайте три правила:

Ключи API передавайте через переменные окружения или менеджер секретов. Не записывайте их в журнал и в checkpoint.json.

Типовые ошибки

Ограничения

Что дальше

Другие практические разборы собраны в гайдах. Термины из статьи — бэкфилл, идемпотентность, мультиарендность — объяснены в глоссарии.