Module 1: Architecture and Composition

LangChain Chains

Learning objectives

  • Explain the core mental model behind LangChain Chains
  • Apply LangChain Chains within Architecture and Composition
  • Identify important boundaries, trade-offs, and failure modes
  • Produce concrete evidence from the practice exercise

Current deterministic composition pattern

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate

model = init_chat_model("openai:gpt-5.4-mini")
prompt = ChatPromptTemplate.from_template("Summarize this precisely: {text}")
pipeline = prompt | model
response = pipeline.invoke({"text": "..."})

Runnable composition remains useful for deterministic pipelines. Older named chain classes are available through langchain-classic; new agentic applications should normally begin with create_agent or LangGraph.

Legacy and migration reference from the vault

Related: LangChain Core Concepts | LangChain Architecture Overview | LangChain Prompts and Templates | LangChain Index


What Is a Chain

A chain is a directed graph of Runnable steps. The simplest chain is sequential: prompt | model | parser.

Engineering analogy: a chain is a shell pipeline. cat file | grep pattern | wc -l is three programs connected by stdin/stdout. A LangChain chain is three runnables connected by typed data flow.


The Pipe Operator

The | operator composes runnables. The output of the left side becomes the input of the right side.

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

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": "cats"})

Under the hood, prompt.invoke returns a PromptValue; model.invoke consumes it and returns an AIMessage; parser.invoke consumes the message and returns a string.


Runnable Types

RunnableSequence (A | B)

Sequential composition. Output of A feeds into B.

RunnableParallel

Run multiple branches with the same input:

from langchain_core.runnables import RunnableParallel

chain = RunnableParallel(
    joke=prompt_joke | model | parser,
    fact=prompt_fact | model | parser,
)

result = chain.invoke({"topic": "cats"})
# {"joke": "...", "fact": "..."}

Engineering analogy: Promise.all in JavaScript or zip in Python. Concurrent, independent branches.

RunnablePassthrough

Pass input through unchanged, useful for merging:

from langchain_core.runnables import RunnablePassthrough

chain = RunnableParallel(
    original=RunnablePassthrough(),
    summary=prompt_summary | model | parser,
)

RunnableLambda

Wrap an arbitrary Python function:

from langchain_core.runnables import RunnableLambda

def add_metadata(input: dict) -> dict:
    input["timestamp"] = "2025-01-01"
    return input

chain = RunnableLambda(add_metadata) | prompt | model | parser

RunnableBranch

Conditional routing:

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: x["topic"] == "math", math_chain),
    (lambda x: x["topic"] == "history", history_chain),
    default_chain,
)

Built-In Chains

LangChain provides prebuilt chains for common patterns:

RetrievalQA

from langchain.chains import RetrievalQA

qa = RetrievalQA.from_chain_type(
    llm=model,
    retriever=retriever,
    chain_type="stuff",  # or "map_reduce", "refine", "map_rerank"
)

LLMMathChain

from langchain.chains import LLMMathChain

math_chain = LLMMathChain.from_llm(llm=model)

SQLDatabaseChain

from langchain.chains import SQLDatabaseChain

sql_chain = SQLDatabaseChain.from_llm(model, db)

Note: Newer code favors composing runnables manually (LCEL) over using prebuilt chain classes. Prebuilt chains are convenient but less flexible.


LCEL: LangChain Expression Language

LCEL is the name for the Runnable composition system. It is not a separate language; it is the Python API for building chains.

Key properties:

  • Streaming: chains built with LCEL support streaming natively
  • Async: ainvoke, abatch, astream are generated automatically
  • Tracing: LangSmith sees the full graph without extra instrumentation
  • Fallbacks: attach fallbacks to any node

Input / Output Mapping

Use itemgetter or dicts to reshape data between steps:

from operator import itemgetter

chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | model
    | parser
)

The dict syntax {"key": runnable} runs each value in parallel and assembles a dict output.


Common Patterns

RAG Chain

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | model
    | parser
)

Conversational RAG

conversational_rag = (
    RunnablePassthrough.assign(
        history=RunnableLambda(load_memory),
        context=retriever | format_docs,
    )
    | chat_prompt
    | model
    | parser
)

Agent Loop

Agents are dynamic chains. The model decides the next step at runtime. See LangChain Agents.


Debugging Chains

Enable tracing to see the execution graph:

result = chain.invoke(
    {"topic": "cats"},
    config={"callbacks": [StdOutCallbackHandler()]}
)

Or use LangSmith for a visual trace.


Related

  • LangChain Core Concepts — the Runnable abstraction
  • LangChain Architecture Overview — where chains sit in the stack
  • LangChain Prompts and Templates — the first node in most chains
  • LangChain Output Parsers — the last node in most chains
  • LangChain Agents — dynamic chains
  • LangChain Index — full topic map

Practice lab

Implement the smallest runnable agent workflow that demonstrates LangChain Chains. Trace inputs, state, model and tool calls, outputs, and cost; inject one failure and add a regression test that prevents it from returning. Add an operational constraint such as concurrency, recovery, security, latency, or cost, and defend the resulting design trade-off.

Review questions

  1. What problem does LangChain Chains solve, and what assumptions does it rely on?
  2. Which boundary or failure case is easiest to miss, and how would you expose it?
  3. What alternative design would you consider, and what trade-off would change the decision?
  4. What artifact, trace, test, or metric proves that your implementation is correct?

Completion evidence

  • A working artifact, annotated trace, or reproducible experiment
  • At least one normal case and one deliberately failing or boundary case
  • A concise explanation of the design choice and its trade-offs
  • Saved output showing how correctness was evaluated