update

libs/langgraph/langgraph/_internal/_runnable.py

@@ -26,6 +26,7 @@ from typing import (
     cast,
 )
 
+from langchain_core.callbacks import AsyncCallbackManager, CallbackManager
 from langchain_core.runnables.base import (
     Runnable,
     RunnableConfig,
@@ -87,6 +88,62 @@ def _process_outputs(processor: Callable[[Any], Any] | None, outputs: Any) -> An
         return {"error": "<trace_outputs processing failed>"}
 
 
+def _filter_langsmith_handlers(
+    callback_manager: CallbackManager,
+) -> CallbackManager:
+    """Filter out LangChainTracer handlers from a callback manager.
+
+    Used when traceable config has enabled=False - we want to fire callbacks
+    to custom handlers but skip LangSmith tracing.
+    """
+    filtered_handlers = [
+        h for h in callback_manager.handlers if not isinstance(h, LangChainTracer)
+    ]
+    filtered_inheritable = [
+        h
+        for h in callback_manager.inheritable_handlers
+        if not isinstance(h, LangChainTracer)
+    ]
+    # Create a new callback manager with filtered handlers
+    return CallbackManager(
+        handlers=filtered_handlers,
+        inheritable_handlers=filtered_inheritable,
+        parent_run_id=callback_manager.parent_run_id,
+        tags=callback_manager.tags,
+        inheritable_tags=callback_manager.inheritable_tags,
+        metadata=callback_manager.metadata,
+        inheritable_metadata=callback_manager.inheritable_metadata,
+    )
+
+
+def _filter_langsmith_handlers_async(
+    callback_manager: AsyncCallbackManager,
+) -> AsyncCallbackManager:
+    """Filter out LangChainTracer handlers from an async callback manager.
+
+    Used when traceable config has enabled=False - we want to fire callbacks
+    to custom handlers but skip LangSmith tracing.
+    """
+    filtered_handlers = [
+        h for h in callback_manager.handlers if not isinstance(h, LangChainTracer)
+    ]
+    filtered_inheritable = [
+        h
+        for h in callback_manager.inheritable_handlers
+        if not isinstance(h, LangChainTracer)
+    ]
+    # Create a new callback manager with filtered handlers
+    return AsyncCallbackManager(
+        handlers=filtered_handlers,
+        inheritable_handlers=filtered_inheritable,
+        parent_run_id=callback_manager.parent_run_id,
+        tags=callback_manager.tags,
+        inheritable_tags=callback_manager.inheritable_tags,
+        metadata=callback_manager.metadata,
+        inheritable_metadata=callback_manager.inheritable_metadata,
+    )
+
+
 def _set_config_context(
     config: RunnableConfig, run: Any = None
 ) -> Token[RunnableConfig | None]:
@@ -579,6 +636,7 @@ class RunnableSeq(Runnable):
         trace_inputs: Callable[[Any], Any] | None = None,
         trace_outputs: Callable[[Any], Any] | None = None,
         trace: bool = True,
+        skip_langsmith: bool = False,
     ) -> None:
         """Create a new RunnableSeq.
 
@@ -588,6 +646,8 @@ class RunnableSeq(Runnable):
             trace_inputs: Optional function to transform inputs before tracing.
             trace_outputs: Optional function to transform outputs before tracing.
             trace: Whether to trace this sequence. Defaults to True.
+            skip_langsmith: If True, filter out LangChainTracer handlers but keep
+                other callbacks. Used when traceable config has enabled=False.
 
         Raises:
             ValueError: If the sequence has less than 2 steps.
@@ -609,6 +669,7 @@ class RunnableSeq(Runnable):
         self.trace_inputs = trace_inputs
         self.trace_outputs = trace_outputs
         self.trace = trace
+        self.skip_langsmith = skip_langsmith
 
     def __or__(
         self,
@@ -669,6 +730,9 @@ class RunnableSeq(Runnable):
         if self.trace:
             # setup callbacks and context
             callback_manager = get_callback_manager_for_config(config)
+            # Filter out LangChainTracer if skip_langsmith is set
+            if self.skip_langsmith:
+                callback_manager = _filter_langsmith_handlers(callback_manager)
             # start the root run
             run_manager = callback_manager.on_chain_start(
                 None,
@@ -728,6 +792,9 @@ class RunnableSeq(Runnable):
         if self.trace:
             # setup callbacks
             callback_manager = get_async_callback_manager_for_config(config)
+            # Filter out LangChainTracer if skip_langsmith is set
+            if self.skip_langsmith:
+                callback_manager = _filter_langsmith_handlers_async(callback_manager)
             # start the root run
             run_manager = await callback_manager.on_chain_start(
                 None,
@@ -804,6 +871,9 @@ class RunnableSeq(Runnable):
         if self.trace:
             # setup callbacks
             callback_manager = get_callback_manager_for_config(config)
+            # Filter out LangChainTracer if skip_langsmith is set
+            if self.skip_langsmith:
+                callback_manager = _filter_langsmith_handlers(callback_manager)
             # start the root run
             run_manager = callback_manager.on_chain_start(
                 None,
@@ -880,6 +950,9 @@ class RunnableSeq(Runnable):
         if self.trace:
             # setup callbacks
             callback_manager = get_async_callback_manager_for_config(config)
+            # Filter out LangChainTracer if skip_langsmith is set
+            if self.skip_langsmith:
+                callback_manager = _filter_langsmith_handlers_async(callback_manager)
             # start the root run
             run_manager = await callback_manager.on_chain_start(
                 None,

libs/langgraph/langgraph/pregel/_call.py

@@ -272,8 +272,9 @@ def get_runnable_for_task(func: Callable[..., Any]) -> Runnable:
             "trace_inputs": trace_inputs,
         }
         if traceable_config:
+            # Filter out LangChainTracer (skip LangSmith) but keep other callbacks
             if traceable_config.get("enabled") is False:
-                seq_kwargs["trace"] = False
+                seq_kwargs["skip_langsmith"] = True
             if traceable_config.get("process_outputs"):
                 seq_kwargs["trace_outputs"] = traceable_config["process_outputs"]
 

libs/langgraph/langgraph/pregel/_read.py

@@ -270,9 +270,9 @@ class PregelNode:
 
         seq_kwargs: dict[str, Any] = {}
         if tc:
-            # Only disable tracing if explicitly set to False
+            # Filter out LangChainTracer (skip LangSmith) but keep other callbacks
             if tc.get("enabled") is False:
-                seq_kwargs["trace"] = False
+                seq_kwargs["skip_langsmith"] = True
             seq_kwargs["trace_inputs"] = tc.get("process_inputs")
             seq_kwargs["trace_outputs"] = tc.get("process_outputs")
 

libs/langgraph/tests/test_traceable_integration.py

@@ -135,7 +135,11 @@ def test_traceable_config_process_outputs():
 
 
 def test_traceable_config_enabled_false():
-    """Test that __traceable_config__ with enabled=False skips trace creation."""
+    """Test that enabled=False skips LangSmith but custom handlers still receive events.
+
+    Note: enabled=False filters out LangChainTracer (LangSmith) specifically,
+    but custom callback handlers like FakeTracer still receive events.
+    """
     tracer = FakeTracer()
 
     def hidden_node(state: SimpleState) -> SimpleState:
@@ -162,12 +166,12 @@ def test_traceable_config_enabled_false():
     runs = tracer.flattened_runs()
     run_names = [r.name for r in runs]
 
-    # visible_node should be traced
+    # Both nodes should be traced to FakeTracer (a custom handler)
+    # enabled=False only filters LangChainTracer, not custom handlers
     assert "visible_node" in run_names, f"Expected visible_node in {run_names}"
-
-    # hidden_node should NOT be traced (enabled=False)
-    assert "hidden_node" not in run_names, (
-        f"hidden_node should not be traced but found in {run_names}"
+    assert "hidden_node" in run_names, (
+        f"hidden_node should be traced to custom handlers, "
+        f"enabled=False only filters LangSmith. Got: {run_names}"
     )
 
 
@@ -498,7 +502,11 @@ def test_entrypoint_traceable_config_process_outputs():
 
 
 def test_entrypoint_traceable_config_enabled_false():
-    """Test that @entrypoint with enabled=False skips trace creation."""
+    """Test that @entrypoint with enabled=False still fires custom callbacks.
+
+    Note: enabled=False filters out LangChainTracer (LangSmith) specifically,
+    but custom callback handlers like FakeTracer still receive events.
+    """
     tracer = FakeTracer()
 
     def hidden_entrypoint(value: str) -> str:
@@ -515,9 +523,11 @@ def test_entrypoint_traceable_config_enabled_false():
     runs = tracer.flattened_runs()
     run_names = [r.name for r in runs]
 
-    # The entrypoint node should NOT be traced (enabled=False)
-    assert "hidden_entrypoint" not in run_names, (
-        f"hidden_entrypoint should not be traced but found in {run_names}"
+    # The entrypoint should still be traced to custom handlers
+    # enabled=False only filters LangChainTracer, not custom handlers
+    assert "hidden_entrypoint" in run_names, (
+        f"hidden_entrypoint should be traced to custom handlers, "
+        f"enabled=False only filters LangSmith. Got: {run_names}"
     )
 
 
@@ -619,7 +629,11 @@ def test_task_traceable_config_process_outputs():
 
 
 def test_task_traceable_config_enabled_false():
-    """Test that @task with enabled=False skips trace creation."""
+    """Test that @task with enabled=False still fires custom callbacks.
+
+    Note: enabled=False filters out LangChainTracer (LangSmith) specifically,
+    but custom callback handlers like FakeTracer still receive events.
+    """
     tracer = FakeTracer()
 
     def hidden_task_func(value: str) -> str:
@@ -647,14 +661,14 @@ def test_task_traceable_config_enabled_false():
     runs = tracer.flattened_runs()
     run_names = [r.name for r in runs]
 
-    # visible_task_func should be traced
+    # Both tasks should be traced to FakeTracer (a custom handler)
+    # enabled=False only filters LangChainTracer, not custom handlers
     assert "visible_task_func" in run_names, (
         f"Expected visible_task_func in {run_names}"
     )
-
-    # hidden_task_func should NOT be traced (enabled=False)
-    assert "hidden_task_func" not in run_names, (
-        f"hidden_task_func should not be traced but found in {run_names}"
+    assert "hidden_task_func" in run_names, (
+        f"hidden_task_func should be traced to custom handlers, "
+        f"enabled=False only filters LangSmith. Got: {run_names}"
     )
 
 
@@ -1086,3 +1100,38 @@ def test_traceable_enabled_false_allows_get_config():
     assert config_thread_id == "test-thread-123", (
         f"Expected thread_id 'test-thread-123', got {config_thread_id}"
     )
+
+
+def test_traceable_enabled_false_still_fires_custom_callbacks():
+    """Test that enabled=False skips LangSmith but fires callbacks to custom handlers.
+
+    This verifies that on_chain_start and on_chain_end are still called for
+    custom callback handlers even when the node has enabled=False in its
+    traceable config.
+    """
+    tracer = FakeTracer()
+
+    def my_node(state: SimpleState) -> SimpleState:
+        return {"value": f"processed_{state['value']}"}
+
+    _set_traceable_config(my_node, enabled=False)
+
+    builder = StateGraph(SimpleState)
+    builder.add_node("my_node", my_node)
+    builder.add_edge("__start__", "my_node")
+    graph = builder.compile()
+
+    result = graph.invoke({"value": "test"}, {"callbacks": [tracer]})
+
+    # Verify the graph executed correctly
+    assert result == {"value": "processed_test"}
+
+    # Verify custom callback handler received events for the node
+    runs = tracer.flattened_runs()
+    run_names = [r.name for r in runs]
+
+    assert "LangGraph" in run_names, f"Missing LangGraph run: {run_names}"
+    assert "my_node" in run_names, (
+        f"enabled=False should still fire callbacks to custom handlers. "
+        f"Got runs: {run_names}"
+    )