LangGraph с участием человека: функция interrupt(), ручное изменение состояний и возможность перемещения во времени.
LangGraph: участие человека в процессе — одобрение вызовов инструментов с использованием функции interrupt(), добавление механизмов обратной связи, ручная корректировка состояния и возврат к предыдущему состоянию с помощью функции перемещения во времени, с примерами реального применения.
Источники: LangGraph Human-in-the-Loop: Pausing, Reviewing, and Rewinding Your Agent
Во всем, что приведено ниже, предполагается, что вы знакомы со стандартной семичастной структурой файла LangGraph (импорты, состояние, инструменты, узлы, ребра, сборка, точка входа) и использовали системы сохранения состояния с параметром thread_id. Каждый из описанных здесь шаблон основан на такой системе сохранения состояния; без нее ничего не сработает.
Почему человек должен постоянно находиться в процессе
Как только ИИ сможет действовать — отправлять электронные письма, удалять файлы, списывать деньги с карты или загружать код, — он также может совершить дорогостоящую ошибку с машинной скоростью, когда рядом никого нет. Присутствие человека в процессе (Human-in-the-Loop, HITL) — это не дополнительная функция, приложенная к агенту; это основа дизайна, позволяющая сделать агентов с реальными инструментами безопасными для использования, и LangGraph был создан с учетом этой проблемы. Его система сохранения состояния в основном предназначена для обеспечения возможности пауз.
Четыре шаблона охватывают большинство реальных систем; они перечислены в порядке частоты их использования:
- Пауза перед рискованным действием с использованием
interrupt()— текущий стандарт. - Гарантированный контрольный пункт в рамках рабочего процесса.
- Редактирование состояния во время паузы для корректировки действий агента перед его продолжением.
- «Путешествие во времени»: повторная обработка и разветвление с любой более ранней точки в разговоре.
Некоторые элементы намеренно не упомянуты, поскольку они редко встречаются в реальных кодовых базах: внутренние механизмы потоковой обработки на уровне токенов, полная библиотека LangGraph Studio SDK и более старое исключение NodeInterrupt, которое заменило interrupt().
Прерывание и возобновление
Каждый из этих подходов сводится к одному и тому же двухэтапному действию:
1. interrupt() → Graph pauses. State is saved. Control returns to your code.
2. Command(resume=...) → Graph picks up exactly where it paused, with your input.
Всё остальное — это вариации: где вызывается функция interrupt(), какие данные передаются ей и что делается с полученным ответом.
Лексика
Функция interrupt(value) импортируется из модуля langgraph.types и вызывается внутри узла. Она мгновенно останавливает работу графа, сохраняет весь его состояние с помощью checkpointer и передаёт значение value тому, кто запустил процесс. Пауза действительная: никакие потоки не работают в фоновом режиме, никакие ресурсы не удерживаются, поэтому поток может оставаться в состоянии паузы в течение месяцев.
Функция Command(resume=value) передаётся в методы graph.invoke() или graph.stream() для возобновления работы графа, находящегося в состоянии паузы. Значение value становится возвращаемым значением вызова interrupt(), как будто interrupt() была синхронной функцией input(), которая только что получила ответ.
__interrupt__ — это ключ, который появляется в выводе при приостановке выполнения. Проверьте его, чтобы узнать, остановился ли агент и требует ли ввода.
Инструмент проверки состояния (MemorySaver, SqliteSaver, PostgresSaver) является обязательным. Без него нет сохранённого состояния, которое можно было бы приостановить или возобновить. Забывание включения такого инструмента при компиляции графа — самая частая причина сбоев кода HITL у начинающих.
Шаблон 1: утверждение вызова инструмента перед его выполнением
Это далеко самое распространённое применение в реальных проектах. Агент принимает решение вызвать инструмент, имеющий реальные последствия — например, отправить электронное письмо, выполнить запрос к базе данных или оформить заказ, — и человек должен сначала его утвердить.
Представьте себе менеджера, который подписывает чек. Сотрудник (LLM) заполняет его и передаёт, и ничего не происходит, пока менеджер (человек) не поставит подпись. Если менеджер откажет, чек становится недействительным.
Узел обзора:
# ── IMPORTS needed ───────────────────────────────────────────
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
# ── MODULE 4: NODES ───────────────────────────────────────────
def review_tool_call(state: AgentState) -> dict:
"""Pauses execution to let a human approve, edit, or reject a tool call
before it actually runs."""
last_message = state["messages"][-1]
tool_call = last_message.tool_calls[0] # Assume one tool call for simplicity
# interrupt() pauses HERE. The dict below is shown to whoever resumes the graph.
human_response = interrupt({
"question": "Approve this action?",
"tool_name": tool_call["name"],
"tool_args": tool_call["args"],
})
# Execution resumes here, with human_response being whatever
# was passed to Command(resume=...)
if human_response["type"] == "approve":
# Do nothing - let the tool call proceed as-is
return {}
elif human_response["type"] == "edit":
# Replace the tool call's arguments with the human's edits
last_message.tool_calls[0]["args"] = human_response["edited_args"]
return {"messages": [last_message]}
elif human_response["type"] == "reject":
# Short-circuit: skip the tool entirely, tell the LLM why
return {
"messages": [
ToolMessage(
content="User rejected this action.",
tool_call_id=tool_call["id"]
)
]
}
Затем подключите его к графу:
graph_builder = StateGraph(AgentState)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("review_tool_call", review_tool_call)
graph_builder.add_node("tools", tool_node)
graph_builder.set_entry_point("agent")
# Agent decides: call a tool (go through review) or finish
graph_builder.add_conditional_edges(
"agent",
should_continue,
{"tools": "review_tool_call", "__end__": END}
)
graph_builder.add_edge("review_tool_call", "tools")
graph_builder.add_edge("tools", "agent")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
Маршрутизация немного отличается от базового агента: запрос на использование инструмента теперь направляется в функцию review_tool_call перед узлом с инструментами. Между выбором агента и фактической обработкой находится проверяющий, что соответствует этапу совместной подписи перед выплатой средств.
Запустить его и возобновить:
config = {"configurable": {"thread_id": "session-001"}}
# ── First call: runs until interrupt() pauses it ─────────────
result = graph.invoke(
{"messages": [HumanMessage(content="Send an email to the team about the delay")]},
config=config
)
# Check if the graph paused
if "__interrupt__" in result:
interrupt_data = result["__interrupt__"][0].value
print(f"Agent wants to: {interrupt_data['tool_name']}")
print(f"With arguments: {interrupt_data['tool_args']}")
# Show this to a human (in a real app: render a UI button, Slack message, etc.)
user_decision = input("Approve? (yes/no): ")
# ── Resume with the human's decision ──────────────────────
if user_decision.lower() == "yes":
final_result = graph.invoke(
Command(resume={"type": "approve"}),
config=config
)
else:
final_result = graph.invoke(
Command(resume={"type": "reject"}),
config=config
)
print(final_result["messages"][-1].content)
That's the whole pattern. What changes elsewhere is the content passed to interrupt() and what you do with the answer that comes back.
Pattern 2: a mandatory feedback gate
Sometimes you don't want a conditional pause depending on what the LLM chose. You want a pause at a fixed stage, every time. A content-generation agent that always stops for review before publishing is the usual example, regardless of what it wrote.
Think of a quality-control station on an assembly line. Each item stops there, no exceptions. The worker can pass it as is, send it back for rework or adjust it directly.
The feedback node:
def human_feedback(state: AgentState) -> dict:
"""A guaranteed pause point. Shows the human the latest draft
and lets them approve, request changes, or edit directly."""
draft = state["messages"][-1].content
feedback = interrupt({
"question": "Review this draft. Approve, or provide edit instructions.",
"draft": draft,
})
if feedback.get("approved"):
return {} # Pass through unchanged
# Feedback contains revision instructions — feed back to the agent
return {
"messages": [
HumanMessage(content=f"Please revise based on this feedback: {feedback['notes']}")
]
}
А вот прокладка контура:
graph_builder = StateGraph(AgentState)
graph_builder.add_node("writer", writer_node)
graph_builder.add_node("human_feedback", human_feedback)
graph_builder.set_entry_point("writer")
graph_builder.add_edge("writer", "human_feedback")
# After feedback, loop back to writer for revisions, OR end if approved
graph_builder.add_conditional_edges(
"human_feedback",
lambda state: "writer" if state["messages"][-1].content.startswith("Please revise") else "__end__",
{"writer": "writer", "__end__": END}
)
graph = graph_builder.compile(checkpointer=MemorySaver())
Это цикл анализа и корректировки, характерный практически для всех агентов, занимающихся обработкой контента, кодом или исследованиями: генерация, пауза для человека, при необходимости корректировка, снова пауза, повторение до получения одобрения.
Шаблон 3: ручная коррекция состояния
Иногда человек не хочет ни одобрять, ни отклонять результат. Он хочет исправить что-то в состоянии агента перед тем, как он продолжит работу, поскольку агент находится в небольшом отклонении, и повторная попытка будет медленнее, чем его коррекция.
Здесь важны три вызова:
graph.get_state(config)возвращает текущийStateSnapshot, включающий.values(словарь состояния) и.next(узел, который будет выполнен при возобновлении работы).graph.update_state(config, values)записывает новые значения в текущую точку контроля. Он создает новую точку контроля, сохраняя при этом историю, что важно позже для перемещения во времени.as_node="node_name"— это необязательный аргумент для функцииupdate_state, который указывает LangGraph на то, что обновление должно рассматриваться как исходящее от данного узла. LangGraph выбирает следующие рёбра на основе исходного узла предыдущего обновления, поэтому ошибка здесь может привести к тому, что обработка графа возобновится в неожиданном месте.
На практике:
# ── Pause first (using interrupt_before at compile time) ────
graph = graph_builder.compile(
checkpointer=memory,
interrupt_before=["agent"] # Always pause before the agent node runs
)
config = {"configurable": {"thread_id": "session-002"}}
# Run until the breakpoint
result = graph.invoke({"messages": [HumanMessage(content="What's 5 + 5?")]}, config)
# ── Inspect what's about to happen ───────────────────────────
snapshot = graph.get_state(config)
print(f"About to run: {snapshot.next}") # e.g. ('agent',)
print(f"Current messages: {snapshot.values['messages']}")
# ── Directly correct the state ───────────────────────────────
# Suppose the human wants to inject a clarification before the agent responds
graph.update_state(
config,
{"messages": [HumanMessage(content="Actually, I meant 5 * 5, not 5 + 5")]},
)
# ── Resume - agent now sees the corrected message ────────────
final = graph.invoke(None, config) # None = "just resume, no new input"
print(final["messages"][-1].content)
Два аспекта могут ввести людей в заблуждение. Во-первых, параметр interrupt_before=[“node_name”] задаётся при компиляции и всегда приводит к паузе перед указанным узлом при каждой экспекуляции, тогда как функция interrupt() находится внутри узла и вызывает паузу только тогда, когда это требуется логикой. Используйте interrupt_before для простой отладки или гарантированных остановок, а interrupt() — когда пауза зависит от условий выполнения, что характерно для большинства реальных приложений.
Во-вторых, вызов graph.invoke(None, config) означает «не добавлять новых входных данных, продолжить с последней точки контроля». Это отличается от функции Command(resume=...), которая обрабатывает незавершённый вызов interrupt(). После использования interrupt_before вместе с update_state для возобновления работы используйте параметр None. После вызова interrupt() используйте Command(resume=...).
Шаблон 4: путешествие во времени
История контрольных точек функционирует аналогично коммитам в Git. Вы можете просмотреть любое прошлое состояние и от него отделиться, чтобы создать альтернативную временную линию, не нарушая оригинала. Ещё одно сравнение: точки сохранения в видеоигре. Функция get_state_history перечисляет их все, и вы можете загрузить любую из них, а не только самую последнюю.
graph.get_state_history(config)— это генератор, который выдает каждыйStateSnapshot, сохранённый для данного потока, начиная с самого нового. Каждый такой снимок содержит собственныйconfigс уникальнымcheckpoint_id.checkpoint_idнаходится по адресуsnapshot.config[„configurable“][„checkpoint_id“]. Передавайте в функцииinvokeилиupdate_stateконфигурацию, содержащую конкретныйcheckpoint_id, чтобы операция была направлена именно на эту точку, а не на самую последнюю.
graph.invoke(None, config_with_old_checkpoint_id). Это приводит к повторной обработке данных начиная с этой точки. Уже сохранённые чекпоинты для данной ветви используются заново, вместо того чтобы их пересчитывать, хотя логика узлов, находящихся дальше по потоку, выполняется снова.update_state с использованием старого чекпоинта, что приводит к созданию нового чекпоинта и новой временной шкалы наряду с первой.config = {"configurable": {"thread_id": "session-003"}}
# Run a few turns normally...
graph.invoke({"messages": [HumanMessage(content="My name is Sam")]}, config)
graph.invoke({"messages": [HumanMessage(content="What's my name?")]}, config)
# ── Find a past checkpoint ────────────────────────────────────
history = list(graph.get_state_history(config))
for snapshot in history:
print(f"Step {snapshot.metadata['step']}: {len(snapshot.values['messages'])} messages")
# Pick an earlier checkpoint - say, right after "My name is Sam"
target_checkpoint = history[-2] # Adjust index based on what you printed above
target_config = target_checkpoint.config
# ── Branch: update state AT that old checkpoint ──────────────
# This creates a NEW checkpoint that forks from the old one.
# The original conversation history is untouched.
branch_config = graph.update_state(
target_config,
{"messages": [HumanMessage(content="Actually, my name is Alex")]},
)
# ── Continue from the new branch ──────────────────────────────
result = graph.invoke(None, branch_config)
print(result["messages"][-1].content)
# Agent now responds based on the BRANCHED history (name = Alex),
# while the original thread (name = Sam) still exists if you go back to it.
Большинству чат-ботов это не нужно во время работы. Его ценность проявляется в трёх аспектах: отладка — повторение всех действий, произошедших в неудачной попытке, для выявления причины сбоя; оценка — повторный запуск той же конверсации с другим запросом или моделью с последующим сравнением, что является методом A/B-тестирования без необходимости повторения всего процесса; восстановление после ошибки — при сбое на определённом этапе возврат к моменту перед сбоем, устранение причины и продолжение работы вместо полного перезапуска всей конверсации.
Полноценный агент с разрешением на использование инструментов
Вот полная шаблонная структура, объединяющая базовую организацию, память и механизм разрешения на вызов инструментов. Это реалистичная форма производственного агента с системой защиты, основанной на участии человека.
# ============================================================
# LANGGRAPH AGENT WITH HUMAN-IN-THE-LOOP TOOL APPROVAL
# Extends: Part 1 (core structure) + Part 2 (memory)
# ============================================================
# ── MODULE 1: IMPORTS & CONFIGURATION ───────────────────────
from typing import Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage, BaseMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# ── MODULE 2: STATE ──────────────────────────────────────────
class State(MessagesState):
pass # messages field is inherited; add custom fields if needed
# ── MODULE 3: TOOLS ──────────────────────────────────────────
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Send an email. This is a SENSITIVE action requiring human approval."""
return f"Email sent to {to} with subject '{subject}'"
tools = [send_email]
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)
# ── MODULE 4: NODES ──────────────────────────────────────────
def agent_node(state: State) -> dict:
messages = [SystemMessage(content="You are a helpful assistant.")] + state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
def review_tool_call(state: State) -> dict:
"""Pauses for human approval before any tool executes."""
last_message = state["messages"][-1]
tool_call = last_message.tool_calls[0]
decision = interrupt({
"tool_name": tool_call["name"],
"tool_args": tool_call["args"],
})
if decision["type"] == "reject":
return {
"messages": [
ToolMessage(
content="Action rejected by user.",
tool_call_id=tool_call["id"]
)
]
}
if decision["type"] == "edit":
last_message.tool_calls[0]["args"] = decision["edited_args"]
return {"messages": [last_message]}
return {} # approved - pass through unchanged
# ── MODULE 5: ROUTING ─────────────────────────────────────────
def should_continue(state: State) -> Literal["review_tool_call", "__end__"]:
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "review_tool_call"
return "__end__"
def after_review(state: State) -> Literal["tools", "agent"]:
"""If the tool call was rejected, the last message is a ToolMessage -
skip the tools node and go straight back to the agent."""
last_message = state["messages"][-1]
if isinstance(last_message, ToolMessage):
return "agent"
return "tools"
# ── MODULE 6: GRAPH ASSEMBLY ──────────────────────────────────
graph_builder = StateGraph(State)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("review_tool_call", review_tool_call)
graph_builder.add_node("tools", tool_node)
graph_builder.add_edge(START, "agent")
graph_builder.add_conditional_edges(
"agent", should_continue,
{"review_tool_call": "review_tool_call", "__end__": END}
)
graph_builder.add_conditional_edges(
"review_tool_call", after_review,
{"tools": "tools", "agent": "agent"}
)
graph_builder.add_edge("tools", "agent")
graph = graph_builder.compile(checkpointer=MemorySaver())
# ── MODULE 7: ENTRYPOINT ──────────────────────────────────────
if __name__ == "__main__":
config = {"configurable": {"thread_id": "session-001"}}
result = graph.invoke(
{"messages": [HumanMessage(content="Email bob@example.com that the meeting moved to 3pm")]},
config
)
while "__interrupt__" in result:
interrupt_info = result["__interrupt__"][0].value
print(f"\n Approval needed:")
print(f" Tool: {interrupt_info['tool_name']}")
print(f" Args: {interrupt_info['tool_args']}")
choice = input("Approve (a) / Edit (e) / Reject (r)? ").strip().lower()
if choice == "a":
result = graph.invoke(Command(resume={"type": "approve"}), config)
elif choice == "e":
new_to = input("New 'to' address: ")
result = graph.invoke(
Command(resume={
"type": "edit",
"edited_args": {**interrupt_info["tool_args"], "to": new_to}
}),
config
)
else:
result = graph.invoke(Command(resume={"type": "reject"}), config)
print(f"\nAgent: {result['messages'][-1].content}")
Краткий справочник
Пауза и возобновление работы. interrupt(value) прерывает выполнение внутри узла, сохраняет его состояние и передаёт value вызывающему коду, затем позже возвращает значение для возобновления. Command(resume=value) обрабатывает ожидающееся прерывание. „__interrupt__“ — это ключ вывода, сигнализирующий о паузе, а result[„__interrupt__“][0].value показывает то, что было возвращено после паузы. interrupt_before=[„node_name“] всегда вызывает паузу перед указанным узлом.
Состояние редактирования. Метод graph.get_state(config) возвращает объект StateSnapshot (.values, .next, .config). Метод graph.update_state(config, values) добавляет точку контроля, не стирая историю. Параметр as_node влияет на то, какие ребра будут использоваться при возобновлении работы. Метод graph.invoke(None, config) продолжает работу без новых данных входа.
Путешествие во времени. Метод graph.get_state_history(config) отображает снимки состояния в порядке от новых к старым. Параметр checkpoint_id используется для идентификации точки сохранения. Функция воспроизведения продолжает развитие событий согласно старой точке контроля, а создание веток приводит к формированию новой точки.
Какой шаблон подходит
- Инструменты, которые тратят средства, отправляют сообщения или изменяют внешние системы: для утверждения используется метод
interrupt(). Для производственных агентов с реальными побочными эффектами это является обязательным требованием.
interrupt_before в сочетании с update_state.get_state_history.Безопасность — это форма графа
Память, краткое изложение, сохранение данных и человеческий надзор в LangGraph выражаются в виде структуры: узел здесь, ребро там, поле в состоянии. Не существует отдельного режима безопасности, который можно включить. Безопасность — это ещё одна форма, которую может принять граф.
Агенты редко терпят неудачу в реальных условиях из-за неверного решения большой языковой модели, что является ожидаемым явлением и поддаётся устранению. Они терпят неудачу, когда между принятием решения и его последствиями отсутствует контрольная точка. interrupt() и есть такая контрольная точка. Она требует дополнительного узла и дополнительной рёбра, и может стать самым важным узлом в графе.
Благодаря стандартной структуре, способу поддержания доступности и непрерывности диалогов, а также защите всего важного, у вас есть основа для практически любого серьёзного приложения LangGraph. Следующими естественными темами являются оркестрация нескольких агентов с использованием надзирателей и подграфов, а также долгосрочная память, такая как профили пользователей между сессиями; оба этих аспекта основаны на состоянии, узлах и механизмах контрольных точек.
Связанная литература
- MCP для агентов LangGraph: пошаговое руководство — серверы, механизмы передачи данных и адаптеры — Изучение протокола Model Context Protocol с точки зрения ролей хоста, клиента и сервера, работа с сервером FastMCP, различия между стандартным вводом-выводом и протоколом HTTP, использование адаптеров langchain-mcp и распространенные ошибки при внедрении в производство в LangGraph.
- LangGraph: графы, состояние, маршрутизация и паттерны для производственного использования — Как LangGraph представляет агентов в виде циклических графов с типизированным состоянием, функциями-редукторами, точками контроля, механизмами маршрутизации, инструментами и циклами ReAct, а также паттерны для применения в производственных условиях.