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

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

До вызывающего кода в четырёх случаях доходят два исключения. В первом примере появляется InvalidUpdateError, в третьем GraphRecursionError, второй завершается успешно, а четвёртый обрабатывает ConnectionError повтором. Я оставила четыре случая и отдельно разобрала ошибки чтения состояния и возобновления без чекпоинтера из другого записанного прогона.

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

🧪 Как устроен прогон

Прогон закреплён на langgraph==1.2.12. В файле используются StateGraph, служебные вершины START и END, типизированные словари и RetryPolicy. Узлы возвращают словари обновлений. Всё внешнее поведение сведено к задержке одного узла и добавлению строк в список вызовов.

Для установки и запуска в комментарии скрипта записаны команды:

1
2

pip install langgraph==1.2.12
python check_claims.py

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

Вывод общего прогона:

1 без редьюсера: исключение InvalidUpdateError At key 'evidence': Can receive only one value per step. Use an Annotated key to handle multiple values.
2 join: {'evidence': ['fast', 'slow'], 'seen': 2}
3 recursion_limit: GraphRecursionError
4 retry_policy: вызовов вопросов = 4 ['вопрос 1', 'вопрос 2', 'вопрос 1', 'вопрос 2']

Это четыре наблюдения на одной закреплённой версии. Строки «вопрос 1» и «вопрос 2» обозначают действия внутри Python-функции; модель таких вопросов не получала. Значение seen=2 показывает, что собирающий узел прочитал оба результата. Оно не измеряет скорость параллельной работы или время сетевого обращения.

Для первоначальной диагностики я составила карту:

В отдельном опыте с подтверждением есть ещё два сообщения: ValueError No checkpointer set и RuntimeError Cannot use Command(resume=...) without checkpointer. Они относятся к операциям после паузы. Сам первый запуск без чекпоинтера в том опыте исключения не вызвал. Если смешать эти события в одну фразу «interrupt падает», получится неверное описание точки отказа.

🧱 Откуда берётся InvalidUpdateError

Первый граф запускает a и b от START. Каждый узел возвращает список под ключом evidence. В схеме состояния это обычный list, без дополнительного правила объединения. Код из проверочного файла:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class S1(TypedDict, total=False):
    evidence: list

def a(s):
    return {"evidence": ["A"]}

def b(s):
    return {"evidence": ["B"]}

g = StateGraph(S1)
g.add_node("a", a)
g.add_node("b", b)
g.add_edge(START, "a")
g.add_edge(START, "b")
g.add_edge("a", END)
g.add_edge("b", END)

try:
    print("1 без редьюсера:", g.compile().invoke({}))
except Exception as e:
    print(
        "1 без редьюсера: исключение",
        type(e).__name__, str(e).splitlines()[0],
    )

Запуск завершается InvalidUpdateError. Сообщение называет ключ evidence и указывает, что за один шаг пришло несколько значений. По этой строке я сначала проверяю места, где узлы возвращают этот ключ, и схему его хранения.

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

Вариант решения из следующего опыта использует Annotated[list, operator.add]. В аннотации задаётся функция, которая объединяет прежнее и поступившее значения. Для списков operator.add даёт конкатенацию. Это правило подходит, когда вы хотите накопить элементы от ветвей в одной коллекции.

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

Есть и другой проектный вариант: ветви пишут в разные поля, а отдельный узел собирает их в итог. Минимальный прогон этот вариант не проверяет. Я называю его как направление изменения схемы, без обещания готового результата на вашем графе. Проверенное исправление здесь относится к списку с operator.add и join из следующего раздела.

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

🔗 Зачем нужен join, если есть редьюсер

Во втором графе поле evidence получает редьюсер. Узел slow задерживается на 0.3 секунды, fast возвращает результат сразу. Узел sup записывает количество элементов, которые увидел. Входящее ребро к нему задаётся списком имён обеих ветвей:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import operator
import time
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END

class S2(TypedDict, total=False):
    evidence: Annotated[list, operator.add]
    seen: int

def slow(s):
    time.sleep(0.3)
    return {"evidence": ["slow"]}

def fast(s):
    return {"evidence": ["fast"]}

def sup(s):
    return {"seen": len(s["evidence"])}

g = StateGraph(S2)
g.add_node("slow", slow)
g.add_node("fast", fast)
g.add_node("sup", sup)
g.add_edge(START, "slow")
g.add_edge(START, "fast")
g.add_edge(["slow", "fast"], "sup")
g.add_edge("sup", END)
print("2 join:", g.compile().invoke({}))

