После релиза LangGraph не может продолжить согласование: как менять имена узлов

Агент готовит отчёт и останавливается, пока руководитель его согласует. За ночь разработчики выпускают новую версию графа и переименовывают узел review в approve_report. Утром руководитель нажимает кнопку. Обработчик нашёл нужный thread_id и передал решение. Теперь важно, сможет ли новая версия вообще найти место, куда его подставить.

Имя узла попадает в контракт с сохранённым состоянием

При interrupt() LangGraph сохраняет точку исполнения через checkpointer. Вызов с тем же thread_id и Command(resume=...) адресует существующее выполнение. При продолжении прерванный узел начинает выполняться с начала. Значение из команды возвращается из соответствующего вызова interrupt(). По этой причине нельзя бездумно менять порядок нескольких interrupt() внутри узла. Состояние сохранённого потока привязано и к имени узла, в который исполнителю предстоит вернуться.

Проверьте точку продолжения через get_state(config).next. Если в сохранённом снимке ожидается review, новое имя само по себе не меняет эту запись. Вход нового запроса и продолжение старого потока требуют разных проверок. Если запускать интеграционные тесты только на новых идентификаторах, переименование пройдёт незамеченным.

Выпускаем граф с двумя точками входа для согласования

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

Пример запускается на langgraph==1.2.20. Один объект InMemorySaver моделирует общее хранилище снимков для обеих сборок. Сначала создадим прерванный поток старого графа. Затем покажем, какой узел ему нужен и присутствует ли он в двух сборках. Наконец продолжим старый поток совместимой сборкой и запустим новый.

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

class State(TypedDict, total=False):
    report: str
    approved: bool

def review(state: State):
    approved = interrupt({"report": state["report"]})
    return {"approved": approved}

def build(entry, keep_old=False):
    graph = StateGraph(State)
    graph.add_node(entry, review)
    graph.add_edge(START, entry)
    graph.add_edge(entry, END)
    if keep_old:
        graph.add_node("review", review)
        graph.add_edge("review", END)
    return graph.compile(checkpointer=saver)

saver = InMemorySaver()
old_cfg = {"configurable": {"thread_id": "old-report"}}
new_cfg = {"configurable": {"thread_id": "new-report"}}
old = build("review")
old.invoke({"report": "Отчёт за октябрь"}, old_cfg)
waiting = old.get_state(old_cfg).next
renamed = build("approve_report")
print("Старый поток ждёт:", waiting)
assert waiting == ("review",)
assert not set(waiting) <= renamed.get_graph().nodes.keys()

compatible = build("approve_report", keep_old=True)
result = compatible.invoke(Command(resume=True), old_cfg)
assert result["approved"] is True
print("Старый отчёт согласован:", result["approved"])
compatible.invoke({"report": "Отчёт за ноябрь"}, new_cfg)
next_node = compatible.get_state(new_cfg).next
assert next_node == ("approve_report",)
print("Новый поток ждёт:", next_node)

У старого потока в next остаётся review. У новой несовместимой сборки этого узла нет. Совместимая сборка успешно завершает старое согласование, а для нового отчёта сохраняет approve_report. На своём графе сделайте такой тест для каждой пользовательской паузы: сохраните снимок кодом текущего релиза, загрузите его кандидатом и проверьте ожидаемый результат после ответа пользователя.

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

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

Практика на Latorn

В пробном уроке AI Agents PRO можно посмотреть, как устроен цикл агентного приложения, прежде чем добавлять сохранение состояния и согласование.

В платном AI Agents PRO вы соберёте граф состояний на LangGraph, подключите инструменты и упакуете агентный сервис для запуска через FastAPI и Docker. После этого у вас будет граф, на котором можно отработать выпуск новой версии с сохранёнными задачами. Открыть программу AI Agents PRO.