When I first used LangChain in 2023, building a simple RAG chain required instantiating four separate classes, configuring each with overlapping parameters, and wiring them together with explicit method calls. The code was verbose, error-prone, and hard to test in isolation. When LangChain introduced LCEL (LangChain Expression Language) in late 2023, I rewrote the same chain in 8 lines. More importantly, the rewritten chain was testable at every component boundary — I could swap the LLM, the retriever, and the output parser independently without touching the surrounding code.

LCEL's core insight: if every AI pipeline component implements the same interface (Runnable), they can be composed freely using a single operator (|). This is not syntactic sugar — it is an architectural decision that unlocks automatic streaming, async execution, parallel branching, and built-in fallbacks without any additional code on your part.

In the Agentic AI Workshop, every chain we build uses LCEL from Day 1. The LangGraph agents we build on Day 3 are themselves composed of LCEL chains at the node level. Understanding LCEL is the foundation for understanding everything that follows.

What Is LCEL and Why It Matters

LCEL is a protocol: every component that participates in an LCEL chain implements the Runnable interface. The Runnable interface has five invocation methods:

Method Use Case Return Type
.invoke(input) Single synchronous call Output object
.ainvoke(input) Single async call (await) Coroutine → Output
.stream(input) Sync streaming (token by token) Iterator of chunks
.astream(input) Async streaming (for FastAPI SSE) AsyncIterator of chunks
.batch(inputs) Process multiple inputs concurrently List of outputs

When you compose a chain with |, the resulting chain also implements Runnable — with all five methods automatically. You do not implement streaming yourself; you compose Runnables and streaming works end-to-end.

The 5 Runnable Primitives

1. ChatPromptTemplate: Converts a dict of variables into a formatted list of messages. The entry point for every LCEL chain. Supports system messages, human messages, and few-shot examples.

2. ChatLLM (ChatOpenAI, ChatOllama, etc.): Takes messages, calls the LLM, returns an AIMessage. The component you swap when changing models. All LLMs expose the same Runnable interface.

3. Output Parsers (StrOutputParser, JsonOutputParser, PydanticOutputParser): Converts AIMessage to the desired Python type — string, dict, or Pydantic model. Streaming-compatible: StrOutputParser streams individual string chunks; JsonOutputParser streams partial JSON as it accumulates.

4. RunnablePassthrough: Passes its input to the output unchanged. Used in RAG chains to preserve the original question alongside retrieved context: {"context": retriever, "question": RunnablePassthrough()}.

5. RunnableLambda: Wraps any Python function as a Runnable. The escape hatch for custom logic. Use for formatting, post-processing, logging, or any transformation that does not fit a built-in component.

Parallel Chain Execution

RunnableParallel runs multiple Runnables concurrently and merges their outputs into a single dict. When called with .ainvoke(), it uses asyncio.gather internally — all branches execute simultaneously, completing in the time of the slowest branch rather than the sum of all branches.

The most common production use case: multi-source RAG. Your query goes to a vector store (semantic search), a keyword search index (BM25 for exact term matching), and a structured database (for factual records) simultaneously. All three results arrive in parallel; a synthesis chain merges them into a final answer. This pattern reduces end-to-end latency from ~9s (3 sequential retrievals × 3s each) to ~3s (3 parallel retrievals).

Streaming and Fallbacks

Streaming: Call .astream() on any LCEL chain to get an async iterator of output chunks. In a chain of prompt | llm | StrOutputParser(), the StrOutputParser passes each token through as it arrives from the LLM. The full chain streams end-to-end with no special configuration. For FastAPI, wrap the async generator in StreamingResponse — covered in the FastAPI post.

Fallbacks: Call .with_fallbacks([backup_llm]) on any Runnable to configure automatic failover. If the primary raises an exception, LangChain transparently retries with the first fallback. This is a production requirement for any agent that uses external LLM APIs — rate limits, network errors, and model availability issues are facts of life. A well-configured fallback chain continues serving users when the primary LLM degrades.

Complete LCEL Code Reference

# pip install langchain langchain-openai langchain-ollama pydantic

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda, RunnableParallel
from langchain_openai import ChatOpenAI
from langchain_ollama import ChatOllama
from pydantic import BaseModel, Field

