Chains in LangChain

Chains represent the foundational design pattern from which LangChain derives its name. In modern Large Language Model (LLM) system architecture, an application is rarely a single call to an API; it is a composition of modular sub-tasks—prompt resolution, context enrichment, inference, structural validation, and conditional routing. Chains transform disjoint components into deterministic, maintainable, and declarative end-to-end execution pipelines.

1. The Need for Chains: From Manual Orchestration to Automated Pipelines

When engineering LLM applications without an orchestration framework, developers invariably encounter the burden of manual step execution. Even a basic workflow consists of distinct, dependent operations:

Prompt Formatting: Transforming raw user parameters into structured model instructions.
Model Invocation: Passing the formatted payload over the network to the model endpoint.
Output Parsing: Extracting raw string content or structured metadata from the model's message object.

The Naive (Manual) Approach

In an unchained implementation, every step requires explicit variable assignment, manual function triggering, and brittle intermediate state handling:

Structure & Concept Flow

User Input ("Cricket")

[Manual] prompt_template.invoke() --> Formatted PromptValue

[Manual] model.invoke() --> AIMessage Object

[Manual] parser.invoke() --> Clean Extracted String

PYTHON

# Naive, decoupled execution (Prone to boilerplate overhead)
formatted_prompt = prompt_template.invoke({"topic": "Cricket"})
raw_response = model.invoke(formatted_prompt)
final_output = parser.invoke(raw_response)

The Problem at Scale

While manageable for a three-step script, this imperative approach breaks down in real-world software engineering:

Error Propagation: Intermediate error handling must be duplicated across every call.
Observability Bottlenecks: Tracing timing, latency, token usage, and payload mutations across 15 discrete function invocations requires verbose boilerplate.
Compositional Rigidity: Adding asynchronous execution, streaming, fallbacks, or branching logic turns the codebase into deeply nested conditional structures.

Chains solve this by abstracting discrete tasks into a unified, executable pipeline where data flows predictably from one stage to the next.

2. Core Mental Model: The Assembly Line Pipeline

The clearest mental model for a chain is an industrial assembly line.

Real-Life Analogy

Think of a manufacturing plant. A conveyor belt transports raw materials through specialized workstations.

Station 1 (Prompt Template) stamps raw sheet metal into a standardized chassis.

Station 2 (The LLM Engine) installs the engine and core machinery.

Station 3 (The Output Parser) inspects the vehicle, strips away protective shipping wraps and diagnostic tags, and hands off a clean, showroom-ready product.

No worker leaves their station to carry parts across the warehouse. The conveyor belt mechanically links Station $N$ 's output directly to Station $N+1$ 's input.

In LangChain, this conveyor belt is defined using the LangChain Expression Language (LCEL) and executed via the pipe operator (|):

Structure & Concept Flow

Input Dict ---> [ Prompt ] ---> [ Model ] ---> [ Output Parser ] ---> Final Result

Each component on the pipeline implements a standard protocol known as a Runnable, guaranteeing that the output type of the left-hand operand matches the expected input type of the right-hand operand.

3. Building a Simple Linear Chain (LCEL Fundamentals)

To build a minimal linear chain, we combine three essential building blocks:

PromptTemplate: Prepares the text input.
ChatOpenAI: The compute engine (LLM).
StrOutputParser: Extracts the clean string payload from the raw AIMessage.

PYTHON

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Load environment variables (OPENAI_API_KEY)
load_dotenv()

# Step 1: Define the Prompt Template
prompt = PromptTemplate(
    template="Generate 5 interesting facts about {topic}.",
    input_variables=["topic"]
)

# Step 2: Initialize the Language Model
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# Step 3: Define the Output Parser
parser = StrOutputParser()

# Step 4: Construct the Chain via LCEL Pipe Syntax
chain = prompt | model | parser

# Step 5: Invoke the Chain with a Single Call
result = chain.invoke({"topic": "cricket"})
print(result)

