LCEL — composing chains with the pipe operator

LCEL — LangChain Expression Language — is the pipe-operator DSL for gluing Runnables together. prompt | model | parser is the smallest example; the same operator scales up to multi-step pipelines with parallelism, branching, and streaming.

Everything that LangChain calls a Runnable plays in this system. ChatModels, prompts, parsers, retrievers, even plain Python functions (wrapped with RunnableLambda) are Runnables. Compose them with | and you get a new Runnable — because | is defined as __or__(self, other) on the Runnable base class and returns a RunnableSequence that itself implements the full Runnable interface. That’s why chaining never breaks the contract: the output is always something you can .invoke(), .batch(), or .stream().

What you get for free

Once you have a chain, every standard execution mode comes built in:

chain = prompt | model | parser

chain.invoke({"x": 1})           # one input, sync
chain.batch([{"x": 1}, {"x": 2}]) # many inputs, parallel under the hood
chain.stream({"x": 1})           # iterator yielding incremental output
await chain.ainvoke({"x": 1})    # async

That uniformity is the actual reason to use LCEL. Writing these four modes yourself for a non-trivial pipeline is a real amount of code, and you’ll get the async + streaming pieces wrong on the first try.

Reading a chain left-to-right

A chain is a pipeline; data flows from left to right. The output of each Runnable becomes the input to the next.

input dict → prompt → list of messages → model → AIMessage → parser → string

If a step expects a dict but receives a string, you’ll get a runtime type error. The shape of what flows between steps is just as important as what each step does:

RunnableParallel — fan out

When two steps need the same input but produce different things, fan them out with RunnableParallel (the dict syntax is shorthand).

from langchain_core.runnables import RunnableParallel

# Each branch sees the same input; the output is a dict keyed by name.
analyze = RunnableParallel({
    "summary":   prompt_summary | model | StrOutputParser(),
    "sentiment": prompt_sent    | model | StrOutputParser(),
    "keywords":  prompt_kw      | model | StrOutputParser(),
})

# LangChain will run all three branches in parallel.
result = analyze.invoke({"text": doc})
# → {"summary": "...", "sentiment": "positive", "keywords": ["...", "..."]}

For LLM calls that don’t depend on each other, this is a real latency win — three sequential calls become one round trip’s worth of wall-clock time.

RunnableBranch — conditional routing

Sometimes the next step depends on the input. RunnableBranch is LCEL’s if/elif/else.

from langchain_core.runnables import RunnableBranch

router = RunnableBranch(
    (lambda x: x["lang"] == "sql",    sql_explain_chain),
    (lambda x: x["lang"] == "python", python_explain_chain),
    default_chain,
)

The first predicate that returns truthy wins. The final argument is the fallback. Each branch is itself a chain, so you can route to arbitrarily complex sub-pipelines.

A four-step LCEL chain

Let’s build the canonical example: extract structured data from a raw doc, summarize it, then format the summary as bullet points.

# Mocked LCEL pipeline. The shapes (prompt | model | parser, RunnableParallel)
# match real LangChain. The mock LLM is deterministic so you can run it.

# --- mocked LCEL primitives ---
class Runnable:
    def invoke(self, x): raise NotImplementedError
    def __or__(self, other): return Pipe(self, other)
    def batch(self, xs): return [self.invoke(x) for x in xs]

class Pipe(Runnable):
    def __init__(self, a, b): self.a, self.b = a, b
    def invoke(self, x): return self.b.invoke(self.a.invoke(x))

class RunnableParallel(Runnable):
    def __init__(self, mapping): self.mapping = mapping
    def invoke(self, x): return {k: v.invoke(x) for k, v in self.mapping.items()}

class Prompt(Runnable):
    def __init__(self, tmpl): self.tmpl = tmpl
    def invoke(self, x): return self.tmpl.format(**x)

