Building Reactive Chains with LCEL

This lesson on Building Reactive Chains with LCEL is hands-on and example-driven. You will construct modular LLM pipelines using LangChain Expression Language (LCEL) and understand its execution model under the hood. You will implement standard LCEL chains, overload Python operators to build custom runnable engines, and orchestrate branching workflows using core primitives.

What You'll Be Able To Do

• Assemble sequential prompt-to-model-to-parser pipelines using LCEL pipe syntax.
• Implement custom Runnable and RunnableSequence classes via Python bitwise OR operator overloading.
• Wrap standard Python transformation functions using RunnableLambda to integrate with LCEL.
• Preserve raw pipeline inputs across intermediate nodes using RunnablePassthrough.
• Construct multi-branch data transformation pipelines using RunnableParallel.

Detailed Concept Walkthrough

1. LCEL Syntax and the Runnable Interface

LCEL standardizes pipeline creation by connecting modular components into a sequence using the pipe operator. Every component implements a uniform execution contract centered on the invoke method.

• Mechanism: Components such as ChatPromptTemplate, ChatOpenAI, and StrOutputParser are composed sequentially with the pipe operator to form an executable chain. Calling invoke on the chain triggers downstream processing where output from one step serves as input to the next.
• Under the Hood: Intermediate data undergoes continuous type conversion across the pipeline. A raw dictionary converts to HumanMessage objects via the prompt template, then to an AIMessage via the chat model, and finally into a plain string via the output parser.
• Best Practice: Standardize pipeline interfaces so each component consumes the exact type produced by the preceding stage. Test individual components by invoking them independently before piping them together.

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

# Compose LCEL sequential chain
prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

chain = prompt | model | parser
result = chain.invoke({"topic": "bears"})

Key Takeaway: LCEL chains enforce a uniform contract where intermediate types flow sequentially from dictionaries to messages to strings.

2. Custom LCEL Engine via Operator Overloading

LCEL leverages Python operator overloading to redefine the bitwise OR symbol into a sequence constructor instead of a logical evaluation.

• Mechanism: Custom runnable classes override Python's native or dunder method. Instead of computing bitwise operations, or returns an instance of a sequence class that holds references to both steps in order.
• Execution Flow: When invoke is called on the resulting sequence object, it executes the first runnable's process logic and immediately passes the output into the next runnable.
• Syntax Rule: Subclassed runnables implement a unified invoke method that delegates execution to internal transformation methods, ensuring full compatibility across nested compositions.

class CRunnable:
    def __or__(self, other):
        return CRunnableSequence(self, other)
    def invoke(self, input_data):
        return self.process(input_data)

class CRunnableSequence(CRunnable):
    def __init__(self, first, second):
        self.first = first
        self.second = second
    def process(self, input_data):
        return self.second.invoke(self.first.invoke(input_data))

class AddTen(CRunnable):
    def process(self, x): return x + 10

class MultiplyTwo(CRunnable):
    def process(self, x): return x * 2

custom_chain = AddTen() | MultiplyTwo()
print(custom_chain.invoke(5))  # Output: 30

Key Takeaway: The pipe operator constructs execution trees by binding runnables into sequence objects via or overloading.

3. Adapting Functions with RunnableLambda

Standard Python functions cannot be piped directly into LCEL chains because they lack the Runnable interface and invoke method.

• Mechanism: RunnableLambda wraps any arbitrary Python function or lambda expression, giving it an invoke method compatible with LCEL sequences.
• Under the Hood: When the parent chain executes, the wrapped function receives the piped intermediate payload as its primary argument and passes its return value to the next stage.
• Best Practice: Use RunnableLambda for data sanitization, formatting, or intermediate parsing steps without needing to subclass custom classes.

from langchain_core.runnables import RunnableLambda

def to_uppercase(text: str) -> str:
    return text.upper()

upper_step = RunnableLambda(to_uppercase)
# Wraps arbitrary function into a runnable interface
chain = upper_step | RunnableLambda(lambda s: f"Result: {s}")
print(chain.invoke("hello world"))  # Output: 'Result: HELLO WORLD'

Key Takeaway: Wrap standard Python functions in RunnableLambda to expose the invoke method required for LCEL composition.

