Building a RAG chain entirely in LCEL

Most RAG examples drown you in boilerplate before you see a working pipeline. This lcel rag chain tutorial shows how to compose retrieval, prompting, and generation as a single declarative chain using LangChain Expression Language, with no custom classes or spaghetti callbacks.

Step 1: Install dependencies and configure credentials

Start with a clean virtual environment. You need the core LangChain packages, a vector store, an embeddings provider, and a document loader.

pip install langchain langchain-community langchain-openai faiss-cpu python-dotenv

Set your OpenAI key (or point to any OpenAI-compatible endpoint) in a .env file:

import os
from dotenv import load_dotenv

load_dotenv()
assert os.environ.get("OPENAI_API_KEY"), "Set OPENAI_API_KEY"

If you later swap ChatOpenAI for a gateway that exposes one OpenAI-compatible endpoint, the rest of this lcel rag chain tutorial stays identical.

Step 2: Load and chunk source documents

Raw files are useless to a retriever. Load them, then split into overlapping chunks so semantic search can hit precise passages.

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = TextLoader("service_spec.txt")
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " "]
)
chunks = splitter.split_documents(documents)
print(f"Created {len(chunks)} chunks")

The RecursiveCharacterTextSplitter respects paragraph and sentence boundaries before falling back to spaces. Tune chunk_size to your model’s context window and embedding dimensionality; 1000 characters is a safe default for most technical docs.

Step 3: Embed and index into a vector store

Embeddings turn text into vectors; the vector store handles similarity search. FAISS is lightweight and runs locally, which keeps the iteration loop fast.

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(chunks, embeddings)

Persist the index if you plan to reuse it across processes:

vectorstore.save_local("faiss_index")
# Reload later:
# vectorstore = FAISS.load_local("faiss_index", embeddings, allow_dangerous_deserialization=True)

For production scale, swap FAISS for Pinecone, pgvector, or Chroma. The LCEL chain below does not care which backend you use as long as it returns Document objects.

Step 4: Build the retriever and format context

A retriever is a Runnable that maps a string query to a list of Documents. You also need a tiny function to flatten those documents into a single context block.

retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

def format_docs(docs):
    return "\n\n".join(f"[{i+1}] {d.page_content}" for i, d in enumerate(docs))

Numbering sources inline ([1], [2]) makes it trivial for the LLM to cite them later. If you need metadata filtering, pass filter inside search_kwargs.

Step 5: Define the prompt and the LLM

Keep the prompt strict. In RAG, hallucination control comes from explicit instructions to use only the provided context.

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template(
    "You are a technical support agent. Answer the question using ONLY the "
    "context below. If the answer is not in the context, say 'Not specified'.\n\n"
    "Context:\n{context}\n\n"
    "Question: {question}\n\n"
    "Answer:"
)

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

temperature=0 reduces stochastic drift. For a chat-style UI, use ChatPromptTemplate.from_messages with a system/user pair; the LCEL composition is unchanged.

Step 6: Compose the LCEL RAG chain

This is the core of the lcel rag chain tutorial. LCEL uses the | operator to pipe runnables. We use RunnableParallel to fan the input question into two branches: one retrieves and formats context, the other passes the question through.

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

chain = (
    RunnableParallel({
        "context": retriever | format_docs,
        "question": RunnablePassthrough()
    })
    | prompt
    | llm
    | StrOutputParser()
)

What happens at runtime:

1. RunnablePassthrough takes the raw string query and copies it to question.
2. The same query hits retriever, which returns documents; format_docs collapses them into a string assigned to context.
3. The dict {context, question} is fed to prompt, producing a BaseMessage.
4. llm generates a response message.
5. StrOutputParser extracts the string content.

No subclassing, no RetrievalQA legacy wrappers. The chain is itself a Runnable, so it gains .invoke, .batch, .stream, and .ainvoke for free.

Step 7: Execute and verify success

Run a single query to confirm the pipeline works end to end:

query = "What is the maximum request payload size?"
answer = chain.invoke(query)
print(answer)

Verification checklist:

• The output is a non-empty string.
• The answer references details present in service_spec.txt (spot-check against the source).
• If you ask something absent from the doc, the model returns “Not specified” instead of guessing.

To prove retrieval fired, inspect the intermediate step:

context_only = (retriever | format_docs).invoke(query)
assert len(context_only) > 0
print(context_only[:500])

Streaming is one line because LCEL propagates stream events:

for token in chain.stream("Explain the retry backoff policy."):
    print(token, end="", flush=True)

If you see tokens appear incrementally and the final text matches the doc, the chain is correct.

Step 8: Add source citations without breaking the chain

A common production requirement is returning which chunks backed the answer. Extend the parallel branch to keep raw docs, then post-process.

from langchain_core.runnables import RunnableLambda

def with_sources(output, inputs):
    # inputs contains the original dict from RunnableParallel
    return output  # placeholder; real impl parses citations

chain_with_sources = (
    RunnableParallel({
        "context": retriever | format_docs,
        "docs": retriever,
        "question": RunnablePassthrough()
    })
    | prompt
    | llm
    | StrOutputParser()
    | RunnableLambda(lambda x: x)  # swap for citation parser
)

Because every step is a Runnable, you can insert logging, caching, or fallback logic (e.g., retry on provider 429) using with_retry or with_fallbacks without touching the core composition.

Step 9: Why this beats legacy RAG classes

The old RetrievalQA chain hid the prompt and parser behind constructor kwargs. Debugging meant reading library source. In this lcel rag chain tutorial you see every transformation explicitly. That matters when you need to:

• Swap the retriever for a hybrid search without rewriting the LLM call.
• Add a reranker by inserting | rerank between retriever and format_docs.
• Batch questions with chain.batch([q1, q2]) and get parallelized fetches.

LCEL chains are also serializable via langchain_core.runnables.RunnableSerializable, so you can dump the graph to JSON and serve it from a gateway.

Step 10: Production hardening notes

• Timeouts: wrap the LLM in ChatOpenAI(timeout=30) and add retriever.with_retry(stop_after_attempt=3).
• Metering: if you route through an OpenAI-compatible inference gateway, per-token usage is usually returned in response headers; capture it in a RunnableLambda after the LLM step.
• Cache: langchain_core.caches.InMemoryCache or Redis can sit on the embeddings call to avoid re-embedding identical queries.
• Evaluation: run chain.batch over a held-out question set and score with langchain_eval or a simple regex on citations.

By the end of this lcel rag chain tutorial you have a declarative, testable RAG pipeline that streams, batches, and degrades gracefully—built entirely from pipe operators and runnables.