The RAG Revolution - Why Custom AI Knowledge Bases Outperform Off-the-Shelf Chatbots by 10x

User asks question
    ↓
AI generates answer based on training data
    ↓
Answer returned

User asks question
    ↓
Retrieve relevant information from knowledge base
    ↓
Send question + retrieved context to AI
    ↓
AI generates answer based on provided context
    ↓
Answer returned with sources

def process_documents(document_path):
    """
    Convert documents into searchable chunks
    """

    # Step 1: Load document
    document = load_document(document_path)

    # Step 2: Clean and normalize
    cleaned = clean_text(document)
    # Remove HTML tags, fix encoding, normalize whitespace

    # Step 3: Chunk intelligently
    chunks = smart_chunking(cleaned, chunk_size=512)
    # Not naive splitting - respects paragraphs, headings, sections

    # Step 4: Add metadata
    for chunk in chunks:
        chunk.metadata = {
            'source': document_path,
            'section': extract_section(chunk),
            'last_updated': document.modified_date,
            'document_type': classify_document_type(document),
            'keywords': extract_keywords(chunk),
            'category': categorize_content(chunk)
        }

    # Step 5: Generate embeddings
    embeddings = generate_embeddings(chunks)
    # Convert text to vector representations

    # Step 6: Store in vector database
    store_in_vectordb(chunks, embeddings, metadata)

    return len(chunks)

def smart_chunking(text, chunk_size=512):
    """
    Chunk text intelligently, not naively
    """

    # First, split by major sections (markdown headers)
    sections = split_by_headers(text)

    chunks = []

    for section in sections:
        # If section is small enough, keep as one chunk
        if len(section) <= chunk_size:
            chunks.append(section)

        # If section is too large, split by paragraphs
        else:
            paragraphs = section.split('\n\n')

            current_chunk = ""
            for paragraph in paragraphs:
                # If adding this paragraph fits in chunk size
                if len(current_chunk) + len(paragraph) <= chunk_size:
                    current_chunk += paragraph + "\n\n"

                # Otherwise, save current chunk and start new one
                else:
                    if current_chunk:
                        chunks.append(current_chunk.strip())
                    current_chunk = paragraph + "\n\n"

            # Add final chunk
            if current_chunk:
                chunks.append(current_chunk.strip())

        # Add context overlap (last 100 chars of previous chunk)
        for i in range(1, len(chunks)):
            overlap = chunks[i-1][-100:]
            chunks[i] = f"[Context: ...{overlap}]\n\n{chunks[i]}"

    return chunks

Chunk 1: "...and then you can configure the settings. Next"
Chunk 2: "step is to save your changes. This allows you to..."

Chunk 1: "...and then you can configure the settings. Next step is to save your changes."
Chunk 2: "[Context: ...Next step is to save your changes.] This allows you to persist your configuration across sessions..."

# Each chunk stored with:
{
    'id': 'doc_123_chunk_5',
    'vector': [0.023, -0.154, 0.089, ...], # 1536 dimensions for OpenAI embeddings
    'metadata': {
        'text': 'Original chunk content...',
        'source': 'docs/api-reference.md',
        'section': 'Authentication',
        'last_updated': '2025-09-15',
        'category': 'technical',
        'keywords': ['API', 'auth', 'tokens'],
        'confidence': 0.95
    }
}

def index_documents(documents, namespace='production'):
    """
    Index all documents into vector database
    """

    index = pinecone.Index('company-knowledge-base')

    vectors_to_upsert = []

    for doc in documents:
        # Process document into chunks
        chunks = process_documents(doc)

        for chunk in chunks:
            # Generate embedding
            embedding = openai.Embedding.create(
                input=chunk.text,
                model="text-embedding-ada-002"
            )['data'][0]['embedding']

            vectors_to_upsert.append({
                'id': chunk.id,
                'values': embedding,
                'metadata': chunk.metadata
            })

            # Batch upsert every 100 vectors
            if len(vectors_to_upsert) >= 100:
                index.upsert(
                    vectors=vectors_to_upsert,
                    namespace=namespace
                )
                vectors_to_upsert = []

    # Upsert remaining vectors
    if vectors_to_upsert:
        index.upsert(vectors=vectors_to_upsert, namespace=namespace)

