diff --git a/libs/langgraph/langgraph/constants.py b/libs/langgraph/langgraph/constants.py index 0ebab0a6f..59e72e9c9 100644 --- a/libs/langgraph/langgraph/constants.py +++ b/libs/langgraph/langgraph/constants.py @@ -102,6 +102,8 @@ CONF = cast(Literal["configurable"], sys.intern("configurable")) # key for the configurable dict in RunnableConfig NULL_TASK_ID = sys.intern("00000000-0000-0000-0000-000000000000") # the task_id to use for writes that are not associated with a task +CONFIG_KEY_RESUME_MAP = sys.intern("__pregel_resume_map") +# holds a mapping of task ns -> resume value for resuming tasks RESERVED = { TAG_HIDDEN, diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index 19cd9d1ff..029cf73be 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -40,6 +40,7 @@ from langgraph.constants import ( CONFIG_KEY_CHECKPOINTER, CONFIG_KEY_PREVIOUS, CONFIG_KEY_READ, + CONFIG_KEY_RESUME_MAP, CONFIG_KEY_SCRATCHPAD, CONFIG_KEY_SEND, CONFIG_KEY_STORE, @@ -594,6 +595,8 @@ def prepare_single_task( config[CONF].get(CONFIG_KEY_SCRATCHPAD), pending_writes, task_id, + xxh3_128_hexdigest(task_checkpoint_ns.encode()), + config[CONF].get(CONFIG_KEY_RESUME_MAP), ), }, ), @@ -704,6 +707,8 @@ def prepare_single_task( config[CONF].get(CONFIG_KEY_SCRATCHPAD), pending_writes, task_id, + xxh3_128_hexdigest(task_checkpoint_ns.encode()), + config[CONF].get(CONFIG_KEY_RESUME_MAP), ), CONFIG_KEY_PREVIOUS: checkpoint["channel_values"].get( PREVIOUS, None @@ -830,6 +835,8 @@ def prepare_single_task( config[CONF].get(CONFIG_KEY_SCRATCHPAD), pending_writes, task_id, + xxh3_128_hexdigest(task_checkpoint_ns.encode()), + config[CONF].get(CONFIG_KEY_RESUME_MAP), ), CONFIG_KEY_PREVIOUS: checkpoint["channel_values"].get( PREVIOUS, None @@ -881,6 +888,8 @@ def _scratchpad( parent_scratchpad: Optional[PregelScratchpad], pending_writes: list[PendingWrite], task_id: str, + namespace_hash: str, + resume_map: Optional[dict[str, Any]], ) -> PregelScratchpad: if len(pending_writes) > 0: # find global resume value @@ -892,17 +901,24 @@ def _scratchpad( # None cannot be used as a resume value, because it would be difficult to # distinguish from missing when used over http null_resume_write = None - # find task-specific resume value - for w in pending_writes: - if w[0] == task_id and w[1] == RESUME: - task_resume_write = w[2] - if not isinstance(task_resume_write, list): - task_resume_write = [task_resume_write] - break + + # find namespace and task-specific resume value + if resume_map and namespace_hash in resume_map: + task_resume_write = resume_map[namespace_hash] + if not isinstance(task_resume_write, list): + task_resume_write = [task_resume_write] else: - task_resume_write = [] - # clear var - del w + # find task-specific resume value + for w in pending_writes: + if w[0] == task_id and w[1] == RESUME: + task_resume_write = w[2] + if not isinstance(task_resume_write, list): + task_resume_write = [task_resume_write] + break + else: + task_resume_write = [] + # clear var + del w else: null_resume_write = None task_resume_write = [] diff --git a/libs/langgraph/langgraph/pregel/loop.py b/libs/langgraph/langgraph/pregel/loop.py index 972699b61..978cbd0df 100644 --- a/libs/langgraph/langgraph/pregel/loop.py +++ b/libs/langgraph/langgraph/pregel/loop.py @@ -47,6 +47,7 @@ from langgraph.constants import ( CONFIG_KEY_DEDUPE_TASKS, CONFIG_KEY_DELEGATE, CONFIG_KEY_ENSURE_LATEST, + CONFIG_KEY_RESUME_MAP, CONFIG_KEY_RESUMING, CONFIG_KEY_SCRATCHPAD, CONFIG_KEY_STREAM, @@ -649,6 +650,8 @@ class PregelLoop(LoopProtocol): # map command to writes if isinstance(self.input, Command): + if self.input.resume_map: + self.config[CONF][CONFIG_KEY_RESUME_MAP] = self.input.resume_map if self.input.resume is not None and not self.checkpointer: raise RuntimeError( "Cannot use Command(resume=...) without checkpointer" @@ -657,7 +660,7 @@ class PregelLoop(LoopProtocol): # group writes by task ID for tid, c, v in map_command(self.input, self.checkpoint_pending_writes): writes[tid].append((c, v)) - if not writes: + if not writes and not self.input.resume_map: raise EmptyInputError("Received empty Command input") # save writes for tid, ws in writes.items(): diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index 5462bafb2..4c088ebe8 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -147,7 +147,7 @@ class Interrupt: """Generate a unique ID for the interrupt based on its namespace.""" if self.ns is None: return "placeholder-id" - return xxh3_128_hexdigest("".join(self.ns).encode()) + return xxh3_128_hexdigest("|".join(self.ns).encode()) class StateUpdate(NamedTuple): @@ -300,12 +300,14 @@ class Command(Generic[N], ToolOutputMixin): - sequence of node names to navigate to next - `Send` object (to execute a node with the input provided) - sequence of `Send` objects + resume_map: mapping of interrupt ids to resume values. To be used together with [`interrupt()`][langgraph.types.interrupt]. """ graph: Optional[str] = None update: Optional[Any] = None resume: Optional[Union[Any, dict[str, Any]]] = None goto: Union[Send, Sequence[Union[Send, str]], str] = () + resume_map: Optional[dict[str, Any]] = None def __repr__(self) -> str: # get all non-None values