Building a Modular RAG Pipeline with LangGraph

Retrieval-Augmented Generation (RAG) enhances Large Language Models (LLMs) by grounding their responses in external, domain-specific data. While traditional RAG implementations often rely on static, linear chains, orchestrating RAG workflows with LangGraph transforms the pipeline into a modular, stateful graph where retrieval, context aggregation, and answer generation operate as discrete, observable, and extensible nodes.

1. The Paradigm Shift: Standard RAG vs. Graph-Orchestrated RAG

In standard sequential pipelines (such as basic LangChain LCEL chains), data flows strictly in a predefined linear execution path. If retrieval fails, context needs custom formatting, or dynamic routing/evaluation is required later, rigid sequential chains become brittle and hard to maintain.

LangGraph breaks the execution cycle into an explicit State Machine:

• State Schema: A single centralized data contract passing between nodes.
• Independent Nodes: Isolated functions dedicated to a single responsibility (e.g., retrieving raw documents, parsing documents into context strings, querying the LLM).
• Explicit Edges: Clear transitions from one node to the next, enabling future additions such as conditional fallback routes, context validators, or human-in-the-loop steps.

2. End-to-End Architectural Blueprint

The complete system is divided into two distinct operational phases:

1. Ingestion Phase (Data Preparation): Loading unstructured files (e.g., PDF medical records), splitting text into semantic chunks, generating vector embeddings, and indexing them into an in-memory vector store.
2. Graph Execution Phase (Query & Reasoning): Initializing the graph state with a user query, traversing the retrieval and formatting nodes, and producing a grounded answer via an LLM.

3. Environment Setup & Dependency Configuration

To implement this pipeline, load your API keys (OpenAI for embeddings, Groq for ultra-fast LLM inference) and import the required modules across LangChain, LangGraph, and Pydantic.

import os
from dotenv import load_dotenv

# Load environment variables (.env containing OPENAI_API_KEY, GROQ_API_KEY)
load_dotenv()

# Document Loading and Text Processing
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Embeddings and Vector Storage
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import InMemoryVectorStore

# Model Provider
from langchain_groq import ChatGroq

# Graph Orchestration
from langgraph.graph import StateGraph, START, END

# Data Validation & State Schema
from pydantic import BaseModel, Field

4. Phase 1: Document Ingestion, Chunking, and Vector Indexing

Before graph traversal begins, external reference documents must be ingested and indexed.

Step 1: Document Loading

PyPDFLoader reads the raw PDF and extracts structured page-level text objects.

# Initialize PDF loader with target document
loader = PyPDFLoader(file_path="../data/medical_report.pdf")
docs = loader.load()

Step 2: Chunking via Recursive Splitting

Large texts exceed context windows and dilute embedding precision. Text is split into $1000$-character chunks with a $200$-character overlap to maintain semantic continuity across boundaries.

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
split_docs = text_splitter.split_documents(docs)

Step 3: Embeddings & Vector Store Population

Chunks are converted into high-dimensional vector representations using text-embedding-3-large and indexed in an InMemoryVectorStore.

# Initialize OpenAI embedding model
embedding_model = OpenAIEmbeddings(model="text-embedding-3-large")

# Index chunked documents into in-memory vector store
vector_store = InMemoryVectorStore.from_documents(
    documents=split_docs,
    embedding=embedding_model
)

5. Defining the Graph State Schema (RagState)

The state object serves as the shared memory across all nodes in the graph. Every node receives the current state, processes data, and returns an updated state slice.

Using Pydantic's BaseModel, we define the schema with default values.

Note on State Initialization: If default values (such as "" or []) are omitted, invoking the graph with only {"question": "..."} will trigger Pydantic validation errors for missing required fields.

class RagState(BaseModel):
    question: str = Field(description="The query input provided by the user")
    documents: list = Field(default=[], description="Retrieved document chunks relevant to query")
    context: str = Field(default="", description="Consolidated context string passed to LLM")
    answer: str = Field(default="", description="Final generated answer from LLM")

6. Engineering the Graph Nodes

Each node in LangGraph is a standard Python function that takes RagState as input and returns an updated RagState (or a dictionary matching state keys).

1. The Retrieval Node (retrieve_node)

Queries the vector database using vector similarity search against state.question.

def retrieve_node(state: RagState) -> RagState:
    # Perform similarity search against the vector store
    matched_docs = vector_store.similarity_search(query=state.question)
    state.documents = matched_docs
    return state

2. The Context Builder Node (create_context_node)

