LCEL & Runnables

This content was generated with the assistance of AI and should be verified against the official documentation of each tool and framework before being relied on in production.

This section’s bibliography lists the reference material consulted while preparing these pages.

This page was written against langchain 1.4.3 and langchain-core 1.6.5 (with langgraph 1.2.12 for the last example), as checked on 2026-09-29. Every snippet was executed with those versions using GenericFakeChatModel from langchain_core instead of a real model, so the outputs shown are what the code printed with a fake model that returns canned replies. The production wiring points at DocsAssistant's Ollama llama3.1:8b chat model; only the model object changes.

Every LangChain building block — chat models, prompt templates, output parsers, retrievers, tools, and the graphs compiled by LangGraph — implements one small interface, the Runnable. The pipe operator that composes runnables into chains is called the LangChain Expression Language (LCEL). In LangChain 1.x, LCEL remains the way to wire a fixed sequence of steps, while the agent loop moved to create_agent (see Agents with create_agent) and the legacy prebuilt chains (LLMChain, create_retrieval_chain and similar) moved to the langchain-classic package (see Ecosystem & Packages). Chats, prompts and parsers are introduced in Chat Models & Messages and Prompts & Templates; this page is about how they connect.

The Runnable protocol

A Runnable is a unit of work that takes an input and produces an output, and exposes the same calling methods whatever it wraps:

Method Behaviour

Method	Behaviour
`invoke` / `ainvoke`	One input in, one output out.
`batch` / `abatch`	Many inputs; by default runs `invoke` for each on a thread pool, in parallel.
`stream` / `astream`	Yields output pieces as they are produced.
`astream_events`	Async stream of the events raised by every step, including nested ones.
`get_graph`	Returns the step graph, which can be drawn (a Mermaid rendering is shown below).

invoke / ainvoke

One input in, one output out.

batch / abatch

Many inputs; by default runs invoke for each on a thread pool, in parallel.

stream / astream

Yields output pieces as they are produced.

astream_events

Async stream of the events raised by every step, including nested ones.

get_graph

Returns the step graph, which can be drawn (a Mermaid rendering is shown below).

All of them take an optional config argument (see Config, tags and callbacks). Because the interface is uniform, a prompt, a model and an output parser can be swapped or reordered without the caller changing. The methods the class exposes are listed in the Runnable reference.

The smallest useful chain is a prompt, a model and an output parser, joined with |:

A prompt-model-parser chain with invoke, batch and stream

from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

model = GenericFakeChatModel(messages=iter([
    AIMessage(content="Use an HNSW index."),
    AIMessage(content="Use IVFFlat."),
    AIMessage(content="Chunk by heading."),
    AIMessage(content="Use IVFFlat."),
]))
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are DocsAssistant. Answer briefly."),
    ("human", "{question}"),
])
chain = prompt | model | StrOutputParser()

print(type(chain).__name__)
print(chain.invoke({"question": "Which pgvector index?"}))
print(chain.batch([{"question": "Second?"}, {"question": "Third?"}]))
for chunk in chain.stream({"question": "Fourth?"}):
    print(repr(chunk), end=" ")

Output with the fake model:

RunnableSequence
Use an HNSW index.
['Use IVFFlat.', 'Chunk by heading.']
'Use' ' ' 'IVFFlat.'

| builds a RunnableSequence, which is itself a Runnable, so a chain can be a step of a bigger chain. In production model becomes ChatOllama(model="llama3.1:8b", temperature=0) from langchain_ollama, which is also a Runnable; nothing else in the chain changes. Streaming works end to end because the fake model, like a real one, emits chunks and StrOutputParser passes them through as they arrive.

See LangChain — Models and the RunnableSequence reference.

The async twins have the same shape:

ainvoke and astream

import asyncio

model = GenericFakeChatModel(messages=iter([AIMessage(content="Use HNSW."), AIMessage(content="Use IVFFlat.")]))
chain = ChatPromptTemplate.from_template("{question}") | model | StrOutputParser()

async def main():
    print(await chain.ainvoke({"question": "Which index?"}))
    async for chunk in chain.astream({"question": "Again?"}):
        print(repr(chunk), end=" ")

asyncio.run(main())

Output with the fake model:

Use HNSW.
'Use' ' ' 'IVFFlat.'

Use the async methods inside a web server (see Streaming); a synchronous invoke called from an async handler blocks the event loop.

See Runnable — reference.

