🔗 LangChain LCEL Primitives: Your AI Building Blocks

The LEGO Analogy

Imagine you have a box of special LEGO blocks. Each block does one thing really well. When you connect them, you build amazing things!

LCEL Primitives are just like those LEGO blocks. Each one has a special job. Connect them with the pipe | symbol, and your data flows through like water in pipes!

🎯 What We’ll Learn

1. RunnablePassthrough - The “Pass It Along” block
2. RunnableLambda - The “Do Anything” block
3. RunnableAssign - The “Add More Stuff” block
4. RunnableParallel - The “Do Many Things At Once” block
5. itemgetter - The “Pick What You Need” block
6. Routing - The “Choose Your Path” block

1. RunnablePassthrough: The “Pass It Along” Block 📦

What Is It?

Think of a relay race. A runner takes the baton and passes it to the next runner without changing it.

RunnablePassthrough does exactly that. It takes your data and passes it along unchanged.

Why Use It?

Sometimes you need to keep the original data while also doing other things with it. It’s like making a photocopy before editing a document!

Simple Example

from langchain_core.runnables import (
    RunnablePassthrough
)

# Create a passthrough
passthrough = RunnablePassthrough()

# Give it data
result = passthrough.invoke("Hello!")

print(result)  # Output: "Hello!"

The data comes in. The same data goes out. Simple!

Real-World Use

You often use it with RunnableParallel to keep original data while transforming it:

chain = RunnableParallel(
    original=RunnablePassthrough(),
    uppercase=lambda x: x.upper()
)

result = chain.invoke("hello")
# {'original': 'hello', 'uppercase': 'HELLO'}

2. RunnableLambda: The “Do Anything” Block 🛠️

What Is It?

Imagine a magic box. You tell it what to do, and it does exactly that to anything you put inside.

RunnableLambda turns any Python function into a chainable block!

Why Use It?

You have a function? Wrap it in RunnableLambda. Now it works in your LCEL chain!

Simple Example

from langchain_core.runnables import (
    RunnableLambda
)

# A simple function
def add_exclamation(text):
    return text + "!"

# Wrap it
exciting = RunnableLambda(add_exclamation)

result = exciting.invoke("Hello")
print(result)  # Output: "Hello!"

Chain Multiple Functions

def double(x):
    return x * 2

def add_ten(x):
    return x + 10

chain = (
    RunnableLambda(double)
    | RunnableLambda(add_ten)
)

result = chain.invoke(5)
# 5 → double → 10 → add_ten → 20
print(result)  # Output: 20

3. RunnableAssign: The “Add More Stuff” Block ➕

What Is It?

Picture a backpack. You have books inside. RunnableAssign lets you add more items without removing what’s already there!

It takes a dictionary and adds new keys to it.

Why Use It?

When you want to enrich your data with new information while keeping the old.

Simple Example

from langchain_core.runnables import (
    RunnableAssign,
    RunnableLambda
)

# Add a greeting based on the name
add_greeting = RunnableAssign(
    greeting=RunnableLambda(
        lambda x: f"Hello, {x['name']}!"
    )
)

data = {"name": "Alice", "age": 25}
result = add_greeting.invoke(data)

print(result)
# {'name': 'Alice', 'age': 25,
#  'greeting': 'Hello, Alice!'}

The original name and age stay. A new greeting is added!

Using .assign() Shortcut

from langchain_core.runnables import (
    RunnablePassthrough
)

chain = RunnablePassthrough().assign(
    doubled=lambda x: x["number"] * 2
)

result = chain.invoke({"number": 5})
# {'number': 5, 'doubled': 10}

4. RunnableParallel: The “Do Many Things At Once” Block ⚡

What Is It?

Imagine you’re cooking. You can boil water, chop vegetables, and heat the pan all at the same time. Much faster than doing them one by one!

RunnableParallel runs multiple operations simultaneously.

Why Use It?

Speed! When tasks don’t depend on each other, run them together.

Simple Example

from langchain_core.runnables import (
    RunnableParallel,
    RunnableLambda
)

parallel = RunnableParallel(
    upper=RunnableLambda(lambda x: x.upper()),
    lower=RunnableLambda(lambda x: x.lower()),
    length=RunnableLambda(lambda x: len(x))
)

result = parallel.invoke("Hello")

print(result)
# {'upper': 'HELLO',
#  'lower': 'hello',
#  'length': 5}

Three operations. One input. All at once!

Dictionary Shorthand

# This also creates RunnableParallel!
parallel = {
    "upper": lambda x: x.upper(),
    "lower": lambda x: x.lower()
}

chain = parallel | RunnableLambda(
    lambda x: f"{x['upper']} and {x['lower']}"
)

result = chain.invoke("Hello")
# "HELLO and hello"

5. itemgetter with LCEL: The “Pick What You Need” Block 🎯

What Is It?

You have a box full of toys. itemgetter reaches in and grabs exactly the toy you want!

It extracts specific keys from a dictionary.

Why Use It?

When you only need certain pieces of your data.

Simple Example

from operator import itemgetter
from langchain_core.runnables import (
    RunnableParallel,
    RunnableLambda
)

