If you need a retrieval-augmented generation pipeline running on your laptop today, langchain faiss local rag is the fastest path to a working prototype. FAISS runs in-process with no external dependencies, LangChain handles the orchestration, and you can swap the vector store later without rewriting your application logic. This guide walks through a complete, runnable implementation from document ingestion through answer generation.
Step 1: Install the minimal dependency set
Keep the environment lean. You need LangChain’s core packages, the FAISS integration, an embedding model, and a local LLM runtime. I use sentence-transformers for embeddings and llama-cpp-python for the generator, but you can substitute any OpenAI-compatible endpoint.
pip install --quiet \
langchain \
langchain-community \
langchain-huggingface \
faiss-cpu \
sentence-transformers \
llama-cpp-pythonIf you’re on Apple Silicon, faiss-cpu works natively. For GPU acceleration on Linux, install faiss-gpu instead and ensure your CUDA toolkit matches the PyTorch build.
Verify: Run python -c "import faiss; print(faiss.__version__)" — you should see a version string like 1.8.0 without errors.
Step 2: Prepare source documents
For a prototype, keep the corpus small and representative. Create a data/ directory with markdown or text files that reflect your actual domain — API docs, internal runbooks, product specs. Avoid PDFs for now; they add parsing complexity without teaching you anything about the RAG pipeline.
mkdir -p data
cat > data/return-policy.md << 'EOF'
# Return Policy
Items may be returned within 30 days of delivery for a full refund.
Items must be in original packaging with all accessories.
Refunds are processed to the original payment method within 5 business days.
Final sale items (marked "Final Sale") cannot be returned.
EOF
cat > data/shipping.md << 'EOF'
# Shipping Information
Standard shipping: 3-5 business days, free on orders over $50.
Express shipping: 1-2 business days, $12.99 flat rate.
International shipping: 7-14 business days, calculated at checkout.
Orders placed before 2 PM EST ship same business day.
EOFVerify: ls data/ shows your files. cat data/*.md renders cleanly.
Step 3: Load and chunk documents
LangChain’s document loaders and text splitters handle the ingestion. Use RecursiveCharacterTextSplitter — it respects markdown structure and falls back gracefully. Chunk size depends on your embedding model’s context window; 500 tokens with 50 overlap works well for all-MiniLM-L6-v2 (384 dimensions, 256 token max).
# ingest.py
from pathlib import Path
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
DATA_DIR = Path("data")
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
def load_documents():
docs = []
for file_path in DATA_DIR.glob("*.md"):
loader = TextLoader(str(file_path), encoding="utf-8")
docs.extend(loader.load())
return docs
def chunk_documents(docs):
splitter = RecursiveCharacterTextSplitter(
chunk_size=CHUNK_SIZE,
chunk_overlap=CHUNK_OVERLAP,
separators=["\n## ", "\n### ", "\n# ", "\n\n", "\n", " ", ""],
length_function=len,
)
return splitter.split_documents(docs)
if __name__ == "__main__":
raw_docs = load_documents()
chunks = chunk_documents(raw_docs)
print(f"Loaded {len(raw_docs)} documents, split into {len(chunks)} chunks")
for i, chunk in enumerate(chunks[:3]):
print(f"\n--- Chunk {i} ---")
print(chunk.page_content[:200])Run it: python ingest.py. You should see 2 documents split into roughly 6-8 chunks depending on content length.
Step 4: Create the FAISS index with embeddings
This is where the langchain faiss local rag pipeline materializes. Initialize the embedding model, embed each chunk, and build the FAISS index in memory. Persist it to disk so you don’t re-embed on every restart.
# build_index.py
from pathlib import Path
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from ingest import load_documents, chunk_documents
INDEX_DIR = Path("faiss_index")
EMBEDDING_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
def build_index():
print("Loading documents...")
raw_docs = load_documents()
chunks = chunk_documents(raw_docs)
print(f"Initializing embeddings: {EMBEDDING_MODEL}")
embeddings = HuggingFaceEmbeddings(
model_name=EMBEDDING_MODEL,
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True},
)
print("Building FAISS index...")
vectorstore = FAISS.from_documents(chunks, embeddings)
print(f"Saving index to {INDEX_DIR}")
INDEX_DIR.mkdir(exist_ok=True)
vectorstore.save_local(str(INDEX_DIR))
print(f"Index built: {vectorstore.index.ntotal} vectors, dimension {vectorstore.index.d}")
return vectorstore
if __name__ == "__main__":
build_index()Run python build_index.py. The output confirms vector count and dimension (384 for MiniLM). The faiss_index/ directory now contains index.faiss and index.pkl.
Verify: ls -la faiss_index/ shows both files. You can also inspect the index directly:
# verify_index.py
import faiss
index = faiss.read_index("faiss_index/index.faiss")
print(f"Vectors: {index.ntotal}, Dimension: {index.d}, Metric: {faiss.MetricType(index.metric_type).name}")Step 5: Configure the retriever
The retriever is your query interface. For a prototype, start with similarity search (k=4) and add a score threshold to filter weak matches. LangChain’s as_retriever() returns a Runnable that plugs directly into chains.
# retriever.py
from pathlib import Path
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
INDEX_DIR = Path("faiss_index")
EMBEDDING_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
def get_retriever(k=4, score_threshold=0.5):
embeddings = HuggingFaceEmbeddings(
model_name=EMBEDDING_MODEL,
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True},
)
vectorstore = FAISS.load_local(
str(INDEX_DIR),
embeddings,
allow_dangerous_deserialization=True, # required for FAISS pickles
)
return vectorstore.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={"k": k, "score_threshold": score_threshold},
)
if __name__ == "__main__":
retriever = get_retriever()
test_queries = [
"What is the return window?",
"How long does express shipping take?",
"Can I return final sale items?",
]
for query in test_queries:
print(f"\nQuery: {query}")
docs = retriever.invoke(query)
print(f"Retrieved {len(docs)} chunks")
for i, doc in enumerate(docs):
print(f" [{i}] {doc.page_content[:150]}...")Run python retriever.py. Each query should return 1-3 relevant chunks. If a query returns zero chunks, lower score_threshold or increase k.
Step 6: Wire the local LLM
Download a quantized GGUF model — Llama-3.2-3B-Instruct-Q4_K_M.gguf (~2 GB) runs comfortably on 8 GB RAM. Place it in models/. Configure llama-cpp-python with a context window large enough for your prompt plus retrieved context (4096 is safe for this prototype).
# llm.py
from pathlib import Path
from langchain_community.llms import LlamaCpp
MODEL_PATH = Path("models/Llama-3.2-3B-Instruct-Q4_K_M.gguf")
N_CTX = 4096
N_GPU_LAYERS = -1 # -1 = offload all possible layers to GPU (Metal on macOS, CUDA on Linux)
TEMPERATURE = 0.1
MAX_TOKENS = 512
def get_llm():
if not MODEL_PATH.exists():
raise FileNotFoundError(f"Model not found at {MODEL_PATH}. Download a GGUF model first.")
return LlamaCpp(
model_path=str(MODEL_PATH),
n_ctx=N_CTX,
n_gpu_layers=N_GPU_LAYERS,
temperature=TEMPERATURE,
max_tokens=MAX_TOKENS,
verbose=False,
streaming=True,
)
if __name__ == "__main__":
llm = get_llm()
response = llm.invoke("Reply with only the word 'ready'")
print(f"LLM test: {response.strip()}")Run python llm.py — you should see ready (or similar) confirming the model loads and generates.
Step 7: Assemble the RAG chain
LangChain’s LCEL (LangChain Expression Language) makes the pipeline explicit. The chain: retrieve → format context → prompt → generate → parse. Use a strict prompt that forces the model to cite sources and refuse when context is insufficient.
# rag_chain.py
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from retriever import get_retriever
from llm import get_llm
PROMPT_TEMPLATE = """You are a customer support assistant. Answer the question using ONLY the provided context.
If the context does not contain the answer, say "I don't have enough information to answer that."
Cite the relevant context sections by referencing their content.
Context:
{context}
Question: {question}
Answer:"""
def format_docs(docs):
"""Format retrieved documents into a single context string."""
if not docs:
return "No relevant context found."
formatted = []
for i, doc in enumerate(docs):
source = doc.metadata.get("source", "unknown")
formatted.append(f"[Source {i+1}: {source}]\n{doc.page_content}")
return "\n\n---\n\n".join(formatted)
def build_rag_chain():
retriever = get_retriever(k=4, score_threshold=0.5)
llm = get_llm()
prompt = ChatPromptTemplate.from_template(PROMPT_TEMPLATE)
chain = (
{"context": retriever | RunnableLambda(format_docs), "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
return chain
if __name__ == "__main__":
chain = build_rag_chain()
test_questions = [
"What is the return policy timeframe?",
"How much does express shipping cost?",
"Can I return a final sale item if it's defective?",
"What's the phone number for customer support?", # not in context
]
for question in test_questions:
print(f"\n{'='*60}")
print(f"Q: {question}")
print(f"{'='*60}")
print("A: ", end="", flush=True)
for token in chain.stream(question):
print(token, end="", flush=True)
print()Run python rag_chain.py. You should see streamed answers for the first three questions, and a refusal for the fourth. The output demonstrates the complete langchain faiss local rag flow: query → retrieval → context formatting → generation.
Step 8: Add a simple CLI for iteration
A REPL lets you test queries rapidly without re-running the script. Wrap the chain in a loop with basic commands.
# cli.py
from rag_chain import build_rag_chain
def main():
print("Loading RAG chain...")
chain = build_rag_chain()
print("Ready. Type 'exit' to quit, 'help' for commands.\n")
while True:
try:
query = input("❯ ").strip()
except (EOFError, KeyboardInterrupt):
print("\nGoodbye.")
break
if not query:
continue
if query.lower() in ("exit", "quit"):
break
if query.lower() == "help":
print("Commands: exit, help, or just ask a question")
continue
print()
for token in chain.stream(query):
print(token, end="", flush=True)
print("\n")
if __name__ == "__main__":
main()Run python cli.py and test interactively.
Step 9: Evaluate retrieval quality systematically
Before trusting the prototype, measure retrieval precision. Create a small eval set of (question, expected_source_file) pairs and compute hit rate.
# eval_retrieval.py
from retriever import get_retriever
EVAL_SET = [
("What is the return window?", "return-policy.md"),
("How long for express shipping?", "shipping.md"),
("Final sale return policy?", "return-policy.md"),
("International shipping time?", "shipping.md"),
("Refund processing time?", "return-policy.md"),
]
def evaluate():
retriever = get_retriever(k=4, score_threshold=0.3)
hits = 0
for question, expected_source in EVAL_SET:
docs = retriever.invoke(question)
retrieved_sources = {doc.metadata.get("source", "") for doc in docs}
hit = any(expected_source in src for src in retrieved_sources)
status = "✓" if hit else "✗"
print(f"{status} Q: {question}")
print(f" Expected: {expected_source}, Got: {retrieved_sources}")
if hit:
hits += 1
print(f"\nHit rate: {hits}/{len(EVAL_SET)} = {hits/len(EVAL_SET):.0%}")
if __name__ == "__main__":
evaluate()Run python eval_retrieval.py. Aim for >80% hit rate on your eval set. If it’s lower, adjust chunk size, overlap, k, or score_threshold and re-run.
Step 10: Persist and share the prototype
The faiss_index/ directory is portable. Zip it with your data/ and models/ (or document the model download URL) and a colleague can run the CLI immediately. For a team setting, consider serving the chain via FastAPI:
# server.py
from fastapi import FastAPI
from pydantic import BaseModel
from rag_chain import build_rag_chain
app = FastAPI(title="Local RAG Prototype")
chain = build_rag_chain()
class Query(BaseModel):
question: str
class Answer(BaseModel):
answer: str
@app.post("/ask", response_model=Answer)
async def ask(query: Query):
answer = chain.invoke(query.question)
return Answer(answer=answer)
# Run: uvicorn server:app --reload --port 8080This exposes POST /ask with JSON body {"question": "..."}. Test with curl -X POST localhost:8080/ask -H "Content-Type: application/json" -d '{"question": "What is the return policy?"}'.
What to swap when you graduate from prototype
FAISS is excellent for local development but has limits: no concurrent writes, no built-in metadata filtering beyond what you implement, single-machine scale. When you need multi-tenancy, horizontal scaling, or managed operations, swap the vector store. The retriever interface stays the same — change FAISS.from_documents to Pinecone.from_documents, Chroma.from_documents, or Weaviate.from_documents and update the connection config.
If you’re already routing inference through a gateway like n4n.ai, you can also swap the local LlamaCpp LLM for any of the 240+ models behind the same OpenAI-compatible endpoint — just change the base URL and model name in the LLM initializer. The RAG chain code doesn’t change.
Common pitfalls
Embedding dimension mismatch: If you switch embedding models, rebuild the index. FAISS stores vectors raw; it doesn’t know the model that produced them.
Score threshold too high: The default cosine similarity on normalized vectors ranges [-1, 1]. A threshold of 0.5 is reasonable for MiniLM but may need tuning per domain.
Context window overflow: With k=4 and 500-token chunks, you’re sending ~2000 tokens of context plus prompt. Leave headroom for generation. Monitor llm.invoke calls for truncation warnings.
Pickle deserialization warning: allow_dangerous_deserialization=True is required for FAISS because it pickles the InMemoryDocstore. Only load indexes you created.
Metal/GPU layers: On macOS, n_gpu_layers=-1 uses Metal. On Linux with CUDA, it offloads to GPU. If you see crashes, set n_gpu_layers=0 to force CPU.
You now have a working langchain faiss local rag prototype that ingests documents, builds a vector index, retrieves relevant context, and generates cited answers — all locally, no API keys required. The code is structured so each component can be replaced independently when requirements change. Start here, measure, then scale the pieces that need it.