LangChain LCEL: Compose Chains with Pipes & Fallbacks

Overview

Gluing an LLM call to your code usually starts as a function that calls the model, then a second one that post-processes the output, then a third that calls the model again. After a few iterations you end up with an imperative script where retries, streaming, and fallbacks each mean rewriting the same wiring. LangChain Expression Language (LCEL) replaces that wiring with the pipe operator (|): each step declares what it does, and the framework handles execution, streaming, batching, and error recovery uniformly.

This recipe builds LCEL chains step by step — from a minimal prompt | model | parser to parallel branches, custom Python components, retries, and fallbacks. For the design-level version of this idea, see the prompt chaining pattern.

When to Use

• Multi-step LLM workflows (summarize then translate, extract then classify).
• Pipelines that need streaming, batching, or retry logic without hand-rolled plumbing.
• Composing prompts, tools, and parsers into reusable, testable chains.
• Not the right fit for stateful agents that need loops and tool-call graphs — that’s LangGraph territory, covered in the LangGraph state machine recipe.

Prerequisites

• Python 3.10+
• langchain and langchain-openai packages
• An OpenAI API key in OPENAI_API_KEY

Solution

1. Install Dependencies

pip install langchain langchain-openai

2. Basic Chain — Prompt | Model | Parser

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise technical writer."),
    ("user", "Explain {topic} in under 100 words."),
])

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
parser = StrOutputParser()

# LCEL pipe operator composes the chain
chain = prompt | model | parser

result = chain.invoke({"topic": "vector databases"})
print(result)

3. Multi-Step Chain — Summarize Then Translate

summarize_prompt = ChatPromptTemplate.from_template(
    "Summarize this text in 3 bullet points:\n\n{text}"
)

translate_prompt = ChatPromptTemplate.from_template(
    "Translate to Spanish:\n\n{summary}"
)

summarize_chain = summarize_prompt | model | StrOutputParser()
translate_chain = translate_prompt | model | StrOutputParser()

# Chain the output of summarize into translate
full_chain = (
    {"summary": summarize_chain}
    | translate_chain
)

result = full_chain.invoke({
    "text": "LangChain is a framework for building applications powered by LLMs..."
})

4. Parallel Branches with RunnableParallel

from langchain_core.runnables import RunnableParallel

# Run multiple chains simultaneously on the same input
classification_prompt = ChatPromptTemplate.from_template(
    "Classify this text as 'technical', 'marketing', or 'other':\n\n{text}"
)

sentiment_prompt = ChatPromptTemplate.from_template(
    "What is the sentiment of this text? (positive, neutral, negative):\n\n{text}"
)

parallel_chain = RunnableParallel(
    classification=classification_prompt | model | StrOutputParser(),
    sentiment=sentiment_prompt | model | StrOutputParser(),
)

result = parallel_chain.invoke({
    "text": "Our new product is amazing and uses cutting-edge AI!"
})
# {'classification': 'marketing', 'sentiment': 'positive'}

5. Custom Runnable Components

from langchain_core.runnables import RunnableLambda

def extract_keywords(text: str) -> list[str]:
    """Extract keywords from text (simplified)."""
    stop_words = {"the", "a", "an", "is", "are", "and", "or", "in", "on"}
    words = [w.lower().strip(".,!?") for w in text.split()]
    return [w for w in words if w not in stop_words and len(w) > 3][:10]

def format_output(data: dict) -> str:
    """Format the final output."""
    return f"Keywords: {', '.join(data['keywords'])}\nSummary: {data['summary']}"

custom_chain = (
    RunnableParallel(
        keywords=RunnableLambda(extract_keywords),
        summary=summarize_chain,
    )
    | RunnableLambda(format_output)
)

result = custom_chain.invoke({
    "text": "Vector databases enable semantic search over embeddings..."
})

6. Adding Retry and Fallback

fast_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
strong_model = ChatOpenAI(model="gpt-4o", temperature=0)

fast_chain = prompt | fast_model | StrOutputParser()
strong_chain = prompt | strong_model | StrOutputParser()

# Try fast model first, fall back to strong model on error
chain_with_fallback = fast_chain.with_fallbacks([strong_chain])

# Add retry logic
chain_with_retry = fast_chain.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
)

7. Streaming Output

# Stream tokens as they arrive
streaming_chain = prompt | model | StrOutputParser()

for chunk in streaming_chain.stream({"topic": "GraphQL federation"}):
    print(chunk, end="", flush=True)

For building a full streaming endpoint around this pattern, see streaming LLM responses.

8. Batch Processing

# Process multiple inputs in parallel
results = chain.batch([
    {"topic": "Redis caching"},
    {"topic": "PostgreSQL indexing"},
    {"topic": "Docker multi-stage builds"},
])

for r in results:
    print(r)
    print("---")

How It Works

Every component in a chain — the prompt, the model, the parser, your Python functions — speaks the same Runnable interface: invoke, stream, and batch. The | operator links them so each step’s output becomes the next step’s input, the same way Unix pipes connect commands. Because everything shares the interface, capabilities like streaming and batching come free on any chain you build.

The diagram below shows how a typical production chain flows, including the parallel branches from step 4:

RunnableParallel is worth the latency win when branches are independent: two 800ms calls run in ~800ms instead of ~1600ms. RunnableLambda wraps any Python function so you can drop custom logic — validation, keyword extraction, API calls — into the middle of a chain without leaving the Runnable interface.

