Composing Chains with LCEL

LangChain Expression Language composes runnable stages with |. Each stage receives the previous stage's output, so input and output shape matter more than how short the expression looks.

#A local, reproducible pipeline

This example needs only langchain-core; it makes no model calls.

pythonCopy
from langchain_core.runnables import RunnableLambda, RunnableParallel

normalize = RunnableLambda(lambda value: value.strip())
inspect = RunnableParallel(
    normalized=RunnableLambda(lambda value: value),
    characters=RunnableLambda(len),
)
chain = normalize | inspect
assert chain.invoke("  hello  ") == {"normalized": "hello", "characters": 5}
print(chain.batch([" one ", " two "], config={"max_concurrency": 2}))

RunnableParallel gives independent branches the same input and returns a dictionary of results. It is not suitable when the second branch needs the first branch's output.

#Preserve the data you still need

In RAG, retain the original question while another branch retrieves documents. RunnablePassthrough passes input through unchanged. A formatting function can join document text and source metadata before the prompt stage.

Name significant subchains such as retrieve, format_context and answer. Inspect intermediate data with redacted diagnostics rather than printing every private prompt.

#Concurrency, streaming and failure

invoke, ainvoke, batch and stream are execution interfaces, not a promise that every component streams incrementally. A blocking lambda or external service can become the bottleneck. Test the actual first-output and completion latency.

Use bounded concurrency and retries. A fallback may help with a transient provider failure, but must not repeat a side effect or change privacy policy silently. Simple routing can use RunnableBranch; explicit state machines or LangGraph are easier to inspect when the workflow loops or pauses for approval.

Runnable reference↗ · LangGraph