# Get specific items
chain = (
    itemgetter("name")
    | RunnableLambda(lambda x: x.upper())
)

data = {"name": "alice", "age": 25}
result = chain.invoke(data)

print(result)  # Output: "ALICE"

Get Multiple Items

# Get multiple values
chain = RunnableParallel(
    name=itemgetter("name"),
    city=itemgetter("city")
)

data = {
    "name": "Bob",
    "age": 30,
    "city": "Paris"
}

result = chain.invoke(data)
# {'name': 'Bob', 'city': 'Paris'}

Real-World Use with Prompts

from langchain_core.prompts import (
    ChatPromptTemplate
)

prompt = ChatPromptTemplate.from_template(
    "Tell me about {topic} in {style} style"
)

chain = (
    {
        "topic": itemgetter("topic"),
        "style": itemgetter("style")
    }
    | prompt
)

result = chain.invoke({
    "topic": "cats",
    "style": "funny",
    "extra": "ignored"
})

6. Routing and Routers: The “Choose Your Path” Block 🚦

What Is It?

Imagine a train station. Different trains go to different cities. A router looks at your ticket and sends you to the right platform!

Routing sends data to different chains based on conditions.

Why Use It?

When different inputs need different processing.

Method 1: RunnableBranch

from langchain_core.runnables import (
    RunnableBranch,
    RunnableLambda
)

# Different handlers
def handle_math(x):
    return "Calculating..."

def handle_text(x):
    return "Processing text..."

def handle_default(x):
    return "General processing..."

router = RunnableBranch(
    # (condition, handler) pairs
    (
        lambda x: x["type"] == "math",
        RunnableLambda(handle_math)
    ),
    (
        lambda x: x["type"] == "text",
        RunnableLambda(handle_text)
    ),
    # Default (no condition)
    RunnableLambda(handle_default)
)

# Test it
result1 = router.invoke({"type": "math"})
# "Calculating..."

result2 = router.invoke({"type": "text"})
# "Processing text..."

result3 = router.invoke({"type": "other"})
# "General processing..."

Method 2: Custom Router Function

from langchain_core.runnables import (
    RunnableLambda
)

# Define different chains
chain_simple = RunnableLambda(
    lambda x: "Simple answer: " + x["question"]
)

chain_complex = RunnableLambda(
    lambda x: "Detailed answer: " + x["question"]
)

# Router function
def route(input):
    if input.get("complexity") == "simple":
        return chain_simple
    else:
        return chain_complex

# Use the router
full_chain = RunnableLambda(route)

result = full_chain.invoke({
    "question": "What is AI?",
    "complexity": "simple"
})
# "Simple answer: What is AI?"

Method 3: Dictionary-Based Routing

from langchain_core.runnables import (
    RunnableLambda
)

# Map of routes
routes = {
    "greeting": RunnableLambda(
        lambda x: "Hello! How can I help?"
    ),
    "farewell": RunnableLambda(
        lambda x: "Goodbye! See you soon!"
    ),
    "question": RunnableLambda(
        lambda x: "Let me think about that..."
    )
}

def router(input):
    route_key = input.get("intent", "question")
    return routes.get(
        route_key,
        routes["question"]
    ).invoke(input)

chain = RunnableLambda(router)

result = chain.invoke({"intent": "greeting"})
# "Hello! How can I help?"

🎨 Putting It All Together

Here’s how these blocks work together in a real chain:

graph TD
    A[Input Data] --> B[itemgetter]
    B --> C[RunnableParallel]
    C --> D1[RunnableLambda]
    C --> D2[RunnablePassthrough]
    D1 --> E[RunnableAssign]
    D2 --> E
    E --> F{Router}
    F --> G1[Chain A]
    F --> G2[Chain B]
    G1 --> H[Output]
    G2 --> H

Complete Example

from operator import itemgetter
from langchain_core.runnables import (
    RunnablePassthrough,
    RunnableLambda,
    RunnableParallel,
    RunnableBranch
)

# Build the chain
chain = (
    # Step 1: Extract and process in parallel
    RunnableParallel(
        name=itemgetter("name"),
        original=RunnablePassthrough()
    )
    # Step 2: Add computed value
    | RunnablePassthrough().assign(
        greeting=lambda x: f"Hi, {x['name']}!"
    )
    # Step 3: Route based on condition
    | RunnableBranch(
        (
            lambda x: x["original"].get("vip"),
            RunnableLambda(
                lambda x: x["greeting"] + " VIP!"
            )
        ),
        RunnableLambda(lambda x: x["greeting"])
    )
)

# Test
result = chain.invoke({
    "name": "Alice",
    "vip": True
})
print(result)  # "Hi, Alice! VIP!"

🧠 Quick Memory Guide

🎯 Key Takeaways

1. LCEL Primitives are building blocks for AI chains
2. Use | to connect them like pipes
3. Each primitive has one job and does it well
4. Combine them to build powerful, flexible chains
5. Routing lets you make smart decisions in your chain

You now have all the LEGO blocks. Go build something amazing! 🚀