Iterates over the retrieved Document objects, extracts page_content, and joins them into a unified string separated by clear line breaks.

def create_context_node(state: RagState) -> RagState:
    context_accumulator = ""
    for doc in state.documents:
        context_accumulator += doc.page_content + "\n\n"
    
    state.context = context_accumulator
    return state

3. The Generation Node (generate_node)

Initializes the LLM (ChatGroq with openai/gpt-oss-20b or equivalent high-throughput models), builds a constrained prompt preventing hallucinations, and invokes the model.

# Initialize Groq LLM client
llm = ChatGroq(model="openai/gpt-oss-20b")

def generate_node(state: RagState) -> RagState:
    prompt = f"""You are a helpful medical assistant. Provide an accurate answer for the user's question based strictly on the provided context.
If you do not find the relevant answer in the context, respond with "I don't know".

Context:
{state.context}

Question:
{state.question}
"""
    response = llm.invoke(prompt)
    state.answer = response.content
    return state

7. Assembling the StateGraph

With the nodes defined, we construct the graph by instantiating StateGraph, registering the nodes, defining the execution edges, and compiling the graph.

# 1. Instantiate graph with state schema
graph_builder = StateGraph(RagState)

# 2. Add individual nodes
graph_builder.add_node("retrieve_node", retrieve_node)
graph_builder.add_node("create_context_node", create_context_node)
graph_builder.add_node("generate_node", generate_node)

# 3. Connect nodes via explicit directional edges
graph_builder.add_edge(START, "retrieve_node")
graph_builder.add_edge("retrieve_node", "create_context_node")
graph_builder.add_edge("create_context_node", "generate_node")
graph_builder.add_edge("generate_node", END)

# 4. Compile the graph
rag_graph = graph_builder.compile()

8. Graph Visualization and Runtime Execution

LangGraph supports graph visualization using Mermaid/PNG rendering via IPython.

from IPython.display import Image, display

# Display the compiled computational graph
display(Image(rag_graph.get_graph().draw_mermaid_png()))

Execution Flow Representation

9. Real-World Debugging: Pydantic State Initialization Pitfalls

A common pitfall when working with Pydantic-backed LangGraph state schemas occurs during initial invocation:

The Problem

When invoking the compiled graph, users typically pass only the initial input:

rag_graph.invoke({"question": "What is the age of patient?"})

If RagState is defined as:

class RagState(BaseModel):
    question: str
    documents: list  # Missing default
    context: str    # Missing default
    answer: str     # Missing default

Pydantic raises a validation error immediately at START because documents, context, and answer are missing from the initial dictionary.

The Solution

Assign explicit default values to all fields populated downstream:

class RagState(BaseModel):
    question: str
    documents: list = []
    context: str = ""
    answer: str = ""

This allows the graph to instantiate cleanly with partial input payloads.

10. Execution Walkthrough & Output Analysis

Once compiled and validated, invoke the graph with queries against the indexed medical record.

Test Query 1: Extracting Patient Information

response = rag_graph.invoke({"question": "What is the age of patient?"})
print("Answer:", response["answer"])

Output:
Answer: The patient is 27 years old.

Test Query 2: Medical Metric Verification

response = rag_graph.invoke({"question": "What is the value of RBC and is it in a range?"})
print("Answer:", response["answer"])

Output:
Answer: The red blood cell (RBC) count is 4.17 million/mcL, which is inside the standard reference range.

Test Query 3: Out-of-Context Fallback

response = rag_graph.invoke({"question": "What is the patient's home address?"})
print("Answer:", response["answer"])

Output:
Answer: I don't know.

Summary and Key Takeaways

• Stateful Decoupling: LangGraph decomposes traditional monolithic RAG chains into discrete functional nodes (retrieve, format, generate), simplifying debugging and state inspection.
• Two-Phase Architecture: Ingestion (loading, splitting, embedding, vector indexing) remains decoupled from query-time graph execution.
• Single Source of Truth: A centralized Pydantic RagState schema ensures type safety and clean data flow between nodes across the execution graph.
• Pydantic Default Value Requirement: All state attributes populated downstream must define defaults (= "" or = []) to avoid validation errors when invoking the graph with just the initial query.
• Strict Grounding via Prompts: Constraining the generator node with strict prompt boundaries prevents hallucinations on unindexed queries.
• Extensible Foundation: The compiled graph structure easily accommodates future nodes, such as dynamic document re-ranking, query transformation, or context validation gates.