LCEL RunnableParallel for multi-step LangChain pipelines

LCEL’s RunnableParallel is the primitive that lets you fan out a single input into multiple concurrent branches, then fan the results back in — without leaving the declarative chain API. If you’ve been stitching together RunnableSequence objects by hand or reaching for asyncio.gather inside custom runnables, this tutorial shows how to replace that glue with a single composable unit that still streams, batches, and traces correctly.

Step 1: Understand what RunnableParallel actually does

RunnableParallel takes a dict mapping string keys to runnables. When invoked, it passes the same input to every branch simultaneously, collects the outputs into a dict with the same keys, and returns that dict downstream. The branches run in a thread pool (sync) or as tasks (async), so I/O-bound steps — provider calls, vector lookups, HTTP fetches — overlap automatically.

from langchain_core.runnables import RunnableParallel, RunnableLambda

parallel = RunnableParallel(
    summary=RunnableLambda(lambda x: f"Summary: {x['text'][:50]}..."),
    entities=RunnableLambda(lambda x: ["entity1", "entity2"]),
    sentiment=RunnableLambda(lambda x: "positive"),
)

result = parallel.invoke({"text": "LangChain Expression Language makes composition explicit."})
# {'summary': 'Summary: LangChain Expression Language makes composition...',
#  'entities': ['entity1', 'entity2'],
#  'sentiment': 'positive'}

Each branch receives the full input dict. If a branch only needs a slice, wrap it with RunnableLambda or itemgetter to pluck the field first.

Step 2: Pluck only the fields each branch needs

Passing the entire input to every branch works, but it couples branches to irrelevant keys and bloats traces. Use operator.itemgetter (or RunnablePick) to narrow the payload before the branch logic runs.

from operator import itemgetter
from langchain_core.runnables import RunnableParallel, RunnableLambda

parallel = RunnableParallel(
    summary=itemgetter("text") | RunnableLambda(lambda t: f"Summary: {t[:50]}..."),
    entities=itemgetter("text") | RunnableLambda(lambda t: ["entity1", "entity2"]),
    sentiment=itemgetter("text") | RunnableLambda(lambda t: "positive"),
)

result = parallel.invoke({"text": "Long document...", "metadata": {"id": "123"}})
# metadata never reaches the three lambdas

itemgetter("text") is itself a runnable that extracts "text" from the input dict. The pipe operator chains it with the branch logic. This pattern keeps branches pure and testable in isolation.

Step 3: Build a real multi-step pipeline with parallel enrichment

A common production pattern: ingest a document, then run classification, extraction, and summarization in parallel, then merge results into a final record. Here’s a complete, runnable pipeline using only LCEL primitives.

import os
from operator import itemgetter
from langchain_core.runnables import RunnableParallel, RunnableLambda, RunnablePassthrough
from langchain_openai import ChatOpenAI

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

classify_prompt = """Classify the document into one category: [legal, financial, technical, marketing].
Return only the category name.
Document: {text}"""

extract_prompt = """Extract all dollar amounts, dates, and organization names from the document.
Return JSON with keys: amounts, dates, organizations.
Document: {text}"""

summarize_prompt = """Summarize the document in three bullet points.
Document: {text}"""

classify_branch = (
    itemgetter("text")
    | RunnableLambda(lambda t: classify_prompt.format(text=t))
    | llm
    | RunnableLambda(lambda msg: msg.content.strip())
)

extract_branch = (
    itemgetter("text")
    | RunnableLambda(lambda t: extract_prompt.format(text=t))
    | llm
    | RunnableLambda(lambda msg: msg.content.strip())
)

summarize_branch = (
    itemgetter("text")
    | RunnableLambda(lambda t: summarize_prompt.format(text=t))
    | llm
    | RunnableLambda(lambda msg: msg.content.strip())
)

enrichment = RunnableParallel(
    category=classify_branch,
    extracted=extract_branch,
    summary=summarize_branch,
)

# Final merge: combine original input with enrichment results
pipeline = RunnablePassthrough.assign(**enrichment)

# Test
doc = """
Acme Corp announced Q3 revenue of $12.5M on October 15, 2024.
The board approved a $2M share buyback program.
"""

result = pipeline.invoke({"text": doc, "source": "sec-filing"})
print(result.keys())
# dict_keys(['text', 'source', 'category', 'extracted', 'summary'])

