Phase 2Intermediate⏱ 60 minutes

Building
Chains with LangChain

This intermediate-level Python tutorial teaches you how to create powerful sequential workflows by chaining operations together in LangChain. Master different chain types and build complex AI pipelines with step-by-step examples and best practices.

🎯

Learning Objectives

Understand the concept of chains and why they're powerful
Build simple chains using the pipe operator (|)
Create sequential chains that pass data between steps
Implement error handling and debugging for chains

📚 Prerequisites for This Intermediate Python Tutorial

This tutorial assumes you have:

• Completed Lessons 5-6 (LangChain Setup & Prompt Templates)
• Intermediate Python programming knowledge
• Understanding of LangChain prompt templates
• Familiarity with async/await concepts (helpful but not required)

🔗

What are Chains?

Chains in LangChain are sequences of components that process data step by step. Think of them as assembly lines where each step transforms the data before passing it to the next.

🔑 Key Benefits:

• Modularity: Break complex tasks into simple, reusable steps
• Flexibility: Easily modify or extend workflows
• Clarity: Clear data flow makes debugging easier
• Reusability: Use the same chain with different inputs

Chain Flow Visualization

Input → [Prompt Template] → [LLM] → [Output Parser] → Result
           ↓                   ↓              ↓
      Format input      Generate text    Structure output

Simple Chains with the Pipe Operator

Your First Chain

The pipe operator (|) is the modern way to create chains in LangChain:

🔍 Understanding the Flow:

1. Prompt Template: Takes your variables ({adjective} and {topic}) and creates a complete prompt
2. Model (LLM): Receives the formatted prompt and generates a response
3. Output Parser: Extracts the text content from the model's response object
4. Result: You get a clean string instead of a complex response object

from langchain.prompts import ChatPromptTemplate
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.output_parsers import StrOutputParser
import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Define components
prompt = ChatPromptTemplate.from_template(
    "Tell me a {adjective} joke about {topic}"
)
model = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    google_api_key=os.getenv("GOOGLE_API_KEY")
)
output_parser = StrOutputParser()

# Create chain using pipe operator
chain = prompt | model | output_parser

# Run the chain
result = chain.invoke({
    "adjective": "funny",
    "topic": "programming"
})

print(result)

Note: The pipe operator (|) creates a chain where output from each component flows into the next. This is more intuitive than the older LLMChain approach.

✨ Why Use Chains?

• Modularity: Each component has a single responsibility
• Reusability: Components can be reused in different chains
• Readability: The flow is clear and easy to understand
• Flexibility: Easy to add, remove, or replace components

Multi-Step Processing

Chains can include multiple processing steps:

🔄 Chain Flow Breakdown:

1. idea_prompt | model: Generate 5 company names as text
2. | list_parser: Parse the text into a Python list ["Name1", "Name2", ...]
3. | RunnableLambda: Transform list to dictionary {"names": "Name1, Name2, ..."}
4. | evaluate_prompt: Create evaluation prompt with the names
5. | model: Get ratings and analysis from the LLM

from langchain.prompts import ChatPromptTemplate
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.output_parsers import CommaSeparatedListOutputParser
from langchain.schema.runnable import RunnableLambda
import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Step 1: Generate ideas
idea_prompt = ChatPromptTemplate.from_template(
    "Generate 5 creative names for a {product_type} company"
)

# Step 2: Evaluate names
evaluate_prompt = ChatPromptTemplate.from_template(
    "Rate these company names from 1-10 for memorability: {names}"
)

# Components
model = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    google_api_key=os.getenv("GOOGLE_API_KEY")
)
list_parser = CommaSeparatedListOutputParser()

# Create multi-step chain
chain = (
    idea_prompt 
    | model 
    | list_parser 
    | RunnableLambda(lambda names: {"names": ", ".join(names)})
    | evaluate_prompt 
    | model
)

# Run the chain
result = chain.invoke({"product_type": "AI software"})
print(result.content)

💡 Key Concepts:

• Data Transformation: RunnableLambda lets you transform data between chain steps
• Parser Integration: Output parsers ensure data is in the right format
• Sequential Processing: Each step depends on the output of the previous one
• Flexible Design: Easy to add, remove, or modify steps in the pipeline

Pro Tip: When building multi-step chains, test each step individually first. This makes debugging much easier when something goes wrong!

📊

Sequential Chains for Complex Workflows

Building a Content Creation Pipeline

Let's build a complete content creation pipeline that generates, improves, and formats content:

🔍 Understanding the Pipeline:

• Step 1: Generate initial article content using the topic, focus areas, and audience
• Step 2: Improve the article by making it more engaging with examples
• Step 3: Generate metadata (title, summary, takeaways) in parallel with finalizing content
• RunnableParallel: Executes multiple operations simultaneously for efficiency

from langchain.prompts import ChatPromptTemplate
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.schema.runnable import RunnablePassthrough, RunnableParallel
import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Initialize model
model = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    temperature=0.7,
    google_api_key=os.getenv("GOOGLE_API_KEY")
)

