diff --git a/libs/langgraph/langgraph/graph/state.py b/libs/langgraph/langgraph/graph/state.py index 9a87f16fa..4c61c287f 100644 --- a/libs/langgraph/langgraph/graph/state.py +++ b/libs/langgraph/langgraph/graph/state.py @@ -1185,6 +1185,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): debug: bool = False, name: str | None = None, transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None, + insert_before: Sequence[str] | None = None, ) -> CompiledStateGraph[StateT, ContextT, InputT, OutputT]: """Compiles the `StateGraph` into a `CompiledStateGraph` object. @@ -1224,6 +1225,12 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): the standard `StreamTransformer` constructor shape by accepting `scope` as their first argument. Appended after the built-in stream transformers. + insert_before: Nodes before which input from `RunControl.submit_input` + may be applied mid-run, e.g. an agent's model node. + + !!! warning "Experimental" + + This parameter may change without notice. Returns: CompiledStateGraph: The compiled `StateGraph`. @@ -1256,6 +1263,13 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): # assign default values interrupt_before = interrupt_before or [] interrupt_after = interrupt_after or [] + if isinstance(insert_before, str): + msg = "`insert_before` takes a sequence of node names, not a string" + raise TypeError(msg) + insert_before = list(insert_before or ()) + if unknown := [node for node in insert_before if node not in self.nodes]: + msg = f"`insert_before` names nodes not in the graph: {unknown}" + raise ValueError(msg) # validate the graph self.validate( @@ -1360,6 +1374,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): checkpointer=checkpointer, interrupt_before_nodes=interrupt_before, interrupt_after_nodes=interrupt_after, + insert_before_nodes=insert_before, auto_validate=False, debug=debug, store=store, @@ -1441,6 +1456,15 @@ class CompiledStateGraph( name=self.get_name("Output"), ) + def _inserted_input_channels(self) -> Sequence[str]: + """The graph's input keys: inserted input may write only these.""" + keys = [ + key + for key, value in self.builder.schemas[self.builder.input_schema].items() + if not is_managed_value(value) + ] + return () if keys == ["__root__"] else keys + def attach_node(self, key: str, node: StateNodeSpec[Any, ContextT] | None) -> None: if key == START: output_keys = [ diff --git a/libs/langgraph/langgraph/pregel/_loop.py b/libs/langgraph/langgraph/pregel/_loop.py index 30ba2b22a..fd71a3de3 100644 --- a/libs/langgraph/langgraph/pregel/_loop.py +++ b/libs/langgraph/langgraph/pregel/_loop.py @@ -122,6 +122,7 @@ from langgraph.pregel._io import ( map_output_values, read_channels, ) +from langgraph.pregel._log import logger from langgraph.pregel._messages import ensure_message_ids from langgraph.pregel._read import PregelNode from langgraph.pregel._task_status import read_task_statuses @@ -187,6 +188,8 @@ class PregelLoop: manager: None | AsyncParentRunManager | ParentRunManager interrupt_after: All | Sequence[str] interrupt_before: All | Sequence[str] + insert_before: Sequence[str] + insert_channels: Sequence[str] durability: Durability retry_policy: Sequence[RetryPolicy] cache_policy: CachePolicy | None @@ -326,6 +329,8 @@ class PregelLoop: durability: Durability, interrupt_after: All | Sequence[str] = EMPTY_SEQ, interrupt_before: All | Sequence[str] = EMPTY_SEQ, + insert_before: Sequence[str] = EMPTY_SEQ, + insert_channels: Sequence[str] = EMPTY_SEQ, manager: None | AsyncParentRunManager | ParentRunManager = None, migrate_checkpoint: Callable[[Checkpoint], None] | None = None, retry_policy: Sequence[RetryPolicy] = (), @@ -347,6 +352,8 @@ class PregelLoop: self.stream_keys = stream_keys self.interrupt_after = interrupt_after self.interrupt_before = interrupt_before + self.insert_before = insert_before + self.insert_channels = insert_channels self.manager = manager self.is_nested = CONFIG_KEY_TASK_ID in self.config.get(CONF, {}) self.is_replaying = CONFIG_KEY_CHECKPOINT_ID in config[CONF] @@ -707,6 +714,11 @@ class PregelLoop: self.status = "draining" return False + # apply input inserted into the running run, then plan the step again + # from the new state + if self._accepts_inserted_input() and self._apply_inserted_input(): + return self.tick() + # produce debug output self._emit("tasks", map_debug_tasks, self.tasks.values()) @@ -801,6 +813,105 @@ class PregelLoop: # private + def _accepts_inserted_input(self) -> bool: + """Whether queued input can be applied before the next step. + + Only in a top-level `StateGraph` with steps left, only when every next + task is a node declared in `insert_before`, and never on the step a + resume continues: the resume applies to the state it paused on. + """ + return ( + self.control is not None + and self.control.has_inputs + and not self.is_nested + and bool(self.insert_before) + and bool(self.insert_channels) + and self.step < self.stop + and not self.config[CONF].get(CONFIG_KEY_RESUMING, False) + and all(task.name in self.insert_before for task in self.tasks.values()) + ) + + def _apply_inserted_input(self) -> bool: + """Apply queued input to the state and save it as an `input` checkpoint. + + Each input gets its own write, so two that set the same plain key don't + raise; the later one wins. Returns whether any was applied, so `tick()` + plans the step again. + """ + applied: list[str] = [] + writes: list[tuple[str, Any]] = [] + inserted_channels: set[str] = set() + for input_id, value in cast(RunControl, self.control).take_inputs(): + if not (input_writes := self._inserted_writes(input_id, value)): + continue + inserted_channels |= apply_writes( + self.checkpoint, + self.channels, + [PregelTaskWrites((), INPUT, input_writes, [])], + self.checkpointer_get_next_version, + self.trigger_to_nodes, + ) + writes.extend(input_writes) + applied.append(input_id) + if not applied: + return False + # snapshot delta channels this input wrote, so they read back without + # their writes stored separately + self._delta_channels_forced_snapshot.update( + channel + for channel, _ in writes + if isinstance(self.specs.get(channel), DeltaChannel) + ) + # add to the last step's updated channels, which trigger the planned tasks + self.updated_channels = (self.updated_channels or set()) | inserted_channels + if not inserted_channels.isdisjoint( + (self.output_keys,) + if isinstance(self.output_keys, str) + else self.output_keys + ): + self._emit( + "values", map_output_values, self.output_keys, writes, self.channels + ) + self._put_checkpoint( + cast( + CheckpointMetadata, + {"source": "input", "inserted_inputs": applied}, + ) + ) + return True + + def _inserted_writes(self, input_id: str, value: Any) -> list[tuple[str, Any]]: + """State updates for inserted input, or none if the state rejects it. + + Written straight to the state channels: through `__start__`, the graph + would route back to its entry. + """ + if not isinstance(value, dict): + logger.warning( + "Skipping inserted input %s: expected a dict of state keys, got %s", + input_id, + type(value).__name__, + ) + return [] + if ignored := [key for key in value if key not in self.insert_channels]: + logger.warning( + "Ignoring keys of inserted input %s that aren't graph input keys: %s", + input_id, + ignored, + ) + writes = [(k, v) for k, v in value.items() if k in self.insert_channels] + try: + for key, val in writes: + self.channels[key].copy().update([val]) + except Exception: + logger.warning( + "Skipping inserted input %s: the state rejected it", + input_id, + exc_info=True, + ) + return [] + return writes + @property def _reapplies_pending_writes(self) -> bool: """Whether the writes loaded with the checkpoint go back to the tasks @@ -1638,6 +1749,8 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager): manager: None | AsyncParentRunManager | ParentRunManager = None, interrupt_after: All | Sequence[str] = EMPTY_SEQ, interrupt_before: All | Sequence[str] = EMPTY_SEQ, + insert_before: Sequence[str] = EMPTY_SEQ, + insert_channels: Sequence[str] = EMPTY_SEQ, input_keys: str | Sequence[str] = EMPTY_SEQ, output_keys: str | Sequence[str] = EMPTY_SEQ, stream_keys: str | Sequence[str] = EMPTY_SEQ, @@ -1660,6 +1773,8 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager): stream_keys=stream_keys, interrupt_after=interrupt_after, interrupt_before=interrupt_before, + insert_before=insert_before, + insert_channels=insert_channels, manager=manager, migrate_checkpoint=migrate_checkpoint, trigger_to_nodes=trigger_to_nodes, @@ -1893,6 +2008,8 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager): durability: Durability, interrupt_after: All | Sequence[str] = EMPTY_SEQ, interrupt_before: All | Sequence[str] = EMPTY_SEQ, + insert_before: Sequence[str] = EMPTY_SEQ, + insert_channels: Sequence[str] = EMPTY_SEQ, manager: None | AsyncParentRunManager | ParentRunManager = None, input_keys: str | Sequence[str] = EMPTY_SEQ, output_keys: str | Sequence[str] = EMPTY_SEQ, @@ -1916,6 +2033,8 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager): stream_keys=stream_keys, interrupt_after=interrupt_after, interrupt_before=interrupt_before, + insert_before=insert_before, + insert_channels=insert_channels, manager=manager, migrate_checkpoint=migrate_checkpoint, trigger_to_nodes=trigger_to_nodes, diff --git a/libs/langgraph/langgraph/pregel/main.py b/libs/langgraph/langgraph/pregel/main.py index f944b1bf3..033e14c06 100644 --- a/libs/langgraph/langgraph/pregel/main.py +++ b/libs/langgraph/langgraph/pregel/main.py @@ -732,6 +732,14 @@ class Pregel( interrupt_before_nodes: All | Sequence[str] + insert_before_nodes: Sequence[str] = () + """Nodes before which input inserted into a running run may be applied. + + !!! warning "Experimental" + + This attribute may change without notice. + """ + input_channels: str | Sequence[str] step_timeout: float | None = None @@ -777,6 +785,7 @@ class Pregel( stream_channels: str | Sequence[str] | None = None, interrupt_after_nodes: All | Sequence[str] = (), interrupt_before_nodes: All | Sequence[str] = (), + insert_before_nodes: Sequence[str] = (), input_channels: str | Sequence[str], step_timeout: float | None = None, debug: bool | None = None, @@ -823,6 +832,7 @@ class Pregel( self.stream_channels = stream_channels self.interrupt_after_nodes = interrupt_after_nodes self.interrupt_before_nodes = interrupt_before_nodes + self.insert_before_nodes = insert_before_nodes self.input_channels = input_channels self.step_timeout = step_timeout self.debug = debug if debug is not None else get_debug() @@ -987,6 +997,10 @@ class Pregel( "image/png": self.get_graph().draw_mermaid_png(), } + def _inserted_input_channels(self) -> Sequence[str]: + """Channels inserted input may write; none for a plain `Pregel`.""" + return () + def copy(self, update: dict[str, Any] | None = None) -> Self: attrs = {k: v for k, v in self.__dict__.items() if k != "__orig_class__"} attrs.update(update or {}) @@ -2802,7 +2816,8 @@ class Pregel( - `"sync"`: Changes are persisted synchronously before the next step starts. - `"async"`: Changes are persisted asynchronously while the next step executes. - `"exit"`: Changes are persisted only when the graph exits. - control: Optional run control used to request cooperative drain. + control: Optional run control used to request cooperative drain or to + insert input into the run (see `RunControl`). subgraphs: Whether to stream events from inside subgraphs, defaults to `False`. If `True`, the events will be emitted as tuples `(namespace, data)`, @@ -3001,6 +3016,8 @@ class Pregel( stream_keys=self.stream_channels_asis, interrupt_before=interrupt_before_, interrupt_after=interrupt_after_, + insert_before=self.insert_before_nodes, + insert_channels=self._inserted_input_channels(), manager=run_manager, durability=durability_, trigger_to_nodes=self.trigger_to_nodes, @@ -3208,7 +3225,8 @@ class Pregel( - `"sync"`: Changes are persisted synchronously before the next step starts. - `"async"`: Changes are persisted asynchronously while the next step executes. - `"exit"`: Changes are persisted only when the graph exits. - control: Optional run control used to request cooperative drain. + control: Optional run control used to request cooperative drain or to + insert input into the run (see `RunControl`). subgraphs: Whether to stream events from inside subgraphs, defaults to `False`. If `True`, the events will be emitted as tuples `(namespace, data)`, @@ -3453,6 +3471,8 @@ class Pregel( stream_keys=self.stream_channels_asis, interrupt_before=interrupt_before_, interrupt_after=interrupt_after_, + insert_before=self.insert_before_nodes, + insert_channels=self._inserted_input_channels(), manager=run_manager, durability=durability_, trigger_to_nodes=self.trigger_to_nodes, @@ -3958,7 +3978,8 @@ class Pregel( - `"sync"`: Changes are persisted synchronously before the next step starts. - `"async"`: Changes are persisted asynchronously while the next step executes. - `"exit"`: Changes are persisted only when the graph exits. - control: Optional run control used to request cooperative drain. + control: Optional run control used to request cooperative drain or to + insert input into the run (see `RunControl`). version: The streaming format version. `"v1"` (default) returns the traditional format, `"v2"` returns `StreamPart` typed dicts when `stream_mode` is not `"values"`. @@ -4135,7 +4156,8 @@ class Pregel( - `"sync"`: Changes are persisted synchronously before the next step starts. - `"async"`: Changes are persisted asynchronously while the next step executes. - `"exit"`: Changes are persisted only when the graph exits. - control: Optional run control used to request cooperative drain. + control: Optional run control used to request cooperative drain or to + insert input into the run (see `RunControl`). version: The streaming format version. `"v1"` (default) returns the traditional format, `"v2"` returns `StreamPart` typed dicts when `stream_mode` is not `"values"`. diff --git a/libs/langgraph/langgraph/runtime.py b/libs/langgraph/langgraph/runtime.py index 9e6d32bf8..827e8213f 100644 --- a/libs/langgraph/langgraph/runtime.py +++ b/libs/langgraph/langgraph/runtime.py @@ -1,5 +1,6 @@ from __future__ import annotations +import threading from collections.abc import Callable from dataclasses import dataclass, field, replace from typing import Any, Generic, cast @@ -77,24 +78,66 @@ class ServerInfo: class RunControl: - """Run-scoped control surface for cooperative draining. + """Run-scoped control surface for cooperative draining and inserted input. Intended for a single graph run. Create a fresh `RunControl` per run; reusing a control after `request_drain()` leaves it drained. Safe to call from any thread: the drain request is represented by a - single attribute write, so no lock is needed for this signal. - If more mutable state is added here, add synchronization. + single attribute write, so no lock is needed for this signal. Inserted + input is guarded by a lock. """ - __slots__ = ("_drain_reason",) + __slots__ = ("_drain_reason", "_inputs", "_inputs_lock") def __init__(self) -> None: self._drain_reason: str | None = None + self._inputs: list[tuple[str, Any]] = [] + self._inputs_lock = threading.Lock() def request_drain(self, reason: str = "shutdown") -> None: self._drain_reason = reason + def submit_input(self, input_id: str, value: Any) -> None: + """Queue input for the running graph to apply at its next safe step boundary. + + !!! warning "Experimental" + + This method may change without notice. + + The graph applies queued input only right before the nodes in its + `insert_before`, the points it declares safe (for an agent, before the + next model call, so input never lands between a tool call and its + result). Only the top-level graph applies it, since a subgraph's state + isn't the caller's, and never on the step a resume continues, since the + resume answers the state it paused on. Input it can't apply (not a dict + of the graph's input keys, or a value the state rejects) is skipped and + logged rather than failing the run. Input the run never reaches stays + queued, so the caller can run it afterwards. + + Args: + input_id: Recorded in the `inserted_inputs` metadata of the + checkpoint that applied it. + value: A dict of the graph's input keys. + """ + with self._inputs_lock: + self._inputs.append((input_id, value)) + + @property + def has_inputs(self) -> bool: + """Whether input is queued and not yet applied.""" + return bool(self._inputs) + + def take_inputs(self) -> list[tuple[str, Any]]: + """Remove and return queued input in the order it was submitted. + + Returns: + `(input_id, value)` pairs. + """ + with self._inputs_lock: + inputs, self._inputs = self._inputs, [] + return inputs + @property def drain_requested(self) -> bool: return self._drain_reason is not None diff --git a/libs/langgraph/tests/test_inserted_input.py b/libs/langgraph/tests/test_inserted_input.py new file mode 100644 index 000000000..989309beb --- /dev/null +++ b/libs/langgraph/tests/test_inserted_input.py @@ -0,0 +1,295 @@ +import operator +from typing import Annotated + +import pytest +from langgraph.checkpoint.memory import InMemorySaver +from typing_extensions import TypedDict + +from langgraph.errors import GraphRecursionError +from langgraph.graph import END, START, StateGraph +from langgraph.runtime import RunControl +from langgraph.types import Command, interrupt + + +class State(TypedDict, total=False): + log: Annotated[list[str], operator.add] + seen: list[str] + + +def _tools_then_model(control: RunControl) -> StateGraph: + """`tools` queues input while it runs; `model` records what it sees.""" + + def tools(state: State) -> State: + control.submit_input("insert-1", {"log": ["inserted"]}) + return {"log": ["tools"]} + + def model(state: State) -> State: + return {"seen": list(state["log"])} + + graph = StateGraph(State) + graph.add_node("tools", tools) + graph.add_node("model", model) + graph.add_edge(START, "tools") + graph.add_edge("tools", "model") + graph.add_edge("model", END) + return graph + + +def test_inserted_input_is_applied_before_an_allowed_node() -> None: + control = RunControl() + checkpointer = InMemorySaver() + graph = _tools_then_model(control).compile( + checkpointer=checkpointer, insert_before=["model"] + ) + config = {"configurable": {"thread_id": "1"}} + + result = graph.invoke({"log": ["user"]}, config, control=control) + + assert result["seen"] == ["user", "tools", "inserted"] + assert not control.has_inputs + inserted = [ + snapshot.metadata + for snapshot in graph.get_state_history(config) + if snapshot.metadata.get("inserted_inputs") + ] + assert len(inserted) == 1 + assert inserted[0]["source"] == "input" + assert inserted[0]["inserted_inputs"] == ["insert-1"] + + +@pytest.mark.anyio +async def test_inserted_input_is_applied_before_an_allowed_node_async() -> None: + control = RunControl() + graph = _tools_then_model(control).compile( + checkpointer=InMemorySaver(), insert_before=["model"] + ) + + result = await graph.ainvoke( + {"log": ["user"]}, {"configurable": {"thread_id": "1"}}, control=control + ) + + assert result["seen"] == ["user", "tools", "inserted"] + assert not control.has_inputs + + +def test_inserted_input_is_applied_without_a_checkpointer() -> None: + control = RunControl() + graph = _tools_then_model(control).compile(insert_before=["model"]) + + result = graph.invoke({"log": ["user"]}, control=control) + + assert result["seen"] == ["user", "tools", "inserted"] + + +@pytest.mark.parametrize("insert_before", [None, ["tools"]]) +def test_inserted_input_waits_when_the_next_node_is_not_allowed( + insert_before: list[str] | None, +) -> None: + control = RunControl() + graph = _tools_then_model(control).compile(insert_before=insert_before) + + result = graph.invoke({"log": ["user"]}, control=control) + + assert result["seen"] == ["user", "tools"] + assert control.take_inputs() == [("insert-1", {"log": ["inserted"]})] + + +def _outer(node: object) -> StateGraph: + """Runs `node` (which runs a subgraph), then `finish`.""" + + def finish(state: State) -> State: + return {"log": ["finished"]} + + outer = StateGraph(State) + outer.add_node("outer", node) + outer.add_node("finish", finish) + outer.add_edge(START, "outer") + outer.add_edge("outer", "finish") + outer.add_edge("finish", END) + return outer + + +def test_inserted_input_is_not_applied_inside_a_subgraph() -> None: + control = RunControl() + inner = _tools_then_model(control).compile(insert_before=["model"]) + + def outer_node(state: State) -> State: + return inner.invoke({"log": state["log"]}) + + result = ( + _outer(outer_node) + .compile(insert_before=["finish"]) + .invoke({"log": ["user"]}, control=control) + ) + + # the subgraph's model didn't see it; the parent applied it before `finish` + assert result["seen"] == ["user", "tools"] + assert result["log"][-2:] == ["inserted", "finished"] + + +@pytest.mark.anyio +async def test_inserted_input_is_not_applied_inside_a_subgraph_async() -> None: + control = RunControl() + inner = _tools_then_model(control).compile(insert_before=["model"]) + + async def outer_node(state: State) -> State: + return await inner.ainvoke({"log": state["log"]}) + + result = await ( + _outer(outer_node) + .compile(insert_before=["finish"]) + .ainvoke({"log": ["user"]}, control=control) + ) + + assert result["seen"] == ["user", "tools"] + assert result["log"][-2:] == ["inserted", "finished"] + + +def test_inserted_input_is_streamed_as_values() -> None: + control = RunControl() + graph = _tools_then_model(control).compile(insert_before=["model"]) + + chunks = list( + graph.stream({"log": ["user"]}, control=control, stream_mode="values") + ) + + assert {"log": ["user", "tools", "inserted"]} in chunks + + +@pytest.mark.parametrize( + "value", ["not a dict", {"seen": ["forged"]}, {"branch:to:model": None}] +) +def test_invalid_inserted_input_is_skipped(value: object) -> None: + class InputState(TypedDict): + log: Annotated[list[str], operator.add] + + control = RunControl() + + def tools(state: State) -> State: + control.submit_input("bad", value) + return {"log": ["tools"]} + + def model(state: State) -> State: + return {"seen": list(state["log"])} + + builder = StateGraph(State, input_schema=InputState) + builder.add_node("tools", tools) + builder.add_node("model", model) + builder.add_edge(START, "tools") + builder.add_edge("tools", "model") + builder.add_edge("model", END) + checkpointer = InMemorySaver() + graph = builder.compile(checkpointer=checkpointer, insert_before=["model"]) + config = {"configurable": {"thread_id": "1"}} + + result = graph.invoke({"log": ["user"]}, config, control=control) + + assert result["seen"] == ["user", "tools"] + assert not any( + snapshot.metadata.get("inserted_inputs") + for snapshot in graph.get_state_history(config) + ) + + +@pytest.mark.parametrize("by_id", [False, True]) +def test_inserted_input_waits_while_resuming(by_id: bool) -> None: + def model(state: State) -> State: + answer = interrupt("approve?") + return {"log": [f"answered {answer}"], "seen": list(state["log"])} + + builder = StateGraph(State) + builder.add_node("model", model) + builder.add_edge(START, "model") + builder.add_edge("model", END) + graph = builder.compile(checkpointer=InMemorySaver(), insert_before=["model"]) + config = {"configurable": {"thread_id": "1"}} + paused = graph.invoke({"log": ["user"]}, config) + interrupt_id = paused["__interrupt__"][0].id + control = RunControl() + control.submit_input("insert-1", {"log": ["inserted"]}) + + result = graph.invoke( + Command(resume={interrupt_id: "yes"} if by_id else "yes"), + config, + control=control, + ) + + # the resume is answered as asked; the input waits for a later boundary + assert result["seen"] == ["user"] + assert result["log"] == ["user", "answered yes"] + assert control.take_inputs() == [("insert-1", {"log": ["inserted"]})] + + +def test_insert_before_rejects_unknown_nodes() -> None: + graph = _tools_then_model(RunControl()) + + with pytest.raises(ValueError, match="missing"): + graph.compile(insert_before=["missing"]) + + +def test_inserts_are_applied_one_by_one_and_bad_values_skipped() -> None: + class InputState(TypedDict, total=False): + log: Annotated[list[str], operator.add] + note: str + + class FullState(InputState, total=False): + seen: list[str] + + control = RunControl() + + def tools(state: FullState) -> FullState: + control.submit_input("first", {"note": "a", "log": ["one"]}) + control.submit_input("bad", {"log": "not a list"}) + control.submit_input("second", {"note": "b", "log": ["two"]}) + return {"log": ["tools"]} + + def model(state: FullState) -> FullState: + return {"seen": [*state["log"], state.get("note", "")]} + + builder = StateGraph(FullState, input_schema=InputState) + builder.add_node("tools", tools) + builder.add_node("model", model) + builder.add_edge(START, "tools") + builder.add_edge("tools", "model") + builder.add_edge("model", END) + checkpointer = InMemorySaver() + graph = builder.compile(checkpointer=checkpointer, insert_before=["model"]) + config = {"configurable": {"thread_id": "1"}} + + result = graph.invoke({"log": ["user"]}, config, control=control) + + assert result["seen"] == ["user", "tools", "one", "two", "b"] + inserted = [ + snapshot.metadata["inserted_inputs"] + for snapshot in graph.get_state_history(config) + if snapshot.metadata.get("inserted_inputs") + ] + assert inserted == [["first", "second"]] + + +def test_inserted_input_waits_while_resuming_from_a_breakpoint() -> None: + control = RunControl() + graph = _tools_then_model(control).compile( + checkpointer=InMemorySaver(), + insert_before=["model"], + interrupt_before=["model"], + ) + config = {"configurable": {"thread_id": "1"}} + graph.invoke({"log": ["user"]}, config) + assert graph.get_state(config).next == ("model",) + + result = graph.invoke(None, config, control=control) + + # the breakpoint is passed once; the input waits for a later boundary + assert result["seen"] == ["user", "tools"] + assert graph.get_state(config).next == () + + +def test_inserted_input_is_not_applied_without_steps_left() -> None: + control = RunControl() + graph = _tools_then_model(control).compile(insert_before=["model"]) + + with pytest.raises(GraphRecursionError): + graph.invoke({"log": ["user"]}, {"recursion_limit": 1}, control=control) + + assert control.take_inputs() == [("insert-1", {"log": ["inserted"]})] diff --git a/libs/sdk-py/langgraph_sdk/_async/cron.py b/libs/sdk-py/langgraph_sdk/_async/cron.py index 824b86f55..860542337 100644 --- a/libs/sdk-py/langgraph_sdk/_async/cron.py +++ b/libs/sdk-py/langgraph_sdk/_async/cron.py @@ -104,7 +104,7 @@ class CronClient: webhook: Webhook to call after LangGraph API call is done. multitask_strategy: Multitask strategy to use. - Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'. end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely. enabled: Whether the cron job is enabled or not. timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``). @@ -222,7 +222,7 @@ class CronClient: after execution. 'keep' creates a new thread for each execution but does not clean them up. Clients are responsible for cleaning up kept threads. multitask_strategy: Multitask strategy to use. - Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'. end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely. enabled: Whether the cron job is enabled or not. timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``). diff --git a/libs/sdk-py/langgraph_sdk/_async/runs.py b/libs/sdk-py/langgraph_sdk/_async/runs.py index 76782b36c..8f8e1c524 100644 --- a/libs/sdk-py/langgraph_sdk/_async/runs.py +++ b/libs/sdk-py/langgraph_sdk/_async/runs.py @@ -252,7 +252,7 @@ class RunsClient: Must be one of 'delete' or 'keep'. webhook: Webhook to call after LangGraph API call is done. multitask_strategy: Multitask strategy to use. - Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'. if_not_exists: How to handle missing thread. Defaults to 'reject'. Must be either 'reject' (raise error if missing), or 'create' (create new thread). after_seconds: The number of seconds to wait before starting the run. @@ -468,7 +468,7 @@ class RunsClient: interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. webhook: Webhook to call after LangGraph API call is done. multitask_strategy: Multitask strategy to use. - Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'. on_completion: Whether to delete or keep the thread created for a stateless run. Must be one of 'delete' or 'keep'. if_not_exists: How to handle missing thread. Defaults to 'reject'. @@ -726,7 +726,7 @@ class RunsClient: on_completion: Whether to delete or keep the thread created for a stateless run. Must be one of 'delete' or 'keep'. multitask_strategy: Multitask strategy to use. - Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'. if_not_exists: How to handle missing thread. Defaults to 'reject'. Must be either 'reject' (raise error if missing), or 'create' (create new thread). after_seconds: The number of seconds to wait before starting the run. diff --git a/libs/sdk-py/langgraph_sdk/_sync/cron.py b/libs/sdk-py/langgraph_sdk/_sync/cron.py index 464edb80d..ffef02fec 100644 --- a/libs/sdk-py/langgraph_sdk/_sync/cron.py +++ b/libs/sdk-py/langgraph_sdk/_sync/cron.py @@ -96,7 +96,7 @@ class SyncCronClient: interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. webhook: Webhook to call after LangGraph API call is done. multitask_strategy: Multitask strategy to use. - Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'. end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely. enabled: Whether the cron job is enabled. By default, it is considered enabled. timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``). @@ -212,7 +212,7 @@ class SyncCronClient: after execution. 'keep' creates a new thread for each execution but does not clean them up. Clients are responsible for cleaning up kept threads. multitask_strategy: Multitask strategy to use. - Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'. end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely. enabled: Whether the cron job is enabled. By default, it is considered enabled. timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``). diff --git a/libs/sdk-py/langgraph_sdk/_sync/runs.py b/libs/sdk-py/langgraph_sdk/_sync/runs.py index 5d2d17086..ff45d33fd 100644 --- a/libs/sdk-py/langgraph_sdk/_sync/runs.py +++ b/libs/sdk-py/langgraph_sdk/_sync/runs.py @@ -252,7 +252,7 @@ class SyncRunsClient: Must be one of 'delete' or 'keep'. webhook: Webhook to call after LangGraph API call is done. multitask_strategy: Multitask strategy to use. - Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'. if_not_exists: How to handle missing thread. Defaults to 'reject'. Must be either 'reject' (raise error if missing), or 'create' (create new thread). after_seconds: The number of seconds to wait before starting the run. @@ -463,7 +463,7 @@ class SyncRunsClient: interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. webhook: Webhook to call after LangGraph API call is done. multitask_strategy: Multitask strategy to use. - Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'. on_completion: Whether to delete or keep the thread created for a stateless run. Must be one of 'delete' or 'keep'. if_not_exists: How to handle missing thread. Defaults to 'reject'. @@ -721,7 +721,7 @@ class SyncRunsClient: on_completion: Whether to delete or keep the thread created for a stateless run. Must be one of 'delete' or 'keep'. multitask_strategy: Multitask strategy to use. - Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'. if_not_exists: How to handle missing thread. Defaults to 'reject'. Must be either 'reject' (raise error if missing), or 'create' (create new thread). after_seconds: The number of seconds to wait before starting the run. diff --git a/libs/sdk-py/langgraph_sdk/auth/types.py b/libs/sdk-py/langgraph_sdk/auth/types.py index 84211878d..791075a6f 100644 --- a/libs/sdk-py/langgraph_sdk/auth/types.py +++ b/libs/sdk-py/langgraph_sdk/auth/types.py @@ -29,7 +29,9 @@ Values: - interrupted: Run was manually interrupted """ -MultitaskStrategy = typing.Literal["reject", "rollback", "interrupt", "enqueue"] +MultitaskStrategy = typing.Literal[ + "reject", "rollback", "interrupt", "enqueue", "insert" +] """Strategy for handling multiple concurrent tasks. Values: @@ -37,6 +39,7 @@ Values: - rollback: Cancel current task and start new one - interrupt: Interrupt current task and start new one - enqueue: Queue new tasks to run after current one + - insert: Add input to current task at its next safe step; held while interrupted """ OnConflictBehavior = typing.Literal["raise", "do_nothing"] diff --git a/libs/sdk-py/langgraph_sdk/schema.py b/libs/sdk-py/langgraph_sdk/schema.py index a16733e60..bf04ad937 100644 --- a/libs/sdk-py/langgraph_sdk/schema.py +++ b/libs/sdk-py/langgraph_sdk/schema.py @@ -78,13 +78,14 @@ Specifies behavior on disconnection: - "continue": Continue the operation even if disconnected. """ -MultitaskStrategy = Literal["reject", "interrupt", "rollback", "enqueue"] +MultitaskStrategy = Literal["reject", "interrupt", "rollback", "enqueue", "insert"] """ Defines how to handle multiple tasks: - "reject": Reject new tasks when busy. - "interrupt": Interrupt current task for new ones. - "rollback": Roll back current task and start new one. - "enqueue": Queue new tasks for later execution. +- "insert": Add input to the current task at its next safe step; held while interrupted. """ OnConflictBehavior = Literal["raise", "do_nothing"]