RunnablePassthrough.assign(**enrichment) spreads the parallel outputs as new keys on the original input dict. The original text and source fields pass through untouched.

Step 4: Share intermediate state between branches with RunnableAssign

Sometimes branch B needs a value produced by branch A, but you still want A and C to run in parallel. RunnableAssign lets you insert a computed field into the stream before the parallel fan-out, so downstream branches see it.

from langchain_core.runnables import RunnableAssign, RunnableParallel, RunnableLambda

# Expensive preprocessing shared by multiple branches
def heavy_preprocess(text: str) -> dict:
    # Simulate NLP pipeline: tokenization, embeddings, etc.
    return {"tokens": text.split(), "length": len(text)}

pipeline = (
    RunnableAssign(preprocessed=itemgetter("text") | RunnableLambda(heavy_preprocess))
    | RunnableParallel(
        stats=itemgetter("preprocessed") | RunnableLambda(lambda p: {"len": p["length"]}),
        first_tokens=itemgetter("preprocessed") | RunnableLambda(lambda p: p["tokens"][:5]),
        # This branch also sees the original input
        original_length=itemgetter("text") | RunnableLambda(len),
    )
)

result = pipeline.invoke({"text": "Hello world from LCEL"})
# {'preprocessed': {'tokens': ['Hello', 'world', 'from', 'LCEL'], 'length': 22},
#  'stats': {'len': 22},
#  'first_tokens': ['Hello', 'world', 'from', 'LCEL'],
#  'original_length': 22}

The RunnableAssign step runs once, then the three parallel branches all receive the enriched dict containing preprocessed. This avoids recomputation while keeping the parallel structure.

Step 5: Add retries and fallbacks per branch

Provider latency spikes happen. Wrap individual branches with .with_retry() and .with_fallbacks() so one flaky call doesn’t stall the whole pipeline.

from langchain_core.runnables import RunnableParallel, RunnableLambda
from langchain_openai import ChatOpenAI

primary_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
fallback_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

def make_branch(prompt: str):
    return (
        itemgetter("text")
        | RunnableLambda(lambda t: prompt.format(text=t))
        | primary_llm.with_retry(stop_after_attempt=2)
        | RunnableLambda(lambda msg: msg.content)
    ).with_fallbacks([
        (
            itemgetter("text")
            | RunnableLambda(lambda t: prompt.format(text=t))
            | fallback_llm
            | RunnableLambda(lambda msg: msg.content)
        )
    ])

parallel = RunnableParallel(
    summary=make_branch("Summarize in one sentence: {text}"),
    keywords=make_branch("Extract 5 keywords: {text}"),
)

result = parallel.invoke({"text": "Your document here..."})

Each branch gets its own retry policy and fallback chain. If the primary model is rate-limited, the branch transparently retries, then falls back to the cheaper model — other branches continue unaffected.

Step 6: Stream results as they arrive

RunnableParallel supports streaming via astream / stream. Each branch yields chunks; the parallel runnable yields partial dicts as branches complete. This is critical for UIs that want to render progressive results.

import asyncio
from langchain_core.runnables import RunnableParallel, RunnableLambda
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, streaming=True)

async def stream_demo():
    parallel = RunnableParallel(
        summary=itemgetter("text") | RunnableLambda(lambda t: f"Summarize: {t}") | llm,
        keywords=itemgetter("text") | RunnableLambda(lambda t: f"Keywords: {t}") | llm,
    )

    async for chunk in parallel.astream({"text": "Long document..."}):
        # chunk is a dict with whatever keys have produced output so far
        print(chunk.keys(), {k: v[:30] if isinstance(v, str) else v for k, v in chunk.items()})

asyncio.run(stream_demo())

Output (truncated):

dict_keys(['summary']) {'summary': 'Summarize: Long document...'}
dict_keys(['summary', 'keywords']) {'summary': 'Summarize: Long document...', 'keywords': 'Keywords: Long document...'}
dict_keys(['summary']) {'summary': 'The document discusses...'}
dict_keys(['keywords']) {'keywords': 'document, discusses, LCEL'}

Branches emit independently. Your consumer decides whether to wait for all keys or render partial updates.

Step 7: Verify the pipeline end to end

