Чалавек у цыкле ў LangGraph: interrupt(), рэдагаванне стану і падарожжа ў часе
Чалавек у цыкле ў LangGraph: пацвярджэнне выклікаў інструментаў праз interrupt(), фільтры зваротнай сувязі, ручное рэдагаванне стану і адкат праз time travel — з рабочымі прыкладамі.
Крыніцы: 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)
Гэта ўсё схэма. Тое, што зміняецца ў іншых месцах, — гэта контэнт, які перэдаецца функцыі interrupt(), і тое, што вы робіце з адпаведным адпаведзеннем.
Схэма 2: обавязковы элемент зворотнага зв’язку
Інодзе вам не трэба умовная пауза, залежная ад таго, калі LLM прыняў рашэнне. Вы хочаце паузу на пэўным этапе ў кожны раз. Тыповым прыкладам ёсць агент для стварэння контэнту, які завжды зупіняецца для перагляду пры публікацыі, незалежна ад таго, што ён напісаў.
Уявіце станцію контролю якосты на лініі зборкі. Кожны віроб зупіняецца там, без выключэнняў. Рабочы можа прыняць яго такім, які ён ёсць, апрацаваць яго занова або змяніць яго безпосерэдна.
Узел зворатнага зв’язку:
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 ідэнтыфікуе пункт перапісву. Паўтарэнне дзеянняў продаўляе хроналагію з старога пункта перапісву, а разгалужэння стварае новы.
Каторы шаблон падходзіць
- Інструменты, якія вітрачаюць грошы, адправляюць паведамленні або зміняюць зовнішнія сістэмы: затверджэнне за дапамою
interrupt(). Для агентаў у працэсе, якія маюць рэальныя парагонныя эфекты, гэта ўваходзіць у прынцыпавыя вакалідаты.
interrupt_before плюс update_state.get_state_history.Безпека — гэта форма графа
Памяць, стварэнне апісу, стойкасць і людская наглядчысць усе выражаюцца як структура ў LangGraph: вузел тут, рэштка там, поле ў стане. Не існуе спецыяльнага рэжыму безпекі, які трэба увімкнуць. Безпека — гэта ўсё тая ж форма, якую можа прыняць граф.
Агенты рэдка калі збываюцца пад час выканання задач, таму што LLM прыняў неправільнае рашэння, што ўсё ж такі ўскладненне, якое можна выправіць. Яны збываюцца тады, калі неправільнае рашэння не мела пункта контролю між самым рашэннем і яго наследкамі. interrupt() — гэты пункт контролю. Ён выклікае неабходнасць у вікорыстоўванні аднаго дадатковага вузла і аднай дадатковай грані, і можа стаць найцэннейшым вузлам у графе.
За дапамогою стандартнай структуры, якая дазволяе падтрымваць дыялогі за прыемнымі цэнамі і забезпечвае ўтрыманне іх, а таксама за дапамогою механізма безпекі для всьго важлівага, утвараецца основа практычна для кожнай серйознай прыменення LangGraph. Наступныя важлівыя тэмы — кількавузловая оркестрацыя з наглядачамі і падграфамі, а таксама дзейнае памяць, такая як прафілі корыстувачаў між сесіямі; обе гэтыя тэмы базуюцца на стане, вузлах і механізмах пунктавага контролю.
Супаўзяныя матэрыялы
- MCP для агентаў LangGraph, па шагах: серверы, срэдства перадачы дадзейнаў і мосты-адаптэры — Вы дазнаецеся пра протакол Model Context Protocol, які выкорыстоўваецца з роляй хоста, кліента і сервера, а таксама пра сервер FastMCP, разлік межы stdio і HTTP, адаптэры langchain-mcp і распалохі пад час працы у продакшэне ў LangGraph.
- LangGraph: графы, стан, маршрутызацыя і шаблоны для продакшэна — Як LangGraph адбірае агентаў у вачынку цыклічных графаў з тыпаваным станам, рэдуктарамі, пунктамі перапыткі, маршрутызацыяй, інструментамі і цыкламі ReAct, а таксама шаблоны для ўжыцця ў продакшэне.