LangChain Expression Language (LCEL): A Practical Guide to the Pipe Operator

Published: 2026-07-28 | By: Carlos Montiel | Reading time: ~4 minutes

LCEL has been the right way to build chains in LangChain since 2024. This guide covers the syntax, composition patterns, and common mistakes we see in production code.

What a Runnable actually is

Everything in LCEL implements the `Runnable` interface: a prompt, a chat model, an output parser, a retriever, even a decorated Python function. This interface guarantees four synchronous methods and their async equivalents: `invoke`, `batch`, `stream`, and their `a*` versions. The `|` operator isn't trivial syntactic sugar: internally it builds a `RunnableSequence` that chains one component's output as the next one's input.

from langchain_core.runnables import RunnableLambda

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

pipeline = double | add_one
print(pipeline.invoke(5))  # 11

This means you can insert pure Python business logic in the middle of an LLM chain without artificial wrappers.

Composing a RAG pipeline with pipe

The most common production pattern is retrieval + generation. With LCEL it's expressed in a linear, readable way:

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

retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

prompt = ChatPromptTemplate.from_template(
    "Context:\n{context}\n\nQuestion: {question}\nAnswer using only the given context."
)

def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | StrOutputParser()
)

answer = rag_chain.invoke("What's the refund policy?")

The input dictionary automatically becomes a `RunnableParallel`: each key runs concurrently, not sequentially, which reduces latency when there are multiple independent data sources.

RunnableParallel and RunnableBranch

When you need to run several sub-chains over the same input — for example, generating a summary and extracting entities at the same time — `RunnableParallel` makes that explicit:

from langchain_core.runnables import RunnableParallel

analysis = RunnableParallel(
    summary=summary_prompt | llm | StrOutputParser(),
    entities=entities_prompt | llm | StrOutputParser(),
)
result = analysis.invoke({"text": document})
# result = {"summary": "...", "entities": "..."}

For simple conditional logic (no cycles, which is LangGraph territory), `RunnableBranch` lets you route based on the input's content, useful for classifying user intent before choosing the right prompt.

Token-by-token streaming

A direct benefit of LCEL is that streaming works consistently across the whole chain, not just on the final call to the model:

for chunk in rag_chain.stream("How long is the warranty?"):
    print(chunk, end="", flush=True)

This is critical for UX in conversational applications: the user sees the response generate in real time instead of waiting for the whole block, even when there are retrieval steps before the model.

Error handling and retries

LCEL exposes `.with_retry()` and `.with_fallbacks()` directly on any `Runnable`, avoiding the need to wrap calls in manual try/except blocks:

robust_llm = ChatOpenAI(model="gpt-4o-mini").with_retry(
    stop_after_attempt=3
).with_fallbacks([ChatOpenAI(model="gpt-4o-mini", temperature=0.2)])

In production, we combine this with `with_fallbacks` pointing to a second provider (for example, Anthropic) for real resilience against API outages — something that required custom code in the old agent layer.

Common mistakes we see in code audits

The most frequent one is mixing `Runnable` with the old `Chain` API in the same project for no reason, which duplicates error-handling patterns. The second is not using `.batch()` when processing batches of documents, instead leaving a `for` loop with sequential `.invoke()` calls, losing the internal parallelism LangChain manages with a `ThreadPoolExecutor`. The third is not typing the chain's input with `RunnableConfig` when you need to pass metadata (like `run_name` or `tags`) for traceability in LangSmith, which complicates debugging later.

Carlos Montiel
Enterprise AI Solutions Architect
Specialist in LLMs, Agents, and Orchestration
guatemalia.com/en/#contact · info@guatemalia.com

← Back to index Spanish blog (355+ articles) →