LangChain SDK Track Part 1: Setup, Chains & LCEL

2. LCEL & The Pipe Operator

2.1 The | Operator

LCEL (LangChain Expression Language) uses Python's | (bitwise OR) operator, overloaded via __or__ on all Runnable objects, to compose chains declaratively:

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

# Each | creates a new RunnableSequence
chain = (
    ChatPromptTemplate.from_template("Tell me a joke about {topic}")
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

# Equivalent to:
# RunnableSequence(first=prompt, middle=[model], last=parser)

2.2 The Runnable Protocol

Every LCEL component implements the Runnable interface, which provides these methods:

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

prompt = ChatPromptTemplate.from_template("Explain {concept} in one sentence.")
model = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | model | StrOutputParser()

# invoke - single call
result = chain.invoke({"concept": "recursion"})

# stream - token by token
for chunk in chain.stream({"concept": "recursion"}):
    print(chunk, end="", flush=True)

# batch - parallel processing
results = chain.batch([
    {"concept": "recursion"},
    {"concept": "memoization"},
    {"concept": "dynamic programming"}
])

2.3 invoke / stream / batch Deep Dive

The power of the Runnable protocol is that streaming and batching come free with any chain. When you call .stream() on a composed chain, each intermediate step streams its output to the next step — you don't need to rewrite your chain for streaming support.

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

prompt = ChatPromptTemplate.from_template("Write a haiku about {topic}")
model = ChatOpenAI(model="gpt-4o-mini", streaming=True)
chain = prompt | model | StrOutputParser()

# Async streaming with event metadata
async def stream_with_events():
    async for event in chain.astream_events(
        {"topic": "programming"},
        version="v2"
    ):
        if event["event"] == "on_chat_model_stream":
            print(event["data"]["chunk"].content, end="")

asyncio.run(stream_with_events())

3. Runnable Primitives

3.1 RunnablePassthrough

Passes the input through unchanged, or assigns new keys while preserving existing ones:

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

# Pass input unchanged (useful in parallel compositions)
chain = RunnablePassthrough() | ChatPromptTemplate.from_template("{input}")

# assign() adds new keys to the input dict
chain = RunnablePassthrough.assign(
    word_count=lambda x: len(x["text"].split()),
    upper=lambda x: x["text"].upper()
)

result = chain.invoke({"text": "Hello world from LangChain"})
# {'text': 'Hello world from LangChain', 'word_count': 5, 'upper': 'HELLO WORLD FROM LANGCHAIN'}
print(result)

3.2 RunnableParallel

Runs multiple Runnables in parallel and collects their outputs into a dictionary:

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

model = ChatOpenAI(model="gpt-4o-mini")

# Define parallel branches
joke_chain = ChatPromptTemplate.from_template("Tell a joke about {topic}") | model | StrOutputParser()
poem_chain = ChatPromptTemplate.from_template("Write a haiku about {topic}") | model | StrOutputParser()
fact_chain = ChatPromptTemplate.from_template("State one fact about {topic}") | model | StrOutputParser()

# Run all three in parallel
parallel = RunnableParallel(
    joke=joke_chain,
    poem=poem_chain,
    fact=fact_chain
)

results = parallel.invoke({"topic": "cats"})
print(results["joke"])
print(results["poem"])
print(results["fact"])

3.3 RunnableLambda

Wraps any Python function as a Runnable, giving it the full invoke/stream/batch interface:

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

# Wrap a custom function
def format_output(text: str) -> str:
    """Add formatting to model output."""
    lines = text.strip().split("\n")
    return "\n".join(f"  • {line}" for line in lines if line.strip())

formatter = RunnableLambda(format_output)

chain = (
    ChatPromptTemplate.from_template("List 3 benefits of {topic}")
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
    | formatter
)

print(chain.invoke({"topic": "exercise"}))

3.4 RunnableBranch

Conditional routing — picks a branch based on the input:

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

model = ChatOpenAI(model="gpt-4o-mini")

# Define specialized chains for each category
tech_chain = ChatPromptTemplate.from_template("As a tech expert: {question}") | model | StrOutputParser()
science_chain = ChatPromptTemplate.from_template("As a scientist: {question}") | model | StrOutputParser()
general_chain = ChatPromptTemplate.from_template("Answer: {question}") | model | StrOutputParser()

# Route based on category field
router = RunnableBranch(
    (lambda x: x.get("category") == "tech", tech_chain),
    (lambda x: x.get("category") == "science", science_chain),
    general_chain  # default fallback
)

result = router.invoke({"category": "tech", "question": "What is LCEL?"})
print(result)

4. Chain Types

4.1 Sequential Chains

Multi-step pipelines where each step's output feeds the next:

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

model = ChatOpenAI(model="gpt-4o-mini")

# Step 1: Generate an outline
outline_chain = (
    ChatPromptTemplate.from_template("Create a 3-point outline for an essay about {topic}")
    | model
    | StrOutputParser()
)

# Step 2: Expand the outline into prose
expand_chain = (
    ChatPromptTemplate.from_template("Expand this outline into a short essay:\n{outline}")
    | model
    | StrOutputParser()
)

# Compose sequentially - output of step 1 becomes input to step 2
full_chain = (
    {"outline": outline_chain, "topic": RunnablePassthrough()}
    | (lambda x: {"outline": x["outline"]})
    | expand_chain
)

essay = full_chain.invoke("renewable energy")
print(essay)

4.2 Transform Chains

Chains that transform data between steps without calling an LLM:

from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

def clean_text(input_dict: dict) -> dict:
    """Remove extra whitespace and normalize text."""
    text = input_dict["raw_text"]
    cleaned = " ".join(text.split())
    return {"text": cleaned, "word_count": len(cleaned.split())}

def add_metadata(input_dict: dict) -> dict:
    """Add processing metadata."""
    return {
        **input_dict,
        "is_long": input_dict["word_count"] > 100,
        "summary_prompt": f"Summarize in {'3' if input_dict['is_long'] else '1'} sentences: {input_dict['text']}"
    }

model = ChatOpenAI(model="gpt-4o-mini")

transform_chain = (
    RunnableLambda(clean_text)
    | RunnableLambda(add_metadata)
    | (lambda x: x["summary_prompt"])
    | ChatPromptTemplate.from_template("{text}")
    | model
    | StrOutputParser()
)

result = transform_chain.invoke({"raw_text": "  This   is   some   messy    text   with   extra   spaces.  "})
print(result)

4.3 Router Chains

Dynamic routing based on LLM classification of the input:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch, RunnableLambda

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Step 1: Classify the input
classifier = (
    ChatPromptTemplate.from_template(
        "Classify this question into exactly one category: tech, science, history.\n"
        "Question: {question}\nCategory:"
    )
    | model
    | StrOutputParser()
    | (lambda x: x.strip().lower())
)

# Step 2: Specialized response chains
tech_chain = ChatPromptTemplate.from_template("As a senior engineer, answer: {question}") | model | StrOutputParser()
science_chain = ChatPromptTemplate.from_template("As a PhD scientist, answer: {question}") | model | StrOutputParser()
history_chain = ChatPromptTemplate.from_template("As a historian, answer: {question}") | model | StrOutputParser()

# Step 3: Route based on classification
def route(info: dict) -> str:
    """Route to the appropriate chain based on classification."""
    category = info["category"]
    question = info["question"]
    if "tech" in category:
        return tech_chain.invoke({"question": question})
    elif "science" in category:
        return science_chain.invoke({"question": question})
    else:
        return history_chain.invoke({"question": question})

# Full router chain
router_chain = (
    {"category": classifier, "question": lambda x: x["question"]}
    | RunnableLambda(route)
)

result = router_chain.invoke({"question": "How does TCP/IP work?"})
print(result)

5. Output Parsers

5.1 Pydantic Output Parser

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

class MovieReview(BaseModel):
    title: str = Field(description="The movie title")
    rating: float = Field(description="Rating from 1-10")
    summary: str = Field(description="One sentence summary")
    genres: list[str] = Field(description="List of genres")

parser = PydanticOutputParser(pydantic_object=MovieReview)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Extract movie review information.\n{format_instructions}"),
    ("human", "{review}")
])

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

