LangChain Expression Language (LCEL): A Practical Guide

What is LangChain Expression Language (LCEL)?

LCEL is a declarative language introduced in LangChain v0.1.0 for composing chains of Large Language Model (LLM) calls, data transformations, and utility components. At its core, LCEL uses the pipe operator (|) to connect runnable objects, creating a graph of operations that can be executed, streamed, batched, or served as an API. Every LCEL expression yields a Runnable object, providing a unified interface for invocation, streaming, async execution, and more.

Instead of writing procedural code to glue together prompts, models, and output parsers, LCEL lets you define the chain as a readable expression, similar to composing functions in a functional pipeline.

Why LCEL Matters

Before LCEL, building LangChain chains often involved subclassing Chain classes and overriding methods, which was verbose and error-prone. LCEL offers several key advantages:

• Simplicity: Build chains with a clean, linear syntax that mirrors data flow.
• Unified Interface: Every LCEL expression is a Runnable, so you can invoke, stream, batch, and deploy them consistently.
• Automatic Parallelism: LCEL automatically runs independent branches in parallel when using RunnablePassthrough.assign(...).
• Streaming & Async Support: Streaming tokens or events works out of the box; async execution is trivial with ainvoke, astream, etc.
• Seamless Integration: LCEL chains are directly deployable via LangServe, and integrate with LangSmith for tracing.
• Optimized Execution: LangChain optimizes the execution graph for batching and caching.

Getting Started with LCEL

To use LCEL, you need LangChain installed (version >= 0.1.0). The core concept is chaining Runnable objects with the pipe operator. Each component in the chain is a Runnable, and the output of one becomes the input of the next.

Basic Syntax

The pipe operator | connects two runnables. For example, prompt | model | output_parser creates a chain that takes user input, formats a prompt, sends it to an LLM, and parses the response.

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

prompt = PromptTemplate.from_template("Tell me a joke about {topic}")
model = ChatOpenAI(model="gpt-3.5-turbo")
parser = StrOutputParser()

chain = prompt | model | parser

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

Here, chain is a RunnableSequence that can be invoked with a dictionary matching the prompt's input variables.

Common Runnable Components

LCEL chains are built from a rich set of built-in runnables:

• PromptTemplate / ChatPromptTemplate: Format prompts from templates and input variables.
• LLMs / Chat Models: OpenAI, ChatOpenAI, Anthropic, etc.
• Output Parsers: StrOutputParser, JsonOutputParser, PydanticOutputParser.
• RunnablePassthrough: Passes input through unchanged or assigns extra keys.
• RunnableLambda: Wraps a plain Python function into a Runnable.
• RunnableBranch: Conditional routing based on input.
• RunnableParallel: Runs multiple runnables in parallel (created automatically by .assign() or dict syntax).

Your First LCEL Chain

Let's build a classic chain: prompt → model → string parser. We'll use ChatPromptTemplate for a chat-style prompt.

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

prompt = ChatPromptTemplate.from_template("Summarize the following text in one sentence:\n\n{text}")
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
parser = StrOutputParser()

chain = prompt | model | parser

summary = chain.invoke({"text": "LangChain is a framework for developing applications powered by language models..."})
print(summary)

Advanced LCEL Patterns

Passing Data Through with RunnablePassthrough

RunnablePassthrough is essential for building chains that need to forward the original input alongside transformed data. It can be used as a placeholder or to assign new keys from other runnables.

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

# Example: Keep original input while adding a processed column
prompt = ChatPromptTemplate.from_template("Write a poem about {topic} in the style of {poet}")
model = ChatOpenAI(model="gpt-3.5-turbo")

# RunnableLambda to extract poet style from a dict
def extract_poet_style(inputs):
    # Suppose we have a mapping elsewhere
    style_map = {"shakespeare": "Shakespearean sonnet", "frost": "rural New England"}
    topic = inputs["topic"]
    poet_key = inputs.get("poet_style_key", "shakespeare")
    return {"topic": topic, "poet": style_map[poet_key]}

# Build chain: pass input through a transformation, then prompt/model/parser
chain = (
    RunnableLambda(extract_poet_style)  # transforms raw input
    | prompt
    | model
    | StrOutputParser()
)

result = chain.invoke({"topic": "autumn", "poet_style_key": "frost"})
print(result)

For more complex flows, RunnablePassthrough.assign(...) lets you add new keys by running other runnables in parallel:

from langchain_core.runnables import RunnablePassthrough