def retrieve_context(question, top_k=5):
    """
    Retrieve relevant context for answering the question
    """

    # Step 1: Generate embedding for the question
    question_embedding = openai.Embedding.create(
        input=question,
        model="text-embedding-ada-002"
    )['data'][0]['embedding']

    # Step 2: Search vector database
    index = pinecone.Index('company-knowledge-base')

    results = index.query(
        vector=question_embedding,
        top_k=top_k,
        include_metadata=True,
        namespace='production'
    )

    # Step 3: Extract and rank results
    contexts = []

    for match in results['matches']:
        # Only include results above confidence threshold
        if match['score'] > 0.70:
            contexts.append({
                'text': match['metadata']['text'],
                'source': match['metadata']['source'],
                'section': match['metadata']['section'],
                'score': match['score'],
                'last_updated': match['metadata']['last_updated']
            })

    # Step 4: Re-rank results (optional but recommended)
    reranked_contexts = rerank_with_cross_encoder(question, contexts)

    return reranked_contexts

def hybrid_search(question, top_k=5):
    """
    Combine semantic search with keyword search
    """

    # Vector search
    vector_results = semantic_search(question, top_k=10)

    # Keyword search (BM25)
    keyword_results = keyword_search(question, top_k=10)

    # Combine and deduplicate
    combined_results = {}

    for result in vector_results:
        combined_results[result['id']] = {
            **result,
            'vector_score': result['score'],
            'keyword_score': 0
        }

    for result in keyword_results:
        if result['id'] in combined_results:
            combined_results[result['id']]['keyword_score'] = result['score']
        else:
            combined_results[result['id']] = {
                **result,
                'vector_score': 0,
                'keyword_score': result['score']
            }

    # Calculate final score (weighted combination)
    for id, result in combined_results.items():
        result['final_score'] = (
            0.7 * result['vector_score'] +
            0.3 * result['keyword_score']
        )

    # Sort by final score
    ranked_results = sorted(
        combined_results.values(),
        key=lambda x: x['final_score'],
        reverse=True
    )

    return ranked_results[:top_k]

def generate_answer(question, contexts):
    """
    Generate answer using retrieved context
    """

    # Build prompt with context
    context_text = "\n\n---\n\n".join([
        f"Source: {c['source']}\nSection: {c['section']}\n\n{c['text']}"
        for c in contexts
    ])

    prompt = f"""You are a helpful customer support assistant.

Answer the user's question based ONLY on the provided context.
If the context doesn't contain enough information, say so.
Always cite your sources.

Context:
{context_text}

Question: {question}

Instructions:
1. Answer the question accurately based on the context
2. Cite specific sources for each claim
3. If information is missing, say "I don't have information about that in my knowledge base"
4. Be concise but complete
5. Use formatting (bullet points, numbered lists) for clarity

Answer:"""

    # Generate response
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "You are a helpful customer support assistant."},
            {"role": "user", "content": prompt}
        ],
        temperature=0.3, # Lower temperature for more factual responses
        max_tokens=800
    )

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

    # Add metadata
    result = {
        'answer': answer,
        'sources': [
            {'source': c['source'], 'section': c['section']}
            for c in contexts
        ],
        'confidence': calculate_confidence(contexts, answer)
    }

    return result

We offer a 30-day money-back guarantee on all purchases. To request
a refund, email support@company.com with your order number. Refunds
are processed within 5-7 business days.

Sources:
- Refund Policy (docs/policies/refunds.md)
- Terms of Service (docs/legal/terms.md)
- Support Guide (docs/support/refund-requests.md)

def calculate_confidence(contexts, answer):
    """
    Determine how confident we are in the answer
    """

    # Factor 1: Retrieval scores
    avg_retrieval_score = sum(c['score'] for c in contexts) / len(contexts)

    # Factor 2: Number of sources
    # More sources = more confidence (up to a point)
    source_factor = min(len(contexts) / 3, 1.0)

    # Factor 3: Recency of sources
    # Newer = more confident
    most_recent = max(c['last_updated'] for c in contexts)
    days_old = (datetime.now() - most_recent).days
    recency_factor = max(1.0 - (days_old / 365), 0.5)

    # Factor 4: Answer specificity
    # Vague answers like "it depends" = lower confidence
    specificity_score = analyze_specificity(answer)

    # Combine factors
    confidence = (
        0.4 * avg_retrieval_score +
        0.2 * source_factor +
        0.2 * recency_factor +
        0.2 * specificity_score
    )

    return round(confidence, 2)

if confidence > 0.85:
    # High confidence - answer directly
    return answer

elif confidence > 0.65:
    # Medium confidence - answer with caveat
    return f"{answer}\n\nNote: Please verify this information is current."

else:
    # Low confidence - escalate to human
    return "I found some information, but I'm not confident in my answer. Let me connect you with a human agent."