Сохранённый результат содержит ['fast', 'slow'] и seen=2. Здесь изменены две части графа. Аннотация поля задаёт объединение обновлений, а ребро со списком источников задаёт ожидание обеих ветвей перед sup. Проверка показывает их совместную работу на этом примере.

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

Например, перед подготовкой объяснения ошибки 1С вам могут понадобиться два независимых результата: записи журнала и сведения об объекте. Для такого проектного случая я проверяла бы отдельно наличие каждого результата на входе узла объяснения. В минимальном опыте эту роль играет счётчик seen; он даёт наблюдаемое условие успешного сбора без обращения к модели.

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

Задержка time.sleep(0.3) нужна для различия ветвей в тесте. Вывод проверяет состав состояния в sup; замера производительности здесь нет.

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

🔁 Когда граф останавливается сам

Третий пример содержит цикл без условия завершения. Узел увеличивает n, а условный переход каждый раз выбирает этот же узел. В запуске указан recursion_limit=12:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

from typing import TypedDict
from langgraph.graph import StateGraph, START

class S3(TypedDict, total=False):
    n: int

def loop(s):
    return {"n": s.get("n", 0) + 1}

g = StateGraph(S3)
g.add_node("loop", loop)
g.add_edge(START, "loop")
g.add_conditional_edges("loop", lambda s: "loop", {"loop": "loop"})

try:
    g.compile().invoke({}, config={"recursion_limit": 12})
except Exception as e:
    print("3 recursion_limit:", type(e).__name__)

Результат: GraphRecursionError. Число в конфигурации ограничивает продвижение графа; его нельзя читать как таймаут HTTP-запроса или размер контекста модели. В этом коде нет рекурсивного вызова Python-функции loop из самой себя. Повтор создаёт переход графа.

Я сначала проверила бы условие выбора следующей вершины. Здесь оно возвращает "loop" при любом состоянии, поэтому увеличение лимита только отодвинет остановку. Для рабочего агента нужно определить состояние, при котором анализ закончен или дальнейший поиск следует прекратить. Затем уже выбирать лимит, соответствующий допустимому пути.

В отдельном прогоне графа с подтверждением recursion_limit=3 также завершился GraphRecursionError. Полный путь с паузой и одобрением укладывался в recursion_limit=6. Эти числа относятся к тому конкретному графу. Из них нельзя получить универсальную настройку для агента с другими узлами, ветвлениями или условиями.

Четвёртая проверка касается другого повторения: функция узла начинает выполняться заново после подходящего исключения. Политика привязана к узлу w. Она допускает две попытки и явно выбирает ConnectionError:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy

class S3(TypedDict, total=False):
    n: int

calls = []

def flaky(s):
    calls.append("вопрос 1")
    calls.append("вопрос 2")
    if len(calls) < 4:
        raise ConnectionError("502")
    return {"n": 1}

g = StateGraph(S3)
g.add_node(
    "w", flaky,
    retry_policy=RetryPolicy(
        max_attempts=2,
        initial_interval=0.01,
        retry_on=ConnectionError,
    ),
)
g.add_edge(START, "w")
g.add_edge("w", END)
g.compile().invoke({})
print("4 retry_policy: вызовов вопросов =", len(calls), calls)

На первой попытке список получает два элемента, затем функция поднимает исключение. На второй оба append выполняются снова. Список содержит четыре строки, и узел возвращает n=1. Таким способом тест наблюдает повтор начала функции без необходимости подключать платный внешний вызов.

Строка "502" здесь только сообщение искусственно созданного ConnectionError. HTTP-сервер не отвечает кодом 502, сетевой запрос не выполняется. Поэтому результат не подтверждает автоматическое повторение любого HTTP-ответа с таким кодом. Проверено правило для класса исключения, который указан в retry_on.

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

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

💤 Что выполняется заново после interrupt

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

1. recursion_limit=3: GraphRecursionError
2. создание RemediationAction в узле: до паузы 1, после resume 2
3. без чекпоинтера: исключения нет; ключи результата: ['__interrupt__', 'audit_trail', 'enterprise_context', 'identified_error_signature', 'is_approved_by_human', 'messages', 'next', 'remediation', 'root_cause_analysis'] | есть __interrupt__: True
   get_state без чекпоинтера: ValueError No checkpointer set
   resume без чекпоинтера: RuntimeError Cannot use Command(resume=...) without checkpointer
