Зачем графу останавливаться посреди работы

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

Технически это значит: граф должен уметь приостановиться в середине, сохранить всё, что успел, и возобновиться позже — может быть, в другом запросе, после того как человек нажал «Подтвердить» в интерфейсе. И здесь критически важен checkpointer из прошлых уроков: без сохранённого состояния «замереть и продолжить потом» невозможно.

ℹ️ Пауза стоит на фундаменте checkpointing

Прерывание работает только с подключённым checkpointer и thread_id. Когда граф встаёт на паузу, его состояние — это обычный checkpoint в треде. Возобновление = загрузка этого checkpoint и продолжение. Нет checkpointer'а — нет и пауз.

Статические и динамические прерывания

В LangGraph два способа остановить граф. Мы уже мельком видели первый в уроке про compile().

Статическое — interrupt_before / after
  • задаётся в compile() на конкретном узле
  • срабатывает всегда на этой границе
  • не несёт данных наружу
  • годится для отладки и простых «стоп перед узлом»
Динамическое — interrupt()
  • вызывается изнутри узла в коде
  • срабатывает по условию (можно в if)
  • отдаёт наружу payload (вопрос, черновик)
  • принимает ответ человека при возобновлении

Статическое прерывание — это «остановись на пороге узла». Динамическое — «остановись вот здесь, покажи человеку вот это и дождись вот такого ответа». Для human-in-the-loop почти всегда нужен второй вариант — ему и посвящён урок.

interrupt(): пауза изнутри узла

Функция interrupt(payload) вызывается прямо в теле узла. Когда выполнение доходит до неё, граф:

  1. замораживается — текущее состояние сохраняется в checkpoint;
  2. отдаёт payload наружу — в результат invoke попадает объект прерывания с этой нагрузкой;
  3. ждёт — управление возвращается вызывающему коду, граф «висит» на этом узле.
100%
колёсико — масштаб  ·  зажать и тянуть — перемещение
START draft ⏸ review interrupt(...) execute END resume Человек / клиент payload: «одобрить ответ?» Command(resume="approve") ⏸ пауза + checkpoint resume → граф висит на узле review, пока не придёт Command(resume=...)
interrupt() внутри узла — пауза с полезной нагрузкой
python 
from langgraph.types import interrupt

def human_review(state: State) -> dict:
    # Граф замирает здесь и отдаёт payload наружу
    decision = interrupt({
        "question": "Одобрить ответ?",
        "draft": state["draft"],
    })
    # ↓ этот код выполнится ТОЛЬКО после возобновления;
    #   в decision окажется то, что передали в Command(resume=...)
    if decision == "approve":
        return {"approved": True}
    else:
        return {"draft": decision, "approved": False}   # человек прислал правку

Ключевое: interrupt() ведёт себя как «функция, которая вернёт значение позже». В первый проход она не возвращает ничего — она прерывает выполнение. Значение появится при возобновлении.

Возобновление через Command(resume=...)

Первый invoke доводит граф до interrupt() и останавливается. Чтобы продолжить, вызывают invoke снова — но вместо обычного входа передают Command(resume=значение). Это значение и станет результатом interrupt().

Полный цикл: пауза → ответ человека → продолжение
python 
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command

graph = builder.compile(checkpointer=MemorySaver())   # пауза требует checkpointer
config = {"configurable": {"thread_id": "review-1"}}

# 1. Первый запуск — доходит до interrupt() и встаёт на паузу
result = graph.invoke({"draft": "Текст письма..."}, config)
print(result["__interrupt__"])
# [Interrupt(value={'question': 'Одобрить ответ?', 'draft': 'Текст письма...'})]

# 2. Показали payload человеку, получили решение...
#    Возобновляем тем же thread_id, передавая ответ в resume
final = graph.invoke(Command(resume="approve"), config)
print(final["approved"])     # True

# Если бы человек прислал правку:
# graph.invoke(Command(resume="Исправленный текст"), config)
#   → interrupt() вернёт "Исправленный текст", узел запишет его в draft

Между шагами 1 и 2 может пройти сколько угодно времени — хоть несколько дней. Состояние лежит в checkpoint'е; пока цел thread_id и хранилище, граф можно возобновить когда угодно, хоть после рестарта сервиса (с персистентным saver'ом из урока про checkpointing).

ℹ️ Как узнать, что граф на паузе

Признак прерывания — ключ __interrupt__ в результате invoke (список объектов Interrupt с их value). Альтернатива — graph.get_state(config).next: если граф ждёт, там будет имя узла, на котором он застрял. По этому признаку приложение понимает, что нужно показать человеку запрос.

Неочевидная семантика: узел запускается заново

Здесь — главный подводный камень. При возобновлении LangGraph выполняет узел с прерыванием заново, с самого начала. Не «с того места, где стоял interrupt()», а целиком. Просто на этот раз interrupt() не останавливает, а сразу возвращает переданное значение.

Код ДО interrupt() выполнится дважды
python 
def risky_node(state: State) -> dict:
    log = expensive_call()          # ⚠️ выполнится И на паузе, И при возобновлении!
    charge_user()                   # ⚠️ ДВАЖДЫ — деньги спишутся два раза

    answer = interrupt({"ask": "ок?"})   # пауза тут

    return {"done": True}           # выполнится только после resume

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

❌ Не ставь сайд-эффекты перед interrupt()

Списания, отправка писем, запись в БД до interrupt() сработают дважды. Лучший приём — узел с interrupt() делает только запрос подтверждения, а необратимое действие живёт в следующем узле, который выполняется уже после возобновления.

Паттерн approval: разрешить / отклонить / править

Самый частый сценарий — три исхода ревью. Удобно вернуть из interrupt() структуру с типом решения и развести логику условным ребром (привет уроку про conditional edges).

Три ветки решения человека
python 
def review(state: State) -> dict:
    decision = interrupt({"draft": state["draft"]})   # {"action": "...", "text": "..."}
    action = decision["action"]
    if action == "approve":
        return {"status": "approved"}
    if action == "edit":
        return {"draft": decision["text"], "status": "edited"}
    return {"status": "rejected"}                      # action == "reject"

def route(state: State) -> str:
    return "send" if state["status"] in ("approved", "edited") else END

builder.add_conditional_edges("review", route, {"send": "send", END: END})

# Возобновление с выбором человека:
graph.invoke(Command(resume={"action": "edit", "text": "Новый текст"}), config)
✅ interrupt можно вызывать несколько раз

В одном графе может быть много точек interrupt() — на каждом рискованном шаге своя. LangGraph сопоставляет ответы прерываниям по порядку их появления в узле. Для надёжности держи в узле по одному interrupt() и не меняй их порядок между запусками.

Типичные ошибки

❌ Ошибка 1: interrupt без checkpointer

interrupt() не работает без подключённого checkpointer и thread_id — паузе негде сохраниться. Будет ошибка. Сначала compile(checkpointer=...), потом прерывания.

❌ Ошибка 2: сайд-эффекты до interrupt()

Узел перезапускается целиком при возобновлении, поэтому код перед interrupt() выполнится дважды. Необратимые действия — только после паузы или в отдельном узле.

❌ Ошибка 3: возобновление обычным входом

Чтобы продолжить, передавай Command(resume=...), а не обычный словарь состояния. Передашь обычный вход — граф не поймёт, что это ответ на прерывание, и попробует начать заново.

❌ Ошибка 4: не проверяют __interrupt__

После invoke приложение должно проверить, не встал ли граф на паузу (ключ __interrupt__ или get_state().next). Иначе ты примешь «остановку на ревью» за финальный ответ.

Шпаргалка

Dynamic interrupts — всё в одном месте
python 
from langgraph.types import interrupt, Command

# 1. Пауза ИЗНУТРИ узла (нужен checkpointer + thread_id)
def review(state) -> dict:
    # ⚠️ код здесь выполнится повторно при возобновлении — держи его чистым
    answer = interrupt({"question": "одобрить?", "draft": state["draft"]})
    return {"approved": answer == "yes"}   # выполнится после resume

graph = builder.compile(checkpointer=saver)
cfg = {"configurable": {"thread_id": "t-1"}}

# 2. Первый запуск → пауза
res = graph.invoke(inp, cfg)
res["__interrupt__"]          # [Interrupt(value={...})] — payload человеку
graph.get_state(cfg).next     # узел, на котором стоим

# 3. Возобновление с ответом человека
graph.invoke(Command(resume="yes"), cfg)

# Статическое прерывание (для сравнения) — на границе узла, без payload:
# builder.compile(checkpointer=saver, interrupt_before=["tools"])

# Правила:
#  • interrupt() требует checkpointer + thread_id
#  • узел перезапускается ЦЕЛИКОМ → сайд-эффекты ПОСЛЕ interrupt()
#  • продолжать только через Command(resume=...)
#  • проверяй __interrupt__, чтобы не спутать паузу с финалом

Практическое задание

Реализуй человека в цикле:

Задание: агент-«отправитель писем» с подтверждением

  1. Состояние: topic, draft, status. Узел write генерирует черновик письма по теме.
  2. Узел review вызывает interrupt() с payload {draft} и возвращает решение в status.
  3. Скомпилируй с MemorySaver, запусти и убедись, что в результате есть __interrupt__, а граф не дошёл до конца.
  4. Возобнови через Command(resume="approve") и проверь, что граф завершился со статусом «одобрено».
  5. Сделай узел send (печатает «письмо отправлено») после review и убедись, что он срабатывает один раз — потому что стоит после паузы, а не до.
  6. Со звёздочкой: поддержи три исхода (approve / edit / reject) через структуру в resume и условное ребро, как в паттерне approval.

Что дальше

Это завершает раздел «Продвинутые возможности». interrupt() — мост к следующему разделу: на нём целиком строится human-in-the-loop, где мы добавим подтверждение рискованных операций и ручное редактирование состояния.

💾

Checkpointing

Фундамент пауз — сохранение состояния

 ⏸️

Human-in-the-loop

interrupt и approval-воркфлоу

 ✏️

Редактирование состояния

update_state во время паузы

 🧩

Все уроки модуля 03

LangGraph целиком