Migrating legacy LangChain chains to LCEL

If you’ve built anything substantial with LangChain before late 2023, your codebase is likely full of LLMChain, SequentialChain, and ConversationChain classes. These legacy chains work, but they’re rigid, hard to debug, and don’t compose well with modern streaming, async, or batch workflows. This guide walks through how to migrate langchain chains to lcel with minimal risk and maximum clarity.

Step 1: Inventory your legacy chains

Before rewriting anything, catalog what you have. Legacy chains typically fall into three categories:

Single-step chains: LLMChain(prompt=..., llm=...) wrapping one prompt + model call
Sequential chains: SequentialChain(chains=[chain1, chain2, ...]) passing outputs forward
Specialized chains: ConversationChain, RetrievalQA, StuffDocumentsChain, MapReduceDocumentsChain, etc.

Run this script to surface them automatically:

# find_legacy_chains.py
import ast
import sys
from pathlib import Path

LEGACY_CLASSES = {
    "LLMChain", "SequentialChain", "ConversationChain",
    "RetrievalQA", "StuffDocumentsChain", "MapReduceDocumentsChain",
    "ReduceDocumentsChain", "MapRerankDocumentsChain",
    "AnalyzeDocumentChain", "VectorDBQA", "ChatVectorDBChain",
}

def find_imports_and_usages(filepath: Path):
    tree = ast.parse(filepath.read_text())
    imports = set()
    usages = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.ImportFrom) and node.module and "langchain" in node.module:
            for alias in node.names:
                if alias.name in LEGACY_CLASSES:
                    imports.add(alias.name)
        elif isinstance(node, ast.Name) and node.id in LEGACY_CLASSES:
            usages.add(node.id)
        elif isinstance(node, ast.Attribute) and node.attr in LEGACY_CLASSES:
            usages.add(node.attr)
    return imports, usages

if __name__ == "__main__":
    root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(".")
    for py_file in root.rglob("*.py"):
        imports, usages = find_imports_and_usages(py_file)
        if imports or usages:
            print(f"{py_file}: imports={imports}, usages={usages}")

Run it: python find_legacy_chains.py ./src. Save the output — this is your migration backlog.

Step 2: Understand the LCEL mental model

LCEL replaces class hierarchies with a single abstraction: Runnable. Every component — prompt, model, parser, retriever, custom function — implements the same interface:

from langchain_core.runnables import Runnable

# All of these are Runnables
prompt = ChatPromptTemplate.from_template("...")
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
retriever = vectorstore.as_retriever()

# Composition via pipe operator
chain = prompt | model | parser

Key differences from legacy chains:

Legacy	LCEL
`chain.run(input)`	`chain.invoke(input)` or `chain.batch([...])`
`chain.arun(input)`	`chain.ainvoke(input)` or `chain.abatch([...])`
Streaming via callbacks	`chain.astream(input)` / `chain.stream(input)`
Fixed input/output keys	Dict in, dict out — explicit and inspectable
Subclass to customize	Compose with `RunnableLambda`, `RunnableParallel`, `RunnablePassthrough`

The pipe operator (|) handles type coercion: if the left side outputs a string and the right expects a dict with key "text", LCEL wraps automatically. But explicit is better — define your schemas.

Step 3: Migrate a single LLMChain

The simplest migration: LLMChain(prompt=prompt, llm=llm, output_parser=parser).

Legacy:

