Пайплайны и процессы на графе

Автоматизируйте запуски по расписанию, DAG с fan-out, ветвления, повторные попытки, подтверждения человеком и сценарии по вебхукам с помощью графовых пайплайнов — визуального холста из узлов и связей, который использует те же функции, что и HTTP-маршруты. Расписания настраиваются как триггеры пайплайна.

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

Визуальный редактор графа#

Редактор графа рассчитан на HTTP-процессы: триггеры ведут в узлы «Функция» и узлы управления потоком, узел Map формирует JSON, а Respond отдаёт HTTP-ответ клиенту.

• В разделе Пайплайны откройте карточку визуального пайплайна и нажмите Открыть визуальный редактор (или создайте пайплайн через Новый визуальный пайплайн).
• Перетаскивайте типы узлов из палитры слева: Триггеры (Manual, HTTP, Cron), Шаги (Функция, If, Set, Map, Respond, Human), группа Parallel (узлы Parallel и Merge) и Note. Соединяйте выходы и входы, чтобы задать порядок выполнения.
• Клик по узлу открывает его настройки в боковой панели (привязка функции, выражения, JSON-шаблоны, коды ответа и т.д.).
• Узлы Заметка только для документации на схеме: они не выполняются и не должны иметь рёбер — это стикеры на холсте.

В визуальном редакторе соединяют триггеры Manual, HTTP и Cron с узлами «Функция», If, Map, Respond и Human; на скриншоте ниже — холст в консоли.

Типы узлов графа (справочник)#

Соединяйте нижние ручки (выходы) с верхней ручкой (входом) следующего узла. У триггеров, Parallel, Map, Set и Merge один выход по умолчанию. У узла «Функция» два выхода — ok и err; у If — true и false. Human gate в режиме approve имеет выходы ok и no (ручки approve/reject); в режиме question — один выход по умолчанию. У Respond и Note исходящих ручек нет. В графе всегда должен быть хотя бы один триггер.

Ручной триггер#

Запускает выполнение из UI пайплайна или через API execute. Дополнительной настройки не требует. Используйте этот узел (или другой триггер) как начало пути.

HTTP-триггер#

Запускает выполнение, когда публичный HTTP-вход пайплайна получает запрос с настроенным методом и путём (например, POST /hook). Узел также хранит настройки авторизации для этого входа (без проверки, API-ключ или bearer); при добавлении из палитры подставляются значения по умолчанию.

Cron-триггер#

Запускает выполнение по расписанию. Задайте пятипольное cron-выражение и строку часового пояса, например UTC. Минимальный интервал — одна минута, а планировщик опрашивает расписание примерно раз в 30 секунд, поэтому запуск срабатывает в пределах примерно полуминуты от намеченного времени. Некорректное cron-выражение — или выражение, срабатывающее чаще минимума, — отклоняется при сохранении или публикации.

Функция#

Вызывает функцию рабочего пространства; выберите её в панели свойств. Настройка поведения при ошибке определяет, что происходит при сбое функции: failPipeline останавливает запуск; continue продолжает по ветке ok, фиксируя сбой; errorBranch направляет на выход err — подключите его для сценариев восстановления. Оставьте переопределение входа пустым, чтобы передать выход предыдущего узла, тело триггера сразу после триггера или структурированный объект ошибки, если предыдущий шаг упал и граф продолжился. JSON переопределения использует те же шаблонные плейсхолдеры, что и другие поля, но input в этом поле не подставляется — используйте пути trigger, vars и steps. Исполнитель поддерживает retryPolicy в JSON графа; в UI настраиваются функция, onError и переопределение входа.

If (ветвление)#

Делит поток по булеву выражению. Подключите обе исходящие ручки — true и false. Условие пишется в синтаксисе, похожем на JavaScript, но платформа сама разбирает и вычисляет его (оно никогда не исполняется как JavaScript); результат приводится к boolean. Поддерживаются: пути вида input.a.b, input["key"], input.items[0], input.a?.b и .length; сравнения (===, !==, ==, !=, >, >=, <, <=); &&, ||, ??, !, typeof, in; арифметика; тернарные выражения и фиксированный набор функций: includes, startsWith, endsWith, indexOf, lastIndexOf, toLowerCase, toUpperCase, trim, Number(), String(), Boolean(), Array.isArray() и Math.min/max/abs/floor/ceil/round/trunc/sign. Другие вызовы функций, стрелочные функции, присваивания, регулярные выражения и шаблонные строки, точки с запятой, переводы строк, комментарии, escape-последовательности \u/\x и свойства constructor, __proto__, prototype отклоняются при сохранении пайплайна. В выражении доступны привязки input, trigger, vars, steps, inputs и ifResult. input.* строится из выхода предыдущего шага: строка, похожая на JSON-объект или массив, разбирается, поэтому её поля работают как JSON-тело Lambda. Значения из узлов Set поверхностно сливаются в тот же input (при совпадении ключей побеждают vars); можно по-прежнему явно писать vars.ИМЯ или steps.NODE_ID.output…. По умолчанию выход шага — это неизменённый входной payload: значение условия хранится отдельно как ifEvaluated на шаге и не смешивается с output. Чтобы передать булево значение в JSON дальше по графу, задайте форму выхода (outputMapping), например с {{ifResult}}. Необязательная форма выхода принимает {{input}}, {{ifResult}}, {{vars…}} и {{steps…}}.

