TopolearnReal pipelines fan out, pass data through, call custom code and must survive flaky APIs.
Beyond linear | chains, langchain_core.runnables offers building blocks:
- RunnableParallel, or simply a dict inside a chain, runs branches concurrently on the same input and returns a dict:
{"context": retriever, "question": RunnablePassthrough()}. - RunnablePassthrough forwards its input unchanged;
RunnablePassthrough.assign(key=runnable)adds keys to an input dict while keeping the rest. - RunnableLambda(fn) wraps a function explicitly (needed when it is the first step, or to add a name).
itemgetter("question")fromoperatoris a handy way to pick a field.
Every runnable also offers wrappers:
.with_retry(stop_after_attempt=3)retries on exceptions with exponential backoff..with_fallbacks([backup_chain])tries alternatives, e.g. another provider, when the primary fails..with_config(run_name=..., tags=[...])names steps for tracing..batch(inputs, config={"max_concurrency": 5})limits parallelism to respect rate limits.
The config dict travels through the whole chain: callbacks, tags, metadata and configurable values reach every step. Use .configurable_fields or init_chat_model(configurable_fields=...) to choose the model per call.
Inspect any chain with chain.get_graph().print_ascii(); it shows exactly which steps run in parallel and in sequence.
Example
from operator import itemgetter
from langchain_core.runnables import RunnableLambda, RunnableParallel, RunnablePassthrough
def flaky(x): # fails on purpose
raise TimeoutError("primary down")
stats = RunnableParallel(
length=RunnableLambda(lambda d: len(d["text"])),
words=RunnableLambda(lambda d: len(d["text"].split())),
)
chain = (
RunnablePassthrough.assign(stats=stats) # keep input, add a key
| RunnablePassthrough.assign(shout=itemgetter("text") | RunnableLambda(str.upper))
)
print(chain.invoke({"text": "hello lcel world"}))
# {'text': 'hello lcel world', 'stats': {'length': 16, 'words': 3}, 'shout': 'HELLO LCEL WORLD'}
robust = RunnableLambda(flaky).with_fallbacks([RunnableLambda(lambda x: f"backup: {x}")])
print(robust.invoke("q")) # backup: q
double = RunnableLambda(lambda x: x * 2)
print(double.batch([1, 2, 3], config={"max_concurrency": 2})) # [2, 4, 6]Remember
- Dicts and RunnableParallel fan out;
RunnablePassthrough.assignadds keys with_retryandwith_fallbacksadd resilience to any step- Config (callbacks, tags, max_concurrency) flows through the whole chain
Watch out for
- Wrapping an entire chain in a single huge lambda, which hides steps from tracing and loses streaming.