From 2d304b8a1b3850904e535a64c76df09bc1554f48 Mon Sep 17 00:00:00 2001 From: William Fu-Hinthorn <13333726+hinthornw@users.noreply.github.com> Date: Wed, 7 Jan 2026 16:02:54 -0800 Subject: [PATCH] cleanup --- .../langgraph/_internal/_runnable.py | 19 ++----- libs/langgraph/langgraph/pregel/_read.py | 51 ++++++++++--------- 2 files changed, 30 insertions(+), 40 deletions(-) diff --git a/libs/langgraph/langgraph/_internal/_runnable.py b/libs/langgraph/langgraph/_internal/_runnable.py index 15938bdae..45e00eaaf 100644 --- a/libs/langgraph/langgraph/_internal/_runnable.py +++ b/libs/langgraph/langgraph/_internal/_runnable.py @@ -66,32 +66,24 @@ logger = logging.getLogger(__name__) def _safe_process_inputs(processor: Callable[[Any], Any] | None, inputs: Any) -> Any: - """Safely process trace inputs, returning error placeholder on failure. - - This prevents PII leakage if a filter function crashes - we return an error - indicator instead of the raw (potentially sensitive) inputs. - """ + """Safely process trace inputs, returning error placeholder on failure.""" if processor is None: return inputs try: return processor(inputs) except Exception: - logger.warning("trace_inputs filter failed", exc_info=True) + logger.exception("trace_inputs filter failed") return {"error": ""} def _safe_process_outputs(processor: Callable[[Any], Any] | None, outputs: Any) -> Any: - """Safely process trace outputs, returning error placeholder on failure. - - This prevents PII leakage if a filter function crashes - we return an error - indicator instead of the raw (potentially sensitive) outputs. - """ + """Safely process trace outputs, returning error placeholder on failure.""" if processor is None: return outputs try: return processor(outputs) except Exception: - logger.warning("trace_outputs filter failed", exc_info=True) + logger.exception("trace_outputs filter failed") return {"error": ""} @@ -708,7 +700,6 @@ class RunnableSeq(Runnable): ) return input else: - # No tracing - just execute the steps directly for i, step in enumerate(self.steps): input = ( step.invoke(input, config, **kwargs) @@ -775,7 +766,6 @@ class RunnableSeq(Runnable): ) return input else: - # No tracing - just execute the steps directly for i, step in enumerate(self.steps): if i == 0: input = await step.ainvoke(input, config, **kwargs) @@ -855,7 +845,6 @@ class RunnableSeq(Runnable): iterator = step.stream(input, config, **kwargs) else: iterator = step.transform(iterator, config) - # consume into final output _consume_iter(iterator) yield diff --git a/libs/langgraph/langgraph/pregel/_read.py b/libs/langgraph/langgraph/pregel/_read.py index 077de0843..f51f988a7 100644 --- a/libs/langgraph/langgraph/pregel/_read.py +++ b/libs/langgraph/langgraph/pregel/_read.py @@ -11,7 +11,11 @@ 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 @@ -26,7 +30,8 @@ class TraceableConfig(TypedDict, total=False): process_inputs: Callable[[Any], Any] | None process_outputs: Callable[[Any], Any] | None - enabled: bool | None # None = honor external tracing context + enabled: bool | None + __unwrapped__: Callable[[Any], Any] | None def _validate_traceable_config(raw: Any) -> TraceableConfig | None: @@ -239,11 +244,12 @@ class PregelNode: 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 - # Get the unwrapped original function (set by functools.wraps) unwrapped = getattr(func, "__wrapped__", None) if unwrapped: config["__unwrapped__"] = unwrapped @@ -253,38 +259,31 @@ class PregelNode: @cached_property def node(self) -> Runnable[Any, Any] | None: """Get a runnable that combines `bound` and `writers`.""" - from langgraph._internal._runnable import coerce_to_runnable - writers = self.flat_writers + if self.bound is DEFAULT_BOUND and not writers: + return None + elif self.bound is DEFAULT_BOUND and len(writers) == 1: + return writers[0] + elif self.bound is DEFAULT_BOUND: + return RunnableSeq(*writers) + elif not writers: + return self.bound tc = self._traceable_config - # Build RunnableSeq kwargs from traceable config seq_kwargs: dict[str, Any] = {} if tc: # Only disable tracing if explicitly set to False - # None means honor external context (keep default trace=True) if tc.get("enabled") is False: seq_kwargs["trace"] = False - # trace_inputs/trace_outputs map to @traceable's process_inputs/process_outputs seq_kwargs["trace_inputs"] = tc.get("process_inputs") seq_kwargs["trace_outputs"] = tc.get("process_outputs") - # Get the bound to use (unwrapped if traceable) bound = self.bound - if tc and tc.get("__unwrapped__"): - # Replace bound with unwrapped function to avoid double-tracing - bound = coerce_to_runnable(tc["__unwrapped__"], name=None, trace=False) + if tc and (unwrapped := tc.get("__unwrapped__")): + # We want to avoid double-tracing. + bound = coerce_to_runnable(unwrapped, name=None, trace=False) - if bound is DEFAULT_BOUND and not writers: - return None - elif bound is DEFAULT_BOUND and len(writers) == 1: - return writers[0] - elif bound is DEFAULT_BOUND: - return RunnableSeq(*writers, **seq_kwargs) - elif writers: - return RunnableSeq(bound, *writers, **seq_kwargs) - else: - return bound + return RunnableSeq(bound, *writers, **seq_kwargs) @cached_property def input_cache_key(self) -> INPUT_CACHE_KEY_TYPE: @@ -292,9 +291,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(