Interrupts

from langgraph.types import interrupt from  langgraph. types import  interrupt def approval_node(state: State): def  approval_node(state:  State):  # Pause and ask for approval  # Pause and ask for approval approved = interrupt("Do you approve this action?")  approved =  interrupt("Do you approve this action?")  # When you resume, Command(resume=...) returns that value here # When you resume, Command(resume=...) returns that value here return {"approved": approved}  return {"approved":  approved}

from langgraph.types import Command from  langgraph. types import  Command # Initial run - hits the interrupt and pauses# Initial run - hits the interrupt and pauses# thread_id is the persistent pointer (stores a stable ID in production)# thread_id is the persistent pointer (stores a stable ID in production)config = {"configurable": {"thread_id": "thread-1"}} config = {"configurable": {"thread_id": "thread-1"}}result = graph.invoke({"input": "data"}, config=config) result =  graph. invoke({"input":  "data"},  config = config) # Check what was interrupted # Check what was interrupted# __interrupt__ contains the payload that was passed to interrupt()# __interrupt__ contains the payload that was passed to interrupt()print(result["__interrupt__"]) print(result[" __interrupt__ "])# > [Interrupt(value='Do you approve this action?')]# > [Interrupt(value='Do you approve this action?')] # Resume with the human's response # Resume with the human's response# The resume payload becomes the return value of interrupt() inside the node# The resume payload becomes the return value of interrupt() inside the nodegraph.invoke(Command(resume=True), config=config) graph. invoke(Command(resume = True),  config = config)

async for metadata, mode, chunk in graph.astream(async  for  metadata,  mode,  chunk in  graph. astream( initial_input,  initial_input, stream_mode=["messages", "updates"],  stream_mode =["messages",  "updates"], subgraphs=True,  subgraphs = True, config=config  config = config):): if mode == "messages":  if  mode ==  "messages":  # Handle streaming message content  # Handle streaming message content msg, _ = chunk  msg,  _ =  chunk if isinstance(msg, AIMessageChunk) and msg.content:  if  isinstance(msg,  AIMessageChunk)  and  msg. content: # Display content in real-time # Display content in real-time display_streaming_content(msg.content)  display_streaming_content(msg. content)  elif mode == "updates":  elif  mode ==  "updates":  # Check for interrupts  # Check for interrupts if "__interrupt__" in chunk:  if  "__interrupt__"  in  chunk:  # Stop streaming display  # Stop streaming display interrupt_info = chunk["__interrupt__"][0].value  interrupt_info =  chunk[" __interrupt__ "][0]. value  # Handle user input  # Handle user input user_response = get_user_input(interrupt_info)  user_response =  get_user_input(interrupt_info)  # Resume graph with updated input  # Resume graph with updated input initial_input = Command(resume=user_response)  initial_input =  Command(resume = user_response)  break  break  else:  else:  # Track node transitions  # Track node transitions current_node = list(chunk.keys())[0]  current_node =  list(chunk. keys())[0]