Composing with the pipe operator

The | operator is shorthand for the pipe method: the output of the left runnable becomes the input of the right one. The two sides may be Runnables or plain values LangChain knows how to coerce:

a plain Python function is wrapped in a RunnableLambda;
a dict whose values are runnables is wrapped in a RunnableParallel (see below).

Wrapping a function with RunnableLambda and mixing it into a chain

from langchain_core.runnables import RunnableLambda

def normalize(q: str) -> dict:
    return {"question": q.strip().rstrip("?") + "?"}

model = GenericFakeChatModel(messages=iter([AIMessage(content="Use HNSW.")]))
pipeline = (
    RunnableLambda(normalize)
    | ChatPromptTemplate.from_template("{question}")
    | model
    | StrOutputParser()
)
print(pipeline.invoke("  which index  "))

Output with the fake model:

Use HNSW.

RunnableLambda also turns a generator function into a streaming step (shown in Retries, fallbacks and streaming through custom steps).

See RunnableLambda reference.

The `@chain` decorator

@chain converts a function into a Runnable without writing RunnableLambda(…) by hand. It is also the way to write a custom step that itself calls other runnables, keeping the calls traceable as child runs.

@chain and composition

from langchain_core.runnables import RunnableLambda, chain

@chain
def shout(text: str) -> str:
    return text.upper()

print(type(shout).__name__, shout.invoke("hnsw"))
print((shout | RunnableLambda(len)).invoke("hnsw"))

# a dict of runnables after a step is coerced to a RunnableParallel
mapped = RunnableLambda(len) | {
    "n": RunnableLambda(lambda n: n * 2),
    "same": RunnableLambda(lambda n: n),
}
print(type(mapped).__name__, mapped.invoke("abc"))

Output:

RunnableLambda HNSW
4
RunnableSequence {'n': 6, 'same': 3}

See the chain reference.

Fan-out with RunnableParallel and RunnablePassthrough

RunnableParallel runs several runnables on the same input and returns a dict with one key per branch; the branches execute concurrently. RunnablePassthrough returns its input unchanged, which is how the original question is carried alongside the retrieved context. The classic retrieval-augmented chain is exactly that shape:

flowchart LR Q[question] --> P{RunnableParallel} P -->|context| R[retrieve] P -->|question| T[RunnablePassthrough] R --> F[ChatPromptTemplate] T --> F F --> M[chat model] M --> O[StrOutputParser] O --> A([answer])

Figure 1. The retrieval chain as a fan-out: one input feeds retrieval and a passthrough

A DocsAssistant retrieval chain using RunnableParallel and RunnablePassthrough

from operator import itemgetter

from langchain_core.runnables import RunnableLambda, RunnableParallel, RunnablePassthrough

def retrieve(question: str) -> str:
    # In DocsAssistant this calls docsassistant.search.search_docs (pgvector) and joins the hits.
    return "[vector-rag/ann-indexes] HNSW is a graph index; IVFFlat clusters vectors."

prompt = ChatPromptTemplate.from_template("Context:\n{context}\n\nQuestion: {question}")
model = GenericFakeChatModel(messages=iter([AIMessage(content="HNSW, per ann-indexes.")]))

rag = (
    RunnableParallel(context=RunnableLambda(retrieve), question=RunnablePassthrough())
    | prompt
    | model
    | StrOutputParser()
)
print(rag.invoke("Which index?"))

# the fan-out step on its own
print(RunnableParallel(context=RunnableLambda(retrieve), question=RunnablePassthrough()).invoke("q"))

# keep the whole input and add a field to it
with_ctx = RunnablePassthrough.assign(context=itemgetter("question") | RunnableLambda(retrieve))
print(with_ctx.invoke({"question": "Which index?"}))

Output with the fake model:

HNSW, per ann-indexes.
{'context': '[vector-rag/ann-indexes] HNSW is a graph index; IVFFlat clusters vectors.', 'question': 'q'}
{'question': 'Which index?', 'context': '[vector-rag/ann-indexes] HNSW is a graph index; IVFFlat clusters vectors.'}

RunnablePassthrough.assign(…) is the form to use when the input is already a dict: it keeps every existing key and adds the computed ones, so later prompt variables can read both. For the retrieval side of this chain, see Retrieval & the RAG Bridge and Integrating with LangChain.

See the RunnableParallel and RunnablePassthrough references.

