Building Production RAG for Under $10/Month: The Complete Stack

from groq import Groq

groq_client = Groq(api_key="gsk_...")

def generate_answer(context: str, question: str) -> str:
    completion = groq_client.chat.completions.create(
        model="llama-3.3-70b-versatile",
        messages=[
            {
                "role": "system",
                "content": "Answer questions using only the provided context. If the answer isn't in the context, say so."
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {question}"
            }
        ],
        temperature=0.1,  # Low temperature for factual answers
        max_tokens=512,
    )
    return completion.choices[0].message.content

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="pcsk_...")

# Create a serverless index (free tier)
pc.create_index(
    name="rag-docs",
    dimension=1536,        # text-embedding-3-small dimension
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1")
)

index = pc.Index("rag-docs")

from openai import OpenAI

openai_client = OpenAI(api_key="sk-...")

def embed(texts: list[str]) -> list[list[float]]:
    response = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=texts
    )
    return [e.embedding for e in response.data]

import hashlib

def ingest_document(doc_id: str, text: str, metadata: dict):
    # Split into chunks
    chunks = split_text(text, chunk_size=500, overlap=50)
    
    # Embed in batches of 100
    vectors = []
    for i, chunk in enumerate(chunks):
        embedding = embed([chunk])[0]
        vectors.append({
            "id": f"{doc_id}_chunk_{i}",
            "values": embedding,
            "metadata": {**metadata, "text": chunk, "chunk_index": i}
        })
    
    # Upsert to Pinecone
    index.upsert(vectors=vectors)
    return len(vectors)

def split_text(text: str, chunk_size: int, overlap: int) -> list[str]:
    words = text.split()
    chunks = []
    for i in range(0, len(words), chunk_size - overlap):
        chunk = " ".join(words[i:i + chunk_size])
        if chunk:
            chunks.append(chunk)
    return chunks

def rag_query(question: str, top_k: int = 5) -> dict:
    # 1. Embed the question
    query_vector = embed([question])[0]
    
    # 2. Retrieve relevant chunks
    results = index.query(
        vector=query_vector,
        top_k=top_k,
        include_metadata=True
    )
    
    # 3. Build context from results
    context_chunks = [
        r["metadata"]["text"] 
        for r in results["matches"] 
        if r["score"] > 0.7  # Relevance threshold
    ]
    
    if not context_chunks:
        return {"answer": "No relevant documents found.", "sources": []}
    
    context = "\n\n---\n\n".join(context_chunks)
    
    # 4. Generate answer
    answer = generate_answer(context, question)
    
    return {
        "answer": answer,
        "sources": [r["metadata"].get("source", "unknown") for r in results["matches"]],
        "retrieval_scores": [r["score"] for r in results["matches"]]
    }

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class QueryRequest(BaseModel):
    question: str
    top_k: int = 5

@app.post("/query")
async def query(req: QueryRequest):
    return rag_query(req.question, req.top_k)

@app.post("/ingest")
async def ingest(doc_id: str, text: str, source: str = ""):
    count = ingest_document(doc_id, text, {"source": source})
    return {"chunks_ingested": count}

railway up