class MockLLM(Runnable):
    """Deterministic 'LLM' — pattern-matches the prompt and returns a fixed answer."""
    def invoke(self, prompt):
        p = prompt.lower()
        if "extract" in p:
            return '{"product": "DataFrame API", "issue": "slow joins on 1B rows"}'
        if "summarize" in p:
            return "Customer reports DataFrame API has slow joins at billion-row scale."
        if "bullet" in p:
            return "- DataFrame API\\n- slow joins\\n- 1B-row scale"
        return "[mock]"

class JsonParser(Runnable):
    def invoke(self, s):
        import json
        return json.loads(s)

class StrParser(Runnable):
    def invoke(self, s): return s

# --- the actual pipeline (this part is the LCEL shape you'd write for real) ---
extract = Prompt("Extract product + issue from this ticket: {ticket}") | MockLLM() | JsonParser()
summarize = Prompt("Summarize this in one sentence: {extracted}") | MockLLM() | StrParser()
bulletize = Prompt("Convert this summary into bullets: {summary}") | MockLLM() | StrParser()

# Wire them together. Each step receives a dict; we adapt between them with small lambdas.
class Lambda(Runnable):
    def __init__(self, fn): self.fn = fn
    def invoke(self, x): return self.fn(x)

pipeline = (
    Lambda(lambda x: {"ticket": x["ticket"]})
    | extract
    | Lambda(lambda extracted: {"extracted": extracted})
    | summarize
    | Lambda(lambda s: {"summary": s})
    | bulletize
)

result = pipeline.invoke({
    "ticket": "Hey — the DataFrame API is way too slow when we join two 1B-row tables.",
})
print(result)

- DataFrame API
- slow joins
- 1B-row scale

Trace the boundaries: the ticket dict becomes a prompt string, the mock LLM returns JSON text, JsonParser makes it a dict, a Lambda re-wraps it for the next prompt, and so on — until bulletize emits the final bullet list. Every | only works because the type leaving one step is the type the next one expects. That’s four LLM calls’ worth of logic in about a dozen lines of pipeline code — and the same chain gets .batch(), .stream(), and .ainvoke() automatically, no extra work.

Streaming for free

Streaming is the place where LCEL most obviously earns its keep.

for chunk in chain.stream({"question": "Explain JOINs"}):
    print(chunk, end="", flush=True)

LangChain figures out which steps in the chain can stream (the model can; a JSON parser usually can’t, since JSON is invalid until complete) and gives you incremental output where it makes sense.

Batching gotcha

.batch([...]) runs inputs in parallel — by default, all of them at once. If your chain hits a rate-limited API, you’ll smack into the limit. Pass config={"max_concurrency": 5} to throttle.

chain.batch(inputs, config={"max_concurrency": 5})

This is the kind of detail that’s easy to miss in development and painful in production.

When plain Python wins

LCEL is not always the right answer. If your pipeline is:

extracted = extract_chain.invoke({"ticket": text})
summary = summarize_chain.invoke({"extracted": extracted})
bullets = bulletize_chain.invoke({"summary": summary})
return bullets

That’s perfectly fine code. It’s debuggable with a normal stacktrace, the type errors are obvious, and a junior on your team can read it in ten seconds. You give up .batch() and .stream() across the whole pipeline, but you can still call them on each individual chain.

The pipe operator becomes worth it when you’re composing parallel branches, conditional routes, or you need streaming across the whole pipeline.

In one breath

• LCEL composes Runnables with |; the result is itself a Runnable, so the chain always exposes .invoke / .batch / .stream / .ainvoke.
• Data flows left to right — each step’s output type must match the next step’s input type, or you get a runtime error.
• RunnableParallel fans the same input out to multiple branches (a real latency win); RunnableBranch is LCEL’s if/elif/else for routing.
• You get batching, streaming, and async for free — but .batch is unbounded by default, so set max_concurrency against rate limits.
• LCEL is for DAGs; the moment you need loops, conditional re-entry, or human-in-the-loop pauses, reach for LangGraph (or plain Python for trivially linear flows).

Quick check

Next

LCEL covers the DAG case. The next chapter introduces LangGraph for when you need real state, branching loops, and persistence.