LangGraph: как исправить INVALID_CONCURRENT_GRAPH_UPDATE

Ассистент поддержки одновременно ищет правила возврата в PDF и текущие сведения на сайте. По отдельности обе ветки работают. После соединения в LangGraph запуск падает с INVALID_CONCURRENT_GRAPH_UPDATE: два узла попытались обновить одно поле состояния за один шаг. Увеличивать тайм-аут бесполезно: конфликт возникает при сборке результатов.

Обычно это проявляется после небольшой переделки последовательного графа. Раньше поиск по PDF записывал список hits, затем веб-поиск заменял его своим списком. Разработчик добавляет два ребра от START, чтобы сократить ожидание, и прежнее правило записи перестаёт подходить. Теперь системе нужно явное решение: какие данные должны сохраниться после одновременных обновлений?

Сначала определите смысл поля

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

У списка документов естественное первое правило — сложение списков. Узел возвращает только собственные находки, а функция operator.add присоединяет их к уже накопленным. Если каждый узел вернёт весь прежний список вместе с новыми элементами, общая история начнёт дублироваться. Поэтому отдельно проверьте return каждого производителя данных, особенно после переноса старого последовательного кода.

Другой смысл требует другого устройства состояния. Если ветки рассчитывают одну итоговую цену, сложение чисел испортит расчёт. Разнесите предложения по полям supplier_price и warehouse_price, а выбор сделайте в отдельном узле. Reducer нужен там, где есть понятное правило объединения, а не просто возможность подавить исключение.

Воспроизведите конфликт на настоящем графе

Сохраните пример в reducer_demo.py. В отдельном окружении установите пакет командой python -m pip install langgraph и выполните файл. Узлы возвращают подготовленные идентификаторы документов: здесь проверяется доставка результатов между ветками, поэтому вызовы поискового сервиса и модели не нужны. В рабочем приложении их место занимают реальные поисковые адаптеры.

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

class Broken(TypedDict):
    hits: list[str]

class Fixed(TypedDict):
    hits: Annotated[list[str], operator.add]

def pdf(state):
    return {"hits": ["pdf:return-policy"]}

def web(state):
    return {"hits": ["web:delivery-status"]}

def build(schema):
    b = StateGraph(schema)
    b.add_node("pdf", pdf)
    b.add_node("web", web)
    b.add_edge(START, "pdf")
    b.add_edge(START, "web")
    b.add_edge("pdf", END)
    b.add_edge("web", END)
    return b.compile()

try:
    build(Broken).invoke({"hits": []})
except InvalidUpdateError:
    print("conflicting updates detected")
else:
    raise AssertionError("Expected an update conflict")

result = build(Fixed).invoke({"hits": []})
assert sorted(result["hits"]) == [
    "pdf:return-policy", "web:delivery-status"
]
print(sorted(result["hits"]))

Первый запуск обязан поймать InvalidUpdateError. Второй сохраняет обе находки; assert сравнивает состав списка после сортировки. Такой тест обнаружит потерю одной ветки, но не потребует конкретного порядка завершения запросов. Сами идентификаторы остаются в состоянии и подходят для последующей загрузки документов или построения ссылок в ответе.

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

Как подключить к существующему поиску

Менять нужно объявление поля в схеме StateGraph. Добавление Annotated к случайной локальной переменной внутри узла не задаёт правило хранения. Найдите класс, который передаётся конструктору графа, и проверьте, что именно его поле hits имеет reducer. Затем убедитесь, что обе ветки возвращают одинаковую структуру элемента: идентификатор, источник и необходимые поисковые метаданные.

После объединения обычно нужен общий узел ранжирования. Если он должен дождаться двух конкретных веток, задайте соединение builder.add_edge(["pdf", "web"], "rank"). Ранжирование получает объединённый набор и выбирает документы для контекста модели. В примере такого узла нет, поскольку он проверяет только конфликт записи, но в приложении граница между сбором и выбором особенно полезна для диагностики.

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

Что проверить перед возвратом параллельности

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

Работа с состоянием становится полезнее на сквозном сервисе. В платном курсе AI Agents PRO вы соберёте RAG-сервис, граф действий, API и контейнер, затем добавите проверки качества и наблюдаемость. Это позволяет связать изменение графа с поведением всего приложения: какие документы дошли до ответа и где потерялось время. Откройте программу курса и посмотрите, как устроен переход от графа к работающему сервису.