update

libs/langgraph/langgraph/_internal/_runnable.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import enum
 import inspect
+import logging
 import sys
 import warnings
 from collections.abc import (
@@ -25,6 +26,7 @@ from typing import (
     cast,
 )
 
+from langchain_core.callbacks import AsyncCallbackManager, CallbackManager
 from langchain_core.runnables.base import (
     Runnable,
     RunnableConfig,
@@ -61,6 +63,86 @@ try:
 except ImportError:
     _StreamingCallbackHandler = None  # type: ignore
 
+logger = logging.getLogger(__name__)
+
+
+def _process_inputs(processor: Callable[[Any], Any] | None, inputs: Any) -> Any:
+    """Safely process trace inputs, returning error placeholder on failure."""
+    if processor is None:
+        return inputs
+    try:
+        return processor(inputs)
+    except Exception:
+        logger.exception("trace_inputs filter failed")
+        return {"error": "<trace_inputs processing failed>"}
+
+
+def _process_outputs(processor: Callable[[Any], Any] | None, outputs: Any) -> Any:
+    """Safely process trace outputs, returning error placeholder on failure."""
+    if processor is None:
+        return outputs
+    try:
+        return processor(outputs)
+    except Exception:
+        logger.exception("trace_outputs filter failed")
+        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
@@ -397,7 +479,9 @@ class RunnableCallable(Runnable):
             else:
                 run_manager.on_chain_end(ret)
         else:
-            ret = self.func(*args, **kwargs)
+            # Still need to set config context for get_config() to work
+            with set_config_context(config, None) as context:
+                ret = context.run(self.func, *args, **kwargs)
         if self.recurse and isinstance(ret, Runnable):
             return ret.invoke(input, config)
         return ret
@@ -470,7 +554,13 @@ class RunnableCallable(Runnable):
             else:
                 await run_manager.on_chain_end(ret)
         else:
-            ret = await self.afunc(*args, **kwargs)
+            # Still need to set config context for get_config() to work
+            coro = cast(Coroutine[None, None, Any], self.afunc(*args, **kwargs))
+            if ASYNCIO_ACCEPTS_CONTEXT:
+                with set_config_context(config, None) as context:
+                    ret = await asyncio.create_task(coro, context=context)
+            else:
+                ret = await coro
         if self.recurse and isinstance(ret, Runnable):
             return await ret.ainvoke(input, config)
         return ret
@@ -544,12 +634,20 @@ class RunnableSeq(Runnable):
         *steps: RunnableLike,
         name: str | None = None,
         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.
 
         Args:
             steps: The steps to include in the sequence.
             name: The name of the `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.
@@ -569,6 +667,9 @@ class RunnableSeq(Runnable):
         self.steps = steps_flat
         self.name = name
         self.trace_inputs = trace_inputs
+        self.trace_outputs = trace_outputs
+        self.trace = trace
+        self.skip_langsmith = skip_langsmith
 
     def __or__(
         self,
@@ -625,43 +726,58 @@ class RunnableSeq(Runnable):
     ) -> Any:
         if config is None:
             config = ensure_config()
-        # setup callbacks and context
-        callback_manager = get_callback_manager_for_config(config)
-        # start the root run
-        run_manager = callback_manager.on_chain_start(
-            None,
-            self.trace_inputs(input) if self.trace_inputs is not None else input,
-            name=config.get("run_name") or self.get_name(),
-            run_id=config.pop("run_id", None),
-        )
-        # invoke all steps in sequence
-        try:
-            for i, step in enumerate(self.steps):
-                # mark each step as a child run
-                config = patch_config(
-                    config, callbacks=run_manager.get_child(f"seq:step:{i + 1}")
-                )
-                # 1st step is the actual node,
-                # others are writers which don't need to be run in context
-                if i == 0:
-                    # get the run object
-                    for h in run_manager.handlers:
-                        if isinstance(h, LangChainTracer):
-                            run = h.run_map.get(str(run_manager.run_id))
-                            break
+
+        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,
+                _process_inputs(self.trace_inputs, input),
+                name=config.get("run_name") or self.get_name(),
+                run_id=config.pop("run_id", None),
+            )
+            # invoke all steps in sequence
+            try:
+                for i, step in enumerate(self.steps):
+                    # mark each step as a child run
+                    config = patch_config(
+                        config, callbacks=run_manager.get_child(f"seq:step:{i + 1}")
+                    )
+                    # 1st step is the actual node,
+                    # others are writers which don't need to be run in context
+                    if i == 0:
+                        # get the run object
+                        for h in run_manager.handlers:
+                            if isinstance(h, LangChainTracer):
+                                run = h.run_map.get(str(run_manager.run_id))
+                                break
+                        else:
+                            run = None
+                        # run in context
+                        with set_config_context(config, run) as context:
+                            input = context.run(step.invoke, input, config, **kwargs)
                     else:
-                        run = None
-                    # run in context
-                    with set_config_context(config, run) as context:
-                        input = context.run(step.invoke, input, config, **kwargs)
-                else:
-                    input = step.invoke(input, config)
-        # finish the root run
-        except BaseException as e:
-            run_manager.on_chain_error(e)
-            raise
+                        input = step.invoke(input, config)
+            # finish the root run
+            except BaseException as e:
+                run_manager.on_chain_error(e)
+                raise
+            else:
+                run_manager.on_chain_end(_process_outputs(self.trace_outputs, input))
+                return input
         else:
-            run_manager.on_chain_end(input)
+            # Still need to set config context for get_config() to work
+            with set_config_context(config, None) as context:
+                for i, step in enumerate(self.steps):
+                    input = (
+                        context.run(step.invoke, input, config, **kwargs)
+                        if i == 0
+                        else step.invoke(input, config)
+                    )
             return input
 
     async def ainvoke(
@@ -672,49 +788,75 @@ class RunnableSeq(Runnable):
     ) -> Any:
         if config is None:
             config = ensure_config()
-        # setup callbacks
-        callback_manager = get_async_callback_manager_for_config(config)
-        # start the root run
-        run_manager = await callback_manager.on_chain_start(
-            None,
-            self.trace_inputs(input) if self.trace_inputs is not None else input,
-            name=config.get("run_name") or self.get_name(),
-            run_id=config.pop("run_id", None),
-        )
 
-        # invoke all steps in sequence
-        try:
-            for i, step in enumerate(self.steps):
-                # mark each step as a child run
-                config = patch_config(
-                    config, callbacks=run_manager.get_child(f"seq:step:{i + 1}")
-                )
-                # 1st step is the actual node,
-                # others are writers which don't need to be run in context
-                if i == 0:
-                    if ASYNCIO_ACCEPTS_CONTEXT:
-                        # get the run object
-                        for h in run_manager.handlers:
-                            if isinstance(h, LangChainTracer):
-                                run = h.run_map.get(str(run_manager.run_id))
-                                break
+        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,
+                _process_inputs(self.trace_inputs, input),
+                name=config.get("run_name") or self.get_name(),
+                run_id=config.pop("run_id", None),
+            )
+
+            # invoke all steps in sequence
+            try:
+                for i, step in enumerate(self.steps):
+                    # mark each step as a child run
+                    config = patch_config(
+                        config, callbacks=run_manager.get_child(f"seq:step:{i + 1}")
+                    )
+                    # 1st step is the actual node,
+                    # others are writers which don't need to be run in context
+                    if i == 0:
+                        if ASYNCIO_ACCEPTS_CONTEXT:
+                            # get the run object
+                            for h in run_manager.handlers:
+                                if isinstance(h, LangChainTracer):
+                                    run = h.run_map.get(str(run_manager.run_id))
+                                    break
+                            else:
+                                run = None
+                            # run in context
+                            with set_config_context(config, run) as context:
+                                input = await asyncio.create_task(
+                                    step.ainvoke(input, config, **kwargs),
+                                    context=context,
+                                )
                         else:
-                            run = None
-                        # run in context
-                        with set_config_context(config, run) as context:
+                            input = await step.ainvoke(input, config, **kwargs)
+                    else:
+                        input = await step.ainvoke(input, config)
+            # finish the root run
+            except BaseException as e:
+                await run_manager.on_chain_error(e)
+                raise
+            else:
+                await run_manager.on_chain_end(
+                    _process_outputs(self.trace_outputs, input)
+                )
+                return input
+        else:
+            # Still need to set config context for get_config() to work
+            if ASYNCIO_ACCEPTS_CONTEXT:
+                with set_config_context(config, None) as context:
+                    for i, step in enumerate(self.steps):
+                        if i == 0:
                             input = await asyncio.create_task(
                                 step.ainvoke(input, config, **kwargs), context=context
                             )
-                    else:
+                        else:
+                            input = await step.ainvoke(input, config)
+            else:
+                for i, step in enumerate(self.steps):
+                    if i == 0:
                         input = await step.ainvoke(input, config, **kwargs)
-                else:
-                    input = await step.ainvoke(input, config)
-        # finish the root run
-        except BaseException as e:
-            await run_manager.on_chain_error(e)
-            raise
-        else:
-            await run_manager.on_chain_end(input)
+                    else:
+                        input = await step.ainvoke(input, config)
             return input
 
     def stream(
@@ -725,80 +867,20 @@ class RunnableSeq(Runnable):
     ) -> Iterator[Any]:
         if config is None:
             config = ensure_config()
-        # setup callbacks
-        callback_manager = get_callback_manager_for_config(config)
-        # start the root run
-        run_manager = callback_manager.on_chain_start(
-            None,
-            self.trace_inputs(input) if self.trace_inputs is not None else input,
-            name=config.get("run_name") or self.get_name(),
-            run_id=config.pop("run_id", None),
-        )
-        # get the run object
-        for h in run_manager.handlers:
-            if isinstance(h, LangChainTracer):
-                run = h.run_map.get(str(run_manager.run_id))
-                break
-        else:
-            run = None
-        # create first step config
-        config = patch_config(
-            config,
-            callbacks=run_manager.get_child(f"seq:step:{1}"),
-        )
-        # run all in context
-        with set_config_context(config, run) as context:
-            try:
-                # stream the last steps
-                # transform the input stream of each step with the next
-                # steps that don't natively support transforming an input stream will
-                # buffer input in memory until all available, and then start emitting output
-                for idx, step in enumerate(self.steps):
-                    if idx == 0:
-                        iterator = step.stream(input, config, **kwargs)
-                    else:
-                        config = patch_config(
-                            config,
-                            callbacks=run_manager.get_child(f"seq:step:{idx + 1}"),
-                        )
-                        iterator = step.transform(iterator, config)
-                # populates streamed_output in astream_log() output if needed
-                if _StreamingCallbackHandler is not None:
-                    for h in run_manager.handlers:
-                        if isinstance(h, _StreamingCallbackHandler):
-                            iterator = h.tap_output_iter(run_manager.run_id, iterator)
-                # consume into final output
-                output = context.run(_consume_iter, iterator)
-                # sequence doesn't emit output, yield to mark as generator
-                yield
-            except BaseException as e:
-                run_manager.on_chain_error(e)
-                raise
-            else:
-                run_manager.on_chain_end(output)
 
-    async def astream(
-        self,
-        input: Input,
-        config: RunnableConfig | None = None,
-        **kwargs: Any | None,
-    ) -> AsyncIterator[Any]:
-        if config is None:
-            config = ensure_config()
-        # setup callbacks
-        callback_manager = get_async_callback_manager_for_config(config)
-        # start the root run
-        run_manager = await callback_manager.on_chain_start(
-            None,
-            self.trace_inputs(input) if self.trace_inputs is not None else input,
-            name=config.get("run_name") or self.get_name(),
-            run_id=config.pop("run_id", None),
-        )
-        # stream the last steps
-        # transform the input stream of each step with the next
-        # steps that don't natively support transforming an input stream will
-        # buffer input in memory until all available, and then start emitting output
-        if ASYNCIO_ACCEPTS_CONTEXT:
+        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,
+                _process_inputs(self.trace_inputs, input),
+                name=config.get("run_name") or self.get_name(),
+                run_id=config.pop("run_id", None),
+            )
             # get the run object
             for h in run_manager.handlers:
                 if isinstance(h, LangChainTracer):
@@ -813,18 +895,136 @@ class RunnableSeq(Runnable):
             )
             # run all in context
             with set_config_context(config, run) as context:
+                try:
+                    # stream the last steps
+                    # transform the input stream of each step with the next
+                    # steps that don't natively support transforming an input stream will
+                    # buffer input in memory until all available, and then start emitting output
+                    for idx, step in enumerate(self.steps):
+                        if idx == 0:
+                            iterator = step.stream(input, config, **kwargs)
+                        else:
+                            config = patch_config(
+                                config,
+                                callbacks=run_manager.get_child(f"seq:step:{idx + 1}"),
+                            )
+                            iterator = step.transform(iterator, config)
+                    # populates streamed_output in astream_log() output if needed
+                    if _StreamingCallbackHandler is not None:
+                        for h in run_manager.handlers:
+                            if isinstance(h, _StreamingCallbackHandler):
+                                iterator = h.tap_output_iter(
+                                    run_manager.run_id, iterator
+                                )
+                    # consume into final output
+                    output = context.run(_consume_iter, iterator)
+                    # sequence doesn't emit output, yield to mark as generator
+                    yield
+                except BaseException as e:
+                    run_manager.on_chain_error(e)
+                    raise
+                else:
+                    run_manager.on_chain_end(
+                        _process_outputs(self.trace_outputs, output)
+                    )
+        else:
+            # No tracing - still need to set config context for get_config() to work
+            with set_config_context(config, None) as context:
+                for idx, step in enumerate(self.steps):
+                    if idx == 0:
+                        iterator = step.stream(input, config, **kwargs)
+                    else:
+                        iterator = step.transform(iterator, config)
+                context.run(_consume_iter, iterator)
+            yield
+
+    async def astream(
+        self,
+        input: Input,
+        config: RunnableConfig | None = None,
+        **kwargs: Any | None,
+    ) -> AsyncIterator[Any]:
+        if config is None:
+            config = ensure_config()
+
+        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,
+                _process_inputs(self.trace_inputs, input),
+                name=config.get("run_name") or self.get_name(),
+                run_id=config.pop("run_id", None),
+            )
+            # stream the last steps
+            # transform the input stream of each step with the next
+            # steps that don't natively support transforming an input stream will
+            # buffer input in memory until all available, and then start emitting output
+            if ASYNCIO_ACCEPTS_CONTEXT:
+                # get the run object
+                for h in run_manager.handlers:
+                    if isinstance(h, LangChainTracer):
+                        run = h.run_map.get(str(run_manager.run_id))
+                        break
+                else:
+                    run = None
+                # create first step config
+                config = patch_config(
+                    config,
+                    callbacks=run_manager.get_child(f"seq:step:{1}"),
+                )
+                # run all in context
+                with set_config_context(config, run) as context:
+                    try:
+                        async with AsyncExitStack() as stack:
+                            for idx, step in enumerate(self.steps):
+                                if idx == 0:
+                                    aiterator = step.astream(input, config, **kwargs)
+                                else:
+                                    config = patch_config(
+                                        config,
+                                        callbacks=run_manager.get_child(
+                                            f"seq:step:{idx + 1}"
+                                        ),
+                                    )
+                                    aiterator = step.atransform(aiterator, config)
+                                if hasattr(aiterator, "aclose"):
+                                    stack.push_async_callback(aiterator.aclose)
+                            # populates streamed_output in astream_log() output if needed
+                            if _StreamingCallbackHandler is not None:
+                                for h in run_manager.handlers:
+                                    if isinstance(h, _StreamingCallbackHandler):
+                                        aiterator = h.tap_output_aiter(
+                                            run_manager.run_id, aiterator
+                                        )
+                            # consume into final output
+                            output = await asyncio.create_task(
+                                _consume_aiter(aiterator), context=context
+                            )
+                            # sequence doesn't emit output, yield to mark as generator
+                            yield
+                    except BaseException as e:
+                        await run_manager.on_chain_error(e)
+                        raise
+                    else:
+                        await run_manager.on_chain_end(
+                            _process_outputs(self.trace_outputs, output)
+                        )
+            else:
                 try:
                     async with AsyncExitStack() as stack:
                         for idx, step in enumerate(self.steps):
+                            config = patch_config(
+                                config,
+                                callbacks=run_manager.get_child(f"seq:step:{idx + 1}"),
+                            )
                             if idx == 0:
                                 aiterator = step.astream(input, config, **kwargs)
                             else:
-                                config = patch_config(
-                                    config,
-                                    callbacks=run_manager.get_child(
-                                        f"seq:step:{idx + 1}"
-                                    ),
-                                )
                                 aiterator = step.atransform(aiterator, config)
                             if hasattr(aiterator, "aclose"):
                                 stack.push_async_callback(aiterator.aclose)
@@ -836,46 +1036,29 @@ class RunnableSeq(Runnable):
                                         run_manager.run_id, aiterator
                                     )
                         # consume into final output
-                        output = await asyncio.create_task(
-                            _consume_aiter(aiterator), context=context
-                        )
+                        output = await _consume_aiter(aiterator)
                         # sequence doesn't emit output, yield to mark as generator
                         yield
                 except BaseException as e:
                     await run_manager.on_chain_error(e)
                     raise
                 else:
-                    await run_manager.on_chain_end(output)
+                    await run_manager.on_chain_end(
+                        _process_outputs(self.trace_outputs, output)
+                    )
         else:
-            try:
-                async with AsyncExitStack() as stack:
-                    for idx, step in enumerate(self.steps):
-                        config = patch_config(
-                            config,
-                            callbacks=run_manager.get_child(f"seq:step:{idx + 1}"),
-                        )
-                        if idx == 0:
-                            aiterator = step.astream(input, config, **kwargs)
-                        else:
-                            aiterator = step.atransform(aiterator, config)
-                        if hasattr(aiterator, "aclose"):
-                            stack.push_async_callback(aiterator.aclose)
-                    # populates streamed_output in astream_log() output if needed
-                    if _StreamingCallbackHandler is not None:
-                        for h in run_manager.handlers:
-                            if isinstance(h, _StreamingCallbackHandler):
-                                aiterator = h.tap_output_aiter(
-                                    run_manager.run_id, aiterator
-                                )
-                    # consume into final output
-                    output = await _consume_aiter(aiterator)
-                    # sequence doesn't emit output, yield to mark as generator
-                    yield
-            except BaseException as e:
-                await run_manager.on_chain_error(e)
-                raise
-            else:
-                await run_manager.on_chain_end(output)
+            # No tracing - just execute the steps directly
+            async with AsyncExitStack() as stack:
+                for idx, step in enumerate(self.steps):
+                    if idx == 0:
+                        aiterator = step.astream(input, config, **kwargs)
+                    else:
+                        aiterator = step.atransform(aiterator, config)
+                    if hasattr(aiterator, "aclose"):
+                        stack.push_async_callback(aiterator.aclose)
+                # consume into final output
+                await _consume_aiter(aiterator)
+                yield
 
 
 def _consume_iter(it: Iterator[Any]) -> Any:

libs/langgraph/langgraph/pregel/_call.py

@@ -21,6 +21,7 @@ from langgraph._internal._runnable import (
     run_in_executor,
 )
 from langgraph.config import get_config
+from langgraph.pregel._read import _validate_traceable_config
 from langgraph.pregel._write import ChannelWrite, ChannelWriteEntry
 from langgraph.types import CachePolicy, RetryPolicy
 
@@ -192,6 +193,19 @@ def get_runnable_for_entrypoint(func: Callable[..., Any]) -> Runnable:
         return CACHE.setdefault(key, run)
 
 
+def _compose_trace_inputs(
+    base_fn: Callable[[Any], Any],
+    process_fn: Callable[[Any], Any],
+) -> Callable[[Any], Any]:
+    """Compose trace_inputs with process_inputs from traceable config."""
+
+    def composed(inputs: Any) -> Any:
+        exploded = base_fn(inputs)
+        return process_fn(exploded)
+
+    return composed
+
+
 def get_runnable_for_task(func: Callable[..., Any]) -> Runnable:
     key = (func, True)
     if key in CACHE:
@@ -206,10 +220,22 @@ def get_runnable_for_task(func: Callable[..., Any]) -> Runnable:
         else:
             name = str(func)
 
-        if is_async_callable(func):
+        # Check for traceable config early to handle __unwrapped__
+        raw_config = getattr(func, "__traceable_config__", None)
+        traceable_config = _validate_traceable_config(raw_config)
+
+        # Use unwrapped function if available to avoid double-tracing
+        # when @traceable and @task are used together
+        func_to_run: Callable[..., Any] = func
+        if traceable_config:
+            unwrapped = traceable_config.get("__unwrapped__")
+            if unwrapped is not None:
+                func_to_run = unwrapped
+
+        if is_async_callable(func_to_run):
             run = RunnableCallable(
                 None,
-                func,
+                func_to_run,
                 explode_args=True,
                 name=name,
                 trace=False,
@@ -217,20 +243,45 @@ def get_runnable_for_task(func: Callable[..., Any]) -> Runnable:
             )
         else:
             run = RunnableCallable(
-                func,
-                functools.wraps(func)(functools.partial(run_in_executor, None, func)),
+                func_to_run,
+                functools.wraps(func_to_run)(
+                    functools.partial(run_in_executor, None, func_to_run)
+                ),
                 explode_args=True,
                 name=name,
                 trace=False,
                 recurse=False,
             )
+
+        # Build trace_inputs - compose with process_inputs if provided
+        # Use original func for signature to match expected args
+        base_trace_inputs = functools.partial(
+            _explode_args_trace_inputs, inspect.signature(func)
+        )
+        process_inputs = (
+            traceable_config.get("process_inputs") if traceable_config else None
+        )
+        if process_inputs is not None:
+            trace_inputs = _compose_trace_inputs(base_trace_inputs, process_inputs)
+        else:
+            trace_inputs = base_trace_inputs
+
+        # Build seq_kwargs
+        seq_kwargs: dict[str, Any] = {
+            "name": name,
+            "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["skip_langsmith"] = True
+            if traceable_config.get("process_outputs"):
+                seq_kwargs["trace_outputs"] = traceable_config["process_outputs"]
+
         seq = RunnableSeq(
             run,
             ChannelWrite([ChannelWriteEntry(RETURN)]),
-            name=name,
-            trace_inputs=functools.partial(
-                _explode_args_trace_inputs, inspect.signature(func)
-            ),
+            **seq_kwargs,
         )
         if not _lookup_module_and_qualname(func):
             return seq

libs/langgraph/langgraph/pregel/_read.py

@@ -7,10 +7,15 @@ from typing import (
 )
 
 from langchain_core.runnables import Runnable, RunnableConfig
+from typing_extensions import TypedDict
 
 from langgraph._internal._config import merge_configs
 from langgraph._internal._constants import CONF, CONFIG_KEY_READ
-from langgraph._internal._runnable import RunnableCallable, RunnableSeq
+from langgraph._internal._runnable import (
+    RunnableCallable,
+    RunnableSeq,
+    coerce_to_runnable,
+)
 from langgraph.pregel._utils import find_subgraph_pregel
 from langgraph.pregel._write import ChannelWrite
 from langgraph.pregel.protocol import PregelProtocol
@@ -20,6 +25,32 @@ READ_TYPE = Callable[[str | Sequence[str], bool], Any | dict[str, Any]]
 INPUT_CACHE_KEY_TYPE = tuple[Callable[..., Any], tuple[str, ...]]
 
 
+class TraceableConfig(TypedDict, total=False):
+    """Configuration extracted from @traceable decorated functions."""
+
+    process_inputs: Callable[[Any], Any] | None
+    process_outputs: Callable[[Any], Any] | None
+    enabled: bool | None
+    __unwrapped__: Callable[[Any], Any] | None
+
+
+def _validate_traceable_config(raw: Any) -> TraceableConfig | None:
+    """Validate __traceable_config__ has expected structure.
+
+    Returns validated config dict or None if invalid.
+    """
+    if not isinstance(raw, dict):
+        return None
+    # Support both "__unwrapped__" (langsmith) and "wrapped" (legacy) keys
+    unwrapped = raw.get("__unwrapped__") or raw.get("wrapped")
+    return {
+        "process_inputs": raw.get("process_inputs"),
+        "process_outputs": raw.get("process_outputs"),
+        "enabled": raw.get("enabled"),  # None means use external context
+        "__unwrapped__": unwrapped,
+    }
+
+
 class ChannelRead(RunnableCallable):
     """Implements the logic for reading state from CONFIG_KEY_READ.
     Usable both as a runnable as well as a static method to call imperatively."""
@@ -176,6 +207,7 @@ class PregelNode:
         attrs = {**self.__dict__, **update}
         # Drop the cached properties
         attrs.pop("flat_writers", None)
+        attrs.pop("_traceable_config", None)
         attrs.pop("node", None)
         attrs.pop("input_cache_key", None)
         return PregelNode(**attrs)
@@ -197,6 +229,31 @@ class PregelNode:
             writers.pop()
         return writers
 
+    @cached_property
+    def _traceable_config(self) -> TraceableConfig | None:
+        """Extract @traceable config if bound wraps a traceable function.
+
+        Returns validated TraceableConfig with:
+        - process_inputs: Optional input processing function
+        - process_outputs: Optional output processing function
+        - enabled: Whether tracing is enabled (None = honor external context)
+        - __unwrapped__: The original unwrapped function (if available)
+        """
+        if not isinstance(self.bound, RunnableCallable):
+            return None
+
+        func = self.bound.func or self.bound.afunc
+        if func is None:
+            return None
+
+        raw_config = getattr(func, "__traceable_config__", None)
+        if raw_config is None:
+            return None
+        config = _validate_traceable_config(raw_config)
+        if config is None:
+            return None
+        return config
+
     @cached_property
     def node(self) -> Runnable[Any, Any] | None:
         """Get a runnable that combines `bound` and `writers`."""
@@ -207,10 +264,24 @@ class PregelNode:
             return writers[0]
         elif self.bound is DEFAULT_BOUND:
             return RunnableSeq(*writers)
-        elif writers:
-            return RunnableSeq(self.bound, *writers)
-        else:
+        elif not writers:
             return self.bound
+        tc = self._traceable_config
+
+        seq_kwargs: dict[str, Any] = {}
+        if tc:
+            # Filter out LangChainTracer (skip LangSmith) but keep other callbacks
+            if tc.get("enabled") is False:
+                seq_kwargs["skip_langsmith"] = True
+            seq_kwargs["trace_inputs"] = tc.get("process_inputs")
+            seq_kwargs["trace_outputs"] = tc.get("process_outputs")
+
+        bound = self.bound
+        if tc and (unwrapped := tc.get("__unwrapped__")):
+            # We want to avoid double-tracing.
+            bound = coerce_to_runnable(unwrapped, name=None, trace=False)
+
+        return RunnableSeq(bound, *writers, **seq_kwargs)
 
     @cached_property
     def input_cache_key(self) -> INPUT_CACHE_KEY_TYPE:
@@ -218,9 +289,11 @@ class PregelNode:
         This is used to avoid calculating the same input multiple times."""
         return (
             self.mapper,
-            tuple(self.channels)
-            if isinstance(self.channels, list)
-            else (self.channels,),
+            (
+                tuple(self.channels)
+                if isinstance(self.channels, list)
+                else (self.channels,)
+            ),
         )
 
     def invoke(