with_fallbacks and with_retry solve different problems. Retry handles transient failures (rate limits, timeouts) by running the same chain again; fallback handles capacity problems by switching to a different chain — typically a stronger or cheaper model — when the primary one keeps failing. Stacking them (fast_chain.with_retry(...).with_fallbacks(...)) gives transient retries first and model failover only when retries exhaust.

batch pushes inputs through the provider’s parallel infrastructure, which is dramatically faster than looping invoke calls, while stream yields chunks as the model produces them so a UI can render the first tokens early. Both are free because every runnable has them built in.

Variants

Conditional Routing

from langchain_core.runnables import RunnableBranch

summary_prompt = ChatPromptTemplate.from_template("Summarize briefly:\n\n{text}")
default_chain = summary_prompt | model | StrOutputParser()

technical_prompt = ChatPromptTemplate.from_template(
    "Write a technical summary preserving jargon:\n\n{text}"
)
marketing_prompt = ChatPromptTemplate.from_template(
    "Write a marketing-friendly summary:\n\n{text}"
)
technical_chain = technical_prompt | model | StrOutputParser()
marketing_chain = marketing_prompt | model | StrOutputParser()

# First matching condition wins; last arg is the fallback
branch_chain = RunnableBranch(
    (lambda x: "technical" in x["text"].lower(), technical_chain),
    (lambda x: "marketing" in x["text"].lower(), marketing_chain),
    default_chain,
)

result = branch_chain.invoke({"text": "This technical guide covers..."})

JSON Output with Structured Parser

from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel

class ProductReview(BaseModel):
    rating: int
    summary: str
    pros: list[str]
    cons: list[str]

json_parser = JsonOutputParser(pydantic_object=ProductReview)

review_prompt = ChatPromptTemplate.from_template(
    "Analyze this review and extract structured data.\n{format_instructions}\n\nReview: {review}"
).partial(format_instructions=json_parser.get_format_instructions())

review_chain = review_prompt | model | json_parser

result = review_chain.invoke({"review": "Great product, fast delivery..."})
# {'rating': 5, 'summary': 'Positive review', 'pros': [...], 'cons': [...]}

For more on getting typed output from models, see structured output with function calling.

Async Execution

import asyncio

async def run_chain():
    # Non-blocking invoke — use inside FastAPI or async workers
    result = await chain.ainvoke({"topic": "vector databases"})
    print(result)

    # Async streaming
    async for chunk in chain.astream({"topic": "Redis caching"}):
        print(chunk, end="", flush=True)

    # Async batch
    results = await chain.abatch([
        {"topic": "Redis caching"},
        {"topic": "PostgreSQL indexing"},
    ])
    print(results)

asyncio.run(run_chain())

Memory-Conversational Chain

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

conversational_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("placeholder", "{history}"),
    ("human", "{input}"),
])

base_chain = conversational_prompt | model | StrOutputParser()

chain_with_history = RunnableWithMessageHistory(
    base_chain,
    lambda session_id: InMemoryChatMessageHistory(),
    input_messages_key="input",
    history_messages_key="history",
)

result = chain_with_history.invoke(
    {"input": "What is Redis?"},
    config={"configurable": {"session_id": "user123"}},
)

Best Practices

• Use StrOutputParser for simple text output — it extracts the string content from the AI message instead of returning the whole message object.
• Break complex chains into smaller components, each doing one thing; a chain you can diagram on a napkin is a chain you can debug.
• Wrap production chains with with_fallbacks — LLM APIs rate-limit and time out at the worst possible moments.
• Prefer stream for anything a user watches in real time; first-token latency matters more than total latency.
• Keep input keys consistent across steps. Most LCEL bugs are a runnable receiving {"text": ...} when it expected {"summary": ...}.

Common Mistakes

Passing the wrong input keys between steps is the number one runtime failure: each runnable expects specific keys, and a mismatch only surfaces when the chain executes, not when you define it.

Running independent branches sequentially wastes latency — if two chains don’t depend on each other’s output, wrap them in RunnableParallel and get both results in the time of the slower one.

Chaining without error handling means a single rate-limit kills the whole pipeline. A with_fallbacks wrapper costs two lines and saves the 3 AM page.

Finally, calling invoke inside async code (FastAPI, async workers) blocks the event loop. Use ainvoke, astream, and abatch — they exist on every runnable.

Troubleshooting

KeyError or missing input errors between steps. The previous runnable outputs a dict key the next prompt doesn’t declare, or the prompt expects a key nobody produces. Print the intermediate output with RunnablePassthrough() or chain.get_graph().print_ascii() to see the data flowing between steps.

A parallel branch fails and kills the whole chain. RunnableParallel raises as soon as any branch raises, and the other results are lost. Wrap each fragile branch in its own with_fallbacks so one failure degrades gracefully instead of aborting the aggregate.

Retries fire on every call and token costs spike. with_retry retries on any exception, including errors your prompt causes (malformed output parsers, invalid input). Log failures before adding retries — a chain that always fails will retry 3x and triple the bill for nothing.

Streaming works but the final output is wrong. When you consume .stream() chunk by chunk, remember to accumulate chunks if downstream code needs the full string; parsers attached to the chain see the complete output, your loop sees fragments.

invoke hangs inside FastAPI or an async worker. Synchronous calls block the event loop. Switch to await chain.ainvoke(...) and the whole chain runs non-blocking, including the model call.

See Also

• Complete Guide to LangChain in Production for alternatives and the broader ecosystem.
• Streaming LLM responses for production streaming endpoints.
• Structured output with function calling for typed model output without LCEL.
• LangChain LCEL documentation for the full Runnable interface reference.

Companion code: LCEL chains project: the chains from this page organized as importable modules, plus a runnable main.py.