# Chain 1: Generate initial content
generate_prompt = ChatPromptTemplate.from_template("""
Write a brief article about {topic}.
Focus on: {focus_areas}
Target audience: {audience}
""")

# Chain 2: Improve the content
improve_prompt = ChatPromptTemplate.from_template("""
Improve this article by making it more engaging and adding specific examples:

{article}

Keep the same topic and audience in mind.
""")

# Chain 3: Create title and summary
metadata_prompt = ChatPromptTemplate.from_template("""
For this article:
{article}

Generate:
1. A catchy title
2. A 2-sentence summary
3. 3 key takeaways

Format as JSON.
""")

# Build the complete pipeline
pipeline = (
    # Step 1: Generate initial content
    RunnablePassthrough.assign(
        article=generate_prompt | model | (lambda x: x.content)
    )
    # Step 2: Improve the content  
    | RunnablePassthrough.assign(
        improved_article=lambda x: model.invoke(improve_prompt.format(article=x["article"]))
    )
    # Step 3: Generate metadata in parallel with final content
    | RunnableParallel(
        final_article=lambda x: x["improved_article"].content,
        metadata=lambda x: model.invoke(metadata_prompt.format(article=x["improved_article"].content))
    )
)

# Run the pipeline
result = pipeline.invoke({
    "topic": "The Future of Remote Work",
    "focus_areas": "productivity, work-life balance, technology",
    "audience": "professionals considering remote work"
})

print("Article:", result["final_article"])
print("\nMetadata:", result["metadata"].content)

⚠️ Common Pitfall:

When using lambda functions in chains, prompt.format() returns a string, not a Runnable. You need to explicitly call model.invoke() instead of using the pipe operator.

✨ Benefits of This Pattern:

• Iterative Improvement: Content gets refined through multiple passes
• Parallel Processing: Metadata generation doesn't block content finalization
• Separation of Concerns: Each step has a clear, focused purpose
• Scalability: Easy to add more processing steps as needed

🎨

Practical Chain Patterns

1. Transformation Chain

Transform data through multiple formats:

🔄 Understanding Transformation Chains:

Transformation chains process data through multiple stages, converting it from one format to another. This pattern is especially useful for complex workflows where each stage builds on the previous one.

• User Story → Technical Spec: Convert high-level requirements to detailed specs
• Spec → Code: Generate implementation based on specifications
• Benefits: Clear separation of concerns, easier debugging, reusable components

from langchain.prompts import ChatPromptTemplate
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.schema.runnable import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Initialize model
model = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    google_api_key=os.getenv("GOOGLE_API_KEY")
)

# Transform user story → technical spec → code
user_story_to_spec = ChatPromptTemplate.from_template("""
Convert this user story into technical specifications:
{user_story}

Include: endpoints, data models, and validation rules.
""")

spec_to_code = ChatPromptTemplate.from_template("""
Generate Python Flask code based on these specifications:
{specifications}

Include proper error handling and documentation.
""")

# Create transformation chain
transform_chain = (
    RunnablePassthrough.assign(
        specifications=user_story_to_spec | model | StrOutputParser()
    )
    | RunnablePassthrough.assign(
        code=lambda x: model.invoke(spec_to_code.format(specifications=x["specifications"]))
    )
)

result = transform_chain.invoke({
    "user_story": "As a user, I want to create and manage todo items with due dates"
})

# Display results
print("Original User Story:")
print(result["user_story"])
print("\n" + "="*50 + "\n")
print("Technical Specifications:")
print(result["specifications"])
print("\n" + "="*50 + "\n")
print("Generated Code:")
print(result["code"].content)

💡 What This Demonstrates:

• Progressive Refinement: Each stage builds on the previous output
• Clear Data Flow: You can see exactly what data is passed between stages
• Debugging-Friendly: Easy to inspect intermediate results
• Reusable Components: Each prompt template can be used in other chains

Pro Tip: When building transformation chains, always test each stage individually first. This helps ensure each transformation produces the expected output format for the next stage.

2. Validation Chain

Generate content and validate it meets requirements:

🔍 Understanding Validation Chains:

Validation chains help ensure generated content meets specific criteria. This pattern separates generation from validation, making it easier to debug and modify each step independently.

• Step 1: Generate initial content based on requirements
• Step 2: Pass generated content to a validator
• Step 3: Get feedback on whether content meets criteria
• Optional: Regenerate if validation fails

from langchain.prompts import ChatPromptTemplate
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.schema.runnable import RunnableLambda
from langchain_core.output_parsers import StrOutputParser
import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Initialize model
model = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    google_api_key=os.getenv("GOOGLE_API_KEY")
)

# Generate and validate email
generate_email = ChatPromptTemplate.from_template("""
Write a professional email for: {purpose}
Tone: {tone}
Key points: {key_points}
""")

