Building LangChain Chains: Multi-Step LLM Pipelines

Chains orchestrate multi-step workflows where the output of one LLM call feeds into the next. Instead of manually threading results through multiple model calls, LangChain's LangChain Expression Language (LCEL) and chain abstractions let you compose reasoning pipelines declaratively. A well-designed chain reads like pseudocode: extract keywords, summarize, classify—each step flows into the next with minimal boilerplate.

In a project analyzing customer feedback, I built a pipeline that extracted sentiment, then generated personalized responses based on that sentiment. Without chains, the orchestration logic was scattered across three functions. With LangChain chains, it was one readable pipeline.

Understanding LCEL (LangChain Expression Language)

LCEL is LangChain's syntax for composing Runnables (any component that has an invoke() method). The pipe operator (|) chains Runnables together:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)
prompt = ChatPromptTemplate.from_template("Classify this as positive or negative: {text}")

# LCEL chain
chain = prompt | model

result = chain.invoke({"text": "This product is amazing!"})
print(result.content)  # "Positive"

When you pipe prompt | model, LangChain creates a chain that:

1. Formats the input dictionary through the prompt
2. Passes the formatted prompt to the model
3. Returns the model's response

You can chain more than two components. Each output becomes the next component's input:

from langchain_core.output_parsers import StrOutputParser

# Three-step chain: prompt → model → string parser
chain = prompt | model | StrOutputParser()
result = chain.invoke({"text": "This product is amazing!"})
print(result)  # String output, not Message object

Building Sequential Multi-Step Chains

For tasks requiring sequential steps, compose multiple prompt-model pairs:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

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

# Step 1: Extract keywords
extract_prompt = ChatPromptTemplate.from_template(
    "Extract 5 key topics from this text: {text}"
)
extract_chain = extract_prompt | model | StrOutputParser()

# Step 2: Summarize based on keywords
summarize_prompt = ChatPromptTemplate.from_template(
    "Given these keywords: {keywords}, write a one-sentence summary."
)
summarize_chain = summarize_prompt | model | StrOutputParser()

# Invoke step 1
text = "LangChain simplifies LLM orchestration. It handles prompts, chains, and memory..."
keywords = extract_chain.invoke({"text": text})
print(f"Keywords: {keywords}")

# Invoke step 2, using step 1's output
summary = summarize_chain.invoke({"keywords": keywords})
print(f"Summary: {summary}")

This works, but you're manually threading outputs. For cleaner composition, use RunnablePassthrough and RunnableLambda:

from langchain_core.runnables import RunnablePassthrough, RunnableLambda

# Create a combined chain where step 2 receives step 1's output
pipeline = (
    {"text": RunnablePassthrough()}  # Pass input through unchanged
    | {"text": RunnablePassthrough(), "keywords": extract_chain}  # Run extract_chain, keep original text
    | RunnableLambda(lambda x: summarize_chain.invoke({"keywords": x["keywords"]}))
)

result = pipeline.invoke({"text": text})
print(result)

However, for most cases, explicit multi-step invocation (as in the first example) is clearer.

Branching and Conditional Logic with RunnableBranch

Some workflows need to route to different chains based on input. Use RunnableBranch:

from langchain_core.runnables import RunnableBranch
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

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

# Classify input
classify_prompt = ChatPromptTemplate.from_template(
    "Is this technical or non-technical? Respond with only 'technical' or 'non-technical': {text}"
)
classifier = classify_prompt | model | StrOutputParser()

# Route based on classification
technical_prompt = ChatPromptTemplate.from_template(
    "Explain this technically: {text}"
)
technical_chain = technical_prompt | model | StrOutputParser()

non_technical_prompt = ChatPromptTemplate.from_template(
    "Explain this simply: {text}"
)
non_technical_chain = non_technical_prompt | model | StrOutputParser()

# Create branching logic
def route(input_dict):
    classification = classifier.invoke(input_dict)
    if "technical" in classification.lower():
        return technical_chain
    else:
        return non_technical_chain

from langchain_core.runnables import RunnableLambda

router = RunnableLambda(route)

result = router.invoke({"text": "How do neural networks work?"})
print(result)

For static routes (no classification needed), RunnableBranch is cleaner:

branch = RunnableBranch(
    (lambda x: "code" in x.get("type", "").lower(), code_chain),
    (lambda x: "explain" in x.get("type", "").lower(), explain_chain),
    default_chain  # Fallback if no condition matches
)

Error Handling and Retries in Chains

Production chains need resilience. LangChain chains support retries via the with_retry() method:

from langchain_core.runnables import RunnableRetry
from langchain_openai import ChatOpenAI

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

# Wrap model with automatic retries on rate limits and transient errors
resilient_model = model.with_retry(
    stop=stop_after_attempt(3),  # Retry up to 3 times
    wait=wait_exponential(multiplier=1, min=4, max=10),  # Exponential backoff
)

chain = prompt | resilient_model | StrOutputParser()

For custom error handling, use RunnableLambda and try-except:

from langchain_core.runnables import RunnableLambda

def safe_invoke(input_dict):
    try:
        return chain.invoke(input_dict)
    except Exception as e:
        print(f"Chain failed: {e}")
        return "Error: Could not process request."

safe_chain = RunnableLambda(safe_invoke)

Debugging Chains with debug=True

When chains misbehave, enable debug mode to trace execution:

# Enable debug globally
import langchain
langchain.debug = True

# Invoke your chain
result = chain.invoke({"text": "sample"})
# Prints detailed logs of each step

Debug output shows each component's input and output, helping you spot where reasoning breaks down.

Chain Comparison Table

Pattern	Use Case	Complexity	When to Use
Simple pipe (prompt | model)	Single LLM call	Very low	Quick experiments
Sequential steps	Extract, then summarize	Low	Multi-stage processing
RunnableBranch	Route based on classification	Medium	Conditional workflows
RunnableLambda	Custom Python logic	Medium	Integration with existing code
LangGraph	Agent loops, state machines	High	Autonomous agents, iterations

Key Takeaways

• LCEL chains components with the pipe operator for readable multi-step workflows
• Sequential chains thread outputs from one step to the next without manual orchestration
• RunnableBranch routes inputs to different chains based on conditions
• Use with_retry() for production resilience against rate limits and transient errors
• Enable langchain.debug = True to trace execution and debug misbehaving chains
• For complex stateful workflows, graduate to LangGraph

Frequently Asked Questions

How do I pass multiple inputs to a chain step?

Use a dictionary to pass multiple values: chain.invoke({"text": "...", "context": "..."}). Prompts access them as template variables.

Can I log outputs from each chain step?

Yes. Insert a RunnableLambda to log: chain | RunnableLambda(lambda x: print(f"Step output: {x}") or x). The or x ensures you return the value for the next step.

What's the difference between a chain and an agent?

Chains execute a fixed sequence of steps. Agents decide which tool to call based on the input and iterate. Chains are deterministic; agents are dynamic. Start with chains; use agents when you need autonomous decision-making.

How do I compose chains from an external configuration (YAML)?

Use LangChain Hub or manually build chains from dictionaries. For simplicity, define chains in Python modules and import them.

What happens if a chain step fails?

By default, the exception propagates. Use with_fallback() to define a backup chain: chain.with_fallback(fallback_chain).

Further Reading

• LCEL Specification
• LangChain Runnables Documentation
• Building Chains: Full Guide
• LangGraph: Stateful Workflows