Visualizing the Chain Execution Graph

LangChain provides built-in introspection tools to inspect the underlying Directed Acyclic Graph (DAG) of any initialized chain:

PYTHON

# Visualize the execution pipeline directly in the console
chain.get_graph().print_ascii()

Structure & Concept Flow

+-------------+

| PromptInput |

+-------------+

+----------------+

| PromptTemplate |

+----------------+

+------------+

| ChatOpenAI |

+------------+

+-----------------+

| StrOutputParser |

+-----------------+

+--------------+

| OutputParser |

+--------------+

When chain.invoke({"topic": "cricket"}) is executed:

The dictionary enters PromptTemplate, which substitutes {topic} and outputs a PromptValue.
The PromptValue automatically streams into ChatOpenAI, which triggers the HTTP request and returns an AIMessage.
The AIMessage flows into StrOutputParser, which discards response metadata (tokens, finish reasons) and extracts the pure string response.

4. Multi-Step Sequential Chains: Feeding LLM Outputs Back into LLMs

A single LLM pass is often insufficient for complex analytical tasks. High-quality outputs often require multi-stage refinement: generating a raw long-form draft first, followed by summarization, critique, or extraction.

Architecture: Two-Stage Sequential Processing

Structure & Concept Flow

User Input: {"topic": "Unemployment in India"}

[ Prompt 1: Generate Detailed Report ]

[ Model 1: ChatOpenAI ]

[ Parser 1: StrOutputParser ]

+---> Raw Detailed Report (Intermediate String Output)

[ Prompt 2: Extract 5-Point Summary ]

[ Model 2: ChatOpenAI ]

[ Parser 2: StrOutputParser ]

Final Output: 5-Pointer Executive Summary

Implementation

PYTHON

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()

# Initialize core primitives
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)
parser = StrOutputParser()

# Stage 1: Generate a comprehensive analysis
prompt1 = PromptTemplate(
    template="Generate a detailed report on {topic}.",
    input_variables=["topic"]
)

# Stage 2: Distill the analysis into key takeaways
prompt2 = PromptTemplate(
    template="Generate a 5-pointer summary from the following text:\n{text}",
    input_variables=["text"]
)

# Constructing the composite linear chain
sequential_chain = (
    prompt1 
    | model 
    | parser 
    | prompt2 
    | model 
    | parser
)

# Execution
summary = sequential_chain.invoke({"topic": "Unemployment in India"})
print(summary)

Note on Contract Matching: For prompt2 to consume the output of the preceding stage directly, prompt2 must accept a single variable if passed a raw string, or the intermediate parser output must align with the input contract of the subsequent step.

5. Parallel Processing with `RunnableParallel`

Linear chains execute sequentially with latency equal to the sum of each step:

T_{\text{total}} = T_{\text{stage 1}} + T_{\text{stage 2}} + \dots + T_{\text{stage } n}

However, independent tasks that share a common context can run simultaneously. For instance, given a complex technical document, an application may need to generate concise study notes and a self-assessment quiz at the same time.

Using RunnableParallel, these operations execute concurrently:

T_{\text{parallel}} \approx \max(T_{\text{notes}}, T_{\text{quiz}}) + T_{\text{merge}}

Structure & Concept Flow

Input Context (Source Document)

+---------------------+---------------------+

| |

v v

[ Branch A: Notes ] [ Branch B: Quiz ]

Prompt -> GPT-4o -> Parser Prompt -> Claude-3 -> Parser

| |

+---------------------+---------------------+

Dictionary: {"notes": ..., "quiz": ...}

[ Stage 3: Merge Chain ]

Prompt -> LLM -> Parser

Consolidated Report

Implementation

PYTHON

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel

load_dotenv()

# Instantiate diverse models for diverse sub-tasks
openai_model = ChatOpenAI(model="gpt-4o-mini")
claude_model = ChatAnthropic(model="claude-3-haiku-20240307")
parser = StrOutputParser()

