Build a RAG System with Python and Pinecone in 45 Minutes

pip install pinecone-client openai tiktoken python-dotenv --break-system-packages

# setup_pinecone.py
import os
from pinecone import Pinecone, ServerlessSpec
from dotenv import load_dotenv

load_dotenv()

pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))

# Create index optimized for OpenAI embeddings
index_name = "knowledge-base"

if index_name not in pc.list_indexes().names():
    pc.create_index(
        name=index_name,
        dimension=1536,  # OpenAI text-embedding-3-small dimension
        metric="cosine",  # Best for semantic similarity
        spec=ServerlessSpec(
            cloud="aws",
            region="us-east-1"
        )
    )

print(f"Index '{index_name}' ready")

python setup_pinecone.py

# ingest_docs.py
import os
from openai import OpenAI
from pinecone import Pinecone
import tiktoken

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))
index = pc.Index("knowledge-base")

def chunk_text(text, max_tokens=512):
    """Split text into chunks that fit embedding limits"""
    encoding = tiktoken.get_encoding("cl100k_base")
    tokens = encoding.encode(text)
    
    chunks = []
    for i in range(0, len(tokens), max_tokens):
        chunk_tokens = tokens[i:i + max_tokens]
        chunks.append(encoding.decode(chunk_tokens))
    
    return chunks

def embed_and_store(doc_id, text, metadata={}):
    """Chunk document, create embeddings, store in Pinecone"""
    chunks = chunk_text(text)
    
    vectors = []
    for i, chunk in enumerate(chunks):
        # Generate embedding
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=chunk
        )
        embedding = response.data[0].embedding
        
        # Prepare vector with metadata
        vectors.append({
            "id": f"{doc_id}_chunk_{i}",
            "values": embedding,
            "metadata": {
                "text": chunk,
                "doc_id": doc_id,
                "chunk_index": i,
                **metadata
            }
        })
    
    # Batch upsert for efficiency
    index.upsert(vectors=vectors)
    print(f"Stored {len(chunks)} chunks for {doc_id}")

# Example: Ingest your documentation
docs = {
    "product_guide": """
    Our API supports both REST and GraphQL endpoints.
    Authentication uses Bearer tokens with 24-hour expiry.
    Rate limits are 1000 requests per hour for free tier.
    """,
    "troubleshooting": """
    If you get 429 errors, you've hit rate limits.
    Wait 60 seconds or upgrade to Pro for 10,000 req/hour.
    For 401 errors, regenerate your API token in settings.
    """
}

for doc_id, content in docs.items():
    embed_and_store(
        doc_id=doc_id,
        text=content,
        metadata={"source": "docs", "category": doc_id}
    )

python ingest_docs.py

# rag_query.py
import os
from openai import OpenAI
from pinecone import Pinecone

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))
index = pc.Index("knowledge-base")

def rag_query(question, top_k=3):
    """Query knowledge base and generate contextualized answer"""
    
    # Step 1: Embed the question
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=question
    )
    query_embedding = response.data[0].embedding
    
    # Step 2: Retrieve relevant chunks
    results = index.query(
        vector=query_embedding,
        top_k=top_k,
        include_metadata=True
    )
    
    # Step 3: Build context from matches
    context_chunks = [
        match['metadata']['text'] 
        for match in results['matches']
    ]
    context = "\n\n".join(context_chunks)
    
    # Step 4: Generate answer with retrieved context
    prompt = f"""Answer the question based only on the following context.
If the context doesn't contain the answer, say "I don't have information about that."

Context:
{context}

Question: {question}

Answer:"""
    
    completion = client.chat.completions.create(
        model="gpt-4o-mini",  # Fast and cheap for RAG
        messages=[
            {"role": "system", "content": "You are a helpful assistant that answers questions based on provided context. Always cite which part of the context you used."},
            {"role": "user", "content": prompt}
        ],
        temperature=0.3  # Lower temp = more factual, less creative
    )
    
    answer = completion.choices[0].message.content
    
    # Return answer with sources for transparency
    return {
        "answer": answer,
        "sources": [
            {
                "doc_id": match['metadata']['doc_id'],
                "chunk": match['metadata']['text'][:100] + "...",
                "score": match['score']
            }
            for match in results['matches']
        ]
    }

# Test it
if __name__ == "__main__":
    result = rag_query("What should I do if I get a 429 error?")
    
    print("Answer:", result['answer'])
    print("\nSources:")
    for source in result['sources']:
        print(f"- {source['doc_id']} (relevance: {source['score']:.2f})")
        print(f"  {source['chunk']}")

python rag_query.py

# Should answer (in your docs)
result1 = rag_query("What's the rate limit for free tier?")
print(result1['answer'])  # Should mention 1000 req/hour

# Shouldn't answer (not in docs)
result2 = rag_query("What's your company's stock price?")
print(result2['answer'])  # Should say "I don't have information about that"

# Requires Pinecone's sparse-dense vectors (available in paid tiers)
from pinecone_text.sparse import BM25Encoder

bm25 = BM25Encoder()
bm25.fit(corpus)  # Your document texts

# Query with both dense and sparse vectors
results = index.query(
    vector=dense_embedding,
    sparse_vector=bm25.encode_queries(question),
    top_k=5,
    alpha=0.5  # Balance between semantic and keyword matching
)

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

# After retrieving top 10 chunks
scores = reranker.predict([(question, chunk) for chunk in chunks])
reranked_chunks = [chunks[i] for i in scores.argsort()[::-1][:3]]

import time

start = time.time()
result = rag_query(question)
latency = time.time() - start

# Log for analysis
print(f"Query latency: {latency:.2f}s")
print(f"Avg relevance: {sum(s['score'] for s in result['sources'])/len(result['sources']):.2f}")