# ===================================================================
# PATTERN 1: Basic chain — prompt | llm | parser
# The fundamental LCEL pattern. Reads left to right like a Unix pipe.
# ===================================================================

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)

basic_chain = (
    ChatPromptTemplate.from_messages([
        ("system", "You are a Kubernetes expert. Be concise."),
        ("human", "{question}")
    ])
    | llm
    | StrOutputParser()
)

# All five execution modes — same chain, zero extra code:
answer_sync   = basic_chain.invoke({"question": "What is a DaemonSet?"})
answers_batch = basic_chain.batch([
    {"question": "What is a DaemonSet?"},
    {"question": "What is a StatefulSet?"},
    {"question": "What is a Deployment?"},
])  # Runs all 3 concurrently

# Streaming — yields tokens as they arrive
for token in basic_chain.stream({"question": "Explain resource limits briefly."}):
    print(token, end="", flush=True)
print()

# ===================================================================
# PATTERN 2: Structured output with Pydantic
# WHY: parse LLM responses as validated Python objects — eliminates
# fragile string parsing and catches schema mismatches at runtime.
# ===================================================================

class TechAnalysis(BaseModel):
    topic: str = Field(description="Main technology topic")
    maturity: str = Field(description="emerging|growing|mature|declining")
    enterprise_readiness: int = Field(description="Score 1-10 for enterprise adoption readiness")
    key_benefits: list[str] = Field(description="Top 3 business benefits")
    risks: list[str] = Field(description="Top 2 adoption risks")

analysis_chain = (
    ChatPromptTemplate.from_template(
        "Analyse this technology for enterprise adoption: {technology}\n"
        "Return structured JSON matching the schema."
    )
    | llm.with_structured_output(TechAnalysis)  # Returns validated TechAnalysis object
)

result: TechAnalysis = analysis_chain.invoke({"technology": "Agentic AI"})
print(f"Topic: {result.topic}")
print(f"Maturity: {result.maturity}")
print(f"Enterprise readiness: {result.enterprise_readiness}/10")
print(f"Benefits: {result.key_benefits}")

# ===================================================================
# PATTERN 3: RunnablePassthrough and RunnableLambda
# WHY RunnablePassthrough: preserve original input alongside processed output
# WHY RunnableLambda: wrap any Python function as a chain component
# ===================================================================

def format_retrieved_docs(docs) -> str:
    """Format retrieved document chunks for the LLM context."""
    return "\n\n---\n\n".join([
        f"[Source: {doc.metadata.get('source', 'unknown')}]\n{doc.page_content}"
        for doc in docs
    ])

# In a RAG chain, the retriever returns Document objects
# RunnableLambda wraps format_retrieved_docs as a Runnable
format_docs_runnable = RunnableLambda(format_retrieved_docs)

# RunnablePassthrough preserves "question" while "context" branch processes it
# This is the standard RAG chain pattern:
rag_prompt = ChatPromptTemplate.from_template("""
Answer the question based only on the following context.
If the answer is not in the context, say "I don't know."

Context:
{context}

Question: {question}
Answer:""")

# Assume `retriever` is a configured LangChain retriever (ChromaDB, FAISS, etc.)
# rag_chain = (
#     {"context": retriever | format_docs_runnable,   # branch 1: retrieve + format
#      "question": RunnablePassthrough()}              # branch 2: pass question through
#     | rag_prompt
#     | llm
#     | StrOutputParser()
# )
# ===================================================================
# PATTERN 4: RunnableParallel — concurrent execution
# WHY: run multiple retrievers or chains simultaneously.
# Latency = max(branch_latencies), not sum(branch_latencies).
# ===================================================================

summary_chain = (
    ChatPromptTemplate.from_template("Summarise in 2 sentences: {topic}")
    | llm | StrOutputParser()
)

use_case_chain = (
    ChatPromptTemplate.from_template("List 3 enterprise use cases for: {topic}")
    | llm | StrOutputParser()
)

risk_chain = (
    ChatPromptTemplate.from_template("List 2 key risks of adopting: {topic}")
    | llm | StrOutputParser()
)

