Building a Complete LangChain RAG Pipeline

In this guide, we assemble a complete, production-grade Retrieval-Augmented Generation (RAG) Pipeline using the modern LangChain Expression Language (LCEL).

You will see how document ingestion, vector retrieval, prompt formatting, model inference, and output parsing chain together in under 15 lines of declarative code.

1. The LCEL RAG Architecture

flowchart LR
    Question["User Question"] --> Passthrough["RunnablePassthrough()"]
    Question --> Retriever["VectorStore.as_retriever()"]

    Retriever --> FormatDocs["format_docs()"]

    FormatDocs --> ContextDict["Context Dict:<br/>{'context': ..., 'question': ...}"]
    Passthrough --> ContextDict

    ContextDict --> Prompt["ChatPromptTemplate"]
    Prompt --> Model["ChatOpenAI / ChatGemini"]
    Model --> Parser["StrOutputParser()"]
    Parser --> FinalAnswer["Streamed Final Answer"]

    style Question fill:#475569,color:#fff
    style Retriever fill:#2563eb,color:#fff
    style ContextDict fill:#7c3aed,color:#fff
    style Prompt fill:#d97706,color:#fff
    style Model fill:#059669,color:#fff
    style FinalAnswer fill:#0284c7,color:#fff

2. Complete Code Walkthrough

import os
from typing import List

# Conceptual End-to-End Implementation of an LCEL RAG Chain

class MockRetriever:
    """Simulates a vector database retriever returning matching chunks."""
    def invoke(self, query: str) -> List[str]:
        return [
            "Snowflake micro-partitions are contiguous units of storage between 50MB and 500MB of uncompressed data.",
            "Snowflake automatically extracts column metadata (min, max, null count) for every micro-partition to enable partition pruning."
        ]

def format_docs(docs: List[str]) -> str:
    """Joins retrieved chunks with clean line breaks."""
    return "

".join(docs)

class MockModel:
    """Simulates streaming LLM response."""
    def invoke(self, prompt_text: str) -> str:
        return (
            "Snowflake micro-partitions are automatically managed storage blocks (50MB–500MB). "
            "Snowflake tracks min/max statistics for each column, allowing the query optimizer "
            "to prune irrelevant partitions without full table scans."
        )

class MockRAGChain:
    def __init__(self, retriever, model):
        self.retriever = retriever
        self.model = model

    def invoke(self, question: str) -> str:
        # Step 1: Parallel retrieval and passthrough
        docs = self.retriever.invoke(question)
        context = format_docs(docs)

        # Step 2: Prompt assembly
        prompt = f"""
You are a Snowflake Technical Advisor. Answer the question using ONLY the context below.

CONTEXT:
{context}

QUESTION:
{question}

ANSWER:
""".strip()

        # Step 3: Model execution and parsing
        return self.model.invoke(prompt)

if __name__ == "__main__":
    retriever = MockRetriever()
    model = MockModel()

    # Instantiate the RAG chain
    rag_pipeline = MockRAGChain(retriever, model)

    question = "How does Snowflake perform partition pruning on micro-partitions?"
    answer = rag_pipeline.invoke(question)

    print("=== Grounded RAG Answer ===")
    print(answer)

Key Takeaways

• The declarative LCEL RAG pattern: ({"context": retriever | format_docs, "question": RunnablePassthrough()} | prompt | model | StrOutputParser())

• Decoupled components: Swap vector databases (ChromaDB to Pinecone) or LLMs (OpenAI to Gemini) without altering the chain pipeline.

• Production reliability: Built-in timeout, fallback, and retry configurations on all Runnables.