mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-10-14 12:15:25 +02:00
fix(langgraph,prebuilt): stop ToolNode from swallowing invalid resume values (#9232)
**Problem.** `interrupt(response_schema=...)` checks a human's answer when the graph resumes. If the answer doesn't match the schema, the resume fails, nothing is saved, and the human can answer again (#8886). That works when the interrupt runs in a graph node, but not when it runs inside a tool: either the tool calls `interrupt()` itself, or it starts a graph that does, such as a subagent called as a tool. There, the resume returns normally and the interrupt disappears. **Why.** `interrupt()` raises a `pydantic.ValidationError` for an invalid answer. When one comes out of a tool, `ToolNode` assumes the model passed bad arguments, so it hands the error to the model ("Error invoking tool … Please fix the error and try again") and the run carries on. The human never sees the error, the model is blamed for it, and the interrupt is no longer pending. **Fix.** `interrupt()` still raises the same `ValidationError`, but marks it so a new helper, `langgraph.errors.is_invalid_resume(error)`, can recognize it. `ToolNode` re-raises these errors instead of handing them to the model, whatever `handle_tool_errors` is set to, so a bad answer fails the resume the same way inside a tool as outside one: nothing is saved and the interrupt can be answered again. Every other `ValidationError` from a tool is handled as before. **Also fixed, in the same error handling.** With `wrap_tool_call` set and a `handle_tool_errors` other than the default (for example `True`), `ToolNode` turned any exception from the wrapper into an error `ToolMessage`, including `GraphInterrupt`. So an interrupt in that setup never paused: the model got `Error: GraphInterrupt(...)` instead. Interrupts now propagate there too, as they already do without a wrapper. `create_agent` uses the default handler, so it was not affected. **Compatibility.** Not a breaking change: the error type is unchanged. Worth a careful look: a wrapped `ToolNode` with a non-default `handle_tool_errors` now pauses on interrupts instead of returning them to the model as errors. With an older `langgraph` that doesn't have `is_invalid_resume`, `ToolNode` keeps today's behavior.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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:
|
||||
|
||||
|
||||
Reference in New Issue
Block a user