Прерывания (interrupts), паттерны одобрения и редактирования, валидация ввода, правила прерываний.
Прерывание останавливает граф в произвольной точке и ждёт внешнего ввода. На этом строится весь паттерн «человек в цикле».
Вызов interrupt() внутри узла приостанавливает выполнение. Состояние графа сохраняется через чекпоинтер, процесс освобождается, а граф может простоять в этом положении сколько угодно: минуту, пока пользователь читает уведомление, или неделю, пока согласование лежит в очереди.
Возобновление происходит новым вызовом с Command(resume=значение). Это значение становится результатом того самого interrupt(), будто функция всё это время ждала.
from langgraph.types import interrupt
def approval_node(state: State):
approved = interrupt("Вы одобряете это действие?")
return {"approved": approved}Для работы прерываний нужны три вещи: чекпоинтер (иначе состояние негде хранить), thread_id в конфигурации (иначе непонятно, какой диалог возобновлять) и JSON-сериализуемое значение в interrupt(), потому что оно тоже сохраняется.
Возобновление не продолжает функцию с той строки, где она остановилась. Узел запускается с начала, и код до interrupt() выполняется повторно.
Это не недоработка, а следствие модели восстановления: LangGraph хранит состояние графа, а не стек Python. Практических выводов два.
Первый: операции до interrupt() должны быть идемпотентными. Создание заказа, отправка письма, списание средств повторятся при каждом возобновлении. Записывайте через операцию вида «создать или обновить», а необратимые действия выполняйте после прерывания.
Второй: сопоставление прерываний строго позиционное. Если узел содержит два вызова interrupt(), при возобновлении первый получит уже известный ответ, а второй остановит граф снова. Менять порядок этих вызовов или пропускать их по условию нельзя: значения разъедутся.
Самый частый сценарий — это пауза перед необратимой операцией. Узел показывает, что собирается сделать, и ждёт решения:
from typing import Literal
from langgraph.types import interrupt, Command
def approval_node(state: State) -> Command[Literal["proceed", "cancel"]]:
decision = interrupt({
"question": "Выполнить перевод?",
"details": state["action_details"],
})
return Command(goto="proceed" if decision else "cancel")Значение внутри interrupt() — это то, что увидит человек. Передавать туда стоит достаточно данных для решения: не «одобрить?», а сумму, получателя и назначение платежа.
Возобновление выглядит так:
config = {"configurable": {"thread_id": "deal-42"}}
graph.invoke(Command(resume=True), config=config) # одобрено
graph.invoke(Command(resume=False), config=config) # отклоненоОдобрение — это частный случай. Часто человеку нужно поправить то, что предложила модель: исправить извлечённые поля, переформулировать письмо, уточнить сумму. Тогда в прерывание передают текущее содержимое, а возвращают исправленное:
def review_draft(state: State):
edited = interrupt({
"action": "Проверьте черновик письма",
"draft": state["draft"],
})
return {"draft": edited}Возобновление передаёт исправленный текст: Command(resume="Исправленный текст письма"). Дальше граф работает уже с ним.
Соблазн написать while True с interrupt() внутри велик: кажется, что так удобно переспрашивать при некорректном вводе. Делать этого нельзя. Узел выполняется заново при каждом возобновлении, поэтому цикл проигрывает все предыдущие итерации, и число прерываний растёт лавинообразно.
Правильная схема — это один interrupt() за вызов узла плюс условное ребро для повтора:
def get_age_node(state: FormState):
question = state.get("pending_question") or "Какой вам возраст?"
answer = interrupt(question) # ровно один раз за вызов
if isinstance(answer, int) and answer > 0:
return {"age": answer, "pending_question": None}
return {"pending_question": f"«{answer}» не похоже на возраст. Введите число."}
def need_retry(state: FormState) -> Literal["get_age", "__end__"]:
return "get_age" if state.get("pending_question") else END
builder.add_conditional_edges("get_age", need_retry)Каждая попытка — это отдельный вход в узел с чистым счётчиком прерываний.
Прерывать можно и в инструментах, а не только в узлах графа. Тогда пауза возникает перед выполнением опасной операции, а модель об этом даже не знает:
from langchain.tools import tool
from langgraph.types import interrupt
@tool
def delete_records(table: str, ids: list[int]) -> str:
"""Удалить записи из таблицы по списку идентификаторов."""
approved = interrupt({
"action": "Удаление записей",
"table": table,
"count": len(ids),
})
if not approved:
return "Удаление отменено пользователем."
db.delete(table, ids)
return f"Удалено записей: {len(ids)}."Обратите внимание на порядок: обращение к базе идёт после interrupt(), а не до. Если поменять их местами, удаление произойдёт до одобрения и повторится при возобновлении.
Когда подтверждать нужно вызовы инструментов, писать прерывания вручную незачем. HumanInTheLoopMiddleware перехватывает вызовы по списку и сама поднимает прерывание:
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
agent = create_agent(
model="openai:gpt-4o-mini",
tools=[write_file, execute_sql, read_data],
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={
"write_file": True,
"execute_sql": {"allowed_decisions": ["approve", "reject"]},
"read_data": False,
},
description_prefix="Требуется подтверждение вызова инструмента",
)
],
checkpointer=InMemorySaver(),
)Значение True требует подтверждения для всех решений, словарь ограничивает допустимые варианты, False пропускает вызов без вопросов. Безопасные операции вроде чтения незачем выносить на подтверждение: лишние паузы обесценивают саму процедуру.
Решений четыре. approve выполняет вызов как есть, edit меняет аргументы перед выполнением, reject отменяет вызов и возвращает модели пояснение, respond подставляет ответ человека вместо результата инструмента.
Возобновление передаёт список решений в порядке приостановленных действий:
from langgraph.types import Command
agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config)Для правки аргументов передают изменённое действие, для отказа можно приложить объяснение:
{"type": "edit", "edited_action": {"name": "write_file", "args": {"path": "/tmp/out.txt"}}}
{"type": "reject", "message": "Не пишем в системные каталоги."}Отказ важен именно тем, что модель получает обратную связь и может предложить другой вариант, а не упереться в ту же операцию.
Когда параллельные ветви останавливаются одновременно, возобновлять их нужно вместе. Значения передают словарём, где ключ — это идентификатор прерывания:
resume_map = {i.id: f"ответ для {i.value}" for i in stream.interrupts}
graph.invoke(Command(resume=resume_map), config=config)Возобновление по одному здесь не сработает: граф ждёт, когда закроются все остановленные ветви супершага.
Не оборачивайте interrupt() в общий try/except. Пауза реализована через специальное исключение, и перехват его блоком «поймать всё» превращает остановку в тихую ошибку. Если обработка исключений нужна, ловите конкретные типы.
Не пропускайте вызовы interrupt() по условию и не меняйте их порядок между запусками. Сопоставление идёт по позиции, а не по смыслу.
Не передавайте в interrupt() объекты, которые не сериализуются в JSON. Значение сохраняется вместе с состоянием.
Не рассчитывайте, что переменные, вычисленные до прерывания, сохранятся сами собой. Они будут пересчитаны при возобновлении; всё, что должно пережить паузу, кладите в состояние.
Для отладки бывает удобно останавливать граф перед конкретным узлом, не вставляя interrupt() в код. Это делается при компиляции:
graph = builder.compile(
checkpointer=checkpointer,
interrupt_before=["risky_node"],
)Такой останов подходит для пошагового разбора: посмотреть состояние через get_state, при необходимости поправить его через update_state и продолжить. В продакшене обычно используют interrupt(), потому что он привязан к логике, а не к структуре графа.
Прерывание без чекпоинтера. Состояние негде сохранить, поэтому запуск падает.
Возобновление без того же thread_id. Граф не найдёт сохранённое состояние и начнёт с начала.
Неидемпотентная операция до interrupt(). Классический симптом — это дубли записей после каждого подтверждения.
Подтверждение вообще всех вызовов инструментов. Человек привыкает нажимать «одобрить» не глядя, и защита перестаёт работать.
Далее: Функциональный API