Параллельные ветки агента нельзя склеивать вслепую

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

У состояния ветвящегося агента есть три разных режима изменений. Одни изменения можно складывать, другие можно принять только при совпадении версии, третьи требуют отдельного решения. Reducer, журнал событий и явный конфликт нужны не как взаимозаменяемые паттерны, а для этих трёх разных режимов. Команда, которая смешивает их в одном merge_state(), почти всегда получает тихую потерю данных, повторные внешние действия или неразбираемые инциденты.

Один объект не равен одному словарю

Объект состояния нельзя считать обычным словарём только потому, что он сериализуется в JSON. У него есть поля с разной семантикой, и один общий алгоритм слияния не способен сохранить эту семантику.

Представьте заявку на возврат средств. Агент запускает параллельно ветку проверки политики, ветку поиска платежа, ветку оценки риска и ветку подготовки ответа оператору. Все они читают ревизию 41 одной заявки. Через несколько секунд они возвращают обновления:

• проверка политики добавляет найденное правило;
• поиск платежа добавляет идентификатор транзакции;
• риск выставляет risk_level = high;
• подготовка ответа предлагает decision = approve.

Первые два изменения независимы. Их можно объединить без потери смысла. Последние два нельзя автоматически трактовать одинаково. Уровень риска может отменить возможность одобрения, а решение ветки ответа часто основано на неполной картине. Если reducer просто склеит поля, объект получит противоречивое состояние. Если алгоритм выберет последнее обновление, заявка станет зависеть от того, какая ветка завершилась позже.

Полезнее делить поля не по техническому типу string, array или object, а по характеру операции:

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

Массив не всегда накопителен, а строка не всегда заменяема. Например, поле status выглядит как строка, но переход closed -> approved может быть запрещён. Список approvers выглядит как массив, но простое склеивание способно повторно добавить одного и того же согласующего. Тип данных не рассказывает, совместимы ли операции.

Документация LangGraph формулирует механику честно: reducer получает накопленное значение слева и новое обновление справа, затем возвращает следующее значение состояния. Если reducer не задан, обновление перезаписывает поле. Это удобная модель исполнения графа, но она не добавляет предметных правил за вас.

Reducer безопасен только для совместимых операций

Reducer стоит применять там, где порядок применения обновлений не меняет допустимый результат или где вы сознательно определили правило порядка. Для агентных веток этого недостаточно проверить на паре примеров. Нужно проверить свойства операции.

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

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

Вот reducer для накопления доказательств, где каждая запись обязана иметь устойчивый id:

from typing import Iterable


def merge_evidence(left: list[dict], right: Iterable[dict]) -> list[dict]:
    by_id = {item['id']: item for item in left}

    for item in right:
        existing = by_id.get(item['id'])
        if existing is None:
            by_id[item['id']] = item
        elif existing != item:
            raise ValueError(f"evidence id collision: {item['id']}")

    return [by_id[item_id] for item_id in sorted(by_id)]

Этот код делает важную вещь, которую часто пропускают. Он не пытается примирить две разные записи с одинаковым идентификатором. Совпадение id с разным содержимым означает ошибку протокола, повторное использование идентификатора или неполную детерминированность ветки. Молчаливо взять одну из записей было бы потерей данных, замаскированной под merge.

Ниже пример состояния, где reducers уместны только для части полей:

state = {
    'case_id': 'refund-918',
    'revision': 41,
    'evidence': [],
    'warnings': [],
    'payment_id': None,
    'risk_level': None,
    'decision': None,
    'effects': []
}

Для evidence и warnings можно задать объединение по идентификаторам. Для payment_id лучше разрешить запись только один раз или потребовать, чтобы новое значение совпадало со старым. Для risk_level правило зависит от шкалы: если уровни действительно упорядочены, допустим reducer max. Но он опасен, когда уровни пришли от разных политик или моделей и не одинаково определены. Для decision автоматического reducer обычно нет.

Популярная ошибка состоит в том, чтобы свести всё к «последняя запись побеждает». Это кажется практичным, потому что код занимает одну строку. Он искажает причину изменения: ветка, которая завершилась последней, не обязательно видела больше данных, имела более высокий приоритет или обладала полномочием принять решение. last write wins годится для некоторых пользовательских предпочтений и кэшей. Для состояния, по которому агент совершает действие, это правило должно быть исключением с явным названием, а не настройкой по умолчанию.

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

Патч и намерение изменения нельзя путать

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

Сравните две записи:

{
  "decision": "approve"
}

и

{
  "event_id": "01JQ2D7W0XK8P7FJ3T9N",
  "case_id": "refund-918",
  "expected_revision": 41,
  "type": "refund_approval_proposed",
  "actor": "policy_branch",
  "reason_codes": ["within_window", "payment_found"],
  "evidence_ids": ["policy-77", "payment-19"]
}

Первый вариант стирает вопрос «почему». Второй хранит автора, ревизию, намерение и доказательства. Когда риск-ветка предложит блокировку, система сможет сопоставить два предложения, а не просто столкнуть два значения строки.