from typing import Annotated, TypedDict from  typing import  Annotated,  TypedDict import operator import  operator from langgraph.checkpoint.memory import InMemorySaver from  langgraph. checkpoint. memory import  InMemorySaverfrom langgraph.graph import START, END, StateGraph from  langgraph. graph import  START,  END,  StateGraphfrom langgraph.types import Command, interrupt from  langgraph. types import  Command,  interrupt class State(TypedDict): class  State(TypedDict): vals: Annotated[list[str], operator.add]  vals:  Annotated[list[str],  operator. add] def node_a(state): def  node_a(state): answer = interrupt("question_a")  answer =  interrupt("question_a") return {"vals": [f"a:{answer}"]}  return {"vals": [f"a:{answer} "]} def node_b(state): def  node_b(state): answer = interrupt("question_b")  answer =  interrupt("question_b") return {"vals": [f"b:{answer}"]}  return {"vals": [f"b:{answer} "]} graph = (graph = ( StateGraph(State)  StateGraph(State) .add_node("a", node_a) . add_node("a",  node_a) .add_node("b", node_b) . add_node("b",  node_b) .add_edge(START, "a") . add_edge(START,  "a") .add_edge(START, "b") . add_edge(START,  "b") .add_edge("a", END) . add_edge("a",  END) .add_edge("b", END) . add_edge("b",  END) .compile(checkpointer=InMemorySaver()) . compile(checkpointer = InMemorySaver()))) config = {"configurable": {"thread_id": "1"}} config = {"configurable": {"thread_id":  "1"}} # Step 1: invoke — both parallel nodes hit interrupt() and pause# Step 1: invoke — both parallel nodes hit interrupt() and pauseinterrupted_result = graph.invoke({"vals": []}, config) interrupted_result =  graph. invoke({"vals": []},  config)print(interrupted_result) print(interrupted_result) """ """{{ 'vals': [], 'vals': [], '__interrupt__': [ '__interrupt__': [ Interrupt(value='question_a', id='bd4f3183600f2c41dddafbf8f0f7be7b'), Interrupt(value='question_a', id='bd4f3183600f2c41dddafbf8f0f7be7b'), Interrupt(value='question_b', id='29963e3d3585f0cef025dd0f14323f55') Interrupt(value='question_b', id='29963e3d3585f0cef025dd0f14323f55') ] ]}} """ """ # Step 2: resume all pending interrupts at once# Step 2: resume all pending interrupts at onceresume_map = {resume_map = { i.id: f"answer for {i.value}"  i. id:  f "answer for {i. value} " for i in interrupted_result["__interrupt__"]  for  i in  interrupted_result[" __interrupt__ "]}}result = graph.invoke(Command(resume=resume_map), config) result =  graph. invoke(Command(resume = resume_map),  config) print("Final state:", result) print("Final state:",  result)#> Final state: {'vals': ['a:answer for question_a', 'b:answer for question_b']}#> Final state: {'vals': ['a:answer for question_a', 'b:answer for question_b']}

from typing import Literal from  typing import  Literalfrom langgraph.types import interrupt, Command from  langgraph. types import  interrupt,  Command def approval_node(state: State) -> Command[Literal["proceed", "cancel"]]: def  approval_node(state:  State) ->  Command[Literal[" proceed ",  " cancel "]]: # Pause execution; payload shows up under result["__interrupt__"] # Pause execution; payload shows up under result["__interrupt__"] is_approved = interrupt({ is_approved =  interrupt({ "question": "Do you want to proceed with this action?",  "question":  "Do you want to proceed with this action?", "details": state["action_details"]  "details":  state[" action_details "] }) })  # Route based on the response  # Route based on the response if is_approved:  if  is_approved: return Command(goto="proceed") # Runs after the resume payload is provided  return  Command(goto = "proceed")  # Runs after the resume payload is provided else:  else: return Command(goto="cancel")  return  Command(goto = "cancel")

# To approve # To approvegraph.invoke(Command(resume=True), config=config) graph. invoke(Command(resume = True),  config = config) # To reject # To rejectgraph.invoke(Command(resume=False), config=config) graph. invoke(Command(resume = False),  config = config)

