Runnables and LangChain Expression Language in depth

RunnableLambda, RunnablePassthrough, parallel and branching runnables, fallbacks, retries, runtime configuration and streaming through a chain.

The Runnable protocol

from langchain_core.runnables import (
    RunnableLambda, RunnablePassthrough, RunnableParallel, RunnableBranch
)

# any function becomes a runnable; keep it pure and small
normalise = RunnableLambda(lambda x: x.strip().lower())
word_count = RunnableLambda(lambda x: len(x.split()))

# RunnableParallel runs its branches on the SAME input and returns a dict
enrich = RunnableParallel(text=RunnablePassthrough(), words=word_count)
print(enrich.invoke("  Hello World  "))

# RunnablePassthrough.assign adds keys while preserving the input
from langchain_core.runnables import RunnablePassthrough as RP
builder = RP.assign(clean=normalise) | RP.assign(length=RunnableLambda(lambda d: len(d["clean"])))
print(builder.invoke({"clean": "raw", "other": 1}))

# branching: the first predicate that returns True wins
router = RunnableBranch(
    (lambda x: "refund" in x["text"].lower(), RunnableLambda(lambda x: {"intent": "refund"})),
    (lambda x: "password" in x["text"].lower(), RunnableLambda(lambda x: {"intent": "account"})),
    RunnableLambda(lambda x: {"intent": "other"}),     # default, must be last
)
print(router.invoke({"text": "I need a password reset"}))

• Every runnable supports invoke, batch, stream, ainvoke, abatch and astream. That uniformity is what lets you swap a function for a model or a retriever freely.
• The pipe operator feeds the left output into the right input. If the right side is a dict, each value becomes a branch fed the same input.
• RunnableBranch needs a default as its final argument, otherwise an unmatched input raises at runtime rather than at construction.
• Keep functions in RunnableLambda pure and quick. A network call inside one is fine; a side effect that mutates shared state is not, because it may run more than once under retries.

Retries, fallbacks and configuration

from langchain_core.runnables import RunnableConfig

# retry a failing step
flaky = model.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
    retry_if_exception_type=(TimeoutError, ConnectionError),
)

# a fallback chain: primary model, then a cheaper one, then a static answer
robust = flaky.with_fallbacks(
    [cheaper_model, RunnableLambda(lambda _: "I could not reach the model. Please retry.")],
    exceptions_to_handle=(Exception,),
)

# runtime configuration flows down to every step that wants it
config: RunnableConfig = {
    "configurable": {"user_id": "u-88", "thread_id": "t-12"},
    "run_name": "support_reply",
    "tags": ["production", "v3"],
    "metadata": {"release": "2026-09-18"},
    "max_concurrency": 4,
}
out = robust.invoke("Summarise the incident.", config=config)

# read the configuration inside a custom runnable
def personalised(prompt: str, config: RunnableConfig):
    user_id = config["configurable"].get("user_id", "anonymous")
    return f"[user {user_id}] {prompt}"

chain = RunnableLambda(personalised) | model

Streaming through a composed chain

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

prompt = ChatPromptTemplate.from_template("Explain {topic} in three sentences.")
chain = prompt | model | StrOutputParser()

# stream the final text out of a multi-step chain
for token in chain.stream({"topic": "idempotency keys"}):
    print(token, end="", flush=True)

# when intermediate steps matter, stream events instead of text
async def trace_chain(topic):
    async for event in chain.astream_events({"topic": topic}, version="v2"):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            print(event["data"]["chunk"].content, end="", flush=True)
        elif kind == "on_chain_start":
            print(f"\n[start {event['name']}]", flush=True)

import asyncio
asyncio.run(trace_chain("rate limiting"))

# event streaming also works inside a retrieval step, so you can show sources first
def stream_with_sources(question):
    return chain.stream({"topic": question})

• chain.stream yields the output of the last step only, and only if every intermediate step can stream. A step that buffers its whole output breaks streaming for the chain.
• astream_events gives you every step: chain starts, model tokens, retriever results, tool calls. It is the tool for building an interface that shows progress.
• Blocking work inside an async chain stalls the event loop. Wrap synchronous clients with asyncio.to_thread or use the native async client.
• Streaming and structured output conflict: a partially parsed JSON object is not a valid object. Stream free text, or stream field-by-field with a tolerant parser.

FAQ

Related

Prompt templates and chains Observability with callbacks and LangSmith

Last refreshed 2026-09-18.