Add a small verification harness that exercises the full graph — including retries, fallbacks, and streaming — without hitting real providers in CI.

import pytest
from unittest.mock import AsyncMock, MagicMock
from langchain_core.runnables import RunnableParallel, RunnableLambda
from langchain_core.messages import AIMessage

@pytest.fixture
def mock_llm():
    llm = MagicMock()
    llm.ainvoke = AsyncMock(return_value=AIMessage(content="mocked output"))
    llm.with_retry.return_value = llm
    llm.with_fallbacks.return_value = llm
    return llm

def test_parallel_branches_receive_same_input(mock_llm):
    parallel = RunnableParallel(
        a=RunnableLambda(lambda x: x["value"] * 2),
        b=RunnableLambda(lambda x: x["value"] + 1),
    )
    result = parallel.invoke({"value": 5})
    assert result == {"a": 10, "b": 6}

@pytest.mark.asyncio
async def test_streaming_yields_partial_dicts(mock_llm):
    parallel = RunnableParallel(
        slow=RunnableLambda(lambda x: x) | mock_llm,
        fast=RunnableLambda(lambda x: x) | mock_llm,
    )
    chunks = []
    async for chunk in parallel.astream({"data": "test"}):
        chunks.append(chunk)
    # At least one partial dict should arrive before completion
    assert any(len(c) == 1 for c in chunks)
    # Final chunk has both keys
    assert set(chunks[-1].keys()) == {"slow", "fast"}

Run with pytest -v. The tests confirm:

• Input distribution works correctly
• Streaming yields partial results
• Retry/fallback wrappers don’t break the graph shape

Step 8: Common pitfalls and how to fix them

Step 9: Wire it into a production endpoint

If you’re serving this behind FastAPI, the runnable becomes your handler — no extra orchestration layer needed.

from fastapi import FastAPI
from pydantic import BaseModel
from langchain_core.runnables import RunnableParallel, RunnableLambda
from langchain_openai import ChatOpenAI

app = FastAPI()
llm = ChatOpenAI(model="gpt-4o-mini")

class DocIn(BaseModel):
    text: str
    request_id: str

classify = itemgetter("text") | RunnableLambda(lambda t: f"Classify: {t}") | llm
extract = itemgetter("text") | RunnableLambda(lambda t: f"Extract: {t}") | llm

pipeline = RunnableParallel(category=classify, extracted=extract)

@app.post("/enrich")
async def enrich(doc: DocIn):
    # request_id passes through automatically via RunnablePassthrough.assign
    result = await pipeline.ainvoke(doc.model_dump())
    return result

The request_id field flows through untouched because RunnableParallel preserves keys it doesn’t consume. Downstream logging or tracing can correlate by that ID.

Step 10: Observe latency with LangSmith or custom callbacks

RunnableParallel emits on_chain_start / on_chain_end events for the parallel node itself, plus nested events for each branch. Attach a callback to measure branch latency distribution.

from langchain_core.tracers import BaseCallbackHandler
from langchain_core.outputs import LLMResult
import time

class BranchLatencyHandler(BaseCallbackHandler):
    def __init__(self):
        self.starts = {}

    def on_chain_start(self, serialized, inputs, *, run_id, parent_run_id, tags, **kwargs):
        if parent_run_id and tags and "parallel_branch" in tags:
            self.starts[run_id] = time.perf_counter()

    def on_chain_end(self, outputs, *, run_id, **kwargs):
        if run_id in self.starts:
            latency = time.perf_counter() - self.starts.pop(run_id)
            print(f"Branch {run_id} took {latency:.3f}s")

handler = BranchLatencyHandler()
result = parallel.invoke({"text": "..."}, config={"callbacks": [handler]})

Tag branches explicitly when constructing the parallel runnable:

RunnableParallel(
    category=classify_branch.with_config(tags=["parallel_branch"]),
    extracted=extract_branch.with_config(tags=["parallel_branch"]),
)

Now you have per-branch latency without instrumenting each LLM call individually.

You now have a complete, production-ready pattern for lcel runnableparallel multi-step pipeline construction: fan-out with RunnableParallel, narrow inputs with itemgetter, share state via RunnableAssign, harden with retries and fallbacks, stream partial results, and verify with unit tests that mock the provider layer. The same graph runs locally, in CI, and behind your API gateway — no translation layer required.