# Prompts for parallel execution
notes_prompt = PromptTemplate(
    template="Generate short and simple notes from the following text:\n{context}",
    input_variables=["context"]
)

quiz_prompt = PromptTemplate(
    template="Generate 5 short question-answers from the following text:\n{context}",
    input_variables=["context"]
)

# Prompt for the synthesis step
merge_prompt = PromptTemplate(
    template=(
        "Merge the provided notes and quiz into a single structured document:\n\n"
        "NOTES:\n{notes}\n\n"
        "QUIZ:\n{quiz}"
    ),
    input_variables=["notes", "quiz"]
)

# 1. Define the Parallel Branch
parallel_branches = RunnableParallel({
    "notes": notes_prompt | openai_model | parser,
    "quiz": quiz_prompt | claude_model | parser
})

# 2. Define the Final Synthesis Chain
merge_chain = merge_prompt | openai_model | parser

# 3. Connect Parallel Execution to the Synthesis Stage
full_pipeline = parallel_branches | merge_chain

# Execution
source_text = """Support Vector Machines (SVMs) are supervised learning models with 
associated learning algorithms that analyze data for classification and regression analysis. 
The objective of SVM is to find a hyperplane in an N-dimensional space that distinctly 
classifies the data points while maximizing the margin between classes."""

result = full_pipeline.invoke({"context": source_text})
print(result)

6. Architecting Conditional Logic: Branching Chains

Complex applications cannot rely exclusively on static execution paths. Production systems must evaluate state at runtime and execute specific sub-routines conditionally.

The Real-World Scenario: Customer Feedback Triage

Consider an automated customer review system for an e-commerce platform:

If customer feedback is positive, route to an appreciative response generator (or prompt for a 5-star public rating).
If customer feedback is negative, route to a remediation response generator (or trigger an incident report to customer support).

Structure & Concept Flow

User Review Text

[ Classification Step ]

Sentiment Output: {"positive" | "negative"}

+--- Is Positive? ---> [ Appreciation Chain ] ---> Output

+--- Is Negative? ---> [ Remediation Chain ] ---> Output

+--- Fallback ---> [ Default Handler ] ---> Output

To implement this reliably, we must first solve a critical engineering challenge: ensuring non-deterministic LLMs return clean, reliable control flags.

7. Deterministic Conditioning via Structured Outputs (`PydanticOutputParser`)

A common mistake when building conditional chains is relying on raw string outputs from an LLM classifier:

PYTHON

# UNRELIABLE: The model might output "Positive", "positive.", 
# "The sentiment is positive", or "Negative (score: 0.8)"
if classification_result == "positive":
    ...

If the model introduces punctuation, conversational fillers, or capitalization differences, downstream string matches will fail silently, triggering unintended fallbacks.

The Solution: Schema Enforcement via Pydantic

We force the classifier model to conform to an explicit schema using PydanticOutputParser and Literal constraints:

PYTHON

from typing import Literal
from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# Step 1: Define the strict domain model
class FeedbackClassification(BaseModel):
    sentiment: Literal["positive", "negative"] = Field(
        description="The classified sentiment of the user review. Must be exactly 'positive' or 'negative'."
    )

# Step 2: Initialize parser and extract formatting instructions
pydantic_parser = PydanticOutputParser(pydantic_object=FeedbackClassification)

# Step 3: Embed schema constraints into the classification prompt
classifier_prompt = PromptTemplate(
    template=(
        "Classify the sentiment of the following feedback text.\n"
        "{format_instructions}\n"
        "Feedback: {feedback}\n"
    ),
    input_variables=["feedback"],
    partial_variables={"format_instructions": pydantic_parser.get_format_instructions()}
)

# Step 4: Assemble the Deterministic Classifier Chain
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
classifier_chain = classifier_prompt | model | pydantic_parser

