Learn LangChain - LCEL: Chains & Composition

#Introduction

In episode 4, you perfected the art of composing prompts and handling modern streaming. Episode 5 is the most defining episode of the entire foundational phase: LCEL. Everything you've learned so far — primitives, chat models, prompts — gets assembled here into genuinely useful pipelines.

You already know the core LCEL concept from episode 2: every component is a Runnable, and runnables are connected with the pipe operator |. Now we'll fully unpack the five main runnables, the entire execution lifecycle, and how applications survive errors.

#Runnable and RunnableSequence

RunnableSequence is the backbone of LCEL — a chain that executes step by step. The pipe operator | is the syntactic sugar for building one:

The most common sequential chain in LangChain

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Kamu adalah penerjemah Inggris-Indonesia."),
    ("human", "Terjemahkan: {teks}"),
])
 
model = ChatOpenAI(model="gpt-4o-mini")
 
rantai = prompt | model | StrOutputParser()
 
print(rantai.invoke({"teks": "Hello, how are you today?"}))

Data flow: the input dict goes into prompt, its result is forwarded to model, then the AI answer is turned into a plain string by StrOutputParser. This expression can also be written explicitly as RunnableSequence — the result is the same, but the pipe form is far more readable. Every element in the chain is a runnable, and the chain itself is also a runnable, so it can be connected to other chains.

#RunnableParallel and RunnablePassthrough

Not every flow is linear. RunnableParallel runs several branches at once over the same input, then collects their results into a single dict:

Parallel branches over one input

from langchain_core.runnables import RunnableParallel, RunnablePassthrough
 
penerjemah = prompt | model | StrOutputParser()
 
cabang = RunnableParallel(
    terjemahan=penerjemah,
    asli=RunnablePassthrough(),
)
 
hasil = cabang.invoke({"teks": "Good morning!"})
print(hasil["terjemahan"])
print(hasil["asli"])

Here RunnablePassthrough plays an important role: it passes the input through unchanged. Useful for inserting original data into the final result, or adding context in the middle of a flow. Both branches above run in parallel — if the translation is slow, the passthrough branch still finishes instantly.

#RunnableLambda and RunnableMap

A chain doesn't always consist solely of LangChain components. RunnableLambda wraps an ordinary Python function into a runnable, so any function — sanitization, validation, calls to another service — can join the pipeline:

Inserting a Python function with RunnableLambda

from langchain_core.runnables import RunnableLambda
 
def bersihkan_teks(teks: str) -> str:
    return teks.strip().lower()
 
def sapaan(nama: str) -> str:
    return f"Halo {nama}, selamat datang di pipeline!"
 
langkah = RunnableLambda(bersihkan_teks) | RunnableLambda(sapaan)
print(langkah.invoke("  BUDI  "))

Each function receives the output of the previous step and passes it to the next. The only rule: the function must be callable with one main argument (unless specially decorated for extra configuration).

Besides RunnableLambda, there's RunnableMap — the dict version of parallelism: you define key names and their functions, then each function's result is collected into the same key:

RunnableMap runs functions and collects results

from langchain_core.runnables import RunnableMap
 
analisa = RunnableMap({
    "panjang": lambda teks: len(teks),
    "kata": lambda teks: teks.split(),
})
 
print(analisa.invoke("LCEL itu menarik"))

The result is a dict with two keys: panjang holding the character count, kata holding the list of words. This pattern is ideal for pre-processing that turns one input into many values the chain will use later.

#Lifecycle: invoke, batch, stream

All runnables share the same three execution modes, and the chains you build are no exception — this is what makes LCEL so consistent:

One chain, three execution modes

jawaban_tunggal = rantai.invoke({"teks": "Hello!"})
 
banyak_jawaban = rantai.batch([
    {"teks": "Hello!"},
    {"teks": "Thank you!"},
    {"teks": "Goodbye!"},
])
 
for potongan in rantai.stream({"teks": "Hello!"}):
    print(potongan, end="", flush=True)

All three have async counterparts — ainvoke, abatch, astream — used when the application runs on a server. Important pattern: because the chain itself is a runnable, you can build chains out of other chains (nested composition), and every layer still supports all three modes.

#Event Streaming

Beyond the execution modes, LCEL exposes an event stream — a detailed record of each step within a chain. When a chain runs with astream_events, you see every event: when the prompt starts assembling, when the model starts streaming, when the parser finishes, complete with timing and ordering. This is the foundation of observability and UI progress:

Observing events per chain step

async def amati():
    async for event in rantai.astream_events(
        {"teks": "Hello!"}, version="v3"
    ):
        if event["event"] == "on_chat_model_stream":
            print("Model mengalir:", event["data"]["chunk"].content, end="")
 
asyncio.run(amati())

With this pattern you can show a "model is typing" indicator in your UI, or record events for debugging. In episode 19 we'll see the full version through LangSmith, which captures this event stream automatically.

#Error Handling and Retry

Production applications must withstand failures — and this is the part most often missed. Two main mechanisms in LCEL: retry for failures that can be repeated (for example transient rate limits), and fallback for failures that need to be diverted to a backup.

Automatic retry with backoff is enabled via with_retry:

Enabling retry with custom rules

from langchain_core.runnables import Runnable
 
rantai_tangguh = rantai.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
)

Fallback uses with_fallbacks — if the primary model fails, the chain automatically tries an alternative:

Fallback to a backup model

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
 
rantai_utama = prompt | ChatOpenAI(model="gpt-4o") | StrOutputParser()
rantai_cadangan = prompt | ChatAnthropic(model="claude-3-5-sonnet-latest") | StrOutputParser()
 
rantai_aman = rantai_utama.with_fallbacks([rantai_cadangan])

Beyond that, errors can still be caught normally with try/except — for example to validate input before it enters the chain, or to record failures to a monitoring system. The combination of retry, fallback, and manual handling makes your application much more resilient to occasionally unstable model providers.

#Conclusion

Episode 5 completed the LCEL foundation: building RunnableSequence with the pipe operator, parallelizing branches with RunnableParallel, passing data through with RunnablePassthrough, wrapping Python functions with RunnableLambda and RunnableMap, running the invoke/batch/stream lifecycle with async variants, monitoring event streaming, and securing the application with retry and fallback.

Key takeaways:

• prompt | model | parser builds a RunnableSequence; the chain itself remains a runnable.
• RunnableParallel runs branches concurrently; RunnablePassthrough passes input through unchanged.
• RunnableLambda inserts Python functions; RunnableMap collects function results into a dict.
• Every runnable consistently supports invoke/batch/stream and their async variants.
• Protect your application with with_retry for retries and with_fallbacks for model backups.

In episode 6 we'll learn to control the shape of model output: output parsers & structured output — StrOutputParser, JsonOutputParser, PydanticOutputParser, JSON-schema-based with_structured_output, streaming parsing, and Pydantic-typed structured results. See you there!