Журнал событий нужен не потому, что «событийная архитектура современнее CRUD». Он нужен, когда история изменений сама входит в требования: аудит, восстановление состояния, разбор ошибочного решения, повторная обработка после исправления правила или раздельные представления одного объекта. Мартин Фаулер описывает event sourcing как сохранение всех изменений состояния последовательностью событий, по которой можно восстановить прошлое состояние. Microsoft отдельно предупреждает, что это сложный паттерн с ценой за миграции, схемы, запросы и обработку конкурентных записей.

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

Минимальный контракт записи для такого журнала выглядит так:

{
  "event_id": "evt_8f6b0c1d",
  "stream_id": "refund-918",
  "expected_version": 41,
  "run_id": "run_2026_04_17_031",
  "branch_id": "risk_review",
  "type": "risk_assessed",
  "payload": {
    "level": "high",
    "reason_codes": ["merchant_mismatch"]
  },
  "causation_id": "cmd_2a16e9",
  "idempotency_key": "run_2026_04_17_031:risk_review:1"
}

expected_version защищает поток от записи поверх уже изменившегося объекта. causation_id отвечает на вопрос, какая команда вызвала событие. idempotency_key защищает от повторной доставки. run_id и branch_id позволяют отличить два параллельных запуска от повторной попытки одной ветки.

Журнал не отменяет reducers. Он сдвигает место слияния. Ветки могут писать независимые события в append-only поток, а проектор уже строит текущее представление: набор доказательств, текущую оценку риска, ожидающие конфликты. Снимок состояния становится производным артефактом, а не единственным местом, где существует правда.

Конфликт должен стать отдельным результатом

Если две ветки предлагают несовместимые изменения, система должна вернуть конфликт как данные, а не прятать его в логах или выбирать победителя по времени.

Есть соблазн называть конфликтом любое одновременное обновление. Это лишняя драматизация. Две ветки могут параллельно добавить разные ссылки, заполнить разные поля или выдать одинаковое решение с разными доказательствами. Конфликт начинается тогда, когда система не может принять оба изменения, сохранив инварианты объекта.

Для заявок на возврат инварианты могут быть такими:

• подтверждённый высокий риск запрещает автоматическое одобрение;
• сумма возврата не превышает подтверждённую сумму платежа;
• после фактической отправки денег решение нельзя заменить без компенсирующего процесса;
• один и тот же платёж нельзя вернуть дважды.

Вместо merge() полезно иметь функцию классификации. Она получает актуальное состояние и предложение ветки, затем возвращает один из трёх исходов: accepted, rejected или conflict.

def classify_proposal(current: dict, proposal: dict) -> dict:
    if proposal['expected_revision'] != current['revision']:
        return {
            'status': 'conflict',
            'reason': 'stale_revision',
            'current_revision': current['revision']
        }

    if proposal['type'] == 'refund_approval_proposed':
        if current.get('risk_level') == 'high':
            return {
                'status': 'conflict',
                'reason': 'approval_conflicts_with_high_risk',
                'required_inputs': ['risk_assessment', 'human_override_or_rejection']
            }
        return {'status': 'accepted'}

    return {'status': 'rejected', 'reason': 'unknown_proposal_type'}

Слово conflict здесь не означает, что система сломалась. Оно означает, что автоматика дошла до границы своих полномочий. Дальше возможны разные обработчики: повторная оценка на новом снимке, более приоритетная ветка, детерминированное доменное правило или передача оператору. Но сам выбор должен быть виден в протоколе.

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

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

Версия защищает объект, но не заменяет правило

Оптимистичная проверка версии отвечает на узкий вопрос: изменился ли объект после того, как ветка его прочитала. Она не отвечает, допустимо ли изменение даже при совпадающей версии.

Допустим, ветка A и ветка B прочитали ревизию 41. A успешно добавила событие и поток стал ревизией 42. B пытается добавить своё событие с expected_version = 41. Хранилище отклоняет запись. Это правильно: B сделала вывод по старому снимку.

Дальше нельзя механически повторить B. Сначала B должна получить ревизию 42, проверить, изменились ли предпосылки её решения, и сформировать новое предложение. Если A добавила безобидный комментарий, B может снова предложить то же действие. Если A изменила лимит, статус или риск, B обязана пересчитать вывод.

В документации Microsoft этот сценарий разбирается на примере одновременной работы с одним потоком: оптимистичный контроль конкурентного доступа отклоняет append, если поток изменился после чтения, а обработчик должен перечитать состояние, заново проверить правила и только потом повторить операцию. Это правильнее, чем слепой retry, который часто встречается в агентных оркестраторах.

Версия также не ловит конфликт между несколькими объектами. Например, ветка резервирует лимит клиента, а другая ветка одновременно создаёт другой заказ, который съедает тот же лимит. У каждого потока может быть корректная локальная версия, но общая сумма нарушит правило. В таких случаях нужен агрегат с общей границей согласованности, транзакция, резервирование или процесс компенсации. Нельзя решить межобъектный инвариант аккуратным reducer для одного JSON-документа.