Now, classifier_chain.invoke({"feedback": "..."}) is guaranteed to return an instance of FeedbackClassification where result.sentiment evaluates strictly to "positive" or "negative".

8. Implementing Routing with `RunnableBranch` and `RunnableLambda`

With deterministic output guaranteed, we build the routing logic using RunnableBranch.

RunnableBranch acts as the LangChain equivalent of a traditional switch-case or if/elif/else control structure:

\text{RunnableBranch}((Condition_1, Chain_1), (Condition_2, Chain_2), \dots, DefaultChain)

Handling Custom Logic with `RunnableLambda`

Every terminal branch must adhere to the Runnable interface. If a fallback branch uses a standard Python function or lambda rather than a chain, it must be wrapped using RunnableLambda.

Complete Branching Implementation

PYTHON

from dotenv import load_dotenv
from typing import Literal
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser, PydanticOutputParser
from langchain_core.runnables import RunnableBranch, RunnableLambda

load_dotenv()

# Models and standard text parser
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
str_parser = StrOutputParser()

# -------------------------------------------------------------
# 1. CLASSIFIER SPECIFICATION
# -------------------------------------------------------------
class FeedbackClassification(BaseModel):
    sentiment: Literal["positive", "negative"] = Field(
        description="Sentiment must be either 'positive' or 'negative'"
    )

classification_parser = PydanticOutputParser(pydantic_object=FeedbackClassification)

classifier_prompt = PromptTemplate(
    template="Classify the sentiment of this feedback text.\n{format_instructions}\nFeedback: {feedback}",
    input_variables=["feedback"],
    partial_variables={"format_instructions": classification_parser.get_format_instructions()}
)

classifier_chain = classifier_prompt | model | classification_parser

# -------------------------------------------------------------
# 2. SPECIALIZED RESPONSE BRANCHES
# -------------------------------------------------------------
positive_prompt = PromptTemplate(
    template="Write a warm, enthusiastic thank-you response to this positive customer feedback:\n{feedback}",
    input_variables=["feedback"]
)
positive_chain = positive_prompt | model | str_parser

negative_prompt = PromptTemplate(
    template="Write an empathetic, professional apology and escalation response to this negative feedback:\n{feedback}",
    input_variables=["feedback"]
)
negative_chain = negative_prompt | model | str_parser

# -------------------------------------------------------------
# 3. CONDITIONAL ROUTING LOGIC
# -------------------------------------------------------------
# The input 'x' passing into the branches contains both:
# 1. The parsed classification object: x["classification"]
# 2. The original raw user feedback:  x["feedback"]

branch_routing = RunnableBranch(
    # Condition 1: Positive sentiment branch
    (
        lambda x: x["classification"].sentiment == "positive",
        (lambda x: {"feedback": x["feedback"]}) | positive_chain
    ),
    # Condition 2: Negative sentiment branch
    (
        lambda x: x["classification"].sentiment == "negative",
        (lambda x: {"feedback": x["feedback"]}) | negative_chain
    ),
    # Fallback branch: Executed if no conditions match
    RunnableLambda(lambda x: "Sentiment could not be verified for automated resolution.")
)

# -------------------------------------------------------------
# 4. COMPOSITE END-TO-END PIPELINE
# -------------------------------------------------------------
# We preserve the original feedback alongside the classification result
evaluation_pipeline = (
    RunnableLambda(lambda user_input: {
        "classification": classifier_chain.invoke(user_input),
        "feedback": user_input["feedback"]
    })
    | branch_routing
)

# -------------------------------------------------------------
# 5. EXECUTION
# -------------------------------------------------------------
negative_review = {"feedback": "This smartphone runs extremely hot and the screen stopped working after two days."}
print("Negative Feedback Output:")
print(evaluation_pipeline.invoke(negative_review))

print("\n" + "="*50 + "\n")

