LCEL RunnableBranch: conditional logic in chains

RunnableBranch is an LCEL primitive that routes an input to one of several branches by evaluating a sequence of (condition, runnable) pairs in order, executing the first runnable whose condition returns true, with an optional default fallback. It implements conditional control flow inside a declarative chain without breaking the Runnable interface, so the result remains composable with other LCEL operators like |, .with_config(), and .with_fallbacks(). Think of it as a type-safe, streamable if/elif/else that lives inside the graph rather than in your application code.

How runnablebranch evaluates conditions

RunnableBranch accepts a variable-length argument list of (condition, runnable) tuples followed by an optional default runnable. Each condition must be a callable that takes the chain input (or the output of the previous step if chained) and returns a boolean. Evaluation is strict left-to-right: the first condition that returns True short-circuits the rest, and its associated runnable receives the full input. If no condition matches and no default is provided, a ValueError is raised at invocation time.

from langchain_core.runnables import RunnableBranch, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

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

def is_question(x: dict) -> bool:
    return x.get("input", "").strip().endswith("?")

def is_code_request(x: dict) -> bool:
    return "code" in x.get("input", "").lower() or "implement" in x.get("input", "").lower()

branch = RunnableBranch(
    (is_question, llm | StrOutputParser()),
    (is_code_request, llm.bind(stop=["```"]) | StrOutputParser()),
    RunnableLambda(lambda x: &#123;"answer": "I only handle questions and code requests."&#125;),
)

chain = branch | StrOutputParser()

Conditions receive the exact input passed to the branch. If you pipe into RunnableBranch, the condition sees the upstream output. This matters when you transform data before routing — see the example below.

Why conditional branching matters in LCEL

Before RunnableBranch, conditional logic meant either writing imperative Python outside the chain (losing streaming, retries, and observability) or abusing RunnableLambda with nested if statements that broke type inference and made testing painful. RunnableBranch keeps the entire graph declarative, which gives you three concrete advantages:

Streaming works end-to-end. The selected branch streams tokens normally because the branch itself is a Runnable that implements stream() and astream(). No buffering, no manual chunk stitching.
Observability is automatic. LangSmith traces show the branch as a single node with the chosen child highlighted. You see exactly which path fired without correlating logs.
Fallbacks compose cleanly. Wrap the whole branch in .with_fallbacks([...]) or attach fallbacks to individual branches. The retry policy applies to whichever runnable actually executes.

# Branch-level fallback: if the selected LLM call fails, retry with a cheaper model
robust_branch = branch.with_fallbacks([
    RunnableBranch(
        (is_question, ChatOpenAI(model="gpt-3.5-turbo") | StrOutputParser()),
        (is_code_request, ChatOpenAI(model="gpt-3.5-turbo") | StrOutputParser()),
    )
])

Concrete example: routing by intent with structured output

A common production pattern is classifying the user’s intent with a fast, cheap model, then routing to a specialized chain. The classifier runs once; the expensive model only runs for the matched intent. This avoids the latency and cost of a single massive prompt that handles every case.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch, RunnableLambda, RunnableParallel
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import Literal

class Intent(BaseModel):
    intent: Literal["summarize", "translate", "code", "chat"] = Field(description="User intent")
    confidence: float = Field(ge=0.0, le=1.0)

classifier_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
classifier_prompt = ChatPromptTemplate.from_template(
    "Classify the user's request. Return JSON with 'intent' and 'confidence'.\n\nRequest: &#123;input&#125;"
)
classifier = classifier_prompt | classifier_llm | JsonOutputParser(pydantic_object=Intent)

# Specialized chains per intent
summarize_chain = (
    ChatPromptTemplate.from_template("Summarize in 3 sentences:\n\n&#123;input&#125;")
    | ChatOpenAI(model="gpt-4o", temperature=0.3)
    | StrOutputParser()
)

translate_chain = (
    ChatPromptTemplate.from_template("Translate to Spanish:\n\n&#123;input&#125;")
    | ChatOpenAI(model="gpt-4o", temperature=0.3)
    | StrOutputParser()
)

code_chain = (
    ChatPromptTemplate.from_template("Write clean, commented Python for:\n\n&#123;input&#125;")
    | ChatOpenAI(model="gpt-4o", temperature=0.2)
    | StrOutputParser()
)

chat_chain = (
    ChatPromptTemplate.from_template("Respond conversationally:\n\n&#123;input&#125;")
    | ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    | StrOutputParser()
)

def route_by_intent(x: dict) -> str:
    return x["intent"]

branch = RunnableBranch(
    (lambda x: x["intent"] == "summarize", summarize_chain),
    (lambda x: x["intent"] == "translate", translate_chain),
    (lambda x: x["intent"] == "code", code_chain),
    chat_chain,  # default
)

# Full pipeline: classify -> route -> execute
pipeline = RunnableParallel(
    intent=classifier,
    input=lambda x: x["input"],
) | RunnableLambda(lambda x: &#123;"intent": x["intent"]["intent"], "input": x["input"]&#125;) | branch

# Usage
result = pipeline.invoke(&#123;"input": "Summarize the key points of the Paris Agreement."&#125;)
print(result)

This pipeline streams the final answer even though classification is non-streaming. The branch itself doesn’t block — once the classifier emits, the selected chain takes over and streams normally.

Common misconceptions

“RunnableBranch is just syntactic sugar for if/else”

It’s not. An if/else in Python executes at graph construction time if the condition is a constant, or at invocation time but outside the Runnable graph if the condition depends on input. RunnableBranch embeds the decision inside the graph. This distinction matters for:

Serialization: branch.get_graph() shows the full conditional structure. An imperative if disappears.
Checkpointing: LangGraph can pause at the branch, resume, and re-evaluate conditions. Imperative logic cannot be checkpointed mid-branch.
Visual debugging: The trace shows the branch node with all possible children, not just the taken path.

“Conditions can be async”

Conditions are synchronous callables. If you need async classification (e.g., calling an embedding service to route), do the async work before the branch and pass the result as part of the input:

# Wrong: condition cannot be async
# RunnableBranch((async_condition, chain), ...)

# Right: async work upstream, sync condition downstream
async_classifier = classifier_prompt | classifier_llm | JsonOutputParser()

async def classify_then_route(input: str):
    intent = await async_classifier.ainvoke(&#123;"input": input&#125;)
    return branch.ainvoke(&#123;"intent": intent["intent"], "input": input&#125;)

“The default branch is optional”

It’s optional at construction but required at runtime if no condition matches. Omitting the default and letting ValueError bubble is a valid design choice for “impossible” states, but in production you almost always want a safe fallback (logging, a generic response, or a handoff chain). Treat the default as your else clause — explicit is better than implicit.

# Explicit default with observability
from langchain_core.runnables import RunnableConfig

def log_unhandled(x: dict, config: RunnableConfig) -> str:
    # config.metadata available for tracing
    return f"Unhandled input type: &#123;x.get('input', '')[:100]&#125;"

branch = RunnableBranch(
    (is_question, qa_chain),
    (is_code_request, code_chain),
    RunnableLambda(log_unhandled),  # explicit default
)

“RunnableBranch only works with string inputs”

Conditions receive whatever type the upstream runnable emits. If your chain passes dicts, Pydantic models, or custom objects, the condition receives that same type. This enables routing on structured fields without string parsing:

from pydantic import BaseModel

class Request(BaseModel):
    user_tier: Literal["free", "pro", "enterprise"]
    payload: str

tier_branch = RunnableBranch(
    (lambda r: r.user_tier == "enterprise", enterprise_chain),
    (lambda r: r.user_tier == "pro", pro_chain),
    free_chain,
)

Composing branches with other LCEL primitives

RunnableBranch implements the full Runnable interface, so it composes with every LCEL operator. A few patterns that appear in real systems:

Parallel pre-processing, then branch

from langchain_core.runnables import RunnableParallel

enriched = RunnableParallel(
    intent=classifier,
    entities=entity_extractor,
    sentiment=sentiment_analyzer,
    input=lambda x: x["input"],
)

branch = RunnableBranch(
    (lambda x: x["intent"]["intent"] == "support" and x["sentiment"]["score"] &lt; -0.5, escalation_chain),
    (lambda x: x["intent"]["intent"] == "support", support_chain),
    general_chain,
)

chain = enriched | branch

Branch inside a branch (nested routing)

code_branch = RunnableBranch(
    (lambda x: "python" in x["input"].lower(), python_chain),
    (lambda x: "javascript" in x["input"].lower(), js_chain),
    generic_code_chain,
)

top_level = RunnableBranch(
    (is_code_request, code_branch),
    (is_question, qa_chain),
    chat_chain,
)

Branch with per-branch config

branch = RunnableBranch(
    (is_question, qa_chain.with_config(&#123;"tags": ["qa"], "metadata": &#123;"cost_center": "support"&#125;&#125;)),
    (is_code_request, code_chain.with_config(&#123;"tags": ["code"], "metadata": &#123;"cost_center": "engineering"&#125;&#125;)),
    default_chain,
)

Each branch inherits the parent’s config but can override tags, metadata, callbacks, and recursion limits independently.

When not to use RunnableBranch

Two static paths known at build time: Use RunnablePassthrough.assign(...) or simple dict mapping. Branch adds indirection you don’t need.
Complex state machines: If you have loops, multi-step workflows, or need to remember previous branches, use LangGraph. RunnableBranch is a single decision point, not a state machine.
High-frequency routing with simple rules: If you route millions of requests on a single field (e.g., model: "gpt-4o"), a dict lookup is faster and clearer:

MODEL_CHAINS = &#123;
    "gpt-4o": gpt4o_chain,
    "gpt-4o-mini": gpt4o_mini_chain,
    "claude-3.5-sonnet": claude_chain,
&#125;

def route_by_model(x: dict):
    return MODEL_CHAINS[x["model"]].invoke(x)

Debugging tips

Print the graph: print(branch.get_graph().draw_ascii()) shows the branch structure with all children.
Log the condition input: Wrap a condition in a lambda that logs before returning:

def logged_condition(x):
    print(f"Condition input: &#123;x&#125;")
    return is_question(x)

Use with_listeners for side effects: Attach on_start/on_end listeners to individual branches to measure latency per path without modifying the chain logic.

from langchain_core.runnables import RunnableConfig
from langchain_core.tracers import Run

def make_measured(chain, name: str):
    def on_start(run: Run, config: RunnableConfig):
        run.metadata["branch"] = name
    return chain.with_listeners(on_start=on_start)

branch = RunnableBranch(
    (is_question, make_measured(qa_chain, "qa")),
    (is_code_request, make_measured(code_chain, "code")),
    make_measured(default_chain, "default"),
)

Summary

RunnableBranch is the idiomatic way to express conditional control flow in LCEL. It keeps the graph declarative, preserves streaming and observability, and composes with the rest of the LCEL ecosystem. Use it when you need to route based on input content, classification results, or structured fields — and when you want that routing to be visible, testable, and checkpointable. For static routing or complex multi-step workflows, reach for simpler dict lookups or LangGraph respectively.