If you’re new to Retrieval-Augmented Generation (RAG), the architecture can seem more complicated than it really is.
At its core, a basic RAG system only performs two tasks:
Retrieve relevant information.
Ask the LLM to answer using that information.
In this tutorial, we’ll build a complete beginner-friendly RAG application using:
Python
LangGraph
OpenAI
FAISS (local vector database)
Everything is contained in a single Python file.
What is RAG?
RAG (Retrieval-Augmented Generation) improves an LLM by giving it relevant documents before it generates an answer.
Instead of relying solely on what the model learned during training, it also draws on your own knowledge base.
Typical use cases include:
Internal company documentation
PDFs
Wikis
Product manuals
Knowledge bases
Research papers
Simple RAG Workflow
Documents
│
▼
Split into Chunks
│
▼
Generate Embeddings
│
▼
Store in Vector Database
│
──────────────────────────────
User Question
│
▼
Generate Query Embedding
│
▼
Similarity Search
│
▼
Retrieve Relevant Chunks
│
▼
Prompt (Context + Question)
│
▼
OpenAI GPT
│
▼
Final AnswerFor a beginner project, that’s the entire pipeline.
Project Structure
Since this is a learning project, everything lives in one file.
simple_rag.py
No complex folder structure.
Install Dependencies
pip install langgraph \
langchain \
langchain-openai \
langchain-community \
langchain-text-splitters \
faiss-cpu \
python-dotenvCode Implementation
import os
import operator
from typing import TypedDict, List
from typing_extensions import Annotated
# LangChain components: Help us handle documents, text-splitting, and vector databases
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain.prompts import ChatPromptTemplate
# LangGraph components: Help us manage the stateful flow of our application
from langgraph.graph import StateGraph, END
# ---------------------------------------------------------------------
# 1. SETUP & CONFIGURATION
# ---------------------------------------------------------------------
# The LLM needs an API key to communicate with OpenAI.
# This line looks for an environment variable, or defaults to your placeholder string.
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "your-api-key-here")
# We declare global variables to hold our vector database and compiled graph.
# This makes it easy for different functions (nodes) to access them during execution.
vector_store = None
rag_app = None
# ---------------------------------------------------------------------
# 2. DOCUMENT INGESTION & PROCESSING
# ---------------------------------------------------------------------
def chunk_documents(documents, chunk_size=1000, chunk_overlap=200):
"""
CONCEPT: Chunking
Large documents (like a 50-page PDF) are too big to feed directly into an LLM at once.
We break them down into smaller, focused 'chunks' (e.g., 1000 characters each).
'chunk_overlap' ensures sentences split at boundaries aren't completely cut in half.
"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
separators=["\n\n", "\n", " ", ""]
)
chunks = text_splitter.split_documents(documents)
print(f"[Ingestion] Successfully cut original documents into {len(chunks)} chunks.")
return chunks
def create_vector_store(chunks):
"""
CONCEPT: Embeddings & Vector Databases
Computers don't understand words; they understand numbers.
1. 'OpenAIEmbeddings' translates our text chunks into lists of numbers (vectors).
2. 'FAISS' is a database optimized to store these vectors.
When a user asks a question, we convert their question into a vector too,
allowing us to mathematically find the chunks with the most 'similar' meaning.
"""
global vector_store
# Initialize the translation model
embeddings_model = OpenAIEmbeddings(model="text-embedding-3-small")
# Process text chunks through the model and save them into the FAISS database
vector_store = FAISS.from_documents(
documents=chunks,
embedding=embeddings_model
)
print("[Database] Vector store created and loaded with chunks successfully.")
return vector_store
# ---------------------------------------------------------------------
# 3. LANGGRAPH STATE DEFINITION
# ---------------------------------------------------------------------
class RAGState(TypedDict):
"""
CONCEPT: Stateful Processing
LangGraph operates like a flowchart. As data moves from one box (Node) to another,
it carries a 'State' dictionary with it. This class defines exactly what keys
are allowed to exist inside that moving package.
Annotated[List[str], operator.add] tells LangGraph: "If multiple nodes write to
the 'context' list, don't overwrite it—instead, append/add the new items to it."
"""
question: str # Stores the user's initial question
context: Annotated[List[str], operator.add] # Stores text chunks retrieved from the DB
answer: str # Stores the final text generated by the LLM
# ---------------------------------------------------------------------
# 4. WORKFLOW GRAPH NODES (The Flowchart Boxes)
# ---------------------------------------------------------------------
def retrieve_node(state: RAGState) -> RAGState:
"""
NODE 1: Retrieval
This function acts as the first step in our pipeline flowchart.
It reads the user's question from the shared 'state', looks up the top 3 most
relevant text chunks in our FAISS database, and updates the state's context list.
"""
user_question = state["question"]
# Search the vector database for the 3 most relevant context document chunks
relevant_docs = vector_store.similarity_search(user_question, k=3)
# Extract just the raw text strings out of those document objects
extracted_text_chunks = [doc.page_content for doc in relevant_docs]
print(f"[Node: Retrieve] Searched DB for: '{user_question}'. Found {len(extracted_text_chunks)} chunks.")
# Return updates to the state. This dictionary merges with the existing graph state.
return {
"context": extracted_text_chunks,
"question": user_question
}
def generate_node(state: RAGState) -> RAGState:
"""
NODE 2: Generation
This function acts as the second step in our pipeline flowchart.
It takes the context chunks we just found in Node 1, combines them with the
original question inside a prompt template, and sends them to the LLM to get a response.
"""
# Initialize our AI model (gpt-4o-mini). Temperature=0 keeps answers strict and factual.
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Pull the required information out of the active graph state
user_question = state["question"]
retrieved_chunks = state["context"]
# Instruct the AI how to behave and feed it our gathered background information
prompt_template = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant. Use the following pieces of retrieved context to answer the question. If you don't know the answer, say you don't know.\n\nContext:\n{context}"),
("human", "{question}")
])
# Join our array of text chunks into one large block of text
combined_context_text = "\n\n".join(retrieved_chunks)
# Pipe the prompt instructions directly into the LLM model
ai_chain = prompt_template | llm
# Invoke (run) the chain
print("[Node: Generate] Sending prompt and context to the LLM...")
response = ai_chain.invoke({"context": combined_context_text, "question": user_question})
# Save the LLM's text output directly into the state's 'answer' key
return {"answer": response.content}
# ---------------------------------------------------------------------
# 5. PIPELINE BUILDER (Connecting the Flowchart Arrows)
# ---------------------------------------------------------------------
def build_rag_graph():
"""
CONCEPT: Graph Construction
Here we define our graph framework, register our functions as valid execution nodes,
and draw the connection lines (edges) instructing data exactly where to go next.
"""
# 1. Initialize a blank graph map that enforces our RAGState structure
workflow = StateGraph(RAGState)
# 2. Register our Python functions as computational boxes (nodes) in the graph
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
# 3. Set the absolute starting point of our pipeline flow
workflow.set_entry_point("retrieve")
# 4. Draw a directional arrow: Once "retrieve" finishes, automatically run "generate"
workflow.add_edge("retrieve", "generate")
# 5. Draw the final arrow: Once "generate" finishes, stop the program entirely (END)
workflow.add_edge("generate", END)
# 6. Compile translates this structural map design into an executable application runtime
return workflow.compile()
# ---------------------------------------------------------------------
# 6. PIPELINE ORCHESTRATION & RUNNERS
# ---------------------------------------------------------------------
def ask_question(question: str):
"""
Helper function to cleanly pass a string question into our compiled graph application.
"""
global rag_app
print(f"\n{'='*60}\n[Pipeline Input] User Question: {question}")
# .invoke() kicks off the graph engine, passing in our initial starting state values
final_output_state = rag_app.invoke({
"question": question,
"context": [], # Starts empty; will be populated dynamically by retrieve_node
"answer": "" # Starts empty; will be populated dynamically by generate_node
})
print(f"\n[Pipeline Output] Final Answer:\n{final_output_state['answer']}\n{'='*60}")
return final_output_state
def complete_rag_pipeline(documents, questions):
"""
The master function that strings the entire workflow lifecycle together.
Loads -> Chunks -> Embeds/Stores -> Connects Graph Workflow -> Queries.
"""
global rag_app
# Step 1: Chunk documents down
chunks = chunk_documents(documents)
# Step 2: Convert chunks into math vectors and store in FAISS database
create_vector_store(chunks)
# Step 3: Build and compile the multi-step LangGraph workflow architecture
rag_app = build_rag_graph()
# Step 4: Loop through and evaluate our sample questions against the operational graph
for question in questions:
ask_question(question)
return rag_app
# ---------------------------------------------------------------------
# 7. AUTOMATED SCRIPT VERIFICATION TEST
# ---------------------------------------------------------------------
if __name__ == "__main__":
from langchain_core.documents import Document
print("--- Running Script Workflow Verification ---")
# Instead of reading from external system files (which can throw "File Not Found" errors),
# we create mock documents directly in memory so the script runs successfully out of the box.
mock_documents = [
Document(page_content="LangGraph is a framework for building stateful, multi-step AI applications using graphs.", metadata={"source": "doc1"}),
Document(page_content="RAG (Retrieval-Augmented Generation) combines external document retrieval with LLM generations.", metadata={"source": "doc2"})
]
test_questions = ["What is LangGraph?"]
# Validation check to ensure you don't run into authentication errors with OpenAI
if os.environ["OPENAI_API_KEY"] == "your-api-key-here":
print("\n[Execution Paused]: Please provide a valid OpenAI API Key on Line 17 before testing.")
else:
# Run the full pipeline process smoothly
complete_rag_pipeline(mock_documents, test_questions)What Happens Internally?
When the user asks:
What is RAG?
The system performs these steps:
Receive the question.
Convert the question into an embedding.
Search the vector database.
Retrieve the most relevant chunks.
Build a prompt using those chunks.
Send the prompt to GPT.
Return the generated answer.
The LLM never searches documents directly.
It only answers using the retrieved context.
Why Use LangGraph?
Although this example has only two nodes, LangGraph scales naturally as your application grows.
You can later add nodes for:
Query rewriting
Web search
Tool calling
Memory
Multi-agent workflows
Human approval
Guardrails
Reranking
Evaluation
Logging
Without changing the overall architecture.
What’s Next?
Once you’re comfortable with this minimal implementation, explore production-ready enhancements such as:
PDF loaders
Chroma or Pinecone
Hybrid search (BM25 + vectors)
Reranking models
Metadata filtering
Parent-child retrieval
Context compression
Streaming responses
Conversation memory
Agentic RAG
Mastering this simple pipeline first makes those advanced techniques much easier to understand.
