Vector Database Tutorial: From Zero to RAG Agent in 2026

import pinecone
from openai import OpenAI
import numpy as np

# Initialize Pinecone
pinecone.init(api_key="your-api-key", environment="us-west1-gcp")

# Create index with 1536 dimensions (OpenAI embeddings)
index_name = "rag-agent-memory"
if index_name not in pinecone.list_indexes():
    pinecone.create_index(
        name=index_name,
        dimension=1536,
        metric="cosine"
    )

index = pinecone.Index(index_name)
client = OpenAI()

def get_embedding(text):
    """Convert text to vector embedding"""
    response = client.embeddings.create(
        input=text,
        model="text-embedding-ada-002"
    )
    return response.data[0].embedding

def store_document(doc_id, content, metadata=None):
    """Store document as vector in database"""
    embedding = get_embedding(content)
    index.upsert([
        {
            "id": doc_id,
            "values": embedding,
            "metadata": {"content": content, **(metadata or {})}
        }
    ])

def search_similar(query, top_k=5):
    """Find similar documents to query"""
    query_embedding = get_embedding(query)
    results = index.query(
        vector=query_embedding,
        top_k=top_k,
        include_metadata=True
    )
    return results.matches

class RAGAgent:
    def __init__(self, index_name="rag-agent"):
        self.index = pinecone.Index(index_name)
        self.client = OpenAI()
        self.conversation_memory = []

    def add_documents(self, documents):
        """Add documents to vector database with chunking"""
        for i, doc in enumerate(documents):
            # Split into chunks (simple approach)
            chunks = self._chunk_text(doc["content"])

            for j, chunk in enumerate(chunks):
                doc_id = f"{doc['id']}_chunk_{j}"
                embedding = get_embedding(chunk)

                self.index.upsert([{
                    "id": doc_id,
                    "values": embedding,
                    "metadata": {
                        "content": chunk,
                        "source": doc["id"],
                        "chunk_index": j
                    }
                }])

    def _chunk_text(self, text, chunk_size=500, overlap=50):
        """Split text into overlapping chunks"""
        words = text.split()
        chunks = []

        for i in range(0, len(words), chunk_size - overlap):
            chunk = " ".join(words[i:i + chunk_size])
            chunks.append(chunk)

        return chunks

    def query(self, question):
        """Query with RAG pipeline"""
        # Retrieve relevant context
        context_docs = search_similar(question, top_k=3)
        context = "\n\n".join([match.metadata["content"] for match in context_docs])

        # Include conversation memory
        memory_context = "\n".join([
            f"User: {msg['user']}\nAssistant: {msg['assistant']}"
            for msg in self.conversation_memory[-3:]  # Last 3 exchanges
        ])

        # Generate response
        prompt = f"""
        Context from knowledge base:
        {context}

        Previous conversation:
        {memory_context}

        Current question: {question}

        Please provide a helpful response based on the context and conversation history.
        """

        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are a helpful assistant that answers questions based on provided context."},
                {"role": "user", "content": prompt}
            ]
        )

        answer = response.choices[0].message.content

        # Store in conversation memory
        self.conversation_memory.append({
            "user": question,
            "assistant": answer
        })

        return answer

class MemoryEnhancedAgent(RAGAgent):
    def __init__(self, index_name="memory-agent"):
        super().__init__(index_name)
        self.user_profile = {}

    def store_interaction(self, user_id, interaction_type, content):
        """Store user interaction as vector for future reference"""
        memory_id = f"{user_id}_{interaction_type}_{len(self.conversation_memory)}"
        embedding = get_embedding(content)

        self.index.upsert([{
            "id": memory_id,
            "values": embedding,
            "metadata": {
                "user_id": user_id,
                "type": interaction_type,
                "content": content,
                "timestamp": int(time.time())
            }
        }])

    def get_user_context(self, user_id, query):
        """Retrieve relevant user history for personalized responses"""
        # Search for relevant past interactions
        results = self.index.query(
            vector=get_embedding(query),
            top_k=5,
            filter={"user_id": {"$eq": user_id}},
            include_metadata=True
        )

        return [match.metadata for match in results.matches]

    def personalized_query(self, user_id, question):
        """Answer with personalized context from user history"""
        # Get user's relevant history
        user_context = self.get_user_context(user_id, question)

        # Combine with knowledge base context
        kb_context = search_similar(question, top_k=3)

        # Generate personalized response
        context_text = "\n".join([
            f"User's past interaction: {ctx['content']}"
            for ctx in user_context[:2]
        ])

        kb_text = "\n".join([
            match.metadata["content"] for match in kb_context
        ])

        prompt = f"""
        User's relevant history:
        {context_text}

        Knowledge base context:
        {kb_text}

        Current question: {question}

        Provide a personalized response considering the user's history and preferences.
        """

        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are a personalized assistant that adapts to user preferences and history."},
                {"role": "user", "content": prompt}
            ]
        )

        answer = response.choices[0].message.content

        # Store this interaction for future reference
        self.store_interaction(user_id, "query_response", f"Q: {question}\nA: {answer}")

        return answer