From 1d2793a26cfb6830f4fd3aaabfee78b60fa6bac8 Mon Sep 17 00:00:00 2001 From: Sydney Runkle Date: Mon, 8 Dec 2025 09:18:15 -0500 Subject: [PATCH] first pass at response schema --- .../langgraph/_internal/_pydantic.py | 36 ++++ libs/langgraph/langgraph/types.py | 68 +++++- libs/langgraph/tests/test_pregel.py | 195 ++++++++++++++++++ libs/sdk-py/langgraph_sdk/schema.py | 4 +- 4 files changed, 295 insertions(+), 8 deletions(-) diff --git a/libs/langgraph/langgraph/_internal/_pydantic.py b/libs/langgraph/langgraph/_internal/_pydantic.py index 0d93f085e..7e7eda3fe 100644 --- a/libs/langgraph/langgraph/_internal/_pydantic.py +++ b/libs/langgraph/langgraph/_internal/_pydantic.py @@ -17,6 +17,7 @@ from pydantic import ( ConfigDict, Field, RootModel, + TypeAdapter, ) from pydantic import ( create_model as _create_model_base, @@ -26,6 +27,7 @@ from pydantic.json_schema import ( DEFAULT_REF_TEMPLATE, GenerateJsonSchema, JsonSchemaMode, + PydanticInvalidForJsonSchema, ) from typing_extensions import TypedDict @@ -273,3 +275,37 @@ def is_supported_by_pydantic(type_: Any) -> bool: if sys.version_info >= (3, 12): return True return False + + +def get_json_schema(typ: type) -> dict[str, Any]: + """Generate a JSON schema for a given type. + + Supports Pydantic BaseModel, TypedDict, dataclass, and any type + supported by Pydantic's TypeAdapter. + + Args: + typ: The type to generate a JSON schema for. + + Returns: + A JSON schema dictionary. + + Raises: + TypeError: If the type cannot be converted to a JSON schema. + """ + try: + return TypeAdapter(typ).json_schema() + except PydanticInvalidForJsonSchema as e: + msg = ( + f"Cannot generate JSON schema for type {typ!r}. " + f"The type must be serializable to JSON. " + f"Pydantic error: {e}" + ) + raise TypeError(msg) from e + except Exception as e: + msg = ( + f"Cannot generate JSON schema for type {typ!r}. " + f"Supported types include: Pydantic BaseModel, TypedDict, " + f"dataclass, and other types supported by Pydantic's TypeAdapter. " + f"Error: {e}" + ) + raise TypeError(msg) from e diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index 18e8854c7..5bf4757b2 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -163,13 +163,18 @@ class Interrupt: id: str """The ID of the interrupt. Can be used to resume the interrupt directly.""" + response_schema: dict[str, Any] | None + """JSON schema describing the expected response format, if specified.""" + 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 @@ -181,8 +186,17 @@ 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) @@ -398,7 +412,11 @@ class Command(Generic[N], ToolOutputMixin): PARENT: ClassVar[Literal["__parent__"]] = "__parent__" -def interrupt(value: Any) -> Any: +# Type variable for type-safe response typing in interrupt() +_ResponseT = TypeVar("_ResponseT") + + +def interrupt(value: Any, *, response_type: type[_ResponseT] | 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 @@ -420,7 +438,7 @@ def interrupt(value: Any) -> Any: To use an `interrupt`, you must enable a checkpointer, as the feature relies on persisting the graph state. - !!! example + !!! example "Basic usage" ```python import uuid @@ -468,7 +486,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='...', response_schema=None),)} command = Command(resume=\"some input from a human!!!\") @@ -479,14 +497,43 @@ def interrupt(value: Any) -> Any: # > {'node': {'human_value': 'some input from a human!!!'}} ``` + !!! example "With response schema" + + ```python + from pydantic import BaseModel + + + class HumanResponse(BaseModel): + name: str + age: int + + + def node(state: State): + # The response_type generates a JSON schema that is included + # in the Interrupt, which can be used by UIs to render forms. + answer = interrupt( + {\"question\": \"Please provide your information\"}, + response_type=HumanResponse, + ) + # answer will be the raw resume value (not validated) + return {\"human_value\": answer} + ``` + Args: value: The value to surface to the client when the graph is interrupted. + response_type: Optional type for the expected response. Can be a Pydantic + BaseModel, TypedDict, dataclass, or any type supported by Pydantic's + TypeAdapter. When provided, a JSON schema is generated and included + in the Interrupt for UI consumption. Returns: - Any: On subsequent invocations within the same node (same task to be precise), returns the value provided during the first invocation + On subsequent invocations within the same node (same task to be precise), + returns the resume value provided via Command. Raises: - GraphInterrupt: On the first invocation within the node, halts execution and surfaces the provided value to the client. + GraphInterrupt: On the first invocation within the node, halts execution + and surfaces the provided value to the client. + TypeError: If response_type cannot be converted to a JSON schema. """ from langgraph._internal._constants import ( CONFIG_KEY_CHECKPOINT_NS, @@ -494,9 +541,15 @@ def interrupt(value: Any) -> Any: CONFIG_KEY_SEND, RESUME, ) + from langgraph._internal._pydantic import get_json_schema from langgraph.config import get_config from langgraph.errors import GraphInterrupt + # Generate JSON schema from response_type if provided + response_schema: dict[str, Any] | None = None + if response_type is not None: + response_schema = get_json_schema(response_type) + conf = get_config()["configurable"] # track interrupt index scratchpad = conf[CONFIG_KEY_SCRATCHPAD] @@ -519,6 +572,7 @@ def interrupt(value: Any) -> Any: Interrupt.from_ns( value=value, ns=conf[CONFIG_KEY_CHECKPOINT_NS], + response_schema=response_schema, ), ) ) diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index 5ede65afc..28a48cd77 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -4982,6 +4982,201 @@ def test_interrupt_functional( assert res == {"a": "foobar", "b": "bar"} +def test_interrupt_response_type_pydantic( + sync_checkpointer: BaseCheckpointSaver, +) -> None: + """Test interrupt with response_type using a Pydantic model.""" + + class HumanResponse(BaseModel): + approved: bool + comment: str | None = None + + class State(TypedDict): + value: str + + def node(state: State) -> State: + response = interrupt( + {"message": "Please approve"}, + response_type=HumanResponse, + ) + return {"value": f"approved={response}"} + + builder = StateGraph(State) + builder.add_node("node", node) + builder.add_edge(START, "node") + graph = builder.compile(checkpointer=sync_checkpointer) + + config = {"configurable": {"thread_id": "1"}} + result = list(graph.stream({"value": "initial"}, config)) + + assert len(result) == 1 + interrupt_data = result[0]["__interrupt__"] + assert len(interrupt_data) == 1 + intr = interrupt_data[0] + assert intr.value == {"message": "Please approve"} + assert intr.response_schema is not None + assert intr.response_schema["type"] == "object" + assert "approved" in intr.response_schema["properties"] + assert "comment" in intr.response_schema["properties"] + assert intr.response_schema["required"] == ["approved"] + + +def test_interrupt_response_type_typeddict( + sync_checkpointer: BaseCheckpointSaver, +) -> None: + """Test interrupt with response_type using a TypedDict.""" + + class FeedbackResponse(TypedDict): + rating: int + feedback: str + + class State(TypedDict): + value: str + + def node(state: State) -> State: + response = interrupt( + "Please provide feedback", + response_type=FeedbackResponse, + ) + return {"value": f"rating={response}"} + + builder = StateGraph(State) + builder.add_node("node", node) + builder.add_edge(START, "node") + graph = builder.compile(checkpointer=sync_checkpointer) + + config = {"configurable": {"thread_id": "1"}} + result = list(graph.stream({"value": "initial"}, config)) + + assert len(result) == 1 + interrupt_data = result[0]["__interrupt__"] + assert len(interrupt_data) == 1 + intr = interrupt_data[0] + assert intr.value == "Please provide feedback" + assert intr.response_schema is not None + assert intr.response_schema["type"] == "object" + assert "rating" in intr.response_schema["properties"] + assert "feedback" in intr.response_schema["properties"] + + +def test_interrupt_response_type_dataclass( + sync_checkpointer: BaseCheckpointSaver, +) -> None: + """Test interrupt with response_type using a dataclass.""" + + @dataclass + class EditResponse: + edited_text: str + confidence: float + + class State(TypedDict): + value: str + + def node(state: State) -> State: + response = interrupt( + {"original": "text"}, + response_type=EditResponse, + ) + return {"value": f"edited={response}"} + + builder = StateGraph(State) + builder.add_node("node", node) + builder.add_edge(START, "node") + graph = builder.compile(checkpointer=sync_checkpointer) + + config = {"configurable": {"thread_id": "1"}} + result = list(graph.stream({"value": "initial"}, config)) + + assert len(result) == 1 + interrupt_data = result[0]["__interrupt__"] + assert len(interrupt_data) == 1 + intr = interrupt_data[0] + assert intr.value == {"original": "text"} + assert intr.response_schema is not None + assert intr.response_schema["type"] == "object" + assert "edited_text" in intr.response_schema["properties"] + assert "confidence" in intr.response_schema["properties"] + + +def test_interrupt_no_response_type( + sync_checkpointer: BaseCheckpointSaver, +) -> None: + """Test interrupt without response_type has response_schema=None.""" + + class State(TypedDict): + value: str + + def node(state: State) -> State: + response = interrupt("Simple question") + return {"value": f"answer={response}"} + + builder = StateGraph(State) + builder.add_node("node", node) + builder.add_edge(START, "node") + graph = builder.compile(checkpointer=sync_checkpointer) + + config = {"configurable": {"thread_id": "1"}} + result = list(graph.stream({"value": "initial"}, config)) + + assert len(result) == 1 + interrupt_data = result[0]["__interrupt__"] + assert len(interrupt_data) == 1 + intr = interrupt_data[0] + assert intr.value == "Simple question" + assert intr.response_schema is None + + +def test_interrupt_response_type_invalid() -> None: + """Test interrupt with invalid response_type raises TypeError.""" + from langgraph._internal._pydantic import get_json_schema + + # A type that cannot be converted to JSON schema + class NonSerializable: + def __init__(self, func): + self.func = func + + with pytest.raises(TypeError, match="Cannot generate JSON schema"): + get_json_schema(NonSerializable) + + +def test_interrupt_response_type_with_resume( + sync_checkpointer: BaseCheckpointSaver, +) -> None: + """Test that interrupt with response_type works correctly with resume.""" + + class ApprovalResponse(BaseModel): + approved: bool + + class State(TypedDict): + result: str + + def node(state: State) -> State: + response = interrupt( + {"action": "approve"}, + response_type=ApprovalResponse, + ) + # response is the raw resume value, not validated + return {"result": f"got: {response}"} + + builder = StateGraph(State) + builder.add_node("node", node) + builder.add_edge(START, "node") + graph = builder.compile(checkpointer=sync_checkpointer) + + config = {"configurable": {"thread_id": "1"}} + + # First invocation - should interrupt + result = list(graph.stream({"result": ""}, config)) + assert len(result) == 1 + assert "__interrupt__" in result[0] + intr = result[0]["__interrupt__"][0] + assert intr.response_schema is not None + + # Resume with a value + result = list(graph.stream(Command(resume={"approved": True}), config)) + assert result == [{"node": {"result": "got: {'approved': True}"}}] + + def test_interrupt_task_functional( sync_checkpointer: BaseCheckpointSaver, snapshot: SnapshotAssertion ) -> None: diff --git a/libs/sdk-py/langgraph_sdk/schema.py b/libs/sdk-py/langgraph_sdk/schema.py index 390adadf0..f487fce56 100644 --- a/libs/sdk-py/langgraph_sdk/schema.py +++ b/libs/sdk-py/langgraph_sdk/schema.py @@ -15,7 +15,7 @@ from typing import ( Union, ) -from typing_extensions import TypedDict +from typing_extensions import NotRequired, TypedDict Json = dict[str, Any] | None """Represents a JSON-like structure, which can be None or a dictionary with string keys and any values.""" @@ -262,6 +262,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 describing the expected response format, if specified.""" class Thread(TypedDict):