Implementation:Langchain ai Langgraph Internal Runnable

Knowledge Sources	Langchain_ai_Langgraph
Domains	Internal, Runtime
Last Updated	2026-02-11 16:00 GMT

Overview

The internal runnable module provides `RunnableCallable` and `RunnableSeq`, the core wrapper classes that make any callable or sequence of callables into LangChain-compatible `Runnable` objects with full tracing, context propagation, and runtime dependency injection.

Description

`RunnableCallable` is the fundamental bridge between plain Python functions (sync or async) and LangChain's `Runnable` protocol. When a function is registered as a graph node, task, or tool, it is wrapped in a `RunnableCallable` that automatically handles: invocation with tracing (creating chain start/end/error callbacks for LangSmith observability), context propagation (setting child config and tracing context via `set_config_context`), and runtime dependency injection for special keyword arguments like `config`, `writer`, `store`, `previous`, and `runtime`. The injection system inspects the function's signature at construction time, matching parameter names and type annotations against `KWARGS_CONFIG_KEYS`, and then provides the appropriate values from the runtime configuration at call time.

`RunnableSeq` is a simplified internal version of LangChain's `RunnableSequence` that chains multiple `Runnable` steps where the output of each becomes the input of the next. It handles tracing for the entire sequence as a single chain run, with each step tracked as a child run. The first step receives full context propagation (running in a copied context with config and tracing set up), while subsequent steps (typically stream writers) run without context overhead. `RunnableSeq` supports `invoke`, `ainvoke`, `stream`, and `astream` operations, with the stream variants consuming iterators/async iterators through transform chains.

The module also provides utility functions: `coerce_to_runnable()` converts arbitrary callable-like objects (functions, async functions, generators, dicts) into proper `Runnable` instances, `is_async_callable()` and `is_async_generator()` provide type-narrowing checks, and `set_config_context()` is a context manager that properly sets and resets both the LangChain config context variable and LangSmith tracing context. The `StrEnum` class provides a Python 3.10-compatible string enum, and `ASYNCIO_ACCEPTS_CONTEXT` flags whether the runtime supports passing context to `asyncio.create_task` (Python 3.11+).

Usage

Use `RunnableCallable` when you need to wrap a function as a graph node or tool with automatic tracing and dependency injection. Use `RunnableSeq` when composing a node with its associated stream writers into a single traced sequence. Use `coerce_to_runnable()` to normalize heterogeneous callable inputs into the `Runnable` interface. These are internal classes used extensively by the LangGraph graph compiler and runtime.

Code Reference

Source Location

• Repository: Langchain_ai_Langgraph
• File: libs/langgraph/langgraph/_internal/_runnable.py

Signature

class RunnableCallable(Runnable):
    """A much simpler version of RunnableLambda that requires sync and async functions."""

    def __init__(
        self,
        func: Callable[..., Any | Runnable] | None,
        afunc: Callable[..., Awaitable[Any | Runnable]] | None = None,
        *,
        name: str | None = None,
        tags: Sequence[str] | None = None,
        trace: bool = True,
        recurse: bool = True,
        explode_args: bool = False,
        **kwargs: Any,
    ) -> None: ...

    def invoke(self, input: Any, config: RunnableConfig | None = None, **kwargs: Any) -> Any: ...
    async def ainvoke(self, input: Any, config: RunnableConfig | None = None, **kwargs: Any) -> Any: ...

class RunnableSeq(Runnable):
    """Sequence of Runnable, where the output of each is the input of the next."""

    def __init__(
        self,
        *steps: RunnableLike,
        name: str | None = None,
        trace_inputs: Callable[[Any], Any] | None = None,
    ) -> None: ...

    def invoke(self, input: Input, config: RunnableConfig | None = None, **kwargs: Any) -> Any: ...
    async def ainvoke(self, input: Input, config: RunnableConfig | None = None, **kwargs: Any) -> Any: ...
    def stream(self, input: Input, config: RunnableConfig | None = None, **kwargs: Any) -> Iterator[Any]: ...
    async def astream(self, input: Input, config: RunnableConfig | None = None, **kwargs: Any) -> AsyncIterator[Any]: ...