validate_email = ChatPromptTemplate.from_template("""
Review this email and check if it:
1. Maintains a {tone} tone
2. Includes all key points: {key_points}
3. Is under 150 words
4. Has proper greeting and closing

Email:
{email}

Respond with: "APPROVED" or "NEEDS_REVISION: [reasons]"
""")

# Validation chain with retry logic
def create_email_chain():
    return (
        generate_email 
        | model 
        | StrOutputParser()
        | RunnableLambda(lambda email: {
            "email": email
        })
    )

# Create validation chain with fixed parameters
email_chain = create_email_chain()

# Run email generation
email_params = {
    "purpose": "Schedule a project review meeting",
    "tone": "friendly but professional", 
    "key_points": "Next Thursday 2pm, Conference Room A, Bring status reports"
}
result = email_chain.invoke(email_params)

# Run validation separately
validation_prompt = validate_email.format(
    email=result["email"],
    tone=email_params["tone"],  # Use original parameters
    key_points=email_params["key_points"]  # Use original parameters
)
validation_result = model.invoke(validation_prompt).content

print(f"Email:\n{result['email']}")
print(f"\nValidation: {validation_result}")

💡 Expected Output:

The validation result should return either "APPROVED" or "NEEDS_REVISION: [reasons]" based on whether the generated email meets all the specified criteria (tone, key points inclusion, length, and format).

Note: This example shows a simple validation pattern. In production, you might want to implement automatic retry logic if validation fails, or use more sophisticated validation criteria.

🛡️

Error Handling and Debugging

Robust Chain Implementation

Build chains with proper error handling and debugging capabilities:

🛡️ Key Concepts:

• Error Wrapper: Catch and log errors at each chain step
• Logging: Track execution flow and identify bottlenecks
• Callbacks: Monitor intermediate outputs for debugging
• Graceful Failure: Return meaningful error messages instead of crashing

from langchain.callbacks import StdOutCallbackHandler
from langchain.schema.runnable import RunnableLambda
from langchain.prompts import ChatPromptTemplate
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.output_parsers import StrOutputParser
import logging
import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize components
model = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    google_api_key=os.getenv("GOOGLE_API_KEY")
)
generate_prompt = ChatPromptTemplate.from_template("Write a short essay about {topic}")
output_parser = StrOutputParser()

# Error handling wrapper
def safe_chain_step(func, step_name):
    def wrapper(*args, **kwargs):
        try:
            logger.info(f"Executing: {step_name}")
            result = func(*args, **kwargs)
            logger.info(f"Completed: {step_name}")
            return result
        except Exception as e:
            logger.error(f"Error in {step_name}: {str(e)}")
            return f"Error in {step_name}: {str(e)}"
    return wrapper

# Create chain with error handling
robust_chain = (
    RunnableLambda(safe_chain_step(
        lambda x: generate_prompt.format(**x), 
        "Format Prompt"
    ))
    | RunnableLambda(safe_chain_step(
        lambda x: model.invoke(x), 
        "LLM Generation"
    ))
    | RunnableLambda(safe_chain_step(
        lambda x: output_parser.parse(x.content), 
        "Parse Output"
    ))
)

# Add callbacks for debugging
callbacks = [StdOutCallbackHandler()]

# Run with debugging
result = robust_chain.invoke(
    {"topic": "AI Safety"},
    config={"callbacks": callbacks}
)

print(f"\nFinal Result:\n{result}")

✨ What This Shows:

• Step-by-Step Logging: Each chain step logs its execution status
• Error Isolation: Errors are caught and reported with specific step names
• Callback Integration: StdOutCallbackHandler shows detailed execution info
• Production-Ready: This pattern scales well to complex chains

🔍 Debugging Tips:

• Use callbacks to see intermediate outputs
• Add logging at each chain step
• Test each component individually first
• Use try-except blocks for graceful failures

Pro Tip: In production, replace print statements with proper logging to a file or monitoring service. Consider using structured logging (JSON format) for easier parsing and analysis.

✨ Chain Best Practices

Design Principles

• Keep chains simple and focused
• Use meaningful variable names
• Document chain purpose and flow
• Test with various inputs

Performance Tips

• Cache repeated LLM calls
• Use streaming for long outputs
• Parallelize independent steps
• Monitor token usage

📚Quick Reference: Chain Types

Simple Chain

prompt | llm | parser

Sequential Chain

step1 | step2 | step3

Parallel Chain

RunnableParallel(a=chain1, b=chain2)

Conditional Chain

RunnableLambda with if/else logic

🎉 Next Step: Adding Memory to Your Chains

Excellent work! You've mastered LangChain chains - the foundation for building complex AI workflows in Python. You can now create sequential operations, combine multiple steps, and build sophisticated AI pipelines.

Ready to make your chains smarter? In the next lesson, you'll learn about Memory and Context Management to create AI applications that remember previous interactions and maintain conversation state.

Previous: Prompt Templates Next: Memory & Context