What is the right way to conditionally interrupt

OSS Product Help LangGraph
1

The langgraph manual said " Do not conditionally skip interrupt calls within a node", but I do have some situations in which I need to conditionally interrupt. For example, I have a very complex tool

def generate_elementary_steps(self,
                           ntimes:int = -1,
                           max_natoms_tot:int = -1, 
                           min_natoms_tot:int = 0, 
                           max_natoms:Dict[str, int] = None, 
                           min_natoms:Dict[str, int] = None, 
                           forbidden_pattern: List[Tuple[str, str]] = None):
        '''
        Generate reactions. 
        Args:
            ntimes: The number of times the elementary steps is sampled. Here, "sample" means use BFS and DFS to search elementary  steps. If `ntimes > 0`, use BFS firstly to get the shortest path and the remaining attempts will be made using DFS (DFS does not always guarantee to find a path, so multiple attempts are necessary). If `ntimes = -1` (default), a BFS-like method will be used to find all possible elementary steps.
            max_natoms_tot: Maximum number of atoms in the intermediates. `-1` means no limitation.
            min_natoms_tot: Minimum number of atoms in the intermediates.
            max_natoms: A `dict` used to limit the maximum number of atoms of each element in the intermediates. If omitted, a default value is calculated from the final products.
            min_natoms: A `dict` used to limit the minimum number of atoms of each element in the intermediates.
            forbidden_pattern: Substructures of intermediates that are prohibited. For example, use `('H,O,C,O,H','0-1,1-2,2-3,3-4')` to represent geminal diol. If omitted, a list of default patterns are generated.
        '''

If the user provided enough information in the initial prompt, then LLM can determine these arguments. However, if they did’t I should interrupt and ask them for the arguments, so I would write something like this


# I will wap this function in a global class, this class stores the objects used by the tool `generate_elementary_steps`
@tool
def generate_elementary_steps(...):
    if max_natoms is None and forbidden_pattern is None:
        response = interrupt({"message": "Ask for arguments",
                                                  "source": "tool.generate_elementary_steps", ...})
        # The user may just click yes, in this case I will return a prompt to the LLM
         if response == "yes":
             return "User ask us to determine the arguments by ourself, let's investigate this chemical system and determine the args."
         else: # The response contains the arguments we need 
              pass

Can I write interrupt in the way? There is another way, I build a interrupt_tool tool and ask the agent to call it when interruption is required:

@tool
def interrupt_tool(message: Dict[str, Any]):
    '''
    Call this tool if interruption is need. For example, ask for the arguments from the user.
    '''
    response = interrupt(message)

But I don’t think this is reliable, although it seems more clean and idempotent by default (only interrupt, no other operations).

2

hi @MSJavaScript

Yes, your first version is fine. The rule "do not conditionally skip interrupt calls” is about conditions that can come out differently between the first run and the resume. On resume, the tool runs again from the start, and resume values are matched to interrupt() calls by position (see interrupt() in langgraph/types.py). Your condition only depends on the tool-call arguments. Those are already saved in the checkpoint as part of the AIMessage, so the replay always takes the same branch. Also, create_agent sends every tool call as a separate Send task (factory.py: [Send(“tools”, [tool_call]) for tool_call in pending_tool_calls]), so each call has its own interrupt counter. And with a single interrupt there’s nothing that can get out of order.

