6. Declarative Pipelines with LCEL

In previous chapters, we've been writing imperative code to orchestrate LLM interactions: create a prompt, invoke the model, parse the response. This works, but as AI applications grow more complex, this approach becomes verbose and harder to maintain. You end up with deeply nested function calls, manual error handling at each step, and difficulty understanding the overall data flow.

LangChain Expression Language (LCEL) solves this by letting you declare what you want to happen, not how to make it happen. Instead of writing procedural code that calls functions in sequence, you compose components using a simple pipe operator (|) that reads like a Unix pipeline. The result is cleaner, more maintainable code that clearly expresses the flow of data through your AI system.

This chapter introduces LCEL for building linear workflows - sequences of operations where data flows from start to finish without branching or loops. We'll cover when to use LCEL, how to compose pipelines, and how to execute them both synchronously and with streaming output.

6.1) Why LCEL?

The Problem LCEL Solves

Let's start with a concrete example. Suppose you're building a customer support assistant that needs to:

1. Take a user question
2. Format it into a prompt with system instructions and the user message
3. Send it to an LLM
4. Parse the response to extract just the text content

Here's how you might write this imperatively (without LCEL):

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support assistant."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
def answer_question(question: str) -> str:
    # Step 1: Format the prompt with the question
    messages = prompt.invoke({"question": question})
    
    # Step 2: Invoke LLM
    response = llm.invoke(messages)
    
    # Step 3: Parse the output to extract text content
    result = parser.invoke(response)
    
    return result
 
# Use it
result = answer_question("How do I reset my password?")
print(result)

Output:

To reset your password, please follow these steps:
1. Go to the login page
2. Click "Forgot Password"
3. Enter your email address
4. Check your email for a reset link
5. Follow the link and create a new password

This works, but notice the problems:

Verbosity: Every step requires explicit variable assignment and function calls. The actual logic (format → invoke → parse) is buried in boilerplate.

Rigid Structure: If you want to add a step (like validating the question or logging the response), you need to insert code in the middle of the function, increasing complexity.

No Built-in Streaming: To stream tokens as they arrive, you'd need to rewrite the entire function to use llm.stream() and handle the async iteration manually.

Unclear Data Flow: Reading the code, it's not immediately obvious that this is a simple pipeline. You have to trace through variable assignments to understand the flow.

Now let's see the LCEL version:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
# Define the components (same as before)
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support assistant."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
# Compose them into a chain using the pipe operator
chain = prompt | llm | parser
 
# Use it
result = chain.invoke({"question": "How do I reset my password?"})
print(result)

Output:

To reset your password, please follow these steps:
1. Go to the login page
2. Click "Forgot Password"
3. Enter your email address
4. Check your email for a reset link
5. Follow the link and create a new password

The output is identical, but the code is dramatically different:

Declarative: chain = prompt | llm | parser expresses the entire flow in one line. Read it left-to-right: prompt → LLM → parser.

Composable: Each component (prompt, llm, parser) is independent and reusable. You can swap components without rewriting the pipeline.

Streaming Built-in: Switch between chain.invoke() and chain.stream() without changing the pipeline definition.

Clear Intent: The | operator makes the data flow obvious at a glance.

Where LCEL fits: Linear workflows

LCEL is designed for linear workflows - sequences where data flows in one direction from start to finish without branching or loops.

This linear pattern covers many AI applications. Consider a document Q&A system: you receive a question → retrieve relevant documents → format a prompt → send to LLM → parse the answer. Each step is a clear sequence where the output of one step becomes the input to the next.

But what if your workflow needs to:

• Have the LLM decide which tool to call based on the question
• Call a tool, observe the result, then decide what to do next
• Retry failed operations with different approaches

These scenarios require loops and conditional branching - things LCEL cannot handle. This is why LangGraph exists (we'll introduce it in Chapter 15).

Here's a visual comparison of the two approaches:

LCEL is perfect for straightforward pipelines where each step processes the output of the previous step. Data flows in one direction only.

LangGraph is for when you need decision-making and loops. The agent can act, observe results, and think again.

For this chapter, we're focusing on LCEL. Why master linear pipelines first?

• Foundation: The pipe operator (|) is LangChain's core syntax. Understanding this makes everything else easier.
• Prerequisite for LangGraph: LangGraph agents use LCEL chains extensively inside their nodes
• Real-world pattern: Complex agents are built by combining LCEL chains

The rest of this chapter will show you how to build these linear pipelines with the pipe operator (|).

6.2) Composing Pipelines with the | Operator