chain = prompt.partial(
    format_instructions=parser.get_format_instructions()
) | model | parser

review = chain.invoke({
    "review": "Inception is a mind-bending sci-fi thriller by Nolan. I'd give it a 9/10."
})
print(f"{review.title}: {review.rating}/10 - {review.genres}")

5.2 with_structured_output()

The modern preferred approach — uses the model's native structured output support:

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

class ExtractedEntity(BaseModel):
    name: str = Field(description="Entity name")
    entity_type: str = Field(description="Type: person, org, or location")
    confidence: float = Field(description="Confidence score 0-1")

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Bind structured output directly to the model
structured_model = model.with_structured_output(ExtractedEntity)

result = structured_model.invoke("Apple Inc. was founded by Steve Jobs in Cupertino.")
print(f"{result.name} ({result.entity_type}) - confidence: {result.confidence}")

5.3 Auto-Fixing & Retry Parsers

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import PydanticOutputParser
from langchain.output_parsers import OutputFixingParser, RetryOutputParser
from pydantic import BaseModel, Field

class StructuredAnswer(BaseModel):
    answer: str = Field(description="The answer")
    confidence: float = Field(description="Confidence 0-1")
    sources: list[str] = Field(description="Source references")

base_parser = PydanticOutputParser(pydantic_object=StructuredAnswer)
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Auto-fixing parser: if parsing fails, asks the LLM to fix the output
fixing_parser = OutputFixingParser.from_llm(parser=base_parser, llm=model)

# Retry parser: re-invokes the original prompt with error context
retry_parser = RetryOutputParser.from_llm(parser=base_parser, llm=model)

# Use in a chain with fallback
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template(
    "Answer this question with sources: {question}\n{format_instructions}"
)

chain = (
    prompt.partial(format_instructions=base_parser.get_format_instructions())
    | model
    | fixing_parser
)

result = chain.invoke({"question": "What is the speed of light?"})
print(f"Answer: {result.answer}")
print(f"Confidence: {result.confidence}")
print(f"Sources: {result.sources}")

Summary & Next Steps

You now have a complete working knowledge of LangChain's core orchestration primitives:

• LCEL and the pipe operator compose any Runnable components into chains with automatic streaming, batching, and async
• The Runnable protocol (invoke/stream/batch/ainvoke/astream/abatch) provides a universal interface for all components
• Runnable primitives (Passthrough, Parallel, Lambda, Branch) handle data flow, parallelism, transforms, and routing
• Chain patterns (Sequential, Transform, Router) cover the vast majority of real-world orchestration needs
• Output parsers (Pydantic, with_structured_output, auto-fixing) turn raw LLM text into validated typed objects