feat(langgraph): add set_node_defaults() to StateGraph

libs/langgraph/langgraph/graph/state.py

@@ -1,3 +1,3 @@
 from __future__ import annotations
 
 import inspect
@@ -1094,6 +1094,7 @@
         debug: bool = False,
         name: str | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
+        default_error_handler: StateNode[Any, ContextT] | None = None,
     ) -> CompiledStateGraph[StateT, ContextT, InputT, OutputT]:
         """Compiles the `StateGraph` into a `CompiledStateGraph` object.
 
@@ -1133,6 +1134,12 @@
                 the standard `StreamTransformer` constructor shape by
                 accepting `scope` as their first argument. Appended after the
                 built-in stream transformers.
+            default_error_handler: Optional fallback node-level error handler
+                invoked when any regular node raises and does not have its own
+                `error_handler` set via `add_node`. Per-node `error_handler`
+                always takes precedence. The default handler is **not** invoked
+                when an error-handler node itself raises -- handler failures
+                fail the run. Not inherited by subgraphs.
 
         Returns:
             CompiledStateGraph: The compiled `StateGraph`.
@@ -1193,10 +1200,35 @@
                 key for key, val in self.channels.items() if not is_managed_value(val)
             ]
         )
+        default_handler_node_name: str | None = None
+        if default_error_handler is not None:
+            _default_eh_name = "__default_error_handler__"
+            if _default_eh_name in self.nodes:
+                raise ValueError(
+                    f"Auto-generated default error handler node "
+                    f"`{_default_eh_name}` already exists."
+                )
+            default_handler_node_name = _default_eh_name
+            self.nodes[default_handler_node_name] = StateNodeSpec[Any, ContextT](
+                coerce_to_runnable(
+                    default_error_handler,  # type: ignore[arg-type]
+                    name=default_handler_node_name,
+                    trace=False,
+                ),
+                metadata=None,
+                input_schema=self.state_schema,
+                retry_policy=None,
+                cache_policy=None,
+                is_error_handler=True,
+            )
+            for _spec in self.nodes.values():
+                if not _spec.is_error_handler and _spec.error_handler_node is None:
+                    _spec.error_handler_node = default_handler_node_name
+
         node_error_handler_map = {
             node_name: spec.error_handler_node
             for node_name, spec in self.nodes.items()
-            if spec.error_handler_node is not None
+            if not spec.is_error_handler and spec.error_handler_node is not None
         }
 
         compiled = CompiledStateGraph[StateT, ContextT, InputT, OutputT](

libs/langgraph/tests/test_retry.py

@@ -15,7 +15,7 @@
 from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
 from langchain_core.messages import AIMessage, AIMessageChunk, BaseMessage, HumanMessage
 from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult
-from langchain_core.runnables import RunnableLambda, RunnableParallel
+from langchain_core.runnables import RunnableConfig, RunnableLambda, RunnableParallel
 from langgraph.checkpoint.memory import InMemorySaver, MemorySaver
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
 from typing_extensions import TypedDict
@@ -2280,3 +2280,191 @@ def fail_without_handler(state: State) -> State:
 
     with pytest.raises(ValueError, match="no handler"):
         graph.invoke({"foo": ""})
+
+
+# ---------------------------------------------------------------------------
+# default_error_handler on compile()
+# ---------------------------------------------------------------------------
+
+
+def test_compile_default_error_handler_catches_all_nodes():
+    class State(TypedDict):
+        route: str
+        foo: Annotated[list[str], operator.add]
+
+    def route_node(state: State) -> Command:
+        return Command(goto=state["route"])
+
+    def fail_a(state: State) -> State:
+        raise RuntimeError("a failed")
+
+    def fail_b(state: State) -> State:
+        raise RuntimeError("b failed")
+
+    captured: dict[str, list[str]] = {"nodes": []}
+
+    def default_handler(state: State, error: NodeError) -> State:
+        captured["nodes"].append(error.node)
+        return {"foo": [f"handled_{error.node}"]}
+
+    graph = (
+        StateGraph(State)
+        .add_node("route_node", route_node)
+        .add_node("fail_a", fail_a)
+        .add_node("fail_b", fail_b)
+        .add_edge(START, "route_node")
+        .add_conditional_edges(
+            "route_node", lambda s: s["route"], path_map=["fail_a", "fail_b"]
+        )
+        .compile(default_error_handler=default_handler)
+    )
+
+    result_a = graph.invoke({"route": "fail_a", "foo": []})
+    result_b = graph.invoke({"route": "fail_b", "foo": []})
+    assert result_a["foo"] == ["handled_fail_a"]
+    assert result_b["foo"] == ["handled_fail_b"]
+    assert "fail_a" in captured["nodes"]
+    assert "fail_b" in captured["nodes"]
+
+
+def test_compile_default_error_handler_overridden_by_node_handler():
+    class State(TypedDict):
+        route: str
+        foo: Annotated[list[str], operator.add]
+
+    def route_node(state: State) -> Command:
+        return Command(goto=state["route"])
+
+    def fail_a(state: State) -> State:
+        raise RuntimeError("a failed")
+
+    def fail_b(state: State) -> State:
+        raise RuntimeError("b failed")
+
+    captured: dict[str, list[str]] = {"handler": []}
+
+    def node_handler(state: State, error: NodeError) -> State:
+        captured["handler"].append(f"node:{error.node}")
+        return {"foo": [f"node_handled_{error.node}"]}
+
+    def default_handler(state: State, error: NodeError) -> State:
+        captured["handler"].append(f"default:{error.node}")
+        return {"foo": [f"default_handled_{error.node}"]}
+
+    graph = (
+        StateGraph(State)
+        .add_node("route_node", route_node)
+        .add_node("fail_a", fail_a, error_handler=node_handler)
+        .add_node("fail_b", fail_b)
+        .add_edge(START, "route_node")
+        .add_conditional_edges(
+            "route_node", lambda s: s["route"], path_map=["fail_a", "fail_b"]
+        )
+        .compile(default_error_handler=default_handler)
+    )
+
+    result_a = graph.invoke({"route": "fail_a", "foo": []})
+    assert result_a["foo"] == ["node_handled_fail_a"]
+    assert "node:fail_a" in captured["handler"]
+    assert "default:fail_a" not in captured["handler"]
+
+    result_b = graph.invoke({"route": "fail_b", "foo": []})
+    assert result_b["foo"] == ["default_handled_fail_b"]
+    assert "default:fail_b" in captured["handler"]
+
+
+def test_compile_default_error_handler_skips_per_node_handler_nodes():
+    """If a per-node error handler itself raises, the default handler must NOT
+    catch it -- the run should fail."""
+
+    class State(TypedDict):
+        foo: str
+
+    def always_failing(state: State) -> State:
+        raise RuntimeError("node boom")
+
+    def broken_handler(state: State, error: NodeError) -> State:
+        raise RuntimeError("handler boom")
+
+    def default_handler(state: State, error: NodeError) -> State:
+        return {"foo": "default recovered"}
+
+    graph = (
+        StateGraph(State)
+        .add_node("always_failing", always_failing, error_handler=broken_handler)
+        .add_edge(START, "always_failing")
+        .compile(default_error_handler=default_handler)
+    )
+
+    with pytest.raises(RuntimeError, match="handler boom"):
+        graph.invoke({"foo": ""})
+
+
+def test_compile_default_error_handler_failure_fails_run():
+    """When the default handler itself raises, the run fails (no infinite
+    recursion, no double-routing)."""
+
+    class State(TypedDict):
+        foo: str
+
+    def always_failing(state: State) -> State:
+        raise RuntimeError("node boom")
+
+    def broken_default_handler(state: State, error: NodeError) -> State:
+        raise RuntimeError("default handler boom")
+
+    graph = (
+        StateGraph(State)
+        .add_node("always_failing", always_failing)
+        .add_edge(START, "always_failing")
+        .compile(default_error_handler=broken_default_handler)
+    )
+
+    with pytest.raises(RuntimeError, match="default handler boom"):
+        graph.invoke({"foo": ""})
+
+
+def test_compile_default_error_handler_receives_runnable_config():
+    class State(TypedDict):
+        foo: str
+
+    def always_failing(state: State) -> State:
+        raise RuntimeError("boom")
+
+    captured: dict[str, Any] = {}
+
+    def default_handler(
+        state: State, error: NodeError, config: RunnableConfig
+    ) -> State:
+        captured["thread_id"] = config["configurable"].get("thread_id")
+        return {"foo": "handled"}
+
+    checkpointer = MemorySaver()
+    graph = (
+        StateGraph(State)
+        .add_node("always_failing", always_failing)
+        .add_edge(START, "always_failing")
+        .compile(checkpointer=checkpointer, default_error_handler=default_handler)
+    )
+
+    thread_id = str(uuid4())
+    result = graph.invoke(
+        {"foo": ""}, config={"configurable": {"thread_id": thread_id}}
+    )
+    assert result["foo"] == "handled"
+    assert captured["thread_id"] == thread_id
+
+
+def test_compile_default_error_handler_collides_with_user_node():
+    class State(TypedDict):
+        foo: str
+
+    def default_handler(state: State, error: NodeError) -> State:
+        return {"foo": "handled"}
+
+    builder = StateGraph(State)
+    builder.add_node("__default_error_handler__", lambda s: s)
+    builder.add_edge(START, "__default_error_handler__")
+
+    with pytest.raises(ValueError, match="__default_error_handler__"):
+        builder.compile(default_error_handler=default_handler)