Project: A Production RAG System with Pinecone

Project description

This is module 7's capstone project and the operational close of your journey through advanced RAG: taking the system you built across modules 2-6 (intelligent chunking, query optimization, hybrid search, re-ranking, metadata filtering) and migrating it from a local development environment to a production architecture on Pinecone serverless with FastAPI, secure multi-tenant caching, and observability.

The goal isn't "use Pinecone" as a technical checklist item. It's to prove you can make infrastructure decisions with data: when to migrate, when not to, how to measure the impact, how to isolate tenants without leaking data between them, and how to deliver a system an engineering team can operate without you being awake at 3am.

By the end you'll have a public repository with modular code, automated tests, a reproducible benchmark, and a technical decision document you can show to an architecture committee or a senior interviewer. It's the portfolio piece that separates an engineer who "read about RAG" from one who "operated RAG in production".

---

The project's objective

Implement a production-grade RAG system on Pinecone with six operational dimensions:

1. A validated migration from ChromaDB with no data loss and no quality degradation
2. Guaranteed multi-tenant isolation via namespaces + TenantContext + automated tests
3. The advanced pipeline preserved: query expansion + hybrid search + re-ranking + metadata filtering
4. An HTTP API with FastAPI: /ask, /health, /metrics endpoints ready for Kubernetes
5. Secure caching that respects the isolation between tenants
6. A benchmark + cost analysis documented with a defensible final recommendation

---

Technical specifications

The mandatory stack

Component	Technology	Why
Vector DB	Pinecone serverless	A managed backend, auto-scaling
Embeddings	OpenAI text-embedding-3-small	The industry standard, 1536 dims
Generation	OpenAI gpt-4o-mini	Balanced cost/quality
API	FastAPI + uvicorn	Native async, automatic OpenAPI
Cache	Redis or in-memory	Reduce redundant read units
Validation	Pydantic v2	Schema enforcement
Observability	structlog + Prometheus	JSON logs + metrics
Tests	pytest + pytest-asyncio	The Python standard
BM25 (optional)	rank-bm25	If you keep the hybrid local

Project setup

mkdir production-rag && cd production-rag
python -m venv venv && source venv/bin/activate
pip install pinecone openai fastapi uvicorn[standard] pydantic redis structlog \
            python-dotenv pytest pytest-asyncio httpx

A suggested directory structure:

production-rag/
├── app/
│   ├── __init__.py
│   ├── main.py              # The FastAPI app
│   ├── config.py            # Pydantic Settings
│   ├── tenant.py            # TenantContext, secure_query
│   ├── retriever.py         # ProductionRetriever
│   ├── pipeline.py          # query → expand → retrieve → rerank → generate
│   ├── cache.py             # The cache layer with tenant isolation
│   ├── metadata.py          # DocMetadata, FilterSpec
│   └── observability.py     # logging, metrics
├── scripts/
│   ├── migrate_from_chroma.py
│   ├── benchmark.py
│   └── seed_test_data.py
├── tests/
│   ├── test_isolation.py
│   ├── test_filters.py
│   └── test_pipeline.py
├── BENCHMARK.md             # The final report
├── DECISION.md              # The technical decision document
├── docker-compose.yml       # Redis + the app
└── pyproject.toml

---

Mandatory features

1) Idempotent index creation (scripts/setup_index.py)

from pinecone import Pinecone, ServerlessSpec
import time

def ensure_production_index(pc: Pinecone, name: str, dimension: int = 1536) -> None:
    existing = [i["name"] for i in pc.list_indexes()]
    if name in existing:
        info = pc.describe_index(name)
        assert info.dimension == dimension, f"Dimension mismatch: {info.dimension} vs {dimension}"
        return
    pc.create_index(
        name=name,
        dimension=dimension,
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1"),
    )
    while not pc.describe_index(name).status["ready"]:
        time.sleep(1)

2) The migration from ChromaDB (scripts/migrate_from_chroma.py)

• Paginated reads from ChromaDB
• Validation with the Pydantic DocMetadata before the upsert
• Upserts in batches of 100 vectors
• Exponential retry on transient errors (429, 5xx)
• A checkpoint on disk so you can resume
• A final validation: assert chroma_count == pinecone_namespace_count

3) Multi-tenant isolation (app/tenant.py)

from dataclasses import dataclass
import re

VALID_TENANT_ID = re.compile(r"^[a-z0-9_-]{1,40}$")

@dataclass(frozen=True)
class TenantContext:
    tenant_id: str
    role: str

    def __post_init__(self):
        if not VALID_TENANT_ID.match(self.tenant_id):
            raise ValueError(f"Invalid tenant_id: {self.tenant_id}")

def tenant_namespace(tenant_id: str) -> str:
    if not VALID_TENANT_ID.match(tenant_id):
        raise ValueError(f"Invalid tenant_id: {tenant_id}")
    return f"workspace-{tenant_id}"

def secure_query(index, vector, tenant: TenantContext, filter_dict: dict | None = None, top_k: int = 10):
    return index.query(
        vector=vector,
        top_k=top_k,
        namespace=tenant_namespace(tenant.tenant_id),
        filter=filter_dict,
        include_metadata=True,
    )

4) The complete RAG pipeline (app/pipeline.py)

It chains the components you already built:

async def rag_pipeline(query: str, tenant: TenantContext, filter_spec: FilterSpec) -> dict:
    expanded = await expand_query(query)         # M03 query optimization
    candidates = await hybrid_retrieve(expanded, tenant, filter_spec, top_k=50)  # M05 hybrid
    reranked = await rerank(query, candidates, top_k=10)  # M04 reranking
    answer = await generate_answer(query, reranked)
    return {"answer": answer, "sources": reranked}

5) The cache layer with isolation (app/cache.py)

import hashlib
import json
from redis.asyncio import Redis

class TenantAwareCache:
    def __init__(self, redis: Redis, ttl_seconds: int = 300):
        self.redis = redis
        self.ttl = ttl_seconds

    def _key(self, tenant_id: str, query: str, filter_dict: dict | None) -> str:
        payload = json.dumps({"q": query, "f": filter_dict or {}}, sort_keys=True)
        digest = hashlib.sha256(payload.encode()).hexdigest()[:16]
        return f"rag:{tenant_id}:{digest}"

    async def get(self, tenant: TenantContext, query: str, filter_dict: dict | None):
        raw = await self.redis.get(self._key(tenant.tenant_id, query, filter_dict))
        return json.loads(raw) if raw else None

    async def set(self, tenant: TenantContext, query: str, filter_dict: dict | None, value: dict):
        await self.redis.setex(
            self._key(tenant.tenant_id, query, filter_dict),
            self.ttl,
            json.dumps(value),
        )

The critical point: the key always includes the tenant_id. Without it, an identical query from two different tenants would share a cache entry → a data leak. This gets tested in tests/test_isolation.py.

6) The FastAPI API (app/main.py)

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel

app = FastAPI(title="Production RAG")

class AskRequest(BaseModel):
    query: str
    types: list[str] | None = None
    tags: list[str] | None = None

class AskResponse(BaseModel):
    answer: str
    sources: list[dict]
    cached: bool

@app.post("/ask", response_model=AskResponse)
async def ask(req: AskRequest, tenant: TenantContext = Depends(get_tenant_from_jwt)):
    spec = FilterSpec(doc_types=req.types, tags_any=req.tags)
    cached = await cache.get(tenant, req.query, spec.build())
    if cached:
        return AskResponse(**cached, cached=True)
    result = await rag_pipeline(req.query, tenant, spec)
    await cache.set(tenant, req.query, spec.build(), result)
    return AskResponse(**result, cached=False)

@app.get("/health")
def health():
    return {"status": "ok"}

@app.get("/metrics")
def metrics():
    return prometheus_client.generate_latest()

7) The benchmark and the costs

Deliver a BENCHMARK.md following capsule 07's methodology. At minimum:

• A p50/p95/p99 table, ChromaDB vs Pinecone, with 1000+ queries
• Throughput at concurrency=[1, 10, 50]
• Recall@10 against a ground truth of 100 queries
• A 12-month cost estimate with the projected growth

---

Validations and error handling

The list of invariants your system must guarantee and your tests must cover:

• Every query requires a valid tenant_id (the regex ^[a-z0-9_-]{1,40}$)
• Every Pinecone query goes through secure_query (no direct calls to index.query)
• The embedding's dimension is validated before the upsert (1536 for OpenAI small)
• Exponential retry with jitter on the upsert (3 attempts: 1s, 2s, 4s + jitter)
• A count mismatch after the migration blocks the deploy
• The cache key includes the tenant_id and a deterministic hash of query+filter
• Every log line includes tenant_id, query_id, latency_ms as structured JSON
• /metrics exposes per-tenant counters: total queries, errors, p95 latency

---

Success criteria

• ✅ A complete migration with no data loss (count validation + a stable recall@10)
• ✅ p95 improves ≥30% over the local ChromaDB baseline at concurrency ≥10
• ✅ The isolation tests pass: there's no case where tenant A sees tenant B's data
• ✅ FastAPI runs in Docker with a green health check
• ✅ The cache cuts latency ≥50% for repeated queries with no leakage between tenants
• ✅ BENCHMARK.md and DECISION.md are versioned in git with real benchmark data

---

Grading rubric (100 points)

Functionality (50 pts)

• (15 pts) A robust, validated migration (the script + checkpoint + count assertion)
• (10 pts) The complete RAG pipeline: expand + hybrid + rerank + filter
• (10 pts) Multi-tenant isolation with dedicated tests
• (10 pts) A cache with tenant_id in the key + no-cross-leak tests
• (5 pts) The FastAPI API with /ask, /health, /metrics

Operations and quality (30 pts)

• (10 pts) Error handling: exponential retry, filter fallbacks, timeout handling
• (10 pts) Modular code: a clear separation between the pipeline, the retriever, the cache, the API
• (10 pts) Structured JSON logging with tenant_id, query_id, latency, status

Benchmark and documentation (20 pts)

• (10 pts) A reproducible BENCHMARK.md with correct percentiles and recall@k
• (10 pts) A DECISION.md with a 12-month cost analysis and a defensible recommendation

Extra credit (+10)

• (+5 pts) A documented rollback plan (how to go back to ChromaDB if Pinecone fails for 24h)
• (+5 pts) Retrieval regression tests (a golden set that catches degradation)

---

Common mistakes and how to avoid them

1. Creating the index with the wrong dimension → the idempotency check must verify dimension == 1536 and fail if it doesn't.
2. Migrating without validating the counts → your CI must block the deploy if chroma_count != pinecone_count.
3. Calling index.query directly → enforce it with a custom linter or a pre-commit hook that catches the pattern.
4. A cache shared between tenants → a mandatory test: two tenants with the same query receive different answers if their data differs.
5. Comparing benchmarks run under different conditions → run from the same machine, the same dataset, the same top_k.
6. Logs with PII → never log the full content of a query; only a hash + operational metadata.
7. Having no rollback plan → if Pinecone has an outage, how do you serve traffic? It must be documented.

---

Documents to deliver

BENCHMARK.md

• The benchmark's setup (hardware, dataset, configuration)
• A table of percentiled latencies
• The throughput vs concurrency curve
• Recall@10, compared
• A 12-month cost estimate

DECISION.md

• The product's context (current scale, expected scale)
• A summary of the benchmark's findings
• The trade-offs evaluated (latency, cost, operations, lock-in)
• The final recommendation with quantified reasons
• The rollback plan

README.md

• Local setup in 5 minutes
• How to run the migration
• How to run the benchmark
• How to run the tests
• How to deploy to production

---

Resources for the project

1. Pinecone Docs - The complete official reference.
2. Upsert Data - Optimized batch ingestion.
3. Query API - The query parameters.
4. Filter by Metadata - The supported operators.
5. Implement Multitenancy - The official patterns.
6. FastAPI Production Deployment - Best practices.
7. Prometheus Python Client - Metrics for /metrics.
8. structlog - Structured JSON logging.

---

Connection with module 8

With this system in production, the queries start generating real traffic and the stakeholders are going to ask you: "but does it answer well?". That question isn't answered with p95 or with cost: it's answered with systematic quality metrics.

In module 8 (RAG Evaluation) you'll build the continuous evaluation system that's missing: versioned golden datasets, RAGAS for automated metrics (faithfulness, answer relevancy, context precision), quality gates in CI/CD that block deploys that degrade quality, and dashboards that catch regressions before a user opens a ticket.

Without continuous evaluation, your production system is a black box that improves when you think it improves. With continuous evaluation, you improve with evidence.

---

Created: March 13, 2026
Version: 2.0