positive_review = {"feedback": "Absolutely phenomenal battery life and the camera quality exceeded my expectations!"}
print("Positive Feedback Output:")
print(evaluation_pipeline.invoke(positive_review))

9. Comparative Architecture: Sequential vs. Parallel vs. Conditional Chains

Architectural Pattern	Core Primitives	Control Flow Behavior	Latency Profile	Primary Use Cases
Linear / Sequential	`	` (LCEL Operator)	Strictly sequential ( $1 \to 1 \to 1$ ). Station $N$ 's output becomes Station $N+1$ 's input.	Additive: $\sum_{i=1}^{n} T_i$
Parallel	`RunnableParallel`	Concurrent split-and-merge ( $1 \to N \to 1$ ). A single input feeds multiple pipelines simultaneously; results merge into a dictionary.	Bounded by slowest branch: $\max(T_1, \dots, T_n) + T_{\text{merge}}$	Multi-perspective document analysis, extracting notes and quizzes concurrently, multi-model evaluation.
Conditional (Branching)	`RunnableBranch`, `RunnableLambda`	Dynamic selective execution ( $1 \to [1 \text{ of } N]$ ). Evaluates runtime state and executes one discrete path.	Single active path: $T_{\text{classifier}} + T_{\text{branch}}$	Intent triage, sentiment-based escalation, guardrail routing, dynamic language translation.

10. Summary and Key Takeaways

Chains as Modular Pipelines: Chains elevate LLM software development from ad-hoc, imperative API calls to structured, declaratively composed Directed Acyclic Graphs (DAGs).
The LCEL Paradigm: LangChain Expression Language leverages the pipe operator (|) to cleanly connect Prompts, Models, Parsers, and Runnables with minimal boilerplate.
Type-Safe Compositions: Every component in an LCEL chain conforms to the standard Runnable protocol, automatically managing state forwarding and interface matching.
Inspectability: Any chain constructed via LCEL can be dynamically inspected and audited using built-in graph visualization tools like chain.get_graph().print_ascii().
Latency Optimization via Concurrency: Using RunnableParallel, computationally independent tasks can run simultaneously, drastically reducing end-to-end latency compared to sequential executions.
Structural Guardrails for Routing: Always place a schema-enforcing parser (such as PydanticOutputParser with Literal types) before a RunnableBranch to prevent unpredictable, non-deterministic branching errors.
Fallback Encapsulation: Dynamic conditions within RunnableBranch require valid Runnable structures across every pathway; custom functional fallbacks should be encapsulated via RunnableLambda.

Chains in LangChain | Generative AI using LangChain | Video 7 | CampusX

Chains in LangChain

1. The Need for Chains: From Manual Orchestration to Automated Pipelines

The Naive (Manual) Approach

The Problem at Scale

2. Core Mental Model: The Assembly Line Pipeline

Real-Life Analogy

3. Building a Simple Linear Chain (LCEL Fundamentals)

Visualizing the Chain Execution Graph

4. Multi-Step Sequential Chains: Feeding LLM Outputs Back into LLMs

Architecture: Two-Stage Sequential Processing

Implementation

5. Parallel Processing with RunnableParallel

Implementation

6. Architecting Conditional Logic: Branching Chains

The Real-World Scenario: Customer Feedback Triage

7. Deterministic Conditioning via Structured Outputs (PydanticOutputParser)

The Solution: Schema Enforcement via Pydantic

8. Implementing Routing with RunnableBranch and RunnableLambda

Handling Custom Logic with RunnableLambda

Complete Branching Implementation

9. Comparative Architecture: Sequential vs. Parallel vs. Conditional Chains

10. Summary and Key Takeaways

5. Parallel Processing with `RunnableParallel`

7. Deterministic Conditioning via Structured Outputs (`PydanticOutputParser`)

8. Implementing Routing with `RunnableBranch` and `RunnableLambda`

Handling Custom Logic with `RunnableLambda`