Пишите булево выражение на JavaScript по input, trigger, vars, steps и inputs. Комбинируйте условия через && / || и тернарные выражения — например input.score > 50 && input.tier == 'pro'. Не оборачивайте выражение в {{ }} — такой синтаксис нужен только в JSON-шаблонах. Вместо NODE_ID подставьте реальный идентификатор узла графа. Примеры:

• Строка из upstream — input.tier == 'pro' (литерал в одинарных или двойных кавычках).
• Число из upstream больше порога — input.score > 50
• Строгое равенство с булевым — input.ok === true (справа также false или null).
• Выход другого узла — steps.NODE_ID.output.status == 'done'
• Переменная пайплайна из Set — vars.environment == 'production'
• Число «больше или равно» — input.retryCount >= 3
• Истинность без оператора: один путь, например input.recordId или steps.NODE_ID.output.items — ветка true, если значение truthy.
• Неравенство — input.code != 404 или input.code !== 404 (== и != допускают приведение типов, === и !== — нет).

Parallel#

Разветвляет одно входящее ребро на несколько исходящих с ручки parallel. Каждая нижестоящая ветка выполняется независимо, пока их не соберёт Merge (или другой способ объединения).

Merge#

Собирает ветки после Parallel. Режим all (единственный поддерживаемый) ждёт завершения каждой непосредственно предшествующей ветки и затем выполняется один раз. Без outputMapping одна входящая ветка передаёт свой payload без изменений, а несколько веток дают объект с картой inputs, где ключи — идентификаторы исходных узлов веток. С outputMapping используйте {{inputs}} для всей карты или {{inputs.someNodeId}} для одной ветки; при единственном входящем ребре также доступен {{input}} с payload этой ветки — как у Map.

Set#

Записывает переменные пайплайна для последующих шаблонных плейсхолдеров. Задайте JSON-объект; значения могут содержать шаблонные строки. Разрешённые записи сливаются в vars. Выход узла объединяет эти значения с upstream-объектом, если тот является обычным объектом.

Map#

Строит JSON для следующего шага через outputMapping и подстановку шаблонов; upstream-payload доступен в шаблонах как input (см. раздел о JSON-шаблонах ниже). Если outputMapping не задан, движок передаёт input без изменений.

Respond#

Завершает запуск по HTTP-триггеру: задайте код ответа, необязательные шаблоны тела ответа и необязательные заголовки (JSON-объект значений заголовков; каждое значение может использовать те же плейсхолдеры, что и тело). Оставьте тело ответа пустым — без outputMapping и без устаревшего bodyMapping, — чтобы вернуть выход предыдущего шага без изменений или тело триггера, если Respond идёт сразу после триггера. Пример объекта заголовков: Content-Type со значением application/json и X-Request-Id со значением {{vars.requestId}}. У узла нет исходящих рёбер. Если клиент должен дождаться этого ответа, предпочитайте синхронный режим выполнения.

Human gate#

Приостанавливает запуск до вызова POST /api/pipeline-graph-executions/:executionId/resume. Режим Approve требует двух исходящих рёбер с метками approve и reject; тело API — { "decision": "approve" } или { "decision": "reject" }. Режим Question использует одно стандартное ребро; тело — { "answer": <любой JSON> }. Необязательный promptTemplate — JSON-шаблон ({{input}} — это upstream-payload), сохраняемый на ожидающем шаге для операторов. Выход завершённого шага объединяет upstream-payload с humanGate: { mode, decision или answer, optional prompt }.

Заметка (Note)#

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

Как запускаются прогоны#

В графе всегда должен быть хотя бы один узел-триггер; триггер определяет, как начинается прогон:

• Ручной триггер — запуск из UI пайплайна или по HTTP через POST /pipeline-graphs/:id/execute с JSON-телом (тело становится payload триггера прогона).
• HTTP-триггер — даёт пайплайну собственный публичный вход на методе и пути (например, POST /hook) с необязательной авторизацией по API-ключу или bearer; подходящий запрос запускает прогон.
• Cron-триггер — запуск по расписанию (например, 0 */6 * * *) с учётом минимального интервала в одну минуту.

Подтверждение человеком#

Чтобы поставить прогон на паузу для человека, добавьте узел Human gate (см. справочник узлов выше). В режиме approve он ждёт, пока кто-нибудь не вызовет POST /api/pipeline-graph-executions/:executionId/resume с { "decision": "approve" } или { "decision": "reject" }, и направляет на соответствующую ручку; в режиме question собирает ответ и продолжает по стандартному ребру.

Графовые пайплайны: JSON-шаблоны#

В визуальном редакторе графа часть узлов принимает JSON, в строковых значениях которого можно использовать плейсхолдеры. Каждый фрагмент, записанный в синтаксисе двойных фигурных скобок (шаблоны ниже), движок заменяет данными текущего запуска.

• {{input}} — Полезная нагрузка с предыдущего узла (или тело триггера после триггера). Вложенные поля: {{input.field}}. Подставляется для Map, Respond (тело и значения заголовков), необязательного выходного JSON узла If и outputMapping узла Merge, если входящая ветка одна.
• {{trigger.body}}, {{trigger.type}} и т.д. — данные триггера запуска (доступны во всех шаблонных полях этих узлов).
• {{vars.name}} — значения, заданные узлами Set ранее в графе.
• {{steps.NODE_ID.output}} — выход завершённого шага; вместо NODE_ID укажите id узла из графа. Работают вложенные пути (например {{steps.abc.output.score}}).
• {{inputs}} — при разрешении outputMapping узла Merge это объект с ключами по id исходного узла каждой входящей ветки. Используйте {{inputs.idУзла}} для полезной нагрузки одной ветки (подставьте реальный id из графа).
• {{ifResult}} — только при разрешении необязательного outputMapping узла If: булево значение условия. В последующих узлах то же значение: {{steps.ID_УЗЛА_IF.ifEvaluated}} (id узла If). В тексте самого условия недоступно.

Если строка состоит только из одного плейсхолдера, подставленное значение сохраняет JSON-тип (объект, массив, число). Внутри более длинной строки значения превращаются в текст.

Как плейсхолдеры применяются к узлам:

• Функция — оставьте Свой шаблон входа пустым, чтобы передать выход предыдущего узла (или payload триггера) без изменений. Если задаёте JSON вручную, используйте {{trigger…}}, {{vars…}} и {{steps…}}. Для переопределения входа узла «Функция» {{input}} не используется.
• Merge — поведение без outputMapping описано выше в разделе Merge. Произвольный outputMapping использует {{inputs…}}; при одном входящем ребре работает и {{input}}.
• If — поле условия без {{ }}. Движок собирает input из выхода предыдущего шага (строки JSON разбираются) и сливает vars из Set. Используйте input.поле, steps.id.output…, vars…. Необязательная форма выхода — JSON как у Map и {{ifResult}}.
• Map — поле outputMapping задаёт объект, который отдаёт узел; {{input}} — вход с предыдущего узла (если поле опущено в JSON, движок по умолчанию пробрасывает {{input}}).
• Respond — необязательный outputMapping формирует тело HTTP; если нет ни outputMapping, ни устаревшего bodyMapping, тело ответа совпадает с входом с предыдущего узла. Значения в headersMapping подчиняются тем же правилам (например Content-Type и application/json).
• Заметка — не участвует в выполнении и в шаблонах; движок её игнорирует. Не подключайте к заметкам рёбра.

// In Map, Respond (body/headers), and Lambda input override, values may be JSON
// with strings like "{{trigger.body.field}}" or "{{steps.lambdaNodeId.output}}".

// Map node — outputMapping (default shape is pass-through of upstream via {{input}})
{
  "data": "{{input}}",
  "userId": "{{input.userId}}",
  "enriched": "{{steps.analyze.output}}"
}

// Respond node — optional outputMapping; omit it to return the upstream body unchanged
{
  "result": "{{input}}",
  "meta": { "source": "pipeline" }
}

// Respond headers (same rules; header values are strings after resolution)
{
  "X-Request-Id": "{{steps.firstLambda.output.requestId}}"
}

// Lambda — input payload override only when you need a custom shape (otherwise leave empty for chain input)
{
  "item": "{{trigger.body}}",
  "prior": "{{steps.stepA.output}}"
}