from langchain.chains import LLMChain
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template(
    "Summarize this text in 3 sentences:\n\n{text}"
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

legacy_chain = LLMChain(prompt=prompt, llm=llm, output_parser=parser)
result = legacy_chain.run(text=long_document)

LCEL equivalent:

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

prompt = ChatPromptTemplate.from_template(
    "Summarize this text in 3 sentences:\n\n{text}"
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

lcel_chain = prompt | llm | parser
result = lcel_chain.invoke({"text": long_document})

Verify: Run both with identical inputs. Compare outputs character-for-character. They should match exactly — same model, same prompt, same parser.

assert legacy_chain.run(text=sample) == lcel_chain.invoke({"text": sample})

Step 4: Migrate SequentialChain

SequentialChain passes outputs from one chain as inputs to the next. In LCEL, use RunnableSequence (the pipe operator) with RunnableParallel or RunnablePassthrough to thread values.

Legacy:

from langchain.chains import LLMChain, SequentialChain

chain1 = LLMChain(
    prompt=ChatPromptTemplate.from_template("Extract key topics from: {text}"),
    llm=llm,
    output_key="topics"
)
chain2 = LLMChain(
    prompt=ChatPromptTemplate.from_template("Write a tweet about: {topics}"),
    llm=llm,
    output_key="tweet"
)

sequential = SequentialChain(
    chains=[chain1, chain2],
    input_variables=["text"],
    output_variables=["topics", "tweet"]
)
result = sequential({"text": article})

LCEL equivalent:

from langchain_core.runnables import RunnablePassthrough

extract_topics = (
    ChatPromptTemplate.from_template("Extract key topics from: {text}")
    | llm
    | StrOutputParser()
)

write_tweet = (
    ChatPromptTemplate.from_template("Write a tweet about: {topics}")
    | llm
    | StrOutputParser()
)

# RunnablePassthrough.assign adds computed fields to the dict
lcel_chain = (
    RunnablePassthrough.assign(topics=extract_topics)
    | RunnablePassthrough.assign(tweet=write_tweet)
)

result = lcel_chain.invoke({"text": article})
# result = {"text": "...", "topics": "...", "tweet": "..."}

Verify: Check that intermediate values (topics) are identical. Add a debug step if needed:

from langchain_core.runnables import RunnableLambda

debug_chain = (
    RunnablePassthrough.assign(topics=extract_topics)
    | RunnableLambda(lambda x: (print(f"topics: {x['topics']}"), x)[1])
    | RunnablePassthrough.assign(tweet=write_tweet)
)

Step 5: Migrate ConversationChain (memory)

Legacy ConversationChain bundles memory internally. LCEL separates memory as a first-class RunnableWithMessageHistory.

Legacy:

from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(return_messages=True)
conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

response1 = conversation.predict(input="Hi, I'm Alice.")
response2 = conversation.predict(input="What's my name?")

LCEL equivalent:

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

chain = prompt | llm | StrOutputParser()

store = {}

def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

config = {"configurable": {"session_id": "user-123"}}
response1 = with_history.invoke({"input": "Hi, I'm Alice."}, config=config)
response2 = with_history.invoke({"input": "What's my name?"}, config=config)

Verify: Run a multi-turn conversation and assert the model recalls earlier context. The config parameter threads session identity — this is explicit in LCEL, not hidden in a memory object.

Step 6: Migrate RetrievalQA and document chains

Legacy RetrievalQA and StuffDocumentsChain combine retrieval + prompt + generation. LCEL makes each step visible and swappable.

Legacy:

from langchain.chains import RetrievalQA
from langchain.chains.combine_documents import StuffDocumentsChain

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(k=4),
    return_source_documents=True,
)
result = qa_chain({"query": "What is the refund policy?"})

LCEL equivalent:

from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

retriever = vectorstore.as_retriever(k=4)

prompt = ChatPromptTemplate.from_template("""Answer the question using only the context below.

Context:
{context}

Question: {question}
""")

def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

rag_chain = (
    RunnableParallel(
        context=retriever | RunnableLambda(format_docs),
        question=RunnablePassthrough(),
    )
    | prompt
    | llm
    | StrOutputParser()
)

result = rag_chain.invoke("What is the refund policy?")
# To get source documents too:
rag_with_sources = RunnableParallel(
    answer=rag_chain,
    context=retriever,
)
result = rag_with_sources.invoke("What is the refund policy?")
# result = {"answer": "...", "context": [Document, ...]}

Verify: Compare retrieved documents and final answers. The RunnableParallel approach lets you inspect context before it hits the prompt — useful for debugging retrieval quality.

Step 7: Handle streaming and async

One major reason to migrate: LCEL streaming works out of the box with no callback plumbing.

Legacy streaming (callback hell):

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

llm = ChatOpenAI(model="gpt-4o-mini", streaming=True, callbacks=[StreamingStdOutCallbackHandler()])
chain = LLMChain(prompt=prompt, llm=llm)
chain.run(text=long_doc)  # prints to stdout, hard to capture

LCEL streaming:

chain = prompt | llm | StrOutputParser()

# Sync streaming
for chunk in chain.stream({"text": long_doc}):
    print(chunk, end="", flush=True)

# Async streaming
async for chunk in chain.astream({"text": long_doc}):
    print(chunk, end="", flush=True)

# Batch with concurrency control
results = chain.batch([{"text": d} for d in documents], config={"max_concurrency": 5})

Verify: Stream a known input and confirm tokens arrive incrementally. For async, run under asyncio and ensure no blocking calls.

Step 8: Add observability — logging, tracing, fallbacks

LCEL’s RunnableConfig carries metadata through the entire chain. Use it for tracing, model fallbacks, or per-request overrides.

from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

primary = ChatOpenAI(model="gpt-4o-mini")
fallback = ChatAnthropic(model="claude-3-haiku-20240307")

chain = prompt | primary.with_fallbacks([fallback]) | StrOutputParser()

# Per-request config: override model, add tags, enable tracing
config = RunnableConfig(
    tags=["production", "summarization"],
    metadata={"user_id": "user-123", "request_id": "req-456"},
    callbacks=[LangChainTracer()],  # if using LangSmith
)
result = chain.invoke({"text": doc}, config=config)

If you’re running inference through a gateway that handles provider fallbacks and usage metering automatically — like n4n.ai — you can skip the manual with_fallbacks and let the gateway handle it. The chain stays clean:

# Gateway handles fallback, retries, cache-control hints
chain = prompt | gateway_model | StrOutputParser()

Step 9: Write regression tests

Don’t rely on manual verification. Add pytest cases that lock in behavior.

# test_migration.py
import pytest
from myapp.chains import legacy_summarize, lcel_summarize

SAMPLES = [
    "Short text.",
    "A" * 5000,  # long context
    "Text with \"quotes\" and \n newlines.",
    "",  # edge case
]

@pytest.mark.parametrize("text", SAMPLES)
def test_summarize_parity(text):
    legacy_out = legacy_summarize(text)
    lcel_out = lcel_summarize(text)
    assert legacy_out == lcel_out, f"Mismatch for input: {text[:50]}..."

@pytest.mark.asyncio
async def test_streaming_emits_tokens():
    chunks = []
    async for chunk in lcel_summarize.astream({"text": "Hello world"}):
        chunks.append(chunk)
    assert len(chunks) > 1
    assert "".join(chunks) == lcel_summarize.invoke({"text": "Hello world"})

Run these in CI. Any drift — model version change, prompt tweak, parser update — fails fast.

Step 10: Incremental rollout with feature flags

Don’t flip a switch. Route traffic gradually.

import random
from myapp.chains import legacy_chain, lcel_chain

def get_chain(user_id: str):
    # 10% to LCEL, ramp up over days
    if hash(user_id) % 100 < 10:
        return lcel_chain
    return legacy_chain

@app.post("/summarize")
def summarize(request: SummarizeRequest):
    chain = get_chain(request.user_id)
    return {"summary": chain.invoke({"text": request.text})}

Monitor latency, error rates, and output quality metrics. If LCEL regresses, rollback is instant — just change the percentage.

Common pitfalls

Implicit dict wrapping: Legacy chains often accept chain.run("raw string"). LCEL requires chain.invoke({"input_key": "raw string"}). Audit call sites.

Output keys: LLMChain(output_key="result") puts output in {"result": "..."}. LCEL returns the parser’s raw output. Adjust downstream consumers.

Memory key names: ConversationChain uses history by default. RunnableWithMessageHistory requires explicit history_messages_key. Match them.

Prompt variable mismatch: Legacy chains sometimes infer variables from the prompt template. LCEL is strict — every {var} in the template must exist in the input dict.

Streaming parser compatibility: StrOutputParser streams. JsonOutputParser does not (it needs complete JSON). Use JsonOutputParser only with invoke/ainvoke, or swap to a streaming JSON parser.

Verification checklist

Before marking a chain migrated, confirm:

Unit tests pass (parity + streaming + async)
Integration tests pass (end-to-end API calls)
Latency within 5% of legacy (run chain.batch with max_concurrency)
Streaming produces tokens incrementally (no buffering)
Error handling: invalid input, model timeout, rate limit
Observability: traces appear in your tracing backend
Fallback behavior verified (kill primary model, confirm fallback fires)
Canary traffic shows no quality regression

What’s next

Migrating to LCEL isn’t a one-time chore — it’s the foundation for composable, observable, production-grade LLM pipelines. Once your chains are Runnables, you can:

Swap models per request via config
Add caching with RunnableCache
Implement retries, timeouts, and circuit breakers at the chain level
Build dynamic chains that assemble themselves at runtime
Use RunnableBranch for conditional logic without spaghetti code

The legacy chain classes aren’t disappearing, but they’re frozen. LCEL is where every new LangChain feature lands first. Migrate now, or migrate later under pressure.