Разберите поломку до того, как она случится в продакшене

Типичный сбой выглядит безобидно. Команда создаёт объект application, запускает четыре ветки и разрешает каждой вернуть частичный словарь. Оркестратор применяет обновления по мере готовности. В тестах всё проходит, потому что заглушки отвечают в постоянном порядке.

В продакшене ветка policy_check завершилась первой и вернула:

{
  "status": "approved",
  "notes": ["policy permits automatic approval"]
}

Через секунду fraud_check вернула:

{
  "status": "manual_review",
  "notes": ["device fingerprint mismatch"]
}

Код использует обычное обновление словаря. В зависимости от порядка прибытия итоговый status становится approved или manual_review. Список notes при этом тоже заменяется, если разработчик не написал отдельный reducer. В журнале исполнения видны оба ответа, а в итоговом объекте остаётся один. Потом ветка отправки письма читает approved и выполняет действие, которое оператор уже не может легко отменить.

Исправление не в том, чтобы применить max к строковым статусам или добавить ещё одну задержку перед отправкой письма. Нужно изменить контракт.

1. Проверочные ветки публикуют факты и предложения, а не записывают финальный status.
2. Один доменный обработчик получает все релевантные факты и применяет правила переходов.
3. При несовместимых предложениях обработчик создаёт объект конфликта с причинами и доказательствами.
4. Ветка внешнего действия запускается только после отдельного события approval_confirmed.

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

Такой владелец не обязан быть отдельным микросервисом. Это может быть функция в том же workflow. Важно, чтобы только она имела право менять поля, несущие инварианты. Разделение полномочий уменьшает число мест, где случайный патч способен запустить необратимое действие.

Смешанная схема обычно лучше чистой идеологии

Большинству агентных приложений не нужен полный event sourcing для каждого шага и не нужен один гигантский reducer. Практичная схема использует все три подхода в своих границах.

Храните краткоживущие технические данные в обычном состоянии запуска: промежуточные выдержки, результаты поиска, трассировку инструментов, временные подсказки маршрутизатору. Эти данные можно удалять по политике хранения, и потеря старого снимка не меняет бизнес-историю.

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

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

Выделяйте явный конфликт для операций, где две правды не могут сосуществовать. Не отправляйте такой конфликт в «ошибки LLM». Это нормальный доменный исход с собственным статусом, SLA и владельцем.

Для моделей разных провайдеров полезно сохранять отдельно технический маршрут вызова и доменное событие. Ответ модели, идентификатор модели, версия промпта и метаданные инструмента нужны для разбора качества. Но событие refund_approval_proposed не должно зависеть от названия модели. Иначе при смене провайдера вы загрязните предметную историю деталями исполнения.

AI Router может помочь сохранить существующий OpenAI-совместимый клиент при маршрутизации запросов к разным моделям, но единый endpoint не заменяет контракт слияния состояния. Смена base_url не должна менять смысл ваших событий, ревизий и конфликтов.

Контракт слияния нужно тестировать как бизнес-логику

Тест, который проверяет только ожидаемый итог в одном порядке, почти ничего не говорит о параллельном workflow. Вам нужны тесты перестановок, повторов и устаревших записей.

Для reducer создайте набор обновлений и прогоните все небольшие перестановки. Итог должен совпасть там, где вы заявляете независимость порядка. Затем повторите одно обновление дважды. Если reducer не идемпотентен, тест должен это зафиксировать и контракт обязан объяснять, почему дубликаты невозможны или как вы их отсекаете.

Для событийного потока проверяйте сценарий «прочитал, потерял гонку, перечитал». Подайте команду на ревизии 41, добавьте конкурентное событие, убедитесь, что append отклонён, затем убедитесь, что повторное вычисление использует ревизию 42. Не подменяйте эту проверку тестом на успешный retry: главное в том, что старое намерение не должно пройти без пересмотра.

Для конфликтов пишите таблицу решений. Она проста, зато предотвращает споры между разработчиками уже после выпуска:

Наконец, тестируйте внешние эффекты отдельно от решения. Обработчик approval_confirmed может получить одну и ту же запись дважды после сбоя очереди. Он обязан определить, отправлял ли уже письмо или создавал перевод, по idempotency_key. Проверка «мы не ожидаем дублей» не является защитой. Дубли появляются именно тогда, когда система восстанавливается после частичного сбоя.

Не прячьте конфликт в поле error

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

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

{
  "conflict_id": "conf_41d9",
  "stream_id": "refund-918",
  "status": "open",
  "kind": "decision_vs_risk",
  "current_revision": 42,
  "proposals": [
    "evt_policy_110",
    "evt_risk_221"
  ],
  "required_action": "human_review",
  "resolution": null
}

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

Параллельность окупается, когда ветки собирают независимые факты одновременно. Она становится опасной, когда каждая ветка получает право объявить финальное состояние объекта. Оставьте reducers для того, что действительно складывается. Записывайте намерения там, где важна история. А в точке, где правила не позволяют принять оба обновления, сохраните конфликт и заставьте систему решить его честно.