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()— це поточний стандарт. - Гарантована точка перевірки у робочому процесі.
- Редагування стану під час зупинки для корекції агента перед його продовженням.
- „Подорож у часі“: повторна обробка та розгалуження від будь-якої попередньої точки розмови.
Деякі елементи навмисно не наведені, оскільки вони рідко зустрічаються у реальних кодових базах: внутрішні механізми потокової обробки на рівні токенів, повний SDK LangGraph Studio та старіша виняткова ситуація 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“]. Передайте конфігурацію, яка містить конкретнийcheckpoint_id, до функційinvokeабоupdate_state, і операція буде спрямована саме на цю точку, а не на найновішу.
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 використовується для ідентифікації точки збереження. Функція Replay продовжує виконання з попередньої контрольної точки, а створення гілок призводить до формування нової.
Яка схема підходить
- Інструменти, які витрачають гроші, надсилають повідомлення чи змінюють зовнішні системи: схвалення з використанням
interrupt(). Для продакшн-агентів із реальними побічними ефектами це є обов’язковим.
interrupt_before разом із update_state.get_state_history.Безпека — це форма графа
Пам’ять, узагальнення, зберігання даних та людський нагляд усі виражаються у структурі LangGraph: вузол тут, ребро там, поле у стані. Не існує окремого режиму безпеки, який можна увімкнути. Безпека — це ще одна форма, яку може набути граф.
Агенти рідко зазнають невдач у реальних умовах через помилкове рішення ШІ, що є очікуваним та піддається виправленню. Вони зазнають невдач, коли помилкове рішення не мало контрольної точки між прийняттям рішення та його наслідками. interrupt() є саме такою контрольною точкою. Він потребує одного додаткового вузла та однієї додаткової грани, і може бути найціннішим вузлом у графі.
За допомогою стандартної структури, яка дозволяє підтримувати розмови за доступною ціною та зберігати їх, а також захисного механізму для всього важливого, у вас є основа майже для кожного серйозного застосунку LangGraph. Наступними природними темами є оркестрація багатьох агентів із керівниками та підграфами, а також довгострокова пам’ять, наприклад профілі користувачів між сеансами, обидві з яких ґрунтуються на стані, вузлах та контрольних точках.
Пов’язана література
- MCP для агентів LangGraph, крок за кроком: сервери, механізми передачі даних та містки адаптерів — Дізнайтеся про протокол Model Context Protocol з точки зору ролей хоста, клієнта та сервера, про сервер FastMCP, відмінності між stdio та HTTP, адаптери langchain-mcp та поширені проблеми під час роботи в LangGraph.
- LangGraph пояснено: графи, стан, маршрутизація та шаблони для продакшну — Як LangGraph моделює агентів у вигляді циклічних графів із типованим станом, функціями-редукторами, точками перевірки, механізмами маршрутизації, інструментами та циклами ReAct, а також шаблони для використання в продакшні.