Chains and LCEL (LangChain Expression Language)

In legacy software, composing functions looks like nested calls: result = parse(model(format(input)))

This becomes unreadable and makes adding asynchronous execution, streaming, and error handling a nightmare.

LangChain Expression Language (LCEL) is a declarative domain-specific language designed to compose LLM primitives into production chains using the Unix pipe operator (|).

1. The Pipe Operator (|) and the Runnable Protocol

Under LCEL, every component (prompts, models, retrievers, output parsers) implements the Runnable Protocol:

flowchart LR
    Input["Input Dictionary:<br/>{'question': 'Spark DPP'}"] --> Prompt["ChatPromptTemplate"]
    Prompt -->|Formatted Messages| Model["ChatModel (LLM)"]
    Model -->|AIMessage Object| Parser["StrOutputParser"]
    Parser --> Output["Final String Answer"]

    style Input fill:#475569,color:#fff
    style Prompt fill:#2563eb,color:#fff
    style Model fill:#7c3aed,color:#fff
    style Parser fill:#059669,color:#fff
    style Output fill:#d97706,color:#fff

In Python code, this is expressed as:

chain = prompt | model | StrOutputParser()
result = chain.invoke({"question": "Explain dynamic partition pruning in Spark"})

The 4 Universal Runnable Methods

Every Runnable component automatically gains four execution methods out of the box:

2. Advanced LCEL Primitives

Data pipelines require parallel branches, branching logic, and custom data transformations. LCEL provides specialized Runnable utilities:

flowchart TD
    Input["User Query:<br/>'Compare Snowflake vs Databricks'"]

    Input --> Fork["RunnableParallel"]
    Fork --> Branch1["Branch 1: Fetch Snowflake Docs (Retriever)"]
    Fork --> Branch2["Branch 2: Fetch Databricks Docs (Retriever)"]

    Branch1 --> Merge["Join Results into Context Dict"]
    Branch2 --> Merge

    Merge --> Prompt["Synthesis Prompt"]
    Prompt --> Model["LLM"]
    Model --> Output["Comparative Analysis"]

    style Input fill:#334155,color:#fff
    style Fork fill:#2563eb,color:#fff
    style Branch1 fill:#059669,color:#fff
    style Branch2 fill:#7c3aed,color:#fff
    style Merge fill:#d97706,color:#fff
    style Output fill:#10b981,color:#fff

Key LCEL Primitives:

1. RunnablePassthrough: Passes the input data through unchanged or dynamically assigns new keys to an input dictionary.
2. RunnableParallel: Executes multiple branches concurrently on separate threads, then bundles their outputs into a single dictionary.
3. RunnableLambda: Converts any regular Python function into an LCEL-compatible Runnable.

3. Hands-on: Building an LCEL Chain with Parallelism

from typing import Any, Callable

class SimpleRunnable:
    """Demonstrates how the LCEL pipe operator works under the hood."""
    def __init__(self, func: Callable[[Any], Any]):
        self.func = func

    def __or__(self, other):
        """Overload the Unix pipe operator |"""
        if isinstance(other, SimpleRunnable):
            return SimpleRunnable(lambda x: other.invoke(self.invoke(x)))
        return NotImplemented

    def invoke(self, input_data: Any) -> Any:
        return self.func(input_data)

# Building a 3-stage SQL Optimization pipeline
extract_tables = SimpleRunnable(
    lambda query: {"query": query, "tables": ["orders", "customers"]}
)

annotate_partitions = SimpleRunnable(
    lambda data: {
        **data,
        "partition_advice": "Ensure 'orders' is filtered by 'order_date' cluster key."
    }
)

format_summary = SimpleRunnable(
    lambda data: (
        f"Query Analysis for Tables {data['tables']}:
"
        f"Recommendation: {data['partition_advice']}"
    )
)

if __name__ == "__main__":
    # LCEL Composition using |
    sql_review_chain = extract_tables | annotate_partitions | format_summary

    test_query = "SELECT * FROM orders JOIN customers ON orders.c_id = customers.id"
    report = sql_review_chain.invoke(test_query)

    print("=== LCEL Chain Output ===")
    print(report)

Key Takeaways

• LCEL replaces messy nested code with clean declarative pipelines using the pipe operator (|).
• All Runnables support streaming, async, and batch automatically without writing custom boilerplate.
• RunnableParallel enables concurrent retrieval across disparate data sources.