A composed chain can draw itself. get_graph().draw_mermaid() renders the steps as Mermaid text, which is useful for checking that the wiring is what you meant:

Rendering a chain’s own graph

print(rag.get_graph().draw_mermaid())

The output is a graph TD listing of the nodes (the parallel input and output, retrieve, Passthrough, ChatPromptTemplate, the model and StrOutputParser) and the edges between them.

See Runnable — reference.

Config, tags and callbacks

Every Runnable method accepts a RunnableConfig, a plain dict that is passed down to every child step, so one call can label and observe a whole chain. The keys used most often:

Key Purpose

Key	Purpose
`run_name`	Name shown in traces for this call.
`tags`	List of strings attached to the run and inherited by children, for filtering.
`metadata`	Free-form `dict` (for example a user ID) attached to the run and inherited by children.
`callbacks`	Handlers notified of start, end and error events of each step.
`max_concurrency`	Upper bound on parallel work in `batch` and `RunnableParallel`.

run_name

Name shown in traces for this call.

tags

List of strings attached to the run and inherited by children, for filtering.

metadata

Free-form dict (for example a user ID) attached to the run and inherited by children.

callbacks

Handlers notified of start, end and error events of each step.

max_concurrency

Upper bound on parallel work in batch and RunnableParallel.

Passing config and a callback handler

from typing import Any

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.runnables import RunnableConfig

class Timer(BaseCallbackHandler):
    def on_chain_start(self, serialized, inputs, *, run_id, tags=None, metadata=None, **kw: Any):
        print("chain_start", kw.get("name"), tags, metadata)

    def on_chat_model_start(self, serialized, messages, *, run_id, tags=None, metadata=None, **kw: Any):
        print("chat_model_start", tags, metadata)

model = GenericFakeChatModel(messages=iter([AIMessage(content="ok")]))
chain = ChatPromptTemplate.from_template("{q}") | model | StrOutputParser()
config: RunnableConfig = {
    "run_name": "docs_answer",
    "tags": ["docsassistant", "prod"],
    "metadata": {"user_id": "u-42"},
    "callbacks": [Timer()],
}
print(chain.invoke({"q": "hi"}, config=config))

Output with the fake model (the ls_* and lc_versions metadata the chat model adds are elided):

chain_start docs_answer ['docsassistant', 'prod'] {'user_id': 'u-42'}
chain_start ChatPromptTemplate ['seq:step:1', 'docsassistant', 'prod'] {'user_id': 'u-42'}
chat_model_start ['seq:step:2', 'docsassistant', 'prod'] {'user_id': 'u-42', ...}
chain_start StrOutputParser ['seq:step:3', 'docsassistant', 'prod'] {'user_id': 'u-42'}
ok

The tags and metadata reached every step, and the framework added a seq:step:N tag to each member of the sequence. The same callbacks mechanism is what tracing integrations plug into: with the langsmith package, setting the LANGSMITH_TRACING environment variable (plus a LANGSMITH_API_KEY supplied from the environment, never from source code) records every run with these names, tags and metadata. That workflow is covered in Observability with LangSmith; the framework-neutral view is in Observability for Agents. See LangChain reference — RunnableConfig and LangChain reference — BaseCallbackHandler.

max_concurrency bounds parallel work. Eight inputs to batch with max_concurrency set to 2 never ran more than two at once in a measurement with a sleeping function:

Bounding batch concurrency

import threading, time

live = peak = 0
lock = threading.Lock()

def work(x):
    global live, peak
    with lock:
        live += 1
        peak = max(peak, live)
    time.sleep(0.05)
    with lock:
        live -= 1
    return x

RunnableLambda(work).batch(list(range(8)), config={"max_concurrency": 2})
print("peak", peak)

Output:

peak 2

Use it to stay under a provider’s or a database’s rate limit when batching over many documents.

See the RunnableConfig and BaseCallbackHandler references.

Retries, fallbacks and streaming through custom steps

Two wrappers make a step more robust without changing the chain around it. with_retry re-runs a failing step; with_fallbacks tries alternatives in order when the primary raises. A generator function used as a step receives the upstream stream and yields transformed chunks, so streaming is not lost:

with_retry, with_fallbacks and a streaming generator step

calls = {"n": 0}

def flaky(x: str) -> str:
    calls["n"] += 1
    if calls["n"] < 3:
        raise TimeoutError("vector store timed out")
    return f"hits for {x}"