4. полный проход (пауза и одобрение) укладывается в recursion_limit=6

По третьему пункту первый вызов вернул результат с __interrupt__. Исключения в этот момент не было. Затем попытка get_state завершилась ValueError, а попытка возобновления через Command(resume=...) завершилась RuntimeError. Это три разных операции, и у каждой свой результат.

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

Во втором пункте счётчик создания RemediationAction увеличился с одного до двух после resume. Это согласуется с описанным в документации возвратом к началу узла при продолжении после interrupt: код, находящийся до паузы, выполняется повторно. Такой повтор следует учитывать независимо от настройки RetryPolicy.

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

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

Для чтения ошибок удобно сохранить различия в таблице:

Увеличение recursion_limit не добавляет чекпоинтер. Настройка повтора не сохраняет состояние для подтверждения человека. Редьюсер списка не препятствует повторному внешнему действию. Эти механизмы встречаются в одном агенте, но проверять их удобнее отдельными минимальными опытами, каждый со своим наблюдаемым результатом.

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

🧭 Ограничения

Четыре минимальных примера я повторила 05.10.2026 на LangGraph 1.2.12, Python 3.12.10, Windows 11: вывод совпал строка в строку. Другие выпуски библиотеки не проверяла; при повторении сохраните свою версию рядом с результатом.

Ни один из четырёх минимальных примеров не вызывает модель или 1С. ConnectionError("502") создан в Python вручную. Не измерялись расходы, задержки внешних сервисов или влияние повтора на документы базы. По списку строк можно подтвердить только повтор действий внутри функции узла.

Дополнительный вывод относится к конкретному графу с подтверждением. Значение 6 описывает успешный проход этого графа; универсальный лимит из него не получается. Счётчик RemediationAction показывает повтор создания объекта, без подтверждения повторного проведения документа или иной записи в 1С.

Успешный join проверен на двух ветвях с одноэлементными списками. Не проверяла универсальный порядок результатов, объединение сложных свидетельств и дедупликацию. operator.add задаёт конкатенацию списков; пригодность этого правила для вашего состояния требует отдельного решения.

Страницы документации открыты как пояснение механизмов. Они могут описывать более новый выпуск библиотеки, чем закреплённый опыт. Для конкретных сообщений исключений и чисел я использовала сохранённые прогоны.

📚 Источники

• INVALID_CONCURRENT_GRAPH_UPDATE: пояснение конфликтующих обновлений и редьюсеров. Точный вывод InvalidUpdateError взят из минимального прогона.
• GRAPH_RECURSION_LIMIT: пояснение остановки графа по лимиту. Значения 12, 3 и 6 относятся к приведённым опытам.
• Use the graph API: справка по построению графа, RetryPolicy и настройкам запуска. Листинги с параметрами перенесены из проверочного файла для версии 1.2.12.
• Interrupts и Persistence: пояснение паузы и сохранения состояния. Наблюдения без чекпоинтера и счётчик RemediationAction взяты из отдельного прогона графа с подтверждением.
• Проверочный файл check_claims.py и сохранённый вывод на LangGraph 1.2.12: источник четырёх минимальных случаев. Вывод дополнительного опыта probe_langgraph.txt: источник наблюдений при паузе и возобновлении.
• Предшественник на Infostart: LangGraph для 1С-разработчика: агент ждёт вашего «да» перед записью, стенд за 3 минуты (Ninel_S, 30.09.2026). Там предметом была сборка агента с паузой; здесь - сообщения ошибок и порядок повторного выполнения.

📎 Файл к статье

В архиве к статье - 13.4-companion.zip: check_claims.py с четырьмя минимальными графами, эталонный вывод output-langgraph-1.2.12.txt, чек-лист chek-list-grafa.md и README. Скрипт запускается без модели и без 1С; сравните четыре строки вывода со своим окружением. Чек-лист - для проверки чекпоинтера, лимита графа и действий перед паузой до подключения внешних сервисов.

 

Нинель Адольевна Щербакова (Ninel_S)

Архитектор • SRE & DevOps инженер • HighLoad 1С

Автор 48 технических публикаций на Infostart. Разработчик NOPik, SRE-Suite-for-1C и мультиагентных ассистентов для экосистемы 1С:Предприятие.

Каталог статей и открытых репозиториев Ninel_S