LangChain Runnables and LCEL: Compose Reliable LLM Pipelines

Why LCEL Matters

LCEL is LangChain Expression Language. It gives you a concise way to compose steps with the pipe operator. Each step is a runnable: it accepts input, returns output, and can be invoked, batched, streamed, retried, or traced.

Once you understand runnables, LangChain becomes easier to reason about. A prompt is a runnable. A model is a runnable. A parser is a runnable. Custom functions can become runnables too. The result is a pipeline that reads like the data flow of the app.

LCEL is not just syntax sugar. It creates a standard execution interface across chains. That means you can call <code>invoke</code> for one input, <code>batch</code> for many inputs, <code>stream</code> for partial output, and <code>with_retry</code> for transient failures.

Use sequences for fixed workflows.
Use maps for parallel preparation of inputs.
Use lambdas for small deterministic transformations.
Use retries around model or network-sensitive steps.

Parallel Context Preparation

This chain prepares two inputs in parallel: a direct question and a generated search query. The final prompt receives both values.

Parallel Context Preparation

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

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

def make_search_query(question: str) -> str:
    return question.lower().replace("?", "").strip()

prepare = RunnableParallel({
    "question": RunnableLambda(lambda x: x["question"]),
    "search_query": RunnableLambda(lambda x: make_search_query(x["question"])),
})

prompt = ChatPromptTemplate.from_template(
    "Answer the question.\nQuestion: {question}\nSearch query to use: {search_query}"
)

chain = prepare | prompt | model | StrOutputParser()
print(chain.invoke({"question": "How does vector search help RAG?"}))

RunnableParallel is useful when multiple prompt variables are derived from the same input.
Use deterministic functions for transformations that do not require model reasoning.

Add Retry to a Chain

Retries should be explicit. They help with transient network failures, not bad prompts or invalid business logic.

Add Retry to a Chain

safe_chain = (prompt | model | StrOutputParser()).with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
)

answer = safe_chain.invoke({"question": "Explain LCEL in one paragraph."})
print(answer)

Do not retry indefinitely. Repeated model calls increase latency and cost.
Pair retries with logging so you can identify recurring failures.

Before you move on