retrieve = RunnableLambda(flaky).with_retry(stop_after_attempt=3, wait_exponential_jitter=False)
print(retrieve.invoke("hnsw"), "after", calls["n"], "attempts")

def broken(_):
    raise RuntimeError("primary model unavailable")

primary = RunnableLambda(broken)
backup = RunnableLambda(lambda x: f"backup answered: {x}")
print(primary.with_fallbacks([backup]).invoke("Which index?"))

def upper_chunks(chunks):
    for chunk in chunks:
        yield chunk.upper()

model = GenericFakeChatModel(messages=iter([AIMessage(content="use hnsw here")]))
streaming = ChatPromptTemplate.from_template("{q}") | model | StrOutputParser() | upper_chunks
print(list(streaming.stream({"q": "hi"})))

Output:

hits for hnsw after 3 attempts
backup answered: Which index?
['USE', ' ', 'HNSW', ' ', 'HERE']

Retrying a step that has side effects (opening an issue, sending a message) repeats the side effect; restrict retries to idempotent steps such as retrieval.

See the with_retry reference.

When to switch to LangGraph

An LCEL chain is a directed acyclic pipeline: every input flows forward through the same steps, once. That is the right tool for a fixed prompt-model-parser sequence, a retrieval chain, or a fan-out that is merged back. Move to a LangGraph graph when the flow needs any of these:

Loops, for example regenerating an answer until a grader accepts it, with a maximum attempt count.
Conditional routing to different steps based on intermediate results.
Persistent state shared between steps, checkpointing, resumable runs or human approval.
Long-running or multi-actor flows where you want to inspect and modify state between steps.

A runnable chain is not thrown away when you do: it becomes a node. This graph loops back to generate (an LCEL chain) until a RunnableLambda grader accepts the answer or three attempts are used:

flowchart LR S([START]) --> G[generate: LCEL chain] G --> C{grounded or attempts >= 3?} C -->|no| G C -->|yes| E([END])

Figure 2. The retry loop: generate until the answer is grounded or three attempts are used

An LCEL chain as a LangGraph node, with a retry loop

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

answer_chain = (
    ChatPromptTemplate.from_template("{question}")
    | GenericFakeChatModel(messages=iter([AIMessage(content="draft"), AIMessage(content="grounded answer")]))
    | StrOutputParser()
)
grade = RunnableLambda(lambda a: "grounded" in a)

class State(TypedDict):
    question: str
    answer: str
    attempts: int

def generate(state: State) -> dict:
    return {"answer": answer_chain.invoke({"question": state["question"]}), "attempts": state["attempts"] + 1}

def route(state: State) -> str:
    return END if grade.invoke(state["answer"]) or state["attempts"] >= 3 else "generate"

g = StateGraph(State)
g.add_node("generate", generate)
g.add_edge(START, "generate")
g.add_conditional_edges("generate", route, ["generate", END])
app = g.compile()
print(app.invoke({"question": "Which index?", "answer": "", "attempts": 0}))

Output with the fake model:

{'question': 'Which index?', 'answer': 'grounded answer', 'attempts': 2}

The compiled graph is itself a Runnable, so invoke, stream and batch work the same way. The graph API is introduced in LangGraph Fundamentals, the standard workflow shapes in LangGraph Workflows, and the framework-neutral theory in Workflow Patterns. For a tool-calling agent, skip both and use create_agent (see Agents with create_agent). Typed model output inside a chain is covered by Structured Output and, at the protocol level, by Structured Output & Tool Calling.

See LangGraph — Choosing between Graph and Functional APIs and LangGraph — Graph API overview.

Version notes

In LangChain 1.x the langchain package no longer contains the legacy chains: import langchain.chains raises ModuleNotFoundError. LLMChain and create_retrieval_chain are importable from langchain_classic.chains, which requires the separate langchain-classic package. Prefer the LCEL forms on this page for new code.
Import RunnablePassthrough, RunnableParallel, RunnableLambda and chain from langchain_core.runnables.
Tracing environment variables are LANGSMITH_TRACING and LANGSMITH_API_KEY; LANGCHAIN_TRACING_V2 is the legacy name.

See LangChain v1 migration guide.

LCEL & Runnables

The Runnable protocol

Composing with the pipe operator

The @chain decorator

Fan-out with RunnableParallel and RunnablePassthrough

Config, tags and callbacks

Retries, fallbacks and streaming through custom steps

When to switch to LangGraph

Version notes

References

The `@chain` decorator