RAG Systems at Production Scale

import pinecone
from openai import OpenAI

class NotionRAG:
    def __init__(self):
        self.pinecone_index = pinecone.Index("notion-docs")
        self.openai = OpenAI()

    def hybrid_search(self, query, user_workspaces, top_k=5):
        """Hybrid search: vector + keyword + permissions."""
        # 1. Embed query
        embedding = self.openai.embeddings.create(
            model="text-embedding-ada-002", input=query,
        ).data[0].embedding

        # 2. Vector search WITH permission filter (pre-VS!)
        vector_results = self.pinecone_index.query(
            vector=embedding,
            top_k=100,
            filter={"workspace_id": {"$in": user_workspaces}},
            include_metadata=True,
        )

        # 3. BM25 keyword search (for exact matches)
        keyword_results = self.keyword_search(query, user_workspaces)

        # 4. Hybrid scoring
        combined = self.combine_scores(
            vector_results, keyword_results, alpha=0.7,
        )

        # 5. Rerank top candidates with GPT-4
        return self.rerank_with_llm(query, combined[:50], top_k=top_k)

    def rerank_with_llm(self, query, candidates, top_k):
        prompt = f"""Query: {query}

Rank these documents by relevance (1 = most relevant):

{self.format_candidates(candidates)}

Return only the document IDs in order."""
        response = self.openai.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
        )
        ranked_ids = self.parse_ranking(response)
        return [c for c in candidates if c.id in ranked_ids[:top_k]]

from langchain.text_splitter import RecursiveCharacterTextSplitter

class NotionChunker:
    """Smart chunking — split at semantic boundaries, preserve structure."""

    def __init__(self):
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=512,
            chunk_overlap=50,
            separators=[
                "\n## ", "\n### ",   # Notion headings
                "\n- ",                # bullet points
                "\n\n", "\n",        # paragraphs
                ". ", " ",
            ],
        )

    def chunk_document(self, document):
        chunks = []
        for i, chunk in enumerate(self.splitter.split_text(document.content)):
            chunks.append({
                'text': chunk,
                'document_id': document.id,
                'chunk_index': i,
                'workspace_id': document.workspace_id,
                'created_by': document.created_by,
                'last_edited': document.last_edited,
                'heading': self.extract_heading(chunk, document),
                'position': i / len(chunks),
            })
        return chunks

    def extract_heading(self, chunk, document):
        headings = self.parse_headings(document.content)
        chunk_start = document.content.find(chunk)
        for heading in reversed(headings):
            if heading['position'] < chunk_start:
                return heading['text']
        return document.title

from weaviate import Client

class IntercomRAG:
    def __init__(self):
        self.weaviate = Client("http://weaviate:8080")

    def retrieve_with_context(self, query, conversation_history, customer_id):
        """Use sliding window of recent messages as additional context."""
        # 1. Build context from last 5 messages
        context_window = conversation_history[-5:]
        context_text = "\n".join(
            f"{msg['role']}: {msg['content']}" for msg in context_window
        )

        # 2. Combine query with conversation context
        enriched_query = f"""Conversation context:
{context_text}

Current question: {query}"""

        # 3. Multi-source retrieval
        results = (
            self.weaviate.query.get(
                "KnowledgeArticle",
                ["title", "content", "last_updated", "source"],
            )
            .with_near_text({"concepts": [enriched_query], "distance": 0.7})
            .with_where({
                "operator": "Or",
                "operands": [
                    {"path": ["source"], "operator": "Equal",
                     "valueString": "knowledge_base"},
                    {"path": ["source"], "operator": "Equal",
                     "valueString": "past_conversations"},
                ],
            })
            .with_additional(["distance", "certainty"])
            .do()
        )

        # 4. Boost recent sources +30%
        return self.rerank_by_recency(results, recency_boost=0.3)[:5]

    def rerank_by_recency(self, results, recency_boost):
        import datetime
        for result in results:
            days_old = (datetime.now() - result['last_updated']).days
            recency_factor = 1.0 + (recency_boost / (1 + days_old / 30))
            result['final_score'] = result['certainty'] * recency_factor
        return sorted(results, key=lambda x: x['final_score'], reverse=True)

from openai import OpenAI

class ConfidenceGate:
    """Only show AI answers when confident; escalate low-confidence to humans."""

    def __init__(self):
        self.openai = OpenAI()
        self.thresholds = {}  # per-customer thresholds

    def generate_with_confidence(self, query, retrieved_docs):
        response = self.openai.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system",
                 "content": "You are a customer support AI. Provide accurate answers based on the knowledge base."},
                {"role": "user",
                 "content": f"""Query: {query}

Relevant articles:
{self.format_docs(retrieved_docs)}

Provide your answer and rate your confidence (0-1)."""},
            ],
            temperature=0,
        )
        answer = response.choices[0].message.content
        confidence = self.calculate_confidence(query, retrieved_docs, answer)
        return {
            'answer': answer,
            'confidence': confidence,
            'retrieved_docs': retrieved_docs,
        }

    def calculate_confidence(self, query, docs, answer):
        """Multi-factor confidence: retrieval quality + consistency + complexity."""
        retrieval_score = docs[0]['certainty'] if docs else 0.0
        consistency_score = self.check_consistency(answer, docs)
        complexity_score = 1.0 - self.estimate_complexity(query)
        return 0.4 * retrieval_score + 0.4 * consistency_score + 0.2 * complexity_score

    def should_escalate(self, confidence, customer_id):
        threshold = self.thresholds.get(customer_id, 0.75)
        if confidence < threshold:
            return True, "Low confidence — escalating to human"
        return False, "High confidence — showing AI answer"