perf: cache inspect.signature results in RunnableCallable.__init__

Avoids repeated signature introspection for the same function (e.g. 1000
ChannelWrite instances all inspecting the same _write/_awrite methods).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

libs/langgraph/langgraph/_internal/_runnable.py

@@ -142,6 +142,9 @@ ANY_TYPE = object()
 
 ASYNCIO_ACCEPTS_CONTEXT = sys.version_info >= (3, 11)
 
+# Cache for inspect.signature results, keyed by function object
+_SIGNATURE_CACHE: dict[Callable, inspect.Signature] = {}
+
 # List of keyword arguments that can be injected into nodes / tasks / tools at runtime.
 # A named argument may appear multiple times if it appears with distinct types.
 KWARGS_CONFIG_KEYS: tuple[tuple[str, tuple[Any, ...], str, Any], ...] = (
@@ -315,7 +318,16 @@ class RunnableCallable(Runnable):
             raise ValueError("At least one of func or afunc must be provided.")
 
         self.func_accepts: dict[str, tuple[str, Any]] = {}
-        params = inspect.signature(cast(Callable, func or afunc)).parameters
+        target = cast(Callable, func or afunc)
+        try:
+            sig = _SIGNATURE_CACHE[target]
+        except (KeyError, TypeError):
+            sig = inspect.signature(target)
+            try:
+                _SIGNATURE_CACHE[target] = sig
+            except TypeError:
+                pass  # unhashable function, skip caching
+        params = sig.parameters
 
         for kw, typ, runtime_key, default in KWARGS_CONFIG_KEYS:
             p = params.get(kw)

libs/langgraph/langgraph/graph/message.py

@@ -191,8 +191,14 @@ def add_messages(
     if not isinstance(right, list):
         right = [right]  # type: ignore[assignment]
     # coerce to message
-    left = [message_chunk_to_message(m) for m in convert_to_messages(left)]
-    right = [message_chunk_to_message(m) for m in convert_to_messages(right)]
+    left = [
+        message_chunk_to_message(cast(BaseMessageChunk, m))
+        for m in convert_to_messages(left)
+    ]
+    right = [
+        message_chunk_to_message(cast(BaseMessageChunk, m))
+        for m in convert_to_messages(right)
+    ]
     # assign missing ids
     for m in left:
         if m.id is None:

libs/langgraph/langgraph/pregel/_algo.py

@@ -235,7 +235,6 @@ def apply_writes(
     tasks: Iterable[WritesProtocol],
     get_next_version: GetNextVersion | None,
     trigger_to_nodes: Mapping[str, Sequence[str]],
-    available_channels: set[str] | None = None,
 ) -> set[str]:
     """Apply writes from a set of tasks (usually the tasks from a Pregel step)
     to the checkpoint and channels, and return managed values writes to be applied
@@ -282,18 +281,6 @@ def apply_writes(
             None,
         )
 
-    # Sync available_channels with channel's actual availability state.
-    # Returns True if the channel is available (for callers that also need
-    # to update updated_channels).
-    def _track(chan: str) -> bool:
-        avail = channels[chan].is_available()
-        if available_channels is not None:
-            if avail:
-                available_channels.add(chan)
-            else:
-                available_channels.discard(chan)
-        return avail
-
     # Consume all channels that were read
     for chan in {
         chan
@@ -303,7 +290,6 @@ def apply_writes(
     }:
         if channels[chan].consume() and next_version is not None:
             checkpoint["channel_versions"][chan] = next_version
-        _track(chan)
 
     # Group writes by channel
     pending_writes_by_channel: dict[str, list[Any]] = defaultdict(list)
@@ -333,28 +319,18 @@ def apply_writes(
             if channels[chan].update(vals) and next_version is not None:
                 checkpoint["channel_versions"][chan] = next_version
                 # unavailable channels can't trigger tasks, so don't add them
-                if _track(chan):
+                if channels[chan].is_available():
                     updated_channels.add(chan)
-            else:
-                _track(chan)
 
     # Channels that weren't updated in this step are notified of a new step
     if bump_step:
-        candidates = (
-            available_channels - updated_channels
-            if available_channels is not None
-            else (
-                chan
-                for chan in channels
-                if channels[chan].is_available() and chan not in updated_channels
-            )
-        )
-        for chan in candidates:
-            if channels[chan].update(EMPTY_SEQ) and next_version is not None:
-                checkpoint["channel_versions"][chan] = next_version
-                # unavailable channels can't trigger tasks, so don't add them
-                if _track(chan):
-                    updated_channels.add(chan)
+        for chan in channels:
+            if channels[chan].is_available() and chan not in updated_channels:
+                if channels[chan].update(EMPTY_SEQ) and next_version is not None:
+                    checkpoint["channel_versions"][chan] = next_version
+                    # unavailable channels can't trigger tasks, so don't add them
+                    if channels[chan].is_available():
+                        updated_channels.add(chan)
 
     # If this is (tentatively) the last superstep, notify all channels of finish
     if bump_step and updated_channels.isdisjoint(trigger_to_nodes):
@@ -362,10 +338,8 @@ def apply_writes(
             if channels[chan].finish() and next_version is not None:
                 checkpoint["channel_versions"][chan] = next_version
                 # unavailable channels can't trigger tasks, so don't add them
-                if _track(chan):
+                if channels[chan].is_available():
                     updated_channels.add(chan)
-            else:
-                _track(chan)
 
     # Return managed values writes to be applied externally
     return updated_channels
@@ -543,7 +517,7 @@ PUSH_TRIGGER = (PUSH,)
 
 
 class _TaskIDFn(Protocol):
-    def __call__(self, namespace: bytes, *parts: str) -> str:
+    def __call__(self, namespace: bytes, *parts: str | bytes) -> str:
         pass
 
 
@@ -1418,37 +1392,32 @@ def _proc_input(
     return val
 
 
-def _uuid5_str(namespace: bytes, *parts: str) -> str:
+def _uuid5_str(namespace: bytes, *parts: str | bytes) -> str:
     """Generate a UUID from the SHA-1 hash of a namespace and str parts."""
 
     sha = sha1(namespace, usedforsecurity=False)
-    sha.update(b"".join(p.encode() for p in parts))
+    sha.update(b"".join(p.encode() if isinstance(p, str) else p for p in parts))
     hex = sha.hexdigest()
     return f"{hex[:8]}-{hex[8:12]}-{hex[12:16]}-{hex[16:20]}-{hex[20:32]}"
 
 
-def _xxhash_str(namespace: bytes, *parts: str) -> str:
+def _xxhash_str(namespace: bytes, *parts: str | bytes) -> str:
     """Generate a UUID from the XXH3 hash of a namespace and str parts."""
-    hex = xxh3_128_hexdigest(namespace + b"".join(p.encode() for p in parts))
+    hex = xxh3_128_hexdigest(
+        namespace + b"".join(p.encode() if isinstance(p, str) else p for p in parts)
+    )
     return f"{hex[:8]}-{hex[8:12]}-{hex[12:16]}-{hex[16:20]}-{hex[20:32]}"
 
 
-def task_path_str(tup: str | int | tuple | list) -> str:
+def task_path_str(tup: str | int | tuple) -> str:
     """Generate a string representation of the task path."""
-    if isinstance(tup, (tuple, list)):
-        parts: list[str] = []
-        for x in tup:
-            if isinstance(x, int):
-                parts.append(f"{x:010d}")
-            elif isinstance(x, (tuple, list)):
-                parts.append(task_path_str(x))
-            else:
-                parts.append(str(x))
-        return f"~{', '.join(parts)}"
-    elif isinstance(tup, int):
-        return f"{tup:010d}"
-    else:
-        return str(tup)
+    return (
+        f"~{', '.join(task_path_str(x) for x in tup)}"
+        if isinstance(tup, (tuple, list))
+        else f"{tup:010d}"
+        if isinstance(tup, int)
+        else str(tup)
+    )
 
 
 LAZY_ATOMIC_COUNTER_LOCK = threading.Lock()

libs/langgraph/langgraph/pregel/_loop.py

@@ -198,7 +198,7 @@ class PregelLoop:
     _migrate_checkpoint: Callable[[Checkpoint], None] | None
     submit: Submit
     channels: Mapping[str, BaseChannel]
-    _available_channels: set[str]
+    _has_untracked_channels: bool
     # Futures from `checkpointer.put_writes` calls that produced delta-channel
     # writes. `_checkpointer_put_after_previous` drains this list (swap to a
     # local `futs` then reset to `[]` and wait/gather) before putting the
@@ -438,9 +438,7 @@ class PregelLoop:
             writes_to_save = writes
 
         # check if any writes are to an UntrackedValue channel
-        if any(
-            isinstance(channel, UntrackedValue) for channel in self.channels.values()
-        ):
+        if self._has_untracked_channels:
             # we do not persist untracked values in checkpoints
             writes_to_save = [
                 # sanitize UntrackedValues that are nested within Send packets
@@ -696,7 +694,6 @@ class PregelLoop:
             self.tasks.values(),
             self.checkpointer_get_next_version,
             self.trigger_to_nodes,
-            available_channels=self._available_channels,
         )
         # produce values output
         if not self.updated_channels.isdisjoint(
@@ -941,7 +938,6 @@ class PregelLoop:
                 [PregelTaskWrites((), INPUT, null_writes, [])],
                 self.checkpointer_get_next_version,
                 self.trigger_to_nodes,
-                available_channels=self._available_channels,
             )
             if updated_channels is not None:
                 updated_channels.update(null_updated_channels)
@@ -1009,7 +1005,6 @@ class PregelLoop:
                 ],
                 self.checkpointer_get_next_version,
                 self.trigger_to_nodes,
-                available_channels=self._available_channels,
             )
             # Input writes go through `apply_writes` directly (above) — they
             # never enter `checkpoint_pending_writes`, so the after_tick
@@ -1165,9 +1160,7 @@ class PregelLoop:
         elif "counters_since_delta_snapshot" in self.checkpoint_metadata:
             del self.checkpoint_metadata["counters_since_delta_snapshot"]
         # sanitize TASK channel in the checkpoint before saving (durability=="exit")
-        if TASKS in self.checkpoint["channel_values"] and any(
-            isinstance(channel, UntrackedValue) for channel in self.channels.values()
-        ):
+        if TASKS in self.checkpoint["channel_values"] and self._has_untracked_channels:
             sanitized_tasks = [
                 sanitize_untracked_values_in_send(value, self.channels)
                 if isinstance(value, Send)
@@ -1353,7 +1346,6 @@ class PregelLoop:
                     self.tasks.values(),
                     self.checkpointer_get_next_version,
                     self.trigger_to_nodes,
-                    available_channels=self._available_channels,
                 )
                 if not updated_channels.isdisjoint(
                     (self.output_keys,)
@@ -1700,9 +1692,9 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             saver=self.checkpointer,
             config=self.checkpoint_config,
         )
-        self._available_channels: set[str] = {
-            k for k, v in self.channels.items() if v.is_available()
-        }
+        self._has_untracked_channels = any(
+            isinstance(ch, UntrackedValue) for ch in self.channels.values()
+        )
         self.stack.push(self._suppress_interrupt)
         self.status = "input"
         self.step = self.checkpoint_metadata["step"] + 1
@@ -1963,9 +1955,9 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             saver=self.checkpointer,
             config=self.checkpoint_config,
         )
-        self._available_channels: set[str] = {
-            k for k, v in self.channels.items() if v.is_available()
-        }
+        self._has_untracked_channels = any(
+            isinstance(ch, UntrackedValue) for ch in self.channels.values()
+        )
         self.stack.push(self._suppress_interrupt)
         self.status = "input"
         self.step = self.checkpoint_metadata["step"] + 1

libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py

@@ -638,18 +638,15 @@ def create_react_agent(
             messages = (
                 _get_state_value(state, "llm_input_messages")
             ) or _get_state_value(state, "messages")
+            error_msg = f"Expected input to call_model to have 'llm_input_messages' or 'messages' key, but got {state}"
         else:
             messages = _get_state_value(state, "messages")
+            error_msg = (
+                f"Expected input to call_model to have 'messages' key, but got {state}"
+            )
 
         if messages is None:
-            if pre_model_hook is not None:
-                raise ValueError(
-                    f"Expected input to call_model to have 'llm_input_messages' or 'messages' key, but got {state}"
-                )
-            else:
-                raise ValueError(
-                    f"Expected input to call_model to have 'messages' key, but got {state}"
-                )
+            raise ValueError(error_msg)
 
         _validate_chat_history(messages)
         # we're passing messages under `messages` key, as this is expected by the prompt