# Suppose we have a retriever that fetches documents
retriever = ...  # a Runnable retriever

chain = (
    RunnablePassthrough.assign(documents=lambda x: retriever.invoke(x["query"]))
    | prompt
    | model
    | parser
)

# Now the input dict gets augmented with "documents" before going to the prompt.

Transforming Data with RunnableLambda

RunnableLambda converts any Python function into a Runnable. The function receives the input from the previous step and must return an output that becomes the input for the next step (or a dict to merge).

from langchain_core.runnables import RunnableLambda

# Simple lambda to uppercase the input string
uppercase = RunnableLambda(lambda x: x.upper())

chain = uppercase | model | parser
# Now input "hello" becomes "HELLO" before hitting the model.

You can also use it for more complex preprocessing:

def preprocess(inputs: dict) -> dict:
    # Clean and validate fields
    return {"cleaned_text": inputs["raw_text"].strip(), "max_tokens": min(inputs.get("max_tokens", 512), 1024)}

preprocess_runnable = RunnableLambda(preprocess)

Branching Logic with RunnableBranch

Conditional routing is possible with RunnableBranch. It takes a list of (condition, runnable) pairs and a default runnable. Conditions are functions that receive the input and return a boolean.

from langchain_core.runnables import RunnableBranch

# Define branches
def is_short(input_text: str) -> bool:
    return len(input_text) < 50

short_chain = prompt_short | model | parser
long_chain = prompt_long | model | parser

branch = RunnableBranch(
    (is_short, short_chain),
    (lambda x: "urgent" in x.lower(), urgent_chain),  # another condition
    long_chain  # default
)

full_chain = RunnableLambda(lambda x: x["text"]) | branch

Handling Multiple Inputs

LCEL chains can accept multiple inputs by passing a dict with keys. The prompt template can reference them. If you need to combine inputs from different sources, use RunnableParallel or the dict literal syntax:

from langchain_core.runnables import RunnableParallel

# Run two retrievers in parallel and combine results
retriever1 = ... # e.g., vector store retriever
retriever2 = ... # another retriever

parallel_retrieval = RunnableParallel(
    docs_from_source1=retriever1,
    docs_from_source2=retriever2
)

# Then merge into a single list
def merge_docs(inputs):
    return {"combined_docs": inputs["docs_from_source1"] + inputs["docs_from_source2"]}

chain = parallel_retrieval | RunnableLambda(merge_docs) | prompt | model | parser

Streaming and Async

LCEL chains support streaming and async natively. Use stream() to get tokens as they arrive, or astream() for async streaming.

# Synchronous streaming
for chunk in chain.stream({"topic": "AI"}):
    print(chunk, end="", flush=True)

# Async invocation
import asyncio

async def main():
    async for chunk in chain.astream({"topic": "AI"}):
        print(chunk, end="", flush=True)

asyncio.run(main())

You can also use ainvoke() and abatch() for async non-streaming.

Best Practices

• Keep Chains Linear and Readable: Prefer the pipe syntax and avoid deeply nested branches when possible. Complex logic can be encapsulated in RunnableLambda functions.
• Use Typed Dictionaries for Inputs: For complex chains, define Pydantic models or TypedDicts for input/output schemas to catch errors early and improve documentation.
• Leverage RunnablePassthrough.assign for Parallelism: When you need to augment the input with data from multiple sources (e.g., retrievers), use .assign() to run them concurrently and merge results automatically.
• Attach Fallbacks for Resilience: Use .with_fallbacks([backup_runnable]) to specify backup models or parsers in case the primary fails.
• Add Configuration with .with_config(): Set tags, metadata, or callbacks to help with tracing and logging. For example, chain.with_config(tags=["production"]).
• Test Chains Thoroughly: Use invoke() with a variety of inputs and inspect intermediate steps using LangSmith or simple print debugging in RunnableLambda.
• Prefer LCEL for Production: LCEL chains are optimized for batching, streaming, and deployment. Avoid custom Chain subclassing in new code.
• Monitor Performance: Use LangSmith to trace LCEL chains and identify bottlenecks.

Conclusion

LangChain Expression Language transforms the way we build LLM applications by providing a declarative, composable, and highly optimized syntax. Whether you're building a simple prompt-response pipeline or a complex multi-step agent, LCEL offers clarity, automatic parallelism, and seamless deployment. By mastering LCEL, you unlock the full potential of LangChain for production-ready, maintainable AI workflows. Start composing your chains with LCEL today and experience the difference in developer productivity and runtime performance.