Chains How to Connect Steps Together
Most useful AI tasks require more than one step. You might need to first prepare the input, then call the AI, then format the output, and then store the result. Doing this manually means writing repetitive glue code and tracking data as it flows between functions. LangChain Chains solve this by letting you connect components into a clean pipeline where the output of one step automatically becomes the input of the next.
The Assembly Line Analogy
A factory assembly line moves a product through a series of stations. At station one, raw metal gets cut to shape. At station two, pieces get welded together. At station three, the product gets painted. At station four, it gets inspected and packaged. Each station has a specific job and passes the result forward. LangChain Chains work the same way — each component does one job and passes its output to the next component in line.
Factory Assembly Line: Raw Metal → [Cut] → [Weld] → [Paint] → [Package] → Finished Product LangChain Chain: User Input → [Prompt Template] → [AI Model] → [Output Parser] → Formatted Response
You define the sequence once. After that, you call the chain with input and it handles every step automatically.
LangChain Expression Language (LCEL) Overview
Modern LangChain uses a syntax called LangChain Expression Language (LCEL). It uses the pipe operator (|) — the vertical bar character — to chain components together. Each component on the left feeds its output into the component on the right.
chain = component_one | component_two | component_three # Reading left to right: # Input flows into component_one # component_one's output flows into component_two # component_two's output flows into component_three # component_three's output is the final result
This syntax is clean, readable, and modular. Adding a step to the pipeline means inserting one more component into the pipe sequence.
Your First Simple Chain
The most common chain in LangChain combines a prompt template and a model. This pattern appears in nearly every LangChain application.
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
load_dotenv()
# Components
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", "You write clear, beginner-friendly definitions."),
("human", "Define the term: {term}")
])
parser = StrOutputParser()
# Chain them together
chain = prompt | model | parser
# Run the chain
result = chain.invoke({"term": "API"})
print(result)
# "An API (Application Programming Interface) is a set of rules that
# allows different software programs to communicate with each other..."
The chain flows: prompt fills the template → model generates a response → parser extracts the text string. The StrOutputParser strips the AIMessage wrapper and returns just the plain text string, which is usually what you want for display or further processing.
Diagram: Three-Component Chain
invoke({"term": "API"})
│
▼
┌───────────────────────────┐
│ ChatPromptTemplate │
│ Fills "term" variable │
│ → produces messages list │
└─────────────┬─────────────┘
│ [SystemMessage, HumanMessage]
▼
┌───────────────────────────┐
│ ChatOpenAI │
│ Sends messages to GPT │
│ → produces AIMessage │
└─────────────┬─────────────┘
│ AIMessage(content="An API is...")
▼
┌───────────────────────────┐
│ StrOutputParser │
│ Extracts .content │
│ → produces plain string │
└─────────────┬─────────────┘
│ "An API is..."
▼
Final Result
Adding Processing Steps with RunnableLambda
Sometimes you need to do something to the data that no built-in LangChain component handles — clean up text, format a number, look something up in a dictionary. RunnableLambda wraps any Python function and makes it a valid step in a chain.
from langchain_core.runnables import RunnableLambda
# A plain Python function that cleans text
def clean_text(text: str) -> str:
return text.strip().lower().replace(" ", " ")
# Wrap it so it works in a chain
cleaner = RunnableLambda(clean_text)
# Use it before the prompt
chain = cleaner | prompt | model | parser
# Now input gets cleaned before the prompt template processes it
result = chain.invoke({"term": " BLOCKCHAIN "})
# "term" gets cleaned to "blockchain" before the prompt sees it
You can insert RunnableLambda at any position in the chain — before the prompt to preprocess input, after the model to post-process output, or anywhere in between.
RunnablePassthrough: Passing Input Unchanged
Some chains need to pass the original input forward while also processing it. RunnablePassthrough passes its input directly to its output without modification. This is useful when you want to combine the original input with processed results further down the chain.
from langchain_core.runnables import RunnablePassthrough
# Example: pass "question" unchanged while also building context
chain = RunnablePassthrough() | prompt | model | parser
# The input dictionary passes through to the prompt unchanged
result = chain.invoke({"topic": "solar panels", "audience": "homeowners"})
Parallel Execution with RunnableParallel
Sometimes you want to run multiple operations on the same input at the same time and combine the results. RunnableParallel achieves this. Both branches run concurrently and produce outputs that you can combine downstream.
from langchain_core.runnables import RunnableParallel
# Define two separate chains
summary_chain = summary_prompt | model | parser
keywords_chain = keywords_prompt | model | parser
# Run both in parallel on the same input
parallel_chain = RunnableParallel(
summary=summary_chain,
keywords=keywords_chain
)
result = parallel_chain.invoke({"text": "Solar energy is a renewable..."})
print(result["summary"]) # Summary paragraph
print(result["keywords"]) # List of key terms
Running steps in parallel cuts wall-clock time significantly when each step involves waiting for an API response. Instead of waiting 2 seconds for step one and then 2 seconds for step two (4 seconds total), both run simultaneously and finish in roughly 2 seconds.
Diagram: Parallel Chain Execution
Input: {"text": "Solar energy..."}
│
├─────────────────┬────────────────────┐
│ │ │
▼ ▼ │
summary_prompt keywords_prompt │
│ │ │
▼ ▼ │
model model │
│ │ │
▼ ▼ │
parser parser │
│ │ │
└────────┬────────┘ │
│ Both finish ~same time │
▼ │
{"summary": "...", "keywords": "..."} │
Conditional Logic with RunnableBranch
Real applications often need to take different actions depending on the input. A customer support chatbot should route billing questions differently than technical questions. RunnableBranch implements this decision logic inside a chain.
from langchain_core.runnables import RunnableBranch
# Two different prompt templates for different question types
billing_chain = billing_prompt | model | parser
technical_chain = technical_prompt | model | parser
general_chain = general_prompt | model | parser
# Decision logic: classify based on input
branch = RunnableBranch(
(lambda x: "billing" in x["question"].lower(), billing_chain),
(lambda x: "error" in x["question"].lower(), technical_chain),
general_chain # default if no condition matches
)
result = branch.invoke({"question": "My billing statement is wrong"})
# Routes to billing_chain because "billing" is in the question
The branch checks each condition in order and uses the first chain whose condition returns True. If no condition matches, it falls through to the default chain.
Chaining Multiple AI Calls
Some tasks benefit from two AI calls where the output of the first becomes the input for the second. A common pattern: first ask the AI to create a plan, then ask it to execute the plan.
from langchain_core.runnables import RunnableLambda
# Step 1: Generate an outline
outline_prompt = ChatPromptTemplate.from_messages([
("system", "Create a three-point outline only. No extra text."),
("human", "Create an outline for an article about: {topic}")
])
# Step 2: Expand the outline into a full article
article_prompt = ChatPromptTemplate.from_messages([
("system", "You write clear, engaging articles."),
("human", "Write an article based on this outline:\n\n{outline}")
])
# Wrap the result so it has the right key for the next prompt
def reformat(outline: str) -> dict:
return {"outline": outline}
# Chain: generate outline, reformat output, generate article
full_chain = (
outline_prompt
| model
| parser
| RunnableLambda(reformat)
| article_prompt
| model
| parser
)
article = full_chain.invoke({"topic": "benefits of daily exercise"})
print(article)
The model runs twice: once to create the outline and once to write the article. Each call produces better results because the second call has a clear, structured input from the first.
Invoking Chains: Three Methods
invoke
Runs the chain once and returns the final result. Waits for the entire pipeline to complete before returning.
result = chain.invoke({"topic": "Python"})
batch
Runs the chain on a list of inputs. Processes all inputs and returns a list of results. LangChain handles parallelism automatically — multiple items can be processed simultaneously.
results = chain.batch([
{"topic": "Python"},
{"topic": "JavaScript"},
{"topic": "SQL"}
])
# Returns a list of three results
stream
Runs the chain and returns tokens one at a time as the model generates them. Use this when you want the user to see text appearing progressively rather than waiting for the full response.
for chunk in chain.stream({"topic": "Python"}):
print(chunk, end="", flush=True)
Error Handling in Chains
API calls fail sometimes — the service is down, you hit a rate limit, or the network drops. LangChain provides with_retry to automatically retry a chain step when it fails.
from langchain_core.runnables import RunnableRetry # Retry the model call up to 3 times if it fails resilient_model = model.with_retry(stop_after_attempt=3) chain = prompt | resilient_model | parser # Now if the API call fails, LangChain retries automatically
You can also use with_fallbacks to switch to a backup model or chain if the primary one fails.
# Try the expensive model first, fall back to the cheap one primary = ChatOpenAI(model="gpt-4o") fallback = ChatOpenAI(model="gpt-3.5-turbo") resilient_model = primary.with_fallbacks([fallback]) chain = prompt | resilient_model | parser
Debugging Chains
When a chain produces unexpected output, you need to see what each step receives and returns. Set the verbose flag on a component to print detailed information during execution.
import langchain
langchain.debug = True
# Now every chain invocation prints detailed step-by-step information:
# - Inputs to each component
# - Outputs from each component
# - Errors if any occur
result = chain.invoke({"term": "blockchain"})
The debug output shows exactly what data flows through each stage. This is the fastest way to find where a chain is going wrong.
Saving and Reusing Chains
Because chains are just Python objects, you create them once and reuse them across your application. Define your chains in a separate module and import them wherever needed.
# chains.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
model = ChatOpenAI(model="gpt-3.5-turbo")
parser = StrOutputParser()
definition_chain = (
ChatPromptTemplate.from_messages([
("system", "Give short, clear definitions."),
("human", "Define: {term}")
])
| model
| parser
)
summary_chain = (
ChatPromptTemplate.from_messages([
("system", "Summarize concisely."),
("human", "Summarize: {text}")
])
| model
| parser
)
# app.py
from chains import definition_chain, summary_chain
# Use them anywhere in your application
print(definition_chain.invoke({"term": "machine learning"}))
print(summary_chain.invoke({"text": long_article_text}))
Chain Composition Patterns Summary
Pattern Components Use Case ────────────────────────────────────────────────────────────── Simple Pipeline template | model Basic Q&A With Parsing template|model|parser Clean string output With Transform lambda | template Input preprocessing Parallel RunnableParallel Multiple outputs Conditional RunnableBranch Route by input type Sequential AI calls chain1 | chain2 Multi-step reasoning Batch .batch([]) Process many inputs Streaming .stream() Real-time output
Summary
LangChain Chains connect components into pipelines using the pipe (|) operator. StrOutputParser extracts plain text from model responses. RunnableLambda wraps any Python function as a chain step. RunnableParallel runs multiple chains simultaneously on the same input. RunnableBranch adds conditional logic. Chains support invoke, batch, and stream execution modes. Error handling uses with_retry and with_fallbacks.