from typing import Literal, Optional, TypedDict from  typing import  Literal,  Optional,  TypedDict from langgraph.checkpoint.memory import MemorySaver from  langgraph. checkpoint. memory import  MemorySaverfrom langgraph.graph import StateGraph, START, END from  langgraph. graph import  StateGraph,  START,  ENDfrom langgraph.types import Command, interrupt from  langgraph. types import  Command,  interrupt class ApprovalState(TypedDict): class  ApprovalState(TypedDict): action_details: str  action_details:  str status: Optional[Literal["pending", "approved", "rejected"]]  status:  Optional[Literal[" pending ",  " approved ",  " rejected "]] def approval_node(state: ApprovalState) -> Command[Literal["proceed", "cancel"]]: def  approval_node(state:  ApprovalState) ->  Command[Literal[" proceed ",  " cancel "]]:  # Expose details so the caller can render them in a UI  # Expose details so the caller can render them in a UI decision = interrupt({ decision =  interrupt({ "question": "Approve this action?",  "question":  "Approve this action?", "details": state["action_details"],  "details":  state[" action_details "], }) })  # Route to the appropriate node after resume  # Route to the appropriate node after resume return Command(goto="proceed" if decision else "cancel")  return  Command(goto = "proceed"  if  decision else  "cancel") def proceed_node(state: ApprovalState): def  proceed_node(state:  ApprovalState): return {"status": "approved"}  return {"status":  "approved"} def cancel_node(state: ApprovalState): def  cancel_node(state:  ApprovalState): return {"status": "rejected"}  return {"status":  "rejected"} builder = StateGraph(ApprovalState) builder =  StateGraph(ApprovalState)builder.add_node("approval", approval_node) builder. add_node("approval",  approval_node)builder.add_node("proceed", proceed_node) builder. add_node("proceed",  proceed_node)builder.add_node("cancel", cancel_node) builder. add_node("cancel",  cancel_node)builder.add_edge(START, "approval") builder. add_edge(START,  "approval")builder.add_edge("proceed", END) builder. add_edge("proceed",  END)builder.add_edge("cancel", END) builder. add_edge("cancel",  END) # Use a more durable checkpointer in production # Use a more durable checkpointer in productioncheckpointer = MemorySaver() checkpointer =  MemorySaver()graph = builder.compile(checkpointer=checkpointer) graph =  builder. compile(checkpointer = checkpointer) config = {"configurable": {"thread_id": "approval-123"}} config = {"configurable": {"thread_id": "approval-123"}}initial = graph.invoke(initial =  graph. invoke( {"action_details": "Transfer $500", "status": "pending"}, {"action_details":  "Transfer $500",  "status":  "pending"}, config=config,  config = config,))print(initial["__interrupt__"]) # -> [Interrupt(value={'question': ..., 'details': ...})] print(initial[" __interrupt__ "]) # -> [Interrupt(value={'question': ..., 'details': ...})] # Resume with the decision; True routes to proceed, False to cancel# Resume with the decision; True routes to proceed, False to cancelresumed = graph.invoke(Command(resume=True), config=config) resumed =  graph. invoke(Command(resume = True),  config = config)print(resumed["status"]) # -> "approved" print(resumed[" status "]) # -> "approved"

from langgraph.types import interrupt from  langgraph. types import  interrupt def review_node(state: State): def  review_node(state:  State): # Pause and show the current content for review (surfaces in result["__interrupt__"]) # Pause and show the current content for review (surfaces in result["__interrupt__"]) edited_content = interrupt({ edited_content =  interrupt({ "instruction": "Review and edit this content",  "instruction":  "Review and edit this content", "content": state["generated_text"]  "content":  state[" generated_text "] }) })  # Update the state with the edited version  # Update the state with the edited version return {"generated_text": edited_content}  return {"generated_text":  edited_content}

