diff --git a/libs/langgraph/langgraph/errors.py b/libs/langgraph/langgraph/errors.py index 65b57d61e..d8ac56200 100644 --- a/libs/langgraph/langgraph/errors.py +++ b/libs/langgraph/langgraph/errors.py @@ -28,6 +28,7 @@ __all__ = ( "ParentCommand", "EmptyInputError", "TaskNotFound", + "is_invalid_resume", ) @@ -239,3 +240,21 @@ class NodeTimeoutError(Exception): self.kind = kind self.idle_timeout = idle_timeout self.run_timeout = run_timeout + + +_INVALID_RESUME = "_langgraph_invalid_resume" + + +def _mark_invalid_resume(error: BaseException) -> None: + setattr(error, _INVALID_RESUME, True) + + +def is_invalid_resume(error: BaseException) -> bool: + """Whether `error` was raised because a resume value didn't match `response_schema`. + + `interrupt()` raises a `pydantic.ValidationError` in that case. `ToolNode` uses + this to tell it apart from invalid tool arguments when a tool calls `interrupt()` + or runs a graph that does, so the resume fails and the interrupt can be answered + again. + """ + return getattr(error, _INVALID_RESUME, False) is True diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index 54510c8fc..8fba752a1 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -20,7 +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 pydantic import TypeAdapter, ValidationError from typing_extensions import ( NotRequired, TypeAliasType, @@ -882,6 +882,16 @@ class Command(Generic[N], ToolOutputMixin): PARENT: ClassVar[Literal["__parent__"]] = "__parent__" +def _validate_resume(adapter: TypeAdapter[Any], value: Any) -> Any: + from langgraph.errors import _mark_invalid_resume + + try: + return adapter.validate_python(value) + except ValidationError as exc: + _mark_invalid_resume(exc) + raise + + @overload def interrupt(value: Any, *, response_schema: type[ResponseT]) -> ResponseT: ... @@ -989,6 +999,7 @@ def interrupt( 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`. + Nothing is saved, so the interrupt can be answered again. `is_invalid_resume` identifies it. """ from langgraph._internal._constants import ( CONFIG_KEY_CHECKPOINT_NS, @@ -1012,14 +1023,14 @@ def interrupt( if scratchpad.resume: if idx < len(scratchpad.resume): v = scratchpad.resume[idx] - validated = adapter.validate_python(v) if adapter else v + validated = _validate_resume(adapter, v) if adapter else v conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume[: idx + 1])]) return validated # 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 + validated = _validate_resume(adapter, v) if adapter else v scratchpad.resume.append(v) conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)]) return validated diff --git a/libs/langgraph/tests/test_interruption.py b/libs/langgraph/tests/test_interruption.py index f4d401e94..50b8ed6d0 100644 --- a/libs/langgraph/tests/test_interruption.py +++ b/libs/langgraph/tests/test_interruption.py @@ -6,6 +6,7 @@ from langgraph.checkpoint.base import BaseCheckpointSaver from pydantic import BaseModel, ValidationError from typing_extensions import TypedDict +from langgraph.errors import is_invalid_resume from langgraph.graph import END, START, StateGraph from langgraph.types import Command, Durability, Interrupt, interrupt from tests.any_str import AnyStr @@ -202,14 +203,22 @@ def test_interrupt_response_schema_rejects_invalid_resume( def resume(value: dict[str, Any]) -> Command: return Command(resume=value if resume_style == "null" else {pending.id: value}) - with pytest.raises(ValidationError, match="approved"): + with pytest.raises(ValidationError, match="approved") as exc_info: graph.invoke(resume({"approved": "nope"}), config) + assert is_invalid_resume(exc_info.value) assert graph.invoke(resume({"approved": False}), config) == { "answer": Decision(approved=False) } +def test_is_invalid_resume_ignores_other_errors() -> None: + with pytest.raises(ValidationError) as exc_info: + Decision.model_validate({"approved": "nope"}) + assert not is_invalid_resume(exc_info.value) + assert not is_invalid_resume(ValueError("nope")) + + @pytest.mark.parametrize("resume_style", ["null", "id_map"]) def test_interrupt_response_schema_invalid_resume_after_earlier_interrupt( sync_checkpointer: BaseCheckpointSaver, resume_style: str diff --git a/libs/prebuilt/langgraph/prebuilt/tool_node.py b/libs/prebuilt/langgraph/prebuilt/tool_node.py index 95e161b90..c0398f8f4 100644 --- a/libs/prebuilt/langgraph/prebuilt/tool_node.py +++ b/libs/prebuilt/langgraph/prebuilt/tool_node.py @@ -99,6 +99,14 @@ if TYPE_CHECKING: from langgraph.runtime import Runtime from pydantic_core import ErrorDetails +try: + from langgraph.errors import is_invalid_resume +except ImportError: # `langgraph` before `is_invalid_resume` never marks resume errors + + def is_invalid_resume(error: BaseException) -> bool: + return False + + # right now we use a dict as the default, can change this to AgentState, but depends # on if this lives in LangChain or LangGraph... ideally would have some typed # messages key @@ -957,6 +965,11 @@ class ToolNode(RunnableCallable): try: response = tool.invoke(call_args, config) except ValidationError as exc: + if is_invalid_resume(exc): + # An `interrupt()` in this tool, or in a graph it ran, got a resume + # value that doesn't match its `response_schema`. That's not a bad + # tool argument: fail the run so the interrupt can be answered again. + raise # Filter out errors for injected arguments injected = self._injected_args.get(call["name"]) filtered_errors = _filter_validation_errors(exc, injected) @@ -982,6 +995,10 @@ class ToolNode(RunnableCallable): except GraphBubbleUp: raise except Exception as e: + # The model can't fix a resume value that doesn't match an interrupt's + # `response_schema`, so no `handle_tool_errors` setting handles it. + if is_invalid_resume(e): + raise # Determine which exception types are handled handled_types: tuple[type[Exception], ...] if isinstance(self._handle_tool_errors, type) and issubclass( @@ -1053,9 +1070,13 @@ class ToolNode(RunnableCallable): # Call wrapper with request and execute callable try: return self._wrap_tool_call(tool_request, execute) + except GraphBubbleUp: + # Interrupts always propagate, as they do without a wrapper. + raise except Exception as e: - # Wrapper threw an exception - if not self._handle_tool_errors: + # Wrapper threw an exception. The model can't fix a resume value that + # doesn't match an interrupt's `response_schema`, so it's never handled. + if not self._handle_tool_errors or is_invalid_resume(e): raise # Convert to error message content = _handle_tool_error(e, flag=self._handle_tool_errors) @@ -1104,6 +1125,11 @@ class ToolNode(RunnableCallable): try: response = await tool.ainvoke(call_args, config) except ValidationError as exc: + if is_invalid_resume(exc): + # An `interrupt()` in this tool, or in a graph it ran, got a resume + # value that doesn't match its `response_schema`. That's not a bad + # tool argument: fail the run so the interrupt can be answered again. + raise # Filter out errors for injected arguments injected = self._injected_args.get(call["name"]) filtered_errors = _filter_validation_errors(exc, injected) @@ -1129,6 +1155,10 @@ class ToolNode(RunnableCallable): except GraphBubbleUp: raise except Exception as e: + # The model can't fix a resume value that doesn't match an interrupt's + # `response_schema`, so no `handle_tool_errors` setting handles it. + if is_invalid_resume(e): + raise # Determine which exception types are handled handled_types: tuple[type[Exception], ...] if isinstance(self._handle_tool_errors, type) and issubclass( @@ -1208,9 +1238,13 @@ class ToolNode(RunnableCallable): # None check was performed above already self._wrap_tool_call = cast("ToolCallWrapper", self._wrap_tool_call) return self._wrap_tool_call(tool_request, _sync_execute) + except GraphBubbleUp: + # Interrupts always propagate, as they do without a wrapper. + raise except Exception as e: - # Wrapper threw an exception - if not self._handle_tool_errors: + # Wrapper threw an exception. The model can't fix a resume value that + # doesn't match an interrupt's `response_schema`, so it's never handled. + if not self._handle_tool_errors or is_invalid_resume(e): raise # Convert to error message content = _handle_tool_error(e, flag=self._handle_tool_errors) diff --git a/libs/prebuilt/tests/test_tool_node.py b/libs/prebuilt/tests/test_tool_node.py index 47ebdcae0..546a41770 100644 --- a/libs/prebuilt/tests/test_tool_node.py +++ b/libs/prebuilt/tests/test_tool_node.py @@ -25,6 +25,7 @@ from langchain_core.runnables.config import RunnableConfig from langchain_core.tools import BaseTool, InjectedToolArg, ToolException from langchain_core.tools import tool as dec_tool from langchain_core.tools.base import InjectedToolCallId +from langgraph.checkpoint.base import BaseCheckpointSaver from langgraph.config import get_stream_writer from langgraph.errors import GraphBubbleUp, GraphInterrupt from langgraph.graph import START, MessagesState, StateGraph @@ -32,8 +33,8 @@ from langgraph.graph.message import REMOVE_ALL_MESSAGES, add_messages from langgraph.runtime import ExecutionInfo, ServerInfo from langgraph.store.base import BaseStore from langgraph.store.memory import InMemoryStore -from langgraph.types import Command, Send -from pydantic import BaseModel +from langgraph.types import Command, Send, interrupt +from pydantic import BaseModel, ValidationError from pydantic.v1 import BaseModel as BaseModelV1 from typing_extensions import TypedDict @@ -626,6 +627,149 @@ def test_tool_node_node_interrupt() -> None: assert exc_info.value == "foo" +class _Approval(BaseModel): + approved: bool + + +class _AskState(TypedDict, total=False): + answer: str + + +def _approval_graph(): + """A graph that asks a human for approval with a typed interrupt.""" + + def ask(state: _AskState) -> _AskState: + approval = interrupt("Approve?", response_schema=_Approval) + return {"answer": f"approved={approval.approved}"} + + return StateGraph(_AskState).add_node("ask", ask).add_edge(START, "ask").compile() + + +def _ask_human_call() -> dict[str, list[AnyMessage]]: + call = ToolCall(name="ask_human", args={}, id="call_1") + return {"messages": [AIMessage("", tool_calls=[call])]} + + +def _handle_any(e): # no annotation: handles every error + return "handled" + + +# A bad answer to an interrupt must fail the run whatever `handle_tool_errors` is, +# including settings that cover `ValidationError` (a `ValueError`), with or without +# a wrapper. `create_agent` always runs tools through a wrapper (its middleware). +_TOOL_NODES = pytest.mark.parametrize( + ("wrapped", "handle_tool_errors"), + [ + (wrapped, handler) + for wrapped in (False, True) + for handler in (None, True, (ValueError,), _handle_any) + ], + ids=[ + f"{wrapped}-{handler}" + for wrapped in ("plain", "wrapped") + for handler in ("default", "handle_true", "handle_value_error", "untyped") + ], +) +# The interrupt either runs in a graph the tool starts (a subagent) or in the tool. +_SHAPE = pytest.mark.parametrize("nested", [True, False], ids=["nested", "direct"]) + + +@_TOOL_NODES +@_SHAPE +def test_tool_node_reraises_invalid_resume( + sync_checkpointer: BaseCheckpointSaver, + wrapped: bool, + nested: bool, + handle_tool_errors: Any, +) -> None: + asker = _approval_graph() + + @dec_tool + def ask_human() -> str: + """Ask a human for approval.""" + if nested: + return asker.invoke({})["answer"] + approval = interrupt("Approve?", response_schema=_Approval) + return f"approved={approval.approved}" + + def pass_through(request, handler): + return handler(request) + + errors = ( + {} if handle_tool_errors is None else {"handle_tool_errors": handle_tool_errors} + ) + graph = ( + StateGraph(MessagesState) + .add_node( + "tools", + ToolNode( + [ask_human], wrap_tool_call=pass_through if wrapped else None, **errors + ), + ) + .add_edge(START, "tools") + .compile(checkpointer=sync_checkpointer) + ) + config: RunnableConfig = {"configurable": {"thread_id": "1"}} + [pending] = graph.invoke(_ask_human_call(), config)["__interrupt__"] + + # A bad answer isn't a bad tool argument: the run fails without saving, so + # the same interrupt can be answered again. + with pytest.raises(ValidationError, match="approved"): + graph.invoke(Command(resume={pending.id: {"approved": "maybe"}}), config) + assert [i.id for i in graph.get_state(config).interrupts] == [pending.id] + + result = graph.invoke(Command(resume={pending.id: {"approved": True}}), config) + assert result["messages"][-1].content == "approved=True" + + +@_TOOL_NODES +@_SHAPE +async def test_tool_node_reraises_invalid_resume_async( + async_checkpointer: BaseCheckpointSaver, + wrapped: bool, + nested: bool, + handle_tool_errors: Any, +) -> None: + asker = _approval_graph() + + @dec_tool + async def ask_human() -> str: + """Ask a human for approval.""" + if nested: + return (await asker.ainvoke({}))["answer"] + approval = interrupt("Approve?", response_schema=_Approval) + return f"approved={approval.approved}" + + async def pass_through(request, handler): + return await handler(request) + + errors = ( + {} if handle_tool_errors is None else {"handle_tool_errors": handle_tool_errors} + ) + graph = ( + StateGraph(MessagesState) + .add_node( + "tools", + ToolNode( + [ask_human], awrap_tool_call=pass_through if wrapped else None, **errors + ), + ) + .add_edge(START, "tools") + .compile(checkpointer=async_checkpointer) + ) + config: RunnableConfig = {"configurable": {"thread_id": "1"}} + [pending] = (await graph.ainvoke(_ask_human_call(), config))["__interrupt__"] + + with pytest.raises(ValidationError, match="approved"): + await graph.ainvoke(Command(resume={pending.id: {"approved": "maybe"}}), config) + state = await graph.aget_state(config) + assert [i.id for i in state.interrupts] == [pending.id] + + resume = Command(resume={pending.id: {"approved": True}}) + result = await graph.ainvoke(resume, config) + assert result["messages"][-1].content == "approved=True" + + @pytest.mark.parametrize("input_type", ["dict", "tool_calls"]) async def test_tool_node_command(input_type: str) -> None: