langchain/openwiki/composability.md

type, title, description, tags, verified, sources, generated

type	title	description	tags	verified	sources	generated
Concept	Composability and LCEL Chains	How Runnable components compose through LCEL operators, creating reusable workflows with automatic async, batch, and streaming support.	composability LCEL runnables chaining operators	by at openwiki/0.5.0 2026-09-21T08:30:16.745Z	id resource openwiki-source-a1981e868973f6fd7f71e12e repo://libs/core/langchain_core/runnables/base.py id resource openwiki-source-48e94bbe49ab4f33ba87e9cb repo://libs/core/langchain_core/runnables/branch.py id resource openwiki-source-f9f4c1dc4f9cdf80d824ce15 repo://libs/core/langchain_core/runnables/fallbacks.py id resource openwiki-source-de6c904bd0171642bd50f6d9 repo://libs/core/langchain_core/runnables/router.py	by at openwiki/0.5.0 2026-09-21T08:30:16.745Z

Overview

Composability is the core feature of LangChain's Runnable protocol: the ability to declaratively chain, parallelize, and conditionally route components. Every composed chain automatically inherits sync (invoke), async (ainvoke), batch (batch/abatch), and streaming (stream/astream) capabilities—with optimizations for efficiency.

The two main composition primitives are RunnableSequence (sequential chaining via the | operator) and RunnableParallel (parallel execution via dict syntax). Conditional routing is achieved with RunnableBranch and RouterRunnable.

Sequential Composition: The | Operator

The pipe operator (|) chains Runnables in sequence, with each step's output becoming the next step's input. This is the most common composition pattern.

from langchain_core.runnables import RunnableLambda

add_one = RunnableLambda(lambda x: x + 1)
mul_two = RunnableLambda(lambda x: x * 2)

sequence = add_one | mul_two
sequence.invoke(1)  # (1 + 1) * 2 = 4

The | operator creates a RunnableSequence, which:

• Invokes each step in order, passing output to the next input
• Flattens nested sequences for efficiency
• Automatically preserves streaming properties if all steps support the transform method
• Supports both sync and async execution

Data Flow

Input → Step 1 → Step 2 → Step 3 → Output

When a dict is piped into a sequence, it becomes a RunnableParallel:

sequence = add_one | {
    "mul_2": RunnableLambda(lambda x: x * 2),
    "mul_5": RunnableLambda(lambda x: x * 5),
}
sequence.invoke(1)  # {'mul_2': 4, 'mul_5': 10}

Parallel Composition: Branching with + and Dict Syntax

Parallel execution invokes multiple Runnables concurrently on the same input. This is achieved via dict literals within a sequence or directly with RunnableParallel.

Dict Literal Syntax

from langchain_core.runnables import RunnableLambda, RunnableParallel

add_one = RunnableLambda(lambda x: x + 1)
mul_two = RunnableLambda(lambda x: x * 2)
mul_three = RunnableLambda(lambda x: x * 3)

# Dict syntax creates a RunnableParallel
sequence = add_one | {
    "mul_2": mul_two,
    "mul_3": mul_three,
}
sequence.invoke(1)
# Output: {'mul_2': 4, 'mul_3': 6}

Explicit RunnableParallel

parallel = RunnableParallel(
    mul_2=mul_two,
    mul_3=mul_three,
)
parallel.invoke(2)
# Output: {'mul_2': 4, 'mul_3': 6}

Concurrent Execution

• RunnableParallel creates independent input copies for each branch using atee (async) or safetee (sync)
• Each branch executes concurrently, with chunks yielded in the order they complete
• For async streaming, tasks are managed with asyncio.wait(return_when=FIRST_COMPLETED) to emit output as soon as any branch produces a chunk
• The final result is a dict combining outputs from all branches

Batching: Parallel Invocation over Multiple Inputs

Batching processes multiple inputs efficiently through a pipeline. Unlike parallel branching, batching applies the same sequence to each input in parallel.

Sync Batch

sequence = add_one | mul_two
results = sequence.batch([1, 2, 3])
# [4, 6, 8]  # Each input processed in parallel via thread pool

Async Batch

results = await sequence.abatch([1, 2, 3])
# [4, 6, 8]

Implementation

• Default batch uses a thread pool executor via get_executor_for_config
• abatch uses asyncio.gather with concurrency control via max_concurrency
• Each step in the sequence batches its inputs independently
• RunnableSequence calls batch on each step in order, feeding outputs to the next

Streaming: Token-by-Token Output

Streaming emits output chunks as they are produced, enabling real-time responses from LLMs and other sequential generators.

Stream Method

for chunk in sequence.stream(1):
    print(chunk)  # Intermediate outputs as they become available

Astream Method (Async)

async for chunk in sequence.astream(1):
    print(chunk)  # Non-blocking iteration

Streaming Pipeline

A RunnableSequence preserves streaming properties:

• If all steps implement transform (which processes Iterator[Input] → Iterator[Output]), streaming passes through the entire pipeline
• If any step doesn't support transform, streaming blocks until that step completes, then resumes
• RunnableLambda does not implement transform by default; use RunnableGenerator for custom streaming logic

Example: Prompt → Model → Parser

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("What is {topic}?")
model = ChatOpenAI()
parser = StrOutputParser()

chain = prompt | model | parser

# Stream tokens as the model generates them
for chunk in chain.stream({"topic": "composability"}):
    print(chunk, end="", flush=True)

In this chain:

1. ChatPromptTemplate formats the input dict into a string prompt
2. ChatOpenAI streams tokens as they arrive from the API
3. StrOutputParser passes tokens through unchanged

Tokens flow end-to-end without waiting for the full response.

Conditional Routing: RunnableBranch and RouterRunnable

Conditional logic routes inputs to different branches based on predicates.

RunnableBranch: Predicate-Based Routing

A RunnableBranch evaluates conditions in order and executes the first matching branch:

from langchain_core.runnables import RunnableBranch, RunnableLambda

branch = RunnableBranch(
    (lambda x: isinstance(x, int), RunnableLambda(lambda x: x * 2)),
    (lambda x: isinstance(x, str), RunnableLambda(lambda x: x.upper())),
    RunnableLambda(lambda x: "unknown"),
)

branch.invoke(5)        # 10
branch.invoke("hello")  # "HELLO"
branch.invoke(None)     # "unknown"

Conditions are evaluated sequentially; the first truthy result selects its corresponding Runnable. If no condition matches, the default branch executes.

RouterRunnable: Key-Based Routing

A RouterRunnable routes based on a string key in the input:

from langchain_core.runnables.router import RouterRunnable

add = RunnableLambda(lambda x: x + 1)
square = RunnableLambda(lambda x: x ** 2)

router = RouterRunnable(runnables={"add": add, "square": square})
router.invoke({"key": "square", "input": 3})  # 9
router.invoke({"key": "add", "input": 3})     # 4

The input is a dict with "key" (which Runnable to route to) and "input" (the data).

Composition with RunnablePassthrough

RunnablePassthrough forwards inputs unchanged or with additional keys, useful for preserving context in parallel branches:

from langchain_core.runnables import RunnablePassthrough

chain = (
    RunnableLambda(lambda x: x + 1)
    | {
        "original": RunnablePassthrough(),
        "modified": RunnableLambda(lambda x: x * 2),
    }
)

chain.invoke(5)
# {'original': 6, 'modified': 12}

Here, the passthrough preserves the intermediate result for reuse by another branch.

Async Equivalents

Every method has an async counterpart:

Sync	Async
invoke(input)	ainvoke(input)
batch(inputs)	abatch(inputs)
stream(input)	astream(input)
transform(Iterator[Input])	atransform(AsyncIterator[Input])

Async methods integrate with the callback system and execute concurrency-aware batching via asyncio.gather.

Variable Binding and Context Flow

In composed chains, data flows through steps along with execution context. Each step receives the output of the previous step as its input.

Context Propagation

When a chain invokes, RunnableSequence creates a callback hierarchy for tracing:

• Each step is marked as a child run using run_manager.get_child(f"seq:step:{i + 1}")
• Callbacks, tags, and metadata flow through the chain via RunnableConfig
• patch_config updates the config for each step while preserving parent context

from langchain_core.runnables import RunnableLambda

# Context flows through each step
step1 = RunnableLambda(lambda x: x + 1)
step2 = RunnableLambda(lambda x: x * 2)
chain = step1 | step2

# Invoke with tracing config
result = chain.invoke(
    5, 
    config={
        "run_name": "my_chain",
        "callbacks": [my_tracer],
        "tags": ["prod"],
    }
)
# Each step runs with inherited config while reporting to callbacks

Dict Composition and Key Selection

When using dict syntax in a sequence, each dict key becomes a separate branch context:

chain = step1 | {
    "result_a": step2,
    "result_b": step3,
}

# Output combines results from both branches
output = chain.invoke(input)  # {'result_a': ..., 'result_b': ...}

Each branch (result_a, result_b) appears as a separate child run in the callback trace.

Fallback Patterns

Fallbacks provide resilience by retrying with alternative Runnables when one fails.

Fallback at Component Level

from langchain_core.runnables import RunnableLambda

primary_llm = ChatOpenAI(model="gpt-4")
fallback_llm = ChatAnthropic(model="claude-3-sonnet")

resilient_llm = primary_llm.with_fallbacks(
    [fallback_llm],
    exceptions_to_handle=(APIConnectionError,),
)

output = resilient_llm.invoke("What is composability?")
# Uses primary_llm; falls back to fallback_llm if APIConnectionError occurs

Fallback at Chain Level

# Construct a chain with fallback
chain_with_fallback = (
    prompt 
    | resilient_llm 
    | parser
).with_fallbacks([
    RunnableLambda(lambda x: "Service unavailable")
])

output = chain_with_fallback.invoke({"topic": "composability"})
# If the entire chain fails, returns fallback response

Multiple Fallbacks

Fallbacks are tried in order until one succeeds:

model = ChatOpenAI().with_fallbacks([
    ChatAnthropic(),        # Try second
    ChatClaude(),          # Try third
    ChatCohere(),          # Try fourth
    RunnableLambda(default_response),  # Final fallback
])

The chain tries each fallback sequentially until one returns successfully or all are exhausted.

Chaining Patterns

Common Pattern: Prompt → Model → Parser

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

chain = (
    ChatPromptTemplate.from_template("What is {topic}?")
    | ChatOpenAI()
    | StrOutputParser()
)

# Single invoke
output = chain.invoke({"topic": "LLMs"})

# Batch process
outputs = chain.batch([{"topic": "LLMs"}, {"topic": "Vectors"}])

# Stream tokens
for chunk in chain.stream({"topic": "LLMs"}):
    print(chunk, end="", flush=True)

Fan-Out / Fan-In: Parallel Processing

from langchain_core.runnables import RunnableLambda, RunnablePassthrough

chain = (
    RunnablePassthrough()
    | {
        "summary": RunnableLambda(summarize),
        "entities": RunnableLambda(extract_entities),
        "sentiment": RunnableLambda(analyze_sentiment),
    }
)

result = chain.invoke(text)
# {'summary': '...', 'entities': [...], 'sentiment': 'positive'}

Conditional Execution

from langchain_core.runnables import RunnableBranch

route_logic = RunnableBranch(
    (lambda x: "math" in x.lower(), math_chain),
    (lambda x: "code" in x.lower(), code_chain),
    general_chain,
)

output = route_logic.invoke("How do I calculate factorial?")

Type Safety and Schema Inference

Chains infer input and output types from their components:

sequence = add_one | mul_two

# Access inferred schemas
print(sequence.input_schema)   # Pydantic model for input
print(sequence.output_schema)  # Pydantic model for output
print(sequence.input_schema.model_json_schema())

This enables validation and documentation without explicit type annotations.

Optimization and Flattening

RunnableSequence automatically flattens nested sequences:

# These are equivalent:
chain1 = step1 | step2 | step3
chain2 = step1 | (step2 | step3)
chain3 = (step1 | step2) | step3

All produce a single flat sequence with steps [step1, step2, step3], avoiding unnecessary nesting overhead.

Serialization and Debugging

Composed chains support serialization via the LangChain serialization system, enabling:

• Persistence: Save and load chains
• Tracing: Automatic callback integration for debugging via LangSmith
• Inspection: Use get_graph() to visualize chain structure

Enable debug output:

from langchain_core.globals import set_debug

set_debug(True)  # Print intermediate results
chain.invoke(input)

# Or use callbacks:
from langchain_core.tracers import ConsoleCallbackHandler

chain.invoke(input, config={"callbacks": [ConsoleCallbackHandler()]})

Extension: Custom Runnables

Implement Runnable to create custom components:

from langchain_core.runnables import Runnable, RunnableConfig
from typing import Iterator

class CustomRunnable(Runnable[str, int]):
    def invoke(self, input: str, config: RunnableConfig | None = None) -> int:
        return len(input)
    
    async def ainvoke(self, input: str, config: RunnableConfig | None = None) -> int:
        return len(input)
    
    def stream(self, input: str, config: RunnableConfig | None = None) -> Iterator[int]:
        # For streaming support, implement transform
        for char in input:
            yield 1
    
    async def astream(self, input: str, config: RunnableConfig | None = None):
        for char in input:
            yield 1

# Immediately composable
chain = CustomRunnable() | another_step

Custom Runnables are automatically compatible with all composition operators.

Summary Table

Operator	Effect	Example
|	Sequential chaining	step1 | step2
Dict in sequence	Parallel branching	step1 | {key1: step2, key2: step3}
RunnableBranch	Conditional routing	RunnableBranch((cond, runnable), default)
RouterRunnable	Key-based routing	RouterRunnable({"key": runnable})
.batch() / .abatch()	Parallel input processing	chain.batch([in1, in2])
.stream() / .astream()	Token-by-token output	for chunk in chain.stream(input):

See the Runnables page for protocol details and method signatures.