From 3913144bdd8bb456728e907767b4e7806f22a18d Mon Sep 17 00:00:00 2001 From: Elior Nataf Lackritz Date: Thu, 10 Sep 2026 12:59:13 -0400 Subject: [PATCH] feat(langgraph): add response_schema to interrupt() Optional keyword-only schema for the resume value, surfaced on Interrupt.response_schema as JSON Schema for clients. Pydantic, TypedDict and dataclass schemas also validate the resume value before it is committed; the validated object is returned. --- libs/langgraph/langgraph/types.py | 45 +++++++++-- libs/langgraph/tests/test_interruption.py | 99 ++++++++++++++++++++++- libs/langgraph/tests/test_pregel.py | 2 + libs/sdk-py/langgraph_sdk/schema.py | 2 + 4 files changed, 140 insertions(+), 8 deletions(-) diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index 2fcf5bef9..6dc0f19fe 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -20,6 +20,7 @@ from warnings import warn from langchain_core.messages import AnyMessage from langchain_core.runnables import Runnable, RunnableConfig from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointMetadata +from pydantic import TypeAdapter from typing_extensions import NotRequired, TypeAliasType, TypedDict, Unpack, deprecated from xxhash import xxh3_128_hexdigest @@ -596,13 +597,19 @@ class Interrupt: id: str """The ID of the interrupt. Can be used to resume the interrupt directly.""" + response_schema: dict[str, Any] | None = None + """JSON Schema for the value expected when resuming this interrupt, if the graph provided one.""" + def __init__( self, value: Any, id: str = _DEFAULT_INTERRUPT_ID, + *, + response_schema: dict[str, Any] | None = None, **deprecated_kwargs: Unpack[DeprecatedKwargs], ) -> None: self.value = value + self.response_schema = response_schema if ( (ns := deprecated_kwargs.get("ns", MISSING)) is not MISSING @@ -614,8 +621,14 @@ class Interrupt: self.id = id @classmethod - def from_ns(cls, value: Any, ns: str) -> Interrupt: - return cls(value=value, id=xxh3_128_hexdigest(ns.encode())) + def from_ns( + cls, value: Any, ns: str, *, response_schema: dict[str, Any] | None = None + ) -> Interrupt: + return cls( + value=value, + id=xxh3_128_hexdigest(ns.encode()), + response_schema=response_schema, + ) @property @deprecated("`interrupt_id` is deprecated. Use `id` instead.", category=None) @@ -848,7 +861,9 @@ class Command(Generic[N], ToolOutputMixin): PARENT: ClassVar[Literal["__parent__"]] = "__parent__" -def interrupt(value: Any) -> Any: +def interrupt( + value: Any, *, response_schema: dict[str, Any] | type[Any] | None = None +) -> Any: """Interrupt the graph with a resumable exception from within a node. The `interrupt` function enables human-in-the-loop workflows by pausing graph @@ -918,7 +933,7 @@ def interrupt(value: Any) -> Any: for chunk in graph.stream({\"foo\": \"abc\"}, config): print(chunk) - # > {'__interrupt__': (Interrupt(value='what is your age?', id='45fda8478b2ef754419799e10992af06'),)} + # > {'__interrupt__': (Interrupt(value='what is your age?', id='45fda8478b2ef754419799e10992af06', response_schema=None),)} command = Command(resume=\"some input from a human!!!\") @@ -931,12 +946,20 @@ def interrupt(value: Any) -> Any: Args: value: The value to surface to the client when the graph is interrupted. + response_schema: Optional schema for the value expected on resume, surfaced + to clients so they can render a typed input form. Accepts a JSON Schema + `dict` (used as-is, resume values are not validated), or a Pydantic model + class, `TypedDict`, or dataclass, which are converted to JSON Schema for + clients and used to validate the resume value; the validated object is + what `interrupt` returns. Returns: - Any: On subsequent invocations within the same node (same task to be precise), returns the value provided during the first invocation + Any: On subsequent invocations within the same node (same task to be precise), returns the value provided during the first invocation, + validated against `response_schema` when one that supports validation was given. Raises: GraphInterrupt: On the first invocation within the node, halts execution and surfaces the provided value to the client. + pydantic.ValidationError: When a resume value does not match a Pydantic model, `TypedDict`, or dataclass `response_schema`. """ from langgraph._internal._constants import ( CONFIG_KEY_CHECKPOINT_NS, @@ -948,6 +971,11 @@ def interrupt(value: Any) -> Any: from langgraph.errors import GraphInterrupt conf = get_config()["configurable"] + adapter = ( + None + if response_schema is None or isinstance(response_schema, dict) + else TypeAdapter(response_schema) + ) # track interrupt index scratchpad = conf[CONFIG_KEY_SCRATCHPAD] idx = scratchpad.interrupt_counter() @@ -955,20 +983,23 @@ def interrupt(value: Any) -> Any: if scratchpad.resume: if idx < len(scratchpad.resume): conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)]) - return scratchpad.resume[idx] + v = scratchpad.resume[idx] + return adapter.validate_python(v) if adapter else v # find current resume value v = scratchpad.get_null_resume(True) if v is not None: assert len(scratchpad.resume) == idx, (scratchpad.resume, idx) + validated = adapter.validate_python(v) if adapter else v scratchpad.resume.append(v) conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)]) - return v + return validated # no resume value found raise GraphInterrupt( ( Interrupt.from_ns( value=value, ns=conf[CONFIG_KEY_CHECKPOINT_NS], + response_schema=adapter.json_schema() if adapter else response_schema, ), ) ) diff --git a/libs/langgraph/tests/test_interruption.py b/libs/langgraph/tests/test_interruption.py index a484d74e5..3d5569149 100644 --- a/libs/langgraph/tests/test_interruption.py +++ b/libs/langgraph/tests/test_interruption.py @@ -1,9 +1,13 @@ +from typing import Any + import pytest from langgraph.checkpoint.base import BaseCheckpointSaver +from pydantic import BaseModel, ValidationError from typing_extensions import TypedDict from langgraph.graph import END, START, StateGraph -from langgraph.types import Durability +from langgraph.types import Command, Durability, Interrupt, interrupt +from tests.any_str import AnyStr pytestmark = pytest.mark.anyio @@ -90,3 +94,96 @@ async def test_interruption_without_state_updates_async( assert (await graph.aget_state(thread)).next == () n_checkpoints = len([c async for c in graph.aget_state_history(thread)]) assert n_checkpoints == (5 if durability != "exit" else 3) + + +class Decision(BaseModel): + approved: bool + note: str | None = None + + +class DecisionDict(TypedDict): + approved: bool + + +RAW_SCHEMA = {"type": "object", "properties": {"approved": {"type": "boolean"}}} + + +@pytest.mark.parametrize( + ("response_schema", "expected_schema", "expected_answer"), + [ + (None, None, {"approved": True, "extra": 1}), + (RAW_SCHEMA, RAW_SCHEMA, {"approved": True, "extra": 1}), + (Decision, Decision.model_json_schema(), Decision(approved=True)), + ( + DecisionDict, + { + "properties": {"approved": {"title": "Approved", "type": "boolean"}}, + "required": ["approved"], + "title": "DecisionDict", + "type": "object", + }, + {"approved": True}, + ), + ], + ids=["none", "raw_dict", "pydantic", "typeddict"], +) +def test_interrupt_response_schema( + sync_checkpointer: BaseCheckpointSaver, + response_schema: Any, + expected_schema: dict[str, Any] | None, + expected_answer: Any, +) -> None: + class State(TypedDict): + answer: Any + + def node(state: State) -> State: + return { + "answer": interrupt( + {"question": "approve?"}, response_schema=response_schema + ) + } + + graph = ( + StateGraph(State) + .add_node("node", node) + .add_edge(START, "node") + .compile(checkpointer=sync_checkpointer) + ) + config = {"configurable": {"thread_id": "1"}} + expected = Interrupt( + value={"question": "approve?"}, id=AnyStr(), response_schema=expected_schema + ) + + assert list(graph.stream({"answer": None}, config)) == [ + {"__interrupt__": (expected,)} + ] + assert graph.get_state(config).tasks[0].interrupts == (expected,) + assert graph.invoke(Command(resume={"approved": True, "extra": 1}), config) == { + "answer": expected_answer + } + + +def test_interrupt_response_schema_rejects_invalid_resume( + sync_checkpointer: BaseCheckpointSaver, +) -> None: + class State(TypedDict): + answer: Any + + def node(state: State) -> State: + return {"answer": interrupt("approve?", response_schema=Decision)} + + graph = ( + StateGraph(State) + .add_node("node", node) + .add_edge(START, "node") + .compile(checkpointer=sync_checkpointer) + ) + config = {"configurable": {"thread_id": "1"}} + graph.invoke({"answer": None}, config) + + with pytest.raises(ValidationError, match="approved"): + graph.invoke(Command(resume={"approved": "nope"}), config) + + assert graph.invoke(Command(resume={"approved": False}), config) == { + "answer": Decision(approved=False) + } diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index c166c5837..2e6661e04 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -5583,6 +5583,7 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver): "interrupts": [ { "id": AnyStr(), + "response_schema": None, "value": "test", }, ], @@ -5627,6 +5628,7 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver): "interrupts": ( { "id": AnyStr(), + "response_schema": None, "value": "test", }, ), diff --git a/libs/sdk-py/langgraph_sdk/schema.py b/libs/sdk-py/langgraph_sdk/schema.py index 18b1b44f3..a16733e60 100644 --- a/libs/sdk-py/langgraph_sdk/schema.py +++ b/libs/sdk-py/langgraph_sdk/schema.py @@ -295,6 +295,8 @@ class Interrupt(TypedDict): """The value associated with the interrupt.""" id: str """The ID of the interrupt. Can be used to resume the interrupt.""" + response_schema: NotRequired[dict[str, Any]] + """JSON Schema for the value expected when resuming this interrupt, if the graph provided one.""" class Thread(TypedDict):