A few things to watch out for, though:

  1. The "yes” branch can loop forever. If you return "determine the arguments yourself”, the LLM may call the tool again with max_natoms=None and forbidden_pattern=None. That triggers the interrupt again, and so on. It’s better to compute the defaults inside the tool and run it right away (your docstring already promises defaults). Another option is an explicit flag like auto_defaults=True that skips the interrupt.
  2. Don’t base the condition on the mutable state of your global class. In-memory objects aren’t checkpointed and can change or disappear between the pause and the resume. Use only the args, runtime.state or other checkpointed data.
  3. Everything before interrupt() runs twice, so put the expensive BFS/DFS after it.
  4. Don’t wrap interrupt() in a broad try/except Exception (in the tool or in a wrap_tool_call middleware). It raises GraphInterrupt, and catching it breaks the pause.
  5. Validate the resume value (a "yes” string or a dict of args) and merge it over the original args.

There’s also a built-in way that keeps the tool itself free of interrupts: HumanInTheLoopMiddleware with a when predicate (langchain>=1.3.3):

def missing_args(request):
    a = request.tool_call["args"]
    return a.get("max_natoms") is None and a.get("forbidden_pattern") is None

HumanInTheLoopMiddleware(interrupt_on={
    "generate_elementary_steps": {
        "allowed_decisions": ["approve", "edit", "respond"],
        "when": missing_args,
    }
})

Your three cases map directly onto its decisions. "approve” means run with defaults, and "edit” means the user supplies the args. "respond” skips the tool and returns your message to the LLM as the ToolMessage, e.g. "investigate the system and pass explicit values”. Docs: Human-in-the-loop - Docs by LangChain

About interrupt_tool: it’s a legitimate pattern. Deep Agents has an ask_user tool that calls interrupt() inside the tool, and the HITL middleware has the "respond” decision for exactly this kind of tool. But you’re right that it’s not reliable for hard requirements, because the LLM decides whether to call it and models often guess missing args instead. I’d use ask_user for open-ended clarification, and a deterministic check (in the tool or via when) for arguments that must never be defaulted silently.

Docs on the interrupt rules: Interrupts - Docs by LangChain

3

One more caveat: models often fill optional args with guessed values even when the user said nothing, so an „is None” check can silently pass. If that happens, add an explicit arg like constraints_confirmed: bool and tell the model in the docstring to set it only when the user gave the limits or explicitly asked it to choose them. Then gate on that (in the tool or in the when predicate). The model can still set it on its own, so a combination of both checks is the safest.

4

My case is a little complex, I have a ReactionSystem object, it has members with types of numpy.ndarry, ase.Atoms and rdkit, none of them is serializable. ReactionSystem is stateful but has a short lifetime, the workflow is: (1) Create ReactionSystem. (2) Set constraints on the reaction network then generate. (3) Combine and delete some reactions. So I need a layout like the following:

# ReactionSystem api are organized as plain and flat tools
# We don't wrap them as a miniwork flow because if the user's don't want to 
# set some options, then we still need to tell the main agent want we want.
# Here we just wrap all the apis as tools, let the main agent read the tool schema
class ReactionSystemToolSets:
    def __init__(self):
    # First key is thread_id, Second key is the name of reaction system, such as CH4+CO2
        self.reactions_systems:Dict[str, Dict[str, ReactionSystem]] = None 
    def create():
    @tool
    def create_reaction_system(system_name, runtime):
        # get thread_id from runtime object
    @tool
    def generate_reactions(system_name, runtime):
        # ... interrupt here, if necessary
    # other tools
    return [create_reaction_system, generate_reactions, ... ]
5

hi @MSJavaScript

That layout can work. An in-process registry keyed by thread_id doesn’t break interrupts by itself, but a few things matter:

  1. While a thread is paused nothing else runs on it, so in the same process your ReactionSystem stays exactly as it was until the resume. Other tool calls that already finished in the same step aren’t re-run on resume either - LangGraph saves their writes and re-applies them (_reapply_writes_to_succeeded_nodes in langgraph/pregel/_loop.py). Only the interrupted tool restarts.

  2. In generate_reactions, call interrupt() before you touch the ReactionSystem (setting constraints, generating), because everything before it runs twice. Even simpler: use HumanInTheLoopMiddleware with when. Then the pause happens before the tool body runs at all, and the tool never pauses and mutates in the same call.

  3. The registry is the weak spot, not the interrupt. It breaks if the process restarts while the thread is paused (a resume can come hours later). It also breaks if you run several workers and the resume lands on another process, or if you rewind or fork from an older checkpoint (the object isn’t rewound). And it grows forever if entries are never removed.

  4. A robust fix is to treat the object as a cache and checkpoint a "recipe” instead. Each tool writes a small serializable description to state via Command(update=…) - reactants, constraints, the combine/delete operations, and the random seed if DFS sampling is random. A helper get_system(thread_id, name) returns the cached object, or rebuilds it by replaying the recipe on a cache miss. Your interrupt guard or when predicate can then read the recipe from runtime.state (checkpointed) instead of the live object.

  5. "Not serializable” is only partly true. LangGraph’s JsonPlusSerializer handles numpy arrays natively and has a pickle_fallback=True option, and ase.Atoms and RDKit Mol are picklable. I still wouldn’t put the whole object in state - it gets copied into every checkpoint, and pickles in a checkpoint DB are a versioning and security risk. The recipe is cheaper and safer.

  6. Small things: initialise self.reactions_systems = {} instead of None. Read the thread id with runtime.config[“configurable”][“thread_id”] from a runtime: ToolRuntime parameter (it’s hidden from the model). And evict an entry when the workflow is done, or use a TTL/LRU cache.