class RAGChatbot:
    def __init__(self):
        self.vector_db = pinecone.Index('company-knowledge-base')
        self.openai = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

    def answer_question(self, question, conversation_history=[]):
        """
        Main entry point for answering questions
        """

        # Step 1: Rephrase question with conversation context
        if conversation_history:
            question = self.rephrase_with_context(
                question,
                conversation_history
            )

        # Step 2: Retrieve relevant contexts
        contexts = self.retrieve_context(question, top_k=5)

        # Step 3: Check if we have sufficient context
        if not contexts or contexts[0]['score'] < 0.65:
            return {
                'answer': "I don't have enough information in my knowledge base to answer that. Let me connect you with a human agent.",
                'confidence': 0.0,
                'escalate': True
            }

        # Step 4: Generate answer
        result = self.generate_answer(question, contexts)

        # Step 5: Log for analytics
        self.log_interaction(question, result)

        return result

    def rephrase_with_context(self, question, history):
        """
        Rephrase follow-up questions with conversation context
        """

        # Example:
        # User: "What's your refund policy?"
        # Bot: "We offer a 30-day..."
        # User: "How do I request one?"
        # Rephrased: "How do I request a refund?"

        prompt = f"""Given this conversation history, rephrase the latest question to be standalone.

History:
{format_conversation_history(history)}

Latest question: {question}

Rephrased standalone question:"""

        response = self.openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            temperature=0
        )

        return response['choices'][0]['message']['content']

from fastapi import FastAPI

app = FastAPI()
chatbot = RAGChatbot()

@app.post("/chat")
async def chat(request: ChatRequest):
    result = chatbot.answer_question(
        question=request.question,
        conversation_history=request.history
    )

    return result

from slack_bolt import App

app = App(token=os.getenv("SLACK_BOT_TOKEN"))
chatbot = RAGChatbot()

@app.message(".*")
def handle_message(message, say):
    question = message['text']
    result = chatbot.answer_question(question)

    say(result['answer'])

// Embed on website
<script>
window.chatbot = {
    async ask(question) {
        const response = await fetch('/api/chat', {
            method: 'POST',
            headers: {'Content-Type': 'application/json'},
            body: JSON.stringify({question})
        });

        const data = await response.json();
        return data.answer;
    }
};
</script>

def decompose_complex_question(question):
    """
    Break complex question into sub-questions
    """

    prompt = f"""Break this complex question into simpler sub-questions:

Question: {question}

Sub-questions (return as JSON array):"""

    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )

    sub_questions = json.loads(response['choices'][0]['message']['content'])

    # Answer each sub-question
    sub_answers = []
    for sub_q in sub_questions:
        answer = answer_simple_question(sub_q)
        sub_answers.append({'question': sub_q, 'answer': answer})

    # Synthesize final answer
    final_answer = synthesize_answers(question, sub_answers)

    return final_answer

Question: "How do I set up authentication and connect to the database?"

Sub-questions:
1. "How do I set up authentication?"
2. "How do I connect to the database?"

[Answer each separately, then synthesize]

def self_reflect(question, answer, contexts):
    """
    AI reviews its own answer for accuracy
    """

    prompt = f"""Review this answer for accuracy:

Question: {question}

Answer: {answer}

Available context: {contexts}

Is this answer:
1. Accurate based on the context?
2. Complete?
3. Missing any important caveats?

Critique (JSON):"""

    critique = get_gpt4_response(prompt)

    if critique['issues_found']:
        # Regenerate with feedback
        improved_answer = regenerate_with_feedback(
            question,
            contexts,
            critique['feedback']
        )
        return improved_answer

    return answer

def answer_with_reasoning(question, contexts):
    """
    Chain-of-thought reasoning
    """

    prompt = f"""Answer this question using step-by-step reasoning:

Context: {contexts}

Question: {question}

Think step-by-step:
1. What information do I need?
2. What does the context say?
3. What can I infer?
4. What's my conclusion?

Reasoning:"""

    response = get_gpt4_response(prompt)

    return response

def detect_hallucination(answer, contexts):
    """
    Check if answer contains information not in contexts
    """

    prompt = f"""Does this answer contain information NOT found in the context?

Context:
{contexts}

Answer:
{answer}

Check each claim in the answer. Is it supported by the context?
Return JSON with flagged claims.
"""

    verification = get_gpt4_response(prompt)

    if verification['unsupported_claims']:
        # Remove unsupported claims
        filtered_answer = remove_unsupported_claims(
            answer,
            verification['unsupported_claims']
        )
        return filtered_answer

    return answer