LangGraph z udziałem człowieka: funkcja interrupt(), edycja stanów i podróże w czasie
LangGraph z udziałem człowieka: zatwierdzanie wywołań narzędzi za pomocą funkcji interrupt(), dodawanie mechanizmów zwrotnych, ręczna edycja stanu oraz cofanie się w czasie, wraz z przykładami zastosowania.
Źródła: LangGraph Human-in-the-Loop: Pausing, Reviewing, and Rewinding Your Agent
Wszystko poniżej zakłada, że znasz standardową siedmioelementową strukturę pliku LangGraph (importy, stan, narzędzia, węzły, krawędzie, montaż, punkt wejścia) oraz że korzystałeś z narzędzi do tworzenia punktów kontrolnych z użyciem thread_id. Każdy z opisanych tu wzorów opiera się na takim punkcie kontrolnym – bez niego nic z tego nie zadziała.
Dlaczego człowiek musi być obecny w procesie
Gdy tylko model językowy będzie mógł działać, wysyłając e-maile, usuwając pliki, dokonując płatności kartą lub wdrażając kod, może również popełnić kosztowny błąd z prędkością maszyny, gdy nikt nie patrzy. Obecność człowieka w procesie (Human-in-the-Loop, HITL) nie jest dodatkową funkcją dołączaną do agenta – to właśnie taka architektura sprawia, że agenty wyposażone w narzędzia z rzeczywistego świata mogą być bezpiecznie wdrażane, a LangGraph został stworzony z myślą o tym problemie. Jego system punktów kontrolnych istnieje głównie po to, by umożliwić przerwy w działaniu.
Cztery wzory obejmują większość rzeczywistych systemów, wymienione według częstotliwości ich występowania:
- Zatrzymanie się przed ryzykowną akcją za pomocą
interrupt(), co jest obecnym standardem. - Gwarantowany punkt kontrolny w procesie pracy.
- Modyfikacja stanu podczas zatrzymania, aby skorygować agenta przed kontynuacją.
- Podróż w czasie: odtwarzanie i rozgałęzianie się od dowolnego wcześniejszego momentu w rozmowie.
Część elementów została celowo pominęta, ponieważ rzadko występują w rzeczywistych bazach kodu: wewnętrzne mechanizmy przepływu danych na poziomie tokenów, pełne SDK LangGraph Studio oraz starszy wyjątek NodeInterrupt, który zastąpił interrupt().
Przerwanie i kontynuacja
Każdy z tych wzorców sprowadza się do tego samego dwuetapowego działania:
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.
Wszystko inne to wariacje: gdzie wywołuje się interrupt(), jakie dane mu przekazuje się oraz co robi się z odpowiedzią.
Słownictwo
interrupt(value) jest importowany z langgraph.types i wywoływany wewnątrz węzła. Zatrzymuje on graf w tym miejscu, zapisuje cały stan za pomocą checkpointera i przekazuje value osobie, która rozpoczęła działanie. Pauza jest rzeczywista – żaden wątek nie pracuje w tle, a żadne zasoby nie są zajmowane, więc wątek może pozostać w stanie pauzy przez miesiące.
Command(resume=value) trafia do graph.invoke() lub graph.stream(), aby przywrócić do życia graf w stanie pauzy. value staje się wartością zwracaną przez wywołanie interrupt(), jakby interrupt() był synchronicznym input(), który właśnie otrzymał odpowiedź.
__interrupt__ to klucz, który pojawia się w wyniku, gdy wykonywanie zostało wstrzymane. Sprawdź go, aby dowiedzieć się, czy agent się zatrzymał i potrzebuje danych wejściowych.
Checkpointer (MemorySaver, SqliteSaver, PostgresSaver) jest obowiązkowy. Bez niego nie ma zapisanego stanu, który można by wstrzymać lub wznowić. Najczęstszą przyczyną niepowodzeń kodu HITL u początkujących jest zapomnienie o skompilowaniu grafu z takim checkpointerem.
Wzorzec 1: zatwierdzenie wywołania narzędzia przed jego uruchomieniem
To zdecydowanie najczęstsze zastosowanie w środowisku produkcyjnym. Agent decyduje się na wywołanie narzędzia, które ma rzeczywiste skutki, takie jak wysłanie e-maila, zapytanie do bazy danych lub złożenie zamówienia, a osoba musi je najpierw zatwierdzić.
Wyobraź sobie menedżera, który podpisuje czek. Pracownik (LLM) wypełnia go i przekazuje, a nic się nie dzieje, dopóki menedżer (człowiek) nie podpisze. Jeśli menedżer odmówi, czek staje się nieważny.
Węzeł recenzji:
# ── 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"]
)
]
}
Następnie podłącz to do wykresu:
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)
Zasady routingu różnią się nieco od podstawowego agenta: żądanie narzędzia trafia teraz do review_tool_call przed węzłem narzędzi. Między wyborem agenta a faktyczną eksploatacją znajduje się osoba sprawdzająca, co stanowi krok współpodpisu przed wypłatą.
Zapuść to i wznow je:
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']}")
]
}
A podłączenia pętli:
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())
To jest cykl przeglądania i korekty, który występuje w niemal każdym agencie do przeglądania treści, kodu lub badań: generowanie, pauza dla osoby, korekta w razie potrzeby, ponowna pauza, powtarzanie aż do zatwierdzenia.
Wzorzec 3: ręczna korekta stanu
Czasami osoba nie chce niczego zatwierdzać ani odrzucać. Chce naprawić coś w stanie agenta przed kontynuacją, ponieważ agent działał nieco niewłaściwie, a ponowna próba jest wolniejsza niż jego korekta.
Tutaj istotne są trzy wywołania:
graph.get_state(config)zwraca aktualnyStateSnapshot, zawierający.values(słownik stanu) oraz.next(węzeł, który miałby zostać uruchomiony po wznowieniu).graph.update_state(config, values)zapisuje nowe wartości w obecnym punkcie kontrolnym. Tworzy nowy punkt kontrolny i zachowuje historię w niezmienionej formie, co jest ważne później przy podróżach w czasie.as_node="node_name"to opcjonalny argument dlaupdate_state, który informuje LangGraph, że aktualizacja pochodzi od tego węzła. LangGraph wybiera kolejne krawędzie na podstawie węzła źródłowego ostatniej aktualizacji, więc błąd w tym zakresie może spowodować kontynuację przetwarzania grafu w nieoczekiwanym miejscu.
W praktyce:
# ── 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)
Dwie różnice mogą wprowadzić ludzi w błąd. Po pierwsze, interrupt_before=[„node_name”] jest ustawiane podczas kompilacji i zawsze powoduje pauzę przed danym węzłem przy każdym uruchomieniu, natomiast interrupt() znajduje się wewnątrz węzła i powoduje pauzę tylko wtedy, gdy tego wymaga logika. Używaj interrupt_before do prostego debugowania lub gwarantowanych zatrzymań, a interrupt() wtedy, gdy pauza zależy od warunków w czasie wykonywania, co dotyczy większości rzeczywistych zastosowań.
Po drugie, graph.invoke(None, config) oznacza „nie dodawać nowych danych wejściowych, kontynuować od ostatniego punktu kontrolnego”. To różni się od Command(resume=...), które odpowiada na nieskończone interrupt(). Po użyciu interrupt_before w połączeniu z update_state, kontynuuj z None. Po interrupt() kontynuuj z Command(resume=...).
Wzorzec 4: podróż w czasie
Historia punktów kontrolnych funkcjonuje podobnie do commitów w Git. Można przejrzeć dowolny wcześniejszy stan i od niego rozgałęzić się, aby stworzyć alternatywną linię czasu, nie zakłócając oryginału. Inne porównanie: punkty zapisu w grze wideo. Funkcja get_state_history wyświetla je wszystkie, a można załadować dowolny z nich, nie tylko najnowszy.
graph.get_state_history(config)to generator, który zwraca każdyStateSnapshotzapisany dla danej nitki, począwszy od najnowszego. Każdy z nich zawiera własnyconfigz unikalnymcheckpoint_id.checkpoint_idznajduje się w adresiesnapshot.config[„configurable“][„checkpoint_id“]. Przekazując do funkcjiinvokelubupdate_statekonfigurację zawierającą określonycheckpoint_id, operacja będzie dotyczyć właśnie tego punktu, a nie najnowszego.
graph.invoke(None, config_with_old_checkpoint_id). Dzięki temu proces jest ponownie uruchamiany od tego momentu. Już zapisane punkty kontrolne dla danej gałęzi są ponownie wykorzystywane zamiast obliczanych na nowo, chociaż logika węzłów późniejszych od tego momentu jest uruchamiana ponownie.update_state na starym punkcie kontrolnym, co powoduje utworzenie nowego punktu kontrolnego oraz nowej osi czasu obok pierwszej.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.
Większość chatbotów nie potrzebuje tego podczas działania. Jego wartość objawia się w trzech obszarach: debugowanie – odtworzenie dokładnie tego, co wydarzyło się podczas nieudanego uruchomienia, aby znaleźć przyczynę awarii; ocena – ponowne przeprowadzenie tej samej rozmowy z innym promptem lub modelem i porównanie wyników, co stanowi test A/B bez konieczności odtwarzania wszystkiego; przywracanie po błędzie – gdy jakiś element zawodzi w trakcie działania, cofnięcie się do momentu przed nim, naprawa przyczyny i kontynuacja pracy zamiast restartowania całej rozmowy.
Pełny agent z zatwierdzeniem użycia narzędzi
Oto kompletny szablon łączący podstawową strukturę, pamięć oraz możliwość zatwierdzania wywołań narzędzi. To realistyczny model agenta produkcyjnego z ludzkim mechanizmem bezpieczeństwa.
# ============================================================
# 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}")
Szybki przewodnik
Pauzowanie i wznowienie działania. interrupt(value) wstrzymuje działanie wewnątrz węzła, zapisuje stan i przekazuje value wywołującemu, a następnie później zwraca wartość służącą do wznowienia. Command(resume=value) odpowiada na oczekujące przerwanie. "__interrupt__" to klucz wyjściowy sygnalizujący pauzę, a result["__interrupt__"][0].value pokazuje, co zostało ujawnione. interrupt_before=["node_name"] zawsze wstrzymuje działanie przed wskazanym węzłem.
Stan edycji. Funkcja graph.get_state(config) zwraca obiekt StateSnapshot (.values, .next, .config). Funkcja graph.update_state(config, values) dodaje punkt kontrolny bez usuwania historii. Parametr as_node wpływa na to, które krawędzie są śledzone podczas wznowienia działania. Funkcja graph.invoke(None, config) kontynuuje pracę bez nowych danych wejściowych.
Podróż w czasie. Funkcja graph.get_state_history(config) wyświetla snapshoty w kolejności od najnowszych. checkpoint_id służy do identyfikacji punktu zapisu. Funkcja replay kontynuuje przebieg zdarzeń od starego punktu kontrolnego, natomiast funkcja branch tworzy nowy.
Jaki wzorzec pasuje
- Narzędzia, które wydają pieniądze, wysyłają wiadomości lub modyfikują zewnętrzne systemy: zatwierdzenie z użyciem
interrupt(). W przypadku agentów produkcyjnych z rzeczywistymi skutkami ubocznymi należy to traktować jako wymaganie.
interrupt_before w połączeniu z update_state.get_state_history.Bезpieczeństwo to kształt grafu
Pamięć, podsumowanie, trwałość danych oraz nadzór ludzki są wszystkie wyrażane jako struktura w LangGraph: węzeł w jednym miejscu, krawędź w innym, pole w stanie. Nie istnieje żaden specjalny tryb bezpieczeństwa do włączenia – bezpieczeństwo to po prostu kolejny kształt, jaki może przybrać graf.
Agenci rzadko zawodzą w produkcji z powodu błędnej decyzji LLM, co jest spodziewane i nadające się do naprawienia. Zawodzą wtedy, gdy błędna decyzja nie miała punktu kontrolnego pomiędzy podjęciem decyzji a jej konsekwencjami. interrupt() jest właśnie tym punktem kontrolnym. Wymaga jednego dodatkowego węzła i jednego dodatkowego krawędzi, a może być najcenniejszym węzłem w grafie.
Dzięki standardowej strukturze, sposobowi na utrzymanie rozmów w rozsądnych kosztach i ich trwałości oraz sieci bezpieczeństwa otaczającej wszystko, co ma znaczenie, masz fundament pod niemal każdym poważnym zastosowaniem LangGraph. Naturalnymi kolejnymi tematami są orkiestracja wielu agentów z nadzorcami i podgrafami oraz pamięć długoterminowa, tak jak profile użytkowników między sesjami – oba te elementy opierają się na stanach, węzłach i mechanizmach punktów kontrolnych.
Literatura pokrewna
- MCP dla agentów LangGraph, krok po kroku: serwery, mechanizmy transportu i most adapterów — Poznaj protokół Model Context Protocol, od ról hosta, klienta i serwera po serwer FastMCP, różnice między stdio a HTTP, adaptery langchain-mcp oraz typowe problemy w produkcji w LangGraph.
- LangGraph wyjaśniony: grafy, stan, routowanie i wzorce stosowane w produkcji — Jak LangGraph traktuje agenty jako cykliczne grafy z zadanym typem stanu, funkcjami redukcyjnymi, punktami kontrolnymi, mechanizmami routowania, narzędziami oraz pętlami ReAct, a także wzorce do wykorzystania w produkcji.