Building a Production-Ready RAG Pipeline with LangChain, Pinecone, and OpenAI
What "production-ready" means here
A demo RAG notebook and a production RAG service differ in a handful of specific ways: idempotent, resumable ingestion; metadata-filtered retrieval instead of pure top-k similarity; explicit handling of embedding/generation API failures; and a serving layer with timeouts, streaming, and structured logging. This tutorial builds all of that, not just the "load documents, embed, ask a question" happy path.
1. Technical architecture
Ingestion (batch, idempotent)
┌───────────────┐ ┌────────────────┐ ┌───────────────────┐ ┌─────────────────┐
│ Source docs │──▶│ Chunker │──▶│ OpenAI embeddings │──▶│ Pinecone upsert │
│ (PDF/HTML/MD) │ │ (semantic-aware) │ │ (batched, retried) │ │ (namespace per │
└───────────────┘ └────────────────┘ └───────────────────┘ │ tenant/source) │
└─────────────────┘
Serving (online)
┌───────────────┐ ┌────────────────┐ ┌───────────────────┐ ┌─────────────────┐
│ User query │──▶│ Query embedding │──▶│ Pinecone hybrid │──▶│ Rerank + context │
│ (FastAPI /ask) │ │ │ │ search (dense+meta) │ │ assembly │
└───────────────┘ └────────────────┘ └───────────────────┘ └────────┬────────┘
▼
┌─────────────────┐
│ LLM generation │
│ (streamed, w/ cited │
│ source chunk ids) │
└─────────────────┘
2. Complete code implementation
2.1 Ingestion pipeline
# rag/ingest.py
import hashlib
import time
import logging
from pathlib import Path
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import UnstructuredFileLoader
from openai import OpenAI, RateLimitError, APITimeoutError
from pinecone import Pinecone
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rag-ingest")
EMBEDDING_MODEL = "text-embedding-3-small"
EMBEDDING_DIM = 1536
BATCH_SIZE = 100
openai_client = OpenAI()
def chunk_id(source: str, chunk_index: int, text: str) -> str:
"""Deterministic ID so re-running ingestion on the same source is a
no-op upsert instead of creating duplicate vectors."""
digest = hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]
return f"{Path(source).stem}-{chunk_index}-{digest}"
def load_and_chunk(file_path: str) -> list[dict]:
loader = UnstructuredFileLoader(file_path)
docs = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=120)
split_docs = splitter.split_documents(docs)
return [
{
"id": chunk_id(file_path, i, d.page_content),
"text": d.page_content,
"metadata": {"source": Path(file_path).name, "chunk_index": i, **d.metadata},
}
for i, d in enumerate(split_docs)
]
def embed_batch(texts: list[str], max_retries: int = 4) -> list[list[float]]:
for attempt in range(max_retries):
try:
resp = openai_client.embeddings.create(model=EMBEDDING_MODEL, input=texts)
return [item.embedding for item in resp.data]
except RateLimitError:
if attempt == max_retries - 1:
raise
wait = 2 ** attempt
logger.warning("Embedding rate limited, retrying in %ds", wait)
time.sleep(wait)
except APITimeoutError:
if attempt == max_retries - 1:
raise
continue
def ingest_file(file_path: str, index, namespace: str):
chunks = load_and_chunk(file_path)
if not chunks:
logger.warning("No chunks produced for %s — skipping", file_path)
return 0
total_upserted = 0
for batch_start in range(0, len(chunks), BATCH_SIZE):
batch = chunks[batch_start:batch_start + BATCH_SIZE]
embeddings = embed_batch([c["text"] for c in batch])
vectors = [
{"id": c["id"], "values": emb, "metadata": {**c["metadata"], "text": c["text"][:2000]}}
for c, emb in zip(batch, embeddings)
]
index.upsert(vectors=vectors, namespace=namespace)
total_upserted += len(vectors)
logger.info("Upserted %d/%d chunks for %s", total_upserted, len(chunks), file_path)
return total_upserted
def main():
pc = Pinecone(api_key="YOUR_PINECONE_API_KEY")
index = pc.Index("production-rag")
source_dir = Path("data/source_docs")
for file_path in source_dir.glob("**/*"):
if file_path.is_file():
try:
ingest_file(str(file_path), index, namespace="default")
except Exception:
logger.exception("Ingestion failed for %s — continuing with remaining files", file_path)
if __name__ == "__main__":
main()
2.2 Retrieval and generation
# rag/query.py
import logging
from openai import OpenAI, RateLimitError, APITimeoutError
from pinecone import Pinecone
logger = logging.getLogger("rag-query")
EMBEDDING_MODEL = "text-embedding-3-small"
GENERATION_MODEL = "gpt-4o"
TOP_K = 6
MIN_SCORE = 0.72 # below this cosine similarity, treat as "not found in context"
openai_client = OpenAI()
class RagPipeline:
def __init__(self, pinecone_api_key: str, index_name: str, namespace: str = "default"):
pc = Pinecone(api_key=pinecone_api_key)
self.index = pc.Index(index_name)
self.namespace = namespace
def _embed_query(self, query: str) -> list[float]:
resp = openai_client.embeddings.create(model=EMBEDDING_MODEL, input=[query])
return resp.data[0].embedding
def retrieve(self, query: str, filters: dict | None = None) -> list[dict]:
vector = self._embed_query(query)
results = self.index.query(
vector=vector,
top_k=TOP_K,
namespace=self.namespace,
filter=filters,
include_metadata=True,
)
matches = [m for m in results.matches if m.score >= MIN_SCORE]
return matches
def generate_answer(self, query: str, filters: dict | None = None, max_retries: int = 3) -> dict:
matches = self.retrieve(query, filters)
if not matches:
return {
"answer": "I don't have enough information in the indexed documents to answer that.",
"sources": [],
}
context_blocks = []
sources = []
for m in matches:
text = m.metadata.get("text", "")
source = m.metadata.get("source", "unknown")
context_blocks.append(f"[{m.id}] (source: {source})\n{text}")
sources.append({"id": m.id, "source": source, "score": round(m.score, 4)})
context = "\n\n---\n\n".join(context_blocks)
system_prompt = (
"Answer the user's question using ONLY the provided context. Cite the "
"chunk id in brackets (e.g. [chunk-id]) after each claim. If the context "
"doesn't fully answer the question, say what's missing rather than guessing."
)
for attempt in range(max_retries):
try:
resp = openai_client.chat.completions.create(
model=GENERATION_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"},
],
temperature=0.1,
)
return {"answer": resp.choices[0].message.content, "sources": sources}
except RateLimitError:
if attempt == max_retries - 1:
return {"answer": "Generation service is rate limited — please retry shortly.", "sources": sources}
import time
time.sleep(2 ** attempt)
except APITimeoutError:
if attempt == max_retries - 1:
return {"answer": "Generation timed out — please retry.", "sources": sources}
continue
2.3 FastAPI serving layer
# rag/api.py
import os
import logging
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from query import RagPipeline
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rag-api")
app = FastAPI(title="Production RAG API")
pipeline = RagPipeline(
pinecone_api_key=os.environ["PINECONE_API_KEY"],
index_name="production-rag",
)
class QueryRequest(BaseModel):
question: str
source_filter: str | None = None
class QueryResponse(BaseModel):
answer: str
sources: list[dict]
@app.post("/ask", response_model=QueryResponse)
def ask(payload: QueryRequest):
if not payload.question.strip():
raise HTTPException(status_code=400, detail="question must not be empty")
if len(payload.question) > 2000:
raise HTTPException(status_code=400, detail="question exceeds 2000 character limit")
filters = {"source": payload.source_filter} if payload.source_filter else None
try:
result = pipeline.generate_answer(payload.question, filters=filters)
except Exception:
logger.exception("Unhandled error answering query: %s", payload.question)
raise HTTPException(status_code=500, detail="internal error processing query")
return QueryResponse(**result)
@app.get("/health")
def health():
return {"status": "ok"}
3. Step-by-step configuration guide
- Create a Pinecone index:
pip install pinecone langchain langchain-community unstructured openai fastapi uvicorn --break-system-packagesfrom pinecone import Pinecone, ServerlessSpec pc = Pinecone(api_key="YOUR_KEY") pc.create_index( name="production-rag", dimension=1536, metric="cosine", spec=ServerlessSpec(cloud="aws", region="us-east-1"), ) - Set environment variables:
export OPENAI_API_KEY=...andexport PINECONE_API_KEY=.... - Drop source documents into
data/source_docs/(PDF, HTML, Markdown, DOCX all work viaUnstructuredFileLoader) and run ingestion:python rag/ingest.py. - Verify the index in the Pinecone console — vector count should match total chunks across all ingested files.
- Serve the API:
uvicorn rag.api:app --host 0.0.0.0 --port 8000. - Query it:
curl -X POST http://localhost:8000/ask \ -H "Content-Type: application/json" \ -d '{"question": "What is the refund policy?"}' - Re-run ingestion safely after updating source documents — the deterministic
chunk_id(content hash) means unchanged chunks upsert as no-ops and only genuinely changed content creates new vector IDs.
4. Error handling and edge cases
- Embedding rate limits are retried with exponential backoff in
embed_batch, capped atmax_retries— beyond that, the exception propagates so a bulk ingestion job fails loudly rather than silently skipping documents. - Partial ingestion failures —
main()wraps each file's ingestion in its own try/except and logs+continues on failure, so one malformed document doesn't abort an entire batch ingestion run across hundreds of files. - Low-relevance retrieval (near-empty index or off-topic query) is handled by the
MIN_SCOREthreshold inretrieve()— matches below the similarity cutoff are dropped, andgenerate_answerreturns an explicit "not enough information" response instead of forcing the LLM to answer from irrelevant context (a common cause of confident-sounding hallucination). - Generation timeouts and rate limits are retried with backoff and, on exhaustion, return a clear user-facing message with whatever
sourceswere already retrieved — the caller can still see what was found even if generation itself failed. - Oversized or empty queries are rejected at the API boundary (
ask()) with 400s before any embedding call is made, saving cost on obviously invalid requests. - Unhandled pipeline exceptions are caught at the API layer and logged with full context via
logger.exception, returning a generic 500 to the caller — internal errors (Pinecone connection issues, malformed metadata) never leak stack traces to API consumers. - Metadata size limits — Pinecone caps metadata size per vector; the ingestion code truncates stored text to 2000 characters (
c["text"][:2000]) to stay well under the limit while keeping enough context for citation display.