# RunnableParallel runs all 3 chains concurrently
# WHY: with 3 × 3-second LLM calls, sequential = 9s, parallel = 3s
parallel_analysis = RunnableParallel({
    "summary": summary_chain,
    "use_cases": use_case_chain,
    "risks": risk_chain,
})

results = parallel_analysis.invoke({"topic": "Agentic AI"})
print("Summary:", results["summary"])
print("Use Cases:", results["use_cases"])
print("Risks:", results["risks"])

# ===================================================================
# PATTERN 5: Fallbacks — automatic failover between LLMs
# WHY: LLM APIs have ~99.9% availability, not 100%. Unhandled API errors
# cascade into agent failures. Fallbacks make your chain resilient.
# ===================================================================

primary_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
fallback_llm = ChatOllama(model="llama3.1:8b", temperature=0.0)  # local, always available

# .with_fallbacks() wraps the primary with automatic retry-on-exception
resilient_llm = primary_llm.with_fallbacks(
    fallbacks=[fallback_llm],
    exceptions_to_handle=(Exception,),  # catch all exceptions from primary
)

# Build a production chain with resilient LLM
production_chain = (
    ChatPromptTemplate.from_messages([
        ("system", "You are a DevOps expert."),
        ("human", "{question}")
    ])
    | resilient_llm  # Uses GPT-4o-mini; falls back to local Llama if it fails
    | StrOutputParser()
)

# This call succeeds even if OpenAI API is down — falls through to local Llama
answer = production_chain.invoke({"question": "What is GitOps?"})
print(answer)

# ===================================================================
# PATTERN 6: Async streaming for FastAPI (see fastapi post for full impl)
# ===================================================================

import asyncio

async def stream_chain_to_console():
    async for chunk in production_chain.astream({"question": "Explain Helm charts."}):
        print(chunk, end="", flush=True)
    print()

# asyncio.run(stream_chain_to_console())

Frequently Asked Questions

What is LCEL in LangChain?

LCEL (LangChain Expression Language) is the composable chain interface that uses Python's | operator to connect Runnable components — prompts, LLMs, parsers, retrievers, custom functions. Every LCEL chain automatically supports .invoke(), .ainvoke(), .stream(), .astream(), and .batch(). It replaced the verbose legacy chain classes (LLMChain, SequentialChain) which are now considered deprecated. LCEL is the current standard for all LangChain chain construction.

How do I run two LangChain chains in parallel?

Use RunnableParallel({"result_a": chain_a, "result_b": chain_b}). When invoked with .ainvoke(), both chains run concurrently via asyncio.gather. Latency equals the slowest branch, not the sum. Common use cases: multi-source RAG (vector search + keyword search + database in parallel), multi-LLM comparison (GPT-4o vs Llama in parallel), and parallel document section generation.

How do I add fallback models to a LangChain chain?

Call .with_fallbacks([backup_llm]) on any Runnable. Example: resilient_llm = ChatOpenAI().with_fallbacks([ChatOllama(model="llama3.1:8b")]). When the primary raises an exception (rate limit, timeout, API error), LangChain automatically retries with the fallback. Chain multiple fallbacks: .with_fallbacks([first_fallback, second_fallback]). For production AI agents, always configure at least one fallback — cloud LLM API availability is not 100%.

What is the difference between RunnablePassthrough and RunnableLambda?

RunnablePassthrough passes input to output unchanged — use it in parallel dicts to preserve the original query alongside processed context: {"context": retriever, "question": RunnablePassthrough()}. RunnableLambda wraps any Python function as a Runnable — use it for custom transformations like formatting retrieved documents, post-processing LLM output, injecting metadata, or any logic that doesn't map to a built-in component.

Conclusion: LCEL Is the Grammar of AI Pipelines

When I teach LangChain in the workshop, I spend 30 minutes on LCEL before touching any other component. The investment pays off immediately: participants understand why chains are structured the way they are, they can debug at component boundaries, and they can extend existing chains by inserting new Runnables without rewriting everything around them.

LCEL is not just a convenience API — it is the architectural foundation of everything in LangChain that is production-grade. LangGraph nodes are LCEL chains. LangChain agent executors are LCEL chains. The multi-agent supervisor we build on Day 3 is a LangGraph StateGraph whose nodes are LCEL chains built on Day 1. Learning LCEL thoroughly is the foundation for everything that follows.