How LCEL Pipes Work

LCEL's pipe operator (|) works because of the Runnable interface.

What is a Runnable?

Runnable is LangChain's standard interface. When a component implements the Runnable interface, it can be chained with other components using the | operator.

Every Runnable provides these methods:

• .invoke(input) - Execute once and get the complete result
• .stream(input) - Execute and receive each word as the LLM generates it
• .batch(inputs) - Execute multiple times with different inputs and get all results

For this chapter, we'll focus on .invoke() and .stream() (we'll cover .batch() later when needed).

Why does | work?

Because the Runnable class implements the | operator using Python's operator overloading. When you write prompt | llm, it creates a new Runnable that executes the two components sequentially.

Most LangChain components are Runnables

This is why you can chain so many different components:

• ChatPromptTemplate is a Runnable
• ChatOpenAI is a Runnable
• StrOutputParser is a Runnable
• Even custom chains you create with | are themselves Runnables!

This means you can build complex pipelines by combining simpler ones.

Connecting Components: Input/Output Types

When connecting components with |, you need to ensure the output type of one component matches the input type of the next.

Key component signatures:

Example flow:

chain = prompt | llm | parser

Here's how the data type transforms as it passes through each component:

• prompt: Takes dict as input and converts to list[BaseMessage]
• llm: Takes list[BaseMessage] as input and converts to AIMessage
• parser: Takes AIMessage as input and converts to str

Let's see this in action:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
 
chain = prompt | llm | parser
 
result = chain.invoke({"question": "What is 2+2?"})
print(result)  # "2+2 equals 4."

What happens at each step:

What happens if types don't match?

If you try to connect incompatible components, you'll get an error:

# ERROR: This won't work
chain = llm | prompt  # LLM outputs AIMessage, but prompt requires dict as input

The error message will tell you what input type the next component expects vs. what input type it actually received.

6.3) Executing a Chain: .invoke() and .stream()

Running a Pipeline

Once you've composed a chain, you execute it using the .invoke() method. This is the synchronous way to run a pipeline - it waits for the entire response before returning the result.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Synchronous invocation
result = chain.invoke({"question": "What is 2+2?"})
print(result)

Output:

2+2 equals 4.

The .invoke() method is straightforward: pass in the input parameter that the first component expects, and get back the output produced by the last component.

Streaming Tokens from the Same Chain

While .invoke() is simple, it has a limitation for user-facing applications: users see nothing until the entire response is complete. For long responses (10-20 seconds), this creates a poor user experience.

Streaming shows tokens immediately as they're generated, rather than waiting for the complete response. It's the typing effect you see in ChatGPT.

Let's see streaming in action:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Stream chunks as they arrive
for chunk in chain.stream({"question": "Explain what Python is in one sentence"}):
    print(chunk, end="", flush=True)

Output (displayed in real-time, token by token):

Python is a versatile, high-level programming language known for its simplicity and readability, widely used in web development, data science, and automation.

Notice that the chain definition is identical to the .invoke() example above. We didn't need to rebuild it - we just called .stream() instead of .invoke().

The .stream() method returns results as chunks while the LLM generates tokens. The print(chunk, end="", flush=True) displays each chunk immediately on screen, creating the live typing effect.

When to Use .invoke() vs .stream()

Use .invoke() when:

• You only need the final result (analysis, translation, classification)
• The response is short and wait time isn't an issue
• You need the complete output before proceeding to the next step

Use .stream() when:

• Users need to see progress (chat, content generation)
• The response is long and wait time would be noticeable
• You're building a UI where real-time feedback matters

Both methods work on the same chain. Define the chain once, then choose the execution mode based on your needs.

In this chapter, you learned LCEL - LangChain's declarative pipeline syntax:

• Build pipelines with the pipe operator: prompt | llm | parser
• Execute flexibly: Use .invoke() for complete results or .stream() for real-time output
• Type safety: Match output types to input types when chaining components

The same chain works for both execution modes - define once, use anywhere.

Next: Chapter 7 covers structured output with Pydantic, enabling you to extract validated JSON data from LLM responses.