4. Data Routing with Passthrough and Parallel

RunnablePassthrough and RunnableParallel enable complex data flows by preserving original inputs and executing multi-branch operations concurrently.

• Mechanism: RunnablePassthrough takes incoming data and forwards it completely unchanged, ensuring upstream payloads remain accessible downstream.
• Execution Flow: RunnableParallel accepts a dictionary mapping of runnables, fans out the identical input payload across all branches simultaneously, and aggregates their outputs into a single dictionary.
• Best Practice: Combine RunnablePassthrough with transformation branches inside RunnableParallel when a downstream prompt requires both raw inputs and transformed contexts.

from langchain_core.runnables import RunnableParallel, RunnablePassthrough, RunnableLambda

# Multi-branch transformation with input retention
branch_chain = RunnableParallel({
    "original": RunnablePassthrough(),
    "upper": RunnableLambda(lambda x: x.upper()),
    "length": RunnableLambda(lambda x: len(x))
})

result = branch_chain.invoke("langchain")
# Output: {'original': 'langchain', 'upper': 'LANGCHAIN', 'length': 9}

Key Takeaway: RunnableParallel fans out single inputs to multiple parallel branches while RunnablePassthrough preserves raw data.

Topics Covered in Building Reactive Chains with LCEL

1. Basic LCEL Chain Construction (0:00 - 2:04) — Chains prompt templates, chat models, and output parsers sequentially using LCEL pipe syntax.
2. The Runnable Interface (2:04 - 3:15) — Demonstrates sequential intermediate data flow across HumanMessage, AIMessage, and string formats.
3. Custom Engine via Operator Overloading (3:15 - 6:33) — Builds a custom runnable sequence engine by overriding Python's bitwise OR dunder method.
4. Testing Custom Runnables (6:33 - 8:24) — Subclasses custom runnables to execute arithmetic operations inside a custom pipe sequence.
5. Identity Mapping with Passthrough (8:24 - 8:59) — Implements RunnablePassthrough to transmit raw inputs unchanged to downstream chain components.
6. Wrapping Functions with RunnableLambda (8:59 - 10:03) — Converts plain Python functions into LCEL-compatible runnables exposing an invoke method.
7. Multi-Branch RunnableParallel (10:03 - 10:41) — Creates dictionary-based branching pipelines to transform identical input payloads concurrently.

LangChain & LangGraph for AI Agents Cheat Sheet

• prompt | model | parser — Sequentially chains LCEL components using pipe syntax

  chain = prompt | model | StrOutputParser()
  

• runnable.invoke(input) — Executes a runnable or chain on input payload

  result = chain.invoke({"topic": "AI"})
  

• __or__(self, other) — Overloads pipe operator to instantiate sequence objects

  def __or__(self, other): return CRunnableSequence(self, other)
  

• RunnablePassthrough() — Forwards input payload unchanged to next stage

  passthrough = RunnablePassthrough()
  

• RunnableLambda(fn) — Wraps arbitrary Python callable into runnable object

  step = RunnableLambda(lambda x: x.strip())
  

• RunnableParallel({"k": runnable}) — Runs multiple branches concurrently on identical input

  branches = RunnableParallel({"raw": RunnablePassthrough()})
  

Comparison Table

Common Pitfalls

• Mistake: Piping standard Python functions directly with the pipe operator without wrapping them. Avoid: Wrap all custom functions in RunnableLambda before chaining.
• Mistake: Expecting RunnableParallel to behave purely as a multi-threaded compute runner. Avoid: Use RunnableParallel specifically for splitting inputs across dictionary-mapped execution branches.
• Mistake: Passing mismatched data types between adjacent nodes in a sequential chain. Avoid: Verify intermediate types like HumanMessage and AIMessage before connecting stages.

FAQs

• Why does the pipe operator work between LangChain objects? LangChain overrides Python's or dunder method on Runnable objects to return a RunnableSequence instance instead of evaluating a bitwise operation.
• Can I use standard Python functions directly in an LCEL chain? No, standard Python functions lack the invoke method and Runnable interface, requiring them to be wrapped inside RunnableLambda.
• When should I use RunnablePassthrough? Use RunnablePassthrough when downstream components or prompt templates need access to the original, untransformed input payload alongside processed outputs.