graph.invoke(graph. invoke( Command(resume="The edited and improved text"), # Value becomes the return from interrupt()  Command(resume = "The edited and improved text"), # Value becomes the return from interrupt() config=config  config = config))

import sqlite3 import  sqlite3 from typing import TypedDict from  typing import  TypedDict from langgraph.checkpoint.memory import MemorySaver from  langgraph. checkpoint. memory import  MemorySaverfrom langgraph.graph import StateGraph, START, END from  langgraph. graph import  StateGraph,  START,  ENDfrom langgraph.types import Command, interrupt from  langgraph. types import  Command,  interrupt class ReviewState(TypedDict): class  ReviewState(TypedDict): generated_text: str  generated_text:  str def review_node(state: ReviewState): def  review_node(state:  ReviewState):  # Ask a reviewer to edit the generated content  # Ask a reviewer to edit the generated content updated = interrupt({ updated =  interrupt({ "instruction": "Review and edit this content",  "instruction":  "Review and edit this content", "content": state["generated_text"],  "content":  state[" generated_text "], }) }) return {"generated_text": updated}  return {"generated_text":  updated} builder = StateGraph(ReviewState) builder =  StateGraph(ReviewState)builder.add_node("review", review_node) builder. add_node("review",  review_node)builder.add_edge(START, "review") builder. add_edge(START,  "review")builder.add_edge("review", END) builder. add_edge("review",  END) checkpointer = MemorySaver() checkpointer =  MemorySaver()graph = builder.compile(checkpointer=checkpointer) graph =  builder. compile(checkpointer = checkpointer) config = {"configurable": {"thread_id": "review-42"}} config = {"configurable": {"thread_id": "review-42"}}initial = graph.invoke({"generated_text": "Initial draft"}, config=config) initial =  graph. invoke({"generated_text":  "Initial draft"},  config = config)print(initial["__interrupt__"]) # -> [Interrupt(value={'instruction': ..., 'content': ...})] print(initial[" __interrupt__ "]) # -> [Interrupt(value={'instruction': ..., 'content': ...})] # Resume with the edited text from the reviewer # Resume with the edited text from the reviewerfinal_state = graph.invoke(final_state =  graph. invoke( Command(resume="Improved draft after review"),  Command(resume = "Improved draft after review"), config=config,  config = config,))print(final_state["generated_text"]) # -> "Improved draft after review" print(final_state[" generated_text "]) # -> "Improved draft after review"

from langchain.tools import tool from  langchain. tools import  toolfrom langgraph.types import interrupt from  langgraph. types import  interrupt @tool @tooldef send_email(to: str, subject: str, body: str): def  send_email(to:  str,  subject:  str,  body:  str): """Send an email to a recipient.""" """Send an email to a recipient."""  # Pause before sending; payload surfaces in result["__interrupt__"] # Pause before sending; payload surfaces in result["__interrupt__"] response = interrupt({ response =  interrupt({ "action": "send_email",  "action":  "send_email", "to": to,  "to":  to, "subject": subject,  "subject":  subject, "body": body,  "body":  body, "message": "Approve sending this email?"  "message":  "Approve sending this email?" }) })  if response.get("action") == "approve":  if  response. get("action") ==  "approve":  # Resume value can override inputs before executing  # Resume value can override inputs before executing final_to = response.get("to", to)  final_to =  response. get("to",  to) final_subject = response.get("subject", subject)  final_subject =  response. get("subject",  subject) final_body = response.get("body", body)  final_body =  response. get("body",  body) return f"Email sent to {final_to} with subject '{final_subject}'"  return  f "Email sent to {final_to}  with subject '{final_subject} '"  return "Email cancelled by user"  return  "Email cancelled by user"

import sqlite3 import  sqlite3 from typing import TypedDict from  typing import  TypedDict from langchain.tools import tool from  langchain. tools import  tool from langchain_anthropic import ChatAnthropic from  langchain_anthropic import  ChatAnthropicfrom langgraph.checkpoint.sqlite import SqliteSaver from  langgraph. checkpoint. sqlite import  SqliteSaverfrom langgraph.graph import StateGraph, START, END from  langgraph. graph import  StateGraph,  START,  ENDfrom langgraph.types import Command, interrupt from  langgraph. types import  Command,  interrupt class AgentState(TypedDict): class  AgentState(TypedDict): messages: list[dict]  messages:  list[dict]  @tool @tooldef send_email(to: str, subject: str, body: str): def  send_email(to:  str,  subject:  str,  body:  str): """Send an email to a recipient.""" """Send an email to a recipient."""  # Pause before sending; payload surfaces in result["__interrupt__"] # Pause before sending; payload surfaces in result["__interrupt__"] response = interrupt({ response =  interrupt({ "action": "send_email",  "action":  "send_email", "to": to,  "to":  to, "subject": subject,  "subject":  subject, "body": body,  "body":  body, "message": "Approve sending this email?",  "message":  "Approve sending this email?", }) })  if response.get("action") == "approve":  if  response. get("action") ==  "approve": final_to = response.get("to", to)  final_to =  response. get("to",  to) final_subject = response.get("subject", subject)  final_subject =  response. get("subject",  subject) final_body = response.get("body", body)  final_body =  response. get("body",  body)  # Actually send the email (your implementation here) # Actually send the email (your implementation here) print(f"[send_email] to={final_to} subject={final_subject} body={final_body}")  print(f"[send_email] to={final_to} subject={final_subject} body={final_body} ") return f"Email sent to {final_to}"  return  f "Email sent to {final_to} "  return "Email cancelled by user"  return  "Email cancelled by user" model = ChatAnthropic(model="claude-sonnet-4-6").bind_tools([send_email]) model =  ChatAnthropic(model ="claude-sonnet-4-6"). bind_tools([send_email]) def agent_node(state: AgentState): def  agent_node(state:  AgentState): # LLM may decide to call the tool; interrupt pauses before sending # LLM may decide to call the tool; interrupt pauses before sending result = model.invoke(state["messages"])  result =  model. invoke(state[" messages "]) return {"messages": state["messages"] + [result]}  return {"messages":  state[" messages "] + [result]} builder = StateGraph(AgentState) builder =  StateGraph(AgentState)builder.add_node("agent", agent_node) builder. add_node("agent",  agent_node)builder.add_edge(START, "agent") builder. add_edge(START,  "agent")builder.add_edge("agent", END) builder. add_edge("agent",  END) checkpointer = SqliteSaver(sqlite3.connect("tool-approval.db")) checkpointer =  SqliteSaver(sqlite3. connect("tool-approval.db"))graph = builder.compile(checkpointer=checkpointer) graph =  builder. compile(checkpointer = checkpointer) config = {"configurable": {"thread_id": "email-workflow"}} config = {"configurable": {"thread_id": "email-workflow"}}initial = graph.invoke(initial =  graph. invoke( { { "messages": [ "messages": [ {"role": "user", "content": "Send an email to alice@example.com about the meeting"} {"role":  "user",  "content": "Send an email to alice@example.com about the meeting"} ] ] }, }, config=config,  config = config,))print(initial["__interrupt__"]) # -> [Interrupt(value={'action': 'send_email', ...})] print(initial[" __interrupt__ "]) # -> [Interrupt(value={'action': 'send_email', ...})] # Resume with approval and optionally edited arguments # Resume with approval and optionally edited argumentsresumed = graph.invoke(resumed =  graph. invoke( Command(resume={"action": "approve", "subject": "Updated subject"}),  Command(resume ={"action":  "approve",  "subject":  "Updated subject"}), config=config,  config = config,))print(resumed["messages"][-1]) # -> Tool result returned by send_email print(resumed[" messages "][- 1]) # -> Tool result returned by send_email

from langgraph.types import interrupt from  langgraph. types import  interrupt def get_age_node(state: State): def  get_age_node(state:  State): prompt = "What is your age?"  prompt =  "What is your age?"  while True:  while  True: answer = interrupt(prompt) # payload surfaces in result["__interrupt__"]  answer =  interrupt(prompt) # payload surfaces in result["__interrupt__"]  # Validate the input  # Validate the input if isinstance(answer, int) and answer > 0:  if  isinstance(answer,  int)  and  answer >  0: # Valid input - continue # Valid input - continue  break  break else:  else: # Invalid input - ask again with a more specific prompt # Invalid input - ask again with a more specific prompt prompt = f"'{answer}' is not a valid age. Please enter a positive number."  prompt =  f "'{answer}' is not a valid age. Please enter a positive number."  return {"age": answer}  return {"age":  answer}

import sqlite3 import  sqlite3 from typing import TypedDict from  typing import  TypedDict from langgraph.checkpoint.sqlite import SqliteSaver from  langgraph. checkpoint. sqlite import  SqliteSaverfrom langgraph.graph import StateGraph, START, END from  langgraph. graph import  StateGraph,  START,  ENDfrom langgraph.types import Command, interrupt from  langgraph. types import  Command,  interrupt class FormState(TypedDict): class  FormState(TypedDict): age: int | None  age:  int  |  None def get_age_node(state: FormState): def  get_age_node(state:  FormState): prompt = "What is your age?"  prompt =  "What is your age?"  while True:  while  True: answer = interrupt(prompt) # payload surfaces in result["__interrupt__"]  answer =  interrupt(prompt) # payload surfaces in result["__interrupt__"]  if isinstance(answer, int) and answer > 0:  if  isinstance(answer,  int)  and  answer >  0: return {"age": answer}  return {"age":  answer}  prompt = f"'{answer}' is not a valid age. Please enter a positive number."  prompt =  f "'{answer}' is not a valid age. Please enter a positive number." builder = StateGraph(FormState) builder =  StateGraph(FormState)builder.add_node("collect_age", get_age_node) builder. add_node("collect_age",  get_age_node)builder.add_edge(START, "collect_age") builder. add_edge(START,  "collect_age")builder.add_edge("collect_age", END) builder. add_edge("collect_age",  END) checkpointer = SqliteSaver(sqlite3.connect("forms.db")) checkpointer =  SqliteSaver(sqlite3. connect("forms.db"))graph = builder.compile(checkpointer=checkpointer) graph =  builder. compile(checkpointer = checkpointer) config = {"configurable": {"thread_id": "form-1"}} config = {"configurable": {"thread_id": "form-1"}}first = graph.invoke({"age": None}, config=config) first =  graph. invoke({"age":  None},  config = config)print(first["__interrupt__"]) # -> [Interrupt(value='What is your age?', ...)] print(first[" __interrupt__ "]) # -> [Interrupt(value='What is your age?', ...)] # Provide invalid data; the node re-prompts# Provide invalid data; the node re-promptsretry = graph.invoke(Command(resume="thirty"), config=config) retry =  graph. invoke(Command(resume = "thirty"),  config = config)print(retry["__interrupt__"]) # -> [Interrupt(value="'thirty' is not a valid age...", ...)] print(retry[" __interrupt__ "]) # -> [Interrupt(value="'thirty' is not a valid age...", ...)] # Provide valid data; loop exits and state updates# Provide valid data; loop exits and state updatesfinal = graph.invoke(Command(resume=30), config=config) final =  graph. invoke(Command(resume = 30),  config = config)print(final["age"]) # -> 30 print(final[" age "]) # -> 30

def node_a(state: State): def  node_a(state:  State): # ✅ Good: interrupting first, then handling # ✅ Good: interrupting first, then handling  # error conditions separately  # error conditions separately interrupt("What's your name?")  interrupt("What's your name?") try:  try: fetch_data() # This can fail  fetch_data()  # This can fail except Exception as e:  except  Exception  as  e: print(e)  print(e)  return state  return  state

def node_a(state: State): def  node_a(state:  State): # ❌ Bad: wrapping interrupt in bare try/except # ❌ Bad: wrapping interrupt in bare try/except  # will catch the interrupt exception  # will catch the interrupt exception try:  try: interrupt("What's your name?")  interrupt("What's your name?") except Exception as e:  except  Exception  as  e: print(e)  print(e)  return state  return  state

def node_a(state: State): def  node_a(state:  State): # ✅ Good: interrupt calls happen in the same order every time # ✅ Good: interrupt calls happen in the same order every time name = interrupt("What's your name?")  name =  interrupt("What's your name?") age = interrupt("What's your age?")  age =  interrupt("What's your age?") city = interrupt("What's your city?")  city =  interrupt("What's your city?")  return { return { "name": name,  "name":  name, "age": age,  "age":  age, "city": city  "city":  city } }

def node_a(state: State): def  node_a(state:  State): # ❌ Bad: conditionally skipping interrupts changes the order # ❌ Bad: conditionally skipping interrupts changes the order name = interrupt("What's your name?")  name =  interrupt("What's your name?")  # On first run, this might skip the interrupt # On first run, this might skip the interrupt # On resume, it might not skip it - causing index mismatch # On resume, it might not skip it - causing index mismatch if state.get("needs_age"):  if  state. get("needs_age"): age = interrupt("What's your age?")  age =  interrupt("What's your age?")  city = interrupt("What's your city?")  city =  interrupt("What's your city?")  return {"name": name, "city": city}  return {"name":  name,  "city":  city}

def node_a(state: State): def  node_a(state:  State): # ✅ Good: passing simple types that are serializable # ✅ Good: passing simple types that are serializable name = interrupt("What's your name?")  name =  interrupt("What's your name?") count = interrupt(42)  count =  interrupt(42) approved = interrupt(True)  approved =  interrupt(True)  return {"name": name, "count": count, "approved": approved}  return {"name":  name,  "count":  count,  "approved":  approved}

def validate_input(value): def  validate_input(value): return len(value) > 0  return  len(value) >  0 def node_a(state: State): def  node_a(state:  State): # ❌ Bad: passing a function to interrupt # ❌ Bad: passing a function to interrupt  # The function cannot be serialized  # The function cannot be serialized response = interrupt({ response =  interrupt({ "question": "What's your name?",  "question":  "What's your name?", "validator": validate_input # This will fail  "validator":  validate_input # This will fail }) }) return {"name": response}  return {"name":  response}

def node_a(state: State): def  node_a(state:  State): # ✅ Good: using upsert operation which is idempotent # ✅ Good: using upsert operation which is idempotent  # Running this multiple times will have the same result  # Running this multiple times will have the same result db.upsert_user( db. upsert_user( user_id=state["user_id"],  user_id = state[" user_id "], status="pending_approval"  status = "pending_approval" ) )  approved = interrupt("Approve this change?")  approved =  interrupt("Approve this change?")  return {"approved": approved}  return {"approved":  approved}

def node_a(state: State): def  node_a(state:  State): # ❌ Bad: creating a new record before interrupt # ❌ Bad: creating a new record before interrupt  # This will create duplicate records on each resume  # This will create duplicate records on each resume audit_id = db.create_audit_log({ audit_id =  db. create_audit_log({ "user_id": state["user_id"],  "user_id":  state[" user_id "], "action": "pending_approval",  "action":  "pending_approval", "timestamp": datetime.now()  "timestamp":  datetime. now() }) })  approved = interrupt("Approve this change?")  approved =  interrupt("Approve this change?")  return {"approved": approved, "audit_id": audit_id}  return {"approved":  approved,  "audit_id":  audit_id}

def node_in_parent_graph(state: State): def  node_in_parent_graph(state:  State): some_code() # <-- This will re-execute when resumed  some_code() # <-- This will re-execute when resumed # Invoke a subgraph as a function. # Invoke a subgraph as a function. # The subgraph contains an `interrupt` call. # The subgraph contains an `interrupt` call. subgraph_result = subgraph.invoke(some_input)  subgraph_result =  subgraph. invoke(some_input) # ... # ... def node_in_subgraph(state: State): def  node_in_subgraph(state:  State): some_other_code() # <-- This will also re-execute when resumed  some_other_code() # <-- This will also re-execute when resumed result = interrupt("What's your name?")  result =  interrupt("What's your name?") # ... # ...

graph = builder.compile(graph =  builder. compile( interrupt_before=["node_a"],  interrupt_before =["node_a"], interrupt_after=["node_b", "node_c"],  interrupt_after =["node_b",  "node_c"], checkpointer=checkpointer,  checkpointer = checkpointer,)) # Pass a thread ID to the graph # Pass a thread ID to the graphconfig = {config = { "configurable": { "configurable": { "thread_id": "some_thread"  "thread_id":  "some_thread" } }}} # Run the graph until the breakpoint # Run the graph until the breakpointgraph.invoke(inputs, config=config) graph. invoke(inputs,  config = config) # Resume the graph # Resume the graphgraph.invoke(None, config=config) graph. invoke(None,  config = config)

config = {config = { "configurable": { "configurable": { "thread_id": "some_thread"  "thread_id":  "some_thread" } }}} # Run the graph until the breakpoint # Run the graph until the breakpointgraph.invoke(graph. invoke( inputs,  inputs, interrupt_before=["node_a"],  interrupt_before =["node_a"], interrupt_after=["node_b", "node_c"],  interrupt_after =["node_b",  "node_c"], config=config,  config = config,)) # Resume the graph # Resume the graphgraph.invoke(None, config=config) graph. invoke(None,  config = config)