def coerce_to_runnable(thing: RunnableLike, *, name: str | None, trace: bool) -> Runnable: ...
def is_async_callable(func: Any) -> TypeGuard[Callable[..., Awaitable]]: ...
def is_async_generator(func: Any) -> TypeGuard[Callable[..., AsyncIterator]]: ...

Import

from langgraph._internal._runnable import RunnableCallable

I/O Contract

RunnableCallable Constructor Parameters
Parameter	Type	Default	Description
`func`	None`	required	Synchronous function to wrap
`afunc`	None`	`None`	Asynchronous function to wrap
`name`	None`	`None`	Name for tracing; auto-detected from function name if not provided
`tags`	None`	`None`	Tags for tracing and callback management
`trace`	`bool`	`True`	Whether to create tracing spans for invocations
`recurse`	`bool`	`True`	Whether to invoke the result if it is itself a `Runnable`
`explode_args`	`bool`	`False`	Whether to unpack input as `(args, kwargs)` tuple
`**kwargs`	`Any`	--	Additional keyword arguments passed through to the wrapped function

Injectable Keyword Arguments (KWARGS_CONFIG_KEYS)
Kwarg Name	Expected Types	Runtime Key	Description
`config`	`RunnableConfig`, `Optional[RunnableConfig]`	N/A (injected directly)	The current `RunnableConfig` for the execution
`writer`	`StreamWriter`	`stream_writer`	A callable for writing to the output stream
`store`	`BaseStore`, `Optional[BaseStore]`	`store`	The long-term memory store instance
`previous`	Any type	`previous`	The previous node return value (for retries/resumption)
`runtime`	Any type	N/A (injected directly)	The full runtime object

RunnableSeq Constructor Parameters
Parameter	Type	Default	Description
`*steps`	`RunnableLike`	required	Two or more steps to chain in sequence
`name`	None`	`None`	Name for the sequence in tracing
`trace_inputs`	None`	`None`	Optional transform applied to inputs before tracing

Usage Examples

from langgraph._internal._runnable import RunnableCallable, RunnableSeq, coerce_to_runnable

# Wrap a sync function as a RunnableCallable
def process_data(state: dict) -> dict:
    return {"result": state["input"] * 2}

runnable = RunnableCallable(process_data, name="process_data")
result = runnable.invoke({"input": 21})  # {"result": 42}

# Wrap an async function
async def async_process(state: dict) -> dict:
    return {"result": state["input"] + 1}

async_runnable = RunnableCallable(None, async_process, name="async_process")
result = await async_runnable.ainvoke({"input": 41})  # {"result": 42}

# Function with dependency injection (config and store)
from langchain_core.runnables import RunnableConfig
from langgraph.store.base import BaseStore

def node_with_deps(state: dict, *, config: RunnableConfig, store: BaseStore) -> dict:
    # config and store are automatically injected at runtime
    thread_id = config["configurable"]["thread_id"]
    item = store.get(("user", thread_id), "profile")
    return {"profile": item}

node_runnable = RunnableCallable(node_with_deps, name="node_with_deps")

# Create a sequence of steps
step1 = RunnableCallable(lambda x: x + 1)
step2 = RunnableCallable(lambda x: x * 2)
seq = RunnableSeq(step1, step2, name="double_after_increment")
result = seq.invoke(5)  # (5 + 1) * 2 = 12

# Coerce a plain function into a Runnable
runnable = coerce_to_runnable(lambda x: x.upper(), name="uppercase", trace=True)
result = runnable.invoke("hello")  # "HELLO"

Related Pages

• Principle:Langchain_ai_Langgraph_Runnable_Wrapping