Retrospective: Our 1-Year Use of LangChain 0.3 and RAGatouille 0.7 Cut RAG Hallucination by 55%

import os
import logging
from typing import List, Dict, Any, Optional

from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.prompts import ChatPromptTemplate
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.llms import HuggingFaceHub
from ragatouille import RAGPretrainedModel
from dotenv import load_dotenv

# Configure logging for pipeline debuggability
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Load environment variables from .env file
load_dotenv()

class ProductionRAGPipeline:
    """Production-ready RAG pipeline using LangChain 0.3 and RAGatouille 0.7."""

    def __init__(self, vectorstore_path: str, reranker_model: str = "colbert-ir/colbertv2.0"):
        self.vectorstore_path = vectorstore_path
        self.reranker_model = reranker_model
        self.embeddings = None
        self.vectorstore = None
        self.reranker = None
        self.llm = None
        self.chain = None

    def _initialize_embeddings(self) -> None:
        """Initialize HuggingFace embeddings with error handling."""
        try:
            # Use all-MiniLM-L6-v2 for fast, accurate embeddings
            self.embeddings = HuggingFaceEmbeddings(
                model_name="sentence-transformers/all-MiniLM-L6-v2",
                model_kwargs={"device": "cpu"}  # Switch to "cuda" for GPU
            )
            logger.info("Embeddings model loaded successfully")
        except Exception as e:
            logger.error(f"Failed to load embeddings model: {str(e)}")
            raise

    def _initialize_vectorstore(self) -> None:
        """Initialize Chroma vectorstore from local path."""
        try:
            if not os.path.exists(self.vectorstore_path):
                raise FileNotFoundError(f"Vectorstore path {self.vectorstore_path} does not exist")

            self.vectorstore = Chroma(
                persist_directory=self.vectorstore_path,
                embedding_function=self.embeddings
            )
            logger.info(f"Vectorstore loaded from {self.vectorstore_path}, contains {self.vectorstore._collection.count()} documents")
        except Exception as e:
            logger.error(f"Failed to load vectorstore: {str(e)}")
            raise

    def _initialize_reranker(self) -> None:
        """Initialize RAGatouille ColBERTv2.0 reranker."""
        try:
            self.reranker = RAGPretrainedModel.from_pretrained(
                self.reranker_model,
                use_gpu=False  # Switch to True for GPU acceleration
            )
            logger.info(f"RAGatouille reranker {self.reranker_model} loaded successfully")
        except Exception as e:
            logger.error(f"Failed to load RAGatouille reranker: {str(e)}")
            raise

    def _initialize_llm(self) -> None:
        """Initialize LLM from HuggingFace Hub with fallback to local model."""
        try:
            hf_token = os.getenv("HUGGINGFACEHUB_API_TOKEN")
            if not hf_token:
                raise ValueError("HUGGINGFACEHUB_API_TOKEN not set in environment")

            self.llm = HuggingFaceHub(
                repo_id="mistralai/Mistral-7B-Instruct-v0.2",
                model_kwargs={"temperature": 0.1, "max_new_tokens": 512},
                huggingfacehub_api_token=hf_token
            )
            logger.info("LLM loaded from HuggingFace Hub")
        except Exception as e:
            logger.error(f"Failed to load LLM: {str(e)}")
            raise

    def _build_chain(self) -> None:
        """Build LangChain 0.3 retrieval chain with RAGatouille reranking."""
        try:
            # Define prompt template with hallucination guardrails
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a helpful assistant that answers questions based only on the provided context. 
                If the context does not contain the answer, say "I don't have enough information to answer that."
                Do not make up information. Context: {context}"""),
                ("human", "{input}")
            ])

            # Create document chain for combining retrieved docs
            document_chain = create_stuff_documents_chain(self.llm, prompt)

            # Create retriever with RAGatouille reranking
            # First get top 20 results from vectorstore, then rerank to top 5
            base_retriever = self.vectorstore.as_retriever(search_kwargs={"k": 20})

            def rerank_documents(query: str, docs: List) -> List:
                """Rerank retrieved documents using RAGatouille."""
                doc_texts = [doc.page_content for doc in docs]
                reranked_results = self.reranker.rerank(query, doc_texts, k=5)
                # Map reranked results back to original documents
                reranked_docs = [docs[i] for i in [r["result_index"] for r in reranked_results]]
                return reranked_docs

            # Wrap retriever with reranking logic
            from langchain.schema.runnable import RunnableLambda
            retriever = RunnableLambda(lambda x: rerank_documents(x["input"], base_retriever.get_relevant_documents(x["input"])))

            # Create final retrieval chain
            self.chain = create_retrieval_chain(retriever, document_chain)
            logger.info("RAG chain built successfully")
        except Exception as e:
            logger.error(f"Failed to build RAG chain: {str(e)}")
            raise

    def initialize(self) -> None:
        """Initialize all pipeline components in order."""
        logger.info("Initializing production RAG pipeline...")
        self._initialize_embeddings()
        self._initialize_vectorstore()
        self._initialize_reranker()
        self._initialize_llm()
        self._build_chain()
        logger.info("Pipeline initialization complete")

    def query(self, query: str) -> Dict[str, Any]:
        """Run a query through the RAG pipeline with error handling."""
        try:
            if not self.chain:
                raise RuntimeError("Pipeline not initialized. Call initialize() first.")

            response = self.chain.invoke({"input": query})
            return {
                "answer": response["answer"],
                "source_documents": [doc.page_content for doc in response["context"]],
                "num_retrieved_docs": len(response["context"])
            }
        except Exception as e:
            logger.error(f"Query failed: {str(e)}")
            return {"error": str(e)}

if __name__ == "__main__":
    # Example usage
    pipeline = ProductionRAGPipeline(vectorstore_path="./chroma_db")
    try:
        pipeline.initialize()
        result = pipeline.query("What was our Q3 2023 enterprise churn rate?")
        print(f"Answer: {result.get('answer', result.get('error'))}")
    except Exception as e:
        logger.error(f"Pipeline failed to run: {str(e)}")

import json
import logging
from typing import List, Dict, Any
from rouge import Rouge
from langchain.evaluation import load_evaluator, EvaluatorType
from ragatouille import RAGPretrainedModel
from dotenv import load_dotenv
import os

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

load_dotenv()

class RAGHallucinationEvaluator:
    """Evaluate RAG pipeline hallucination rate using benchmark datasets."""

    def __init__(self, pipeline, reranker: RAGPretrainedModel, benchmark_path: str = "benchmark_qa.json"):
        self.pipeline = pipeline
        self.reranker = reranker
        self.benchmark_path = benchmark_path
        self.benchmark_data = None
        self.rouge = Rouge()
        self.langchain_evaluator = None

    def _load_benchmark_data(self) -> None:
        """Load benchmark QA dataset with error handling."""
        try:
            if not os.path.exists(self.benchmark_path):
                raise FileNotFoundError(f"Benchmark file {self.benchmark_path} not found")

            with open(self.benchmark_path, "r") as f:
                self.benchmark_data = json.load(f)

            logger.info(f"Loaded {len(self.benchmark_data)} benchmark questions")
        except json.JSONDecodeError as e:
            logger.error(f"Invalid JSON in benchmark file: {str(e)}")
            raise
        except Exception as e:
            logger.error(f"Failed to load benchmark data: {str(e)}")
            raise

    def _initialize_evaluators(self) -> None:
        """Initialize LangChain and ROUGE evaluators."""
        try:
            # Load LangChain's hallucination evaluator
            self.langchain_evaluator = load_evaluator(
                EvaluatorType.HALLUCINATION,
                llm=self.pipeline.llm
            )
            logger.info("LangChain hallucination evaluator loaded")
        except Exception as e:
            logger.error(f"Failed to load LangChain evaluator: {str(e)}")
            raise

    def _calculate_rouge_score(self, generated: str, reference: str) -> Dict[str, float]:
        """Calculate ROUGE scores between generated and reference answers."""
        try:
            scores = self.rouge.get_scores(generated, reference, avg=True)
            return scores
        except Exception as e:
            logger.warning(f"ROUGE calculation failed: {str(e)}")
            return {"rouge-l": {"f": 0.0}}

    def _is_hallucinated(self, query: str, generated: str, reference: str, context: List[str]) -> bool:
        """Determine if a generated answer is hallucinated using multiple signals."""
        try:
            # Signal 1: LangChain hallucination evaluator
            eval_result = self.langchain_evaluator.evaluate_strings(
                prediction=generated,
                reference=reference,
                input=query
            )
            langchain_hallucinated = eval_result.get("score", 0) > 0.5  # Score >0.5 means hallucinated

            # Signal 2: ROUGE-L score < 0.3 means low overlap with reference
            rouge_scores = self._calculate_rouge_score(generated, reference)
            rouge_hallucinated = rouge_scores["rouge-l"]["f"] < 0.3

            # Signal 3: No relevant context for answer
            context_text = " ".join(context)
            reranked_context = self.reranker.rerank(query, [context_text], k=1)
            context_relevant = reranked_context[0]["score"] > 0.7  # Reranker score >0.7 means relevant
            context_hallucinated = not context_relevant

            # Combine signals: hallucinated if 2+ signals trigger
            hallucination_signals = [langchain_hallucinated, rouge_hallucinated, context_hallucinated]
            return sum(hallucination_signals) >= 2
        except Exception as e:
            logger.error(f"Hallucination check failed for query '{query}': {str(e)}")
            return True  # Assume hallucinated if check fails

    def run_evaluation(self, sample_size: Optional[int] = None) -> Dict[str, Any]:
        """Run full evaluation on benchmark dataset."""
        try:
            self._load_benchmark_data()
            self._initialize_evaluators()

            # Sample data if sample_size is provided
            eval_data = self.benchmark_data[:sample_size] if sample_size else self.benchmark_data
            logger.info(f"Running evaluation on {len(eval_data)} questions")

            results = []
            hallucinated_count = 0

            for idx, item in enumerate(eval_data):
                query = item["query"]
                reference = item["reference_answer"]

                # Run query through pipeline
                pipeline_result = self.pipeline.query(query)
                if "error" in pipeline_result:
                    logger.warning(f"Query {idx} failed: {pipeline_result['error']}")
                    continue

                generated = pipeline_result["answer"]
                context = pipeline_result["source_documents"]

                # Check if hallucinated
                is_hallucinated = self._is_hallucinated(query, generated, reference, context)
                if is_hallucinated:
                    hallucinated_count += 1

                results.append({
                    "query": query,
                    "reference": reference,
                    "generated": generated,
                    "is_hallucinated": is_hallucinated,
                    "context": context
                })

                if (idx + 1) % 10 == 0:
                    logger.info(f"Processed {idx + 1}/{len(eval_data)} queries")

            # Calculate metrics
            total_queries = len(results)
            hallucination_rate = (hallucinated_count / total_queries) * 100 if total_queries > 0 else 0

            return {
                "total_queries": total_queries,
                "hallucinated_queries": hallucinated_count,
                "hallucination_rate_percent": round(hallucination_rate, 2),
                "results": results
            }
        except Exception as e:
            logger.error(f"Evaluation failed: {str(e)}")
            raise

    def save_results(self, results: Dict[str, Any], output_path: str = "evaluation_results.json") -> None:
        """Save evaluation results to JSON file."""
        try:
            with open(output_path, "w") as f:
                json.dump(results, f, indent=2)
            logger.info(f"Results saved to {output_path}")
        except Exception as e:
            logger.error(f"Failed to save results: {str(e)}")
            raise

if __name__ == "__main__":
    # Example usage: assumes pipeline is already initialized
    from code_example1 import ProductionRAGPipeline  # In practice, import your pipeline

    pipeline = ProductionRAGPipeline(vectorstore_path="./chroma_db")
    pipeline.initialize()

    evaluator = RAGHallucinationEvaluator(
        pipeline=pipeline,
        reranker=pipeline.reranker,
        benchmark_path="./benchmark_qa.json"
    )

    results = evaluator.run_evaluation(sample_size=100)
    evaluator.save_results(results)
    print(f"Hallucination Rate: {results['hallucination_rate_percent']}%")

import time
import logging
from typing import Dict, Any, List
from prometheus_client import Counter, Histogram, Gauge, start_http_server
from langchain.callbacks.base import BaseCallbackHandler
from ragatouille import RAGPretrainedModel
from dotenv import load_dotenv
import os

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

load_dotenv()

# Prometheus metrics definitions
QUERY_COUNTER = Counter(
    "rag_queries_total",
    "Total number of RAG queries processed",
    ["pipeline_version", "status"]
)
LATENCY_HISTOGRAM = Histogram(
    "rag_query_latency_seconds",
    "RAG query latency in seconds",
    ["pipeline_version"],
    buckets=[0.5, 1.0, 1.5, 2.0, 2.5, 3.0, 5.0]
)
HALLUCINATION_GAUGE = Gauge(
    "rag_hallucination_rate_percent",
    "Current RAG hallucination rate percentage",
    ["pipeline_version"]
)
RERANKER_SCORE_GAUGE = Gauge(
    "rag_reranker_avg_score",
    "Average RAGatouille reranker score for queries",
    ["pipeline_version"]
)
INFERENCE_COST_GAUGE = Gauge(
    "rag_inference_cost_usd_per_1k_queries",
    "Estimated inference cost per 1000 queries",
    ["pipeline_version"]
)

class RAGMonitoringCallback(BaseCallbackHandler):
    """LangChain callback handler to track RAG pipeline metrics."""

    def __init__(self, pipeline_version: str = "langchain-0.3.12-ragatouille-0.7.4"):
        self.pipeline_version = pipeline_version
        self.query_start_time = None
        self.current_query = None

    def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs) -> None:
        """Record query start time and input."""
        self.query_start_time = time.time()
        self.current_query = inputs.get("input", "unknown")
        logger.debug(f"Chain started for query: {self.current_query}")

    def on_chain_end(self, outputs: Dict[str, Any], **kwargs) -> None:
        """Record query latency and success status."""
        if self.query_start_time:
            latency = time.time() - self.query_start_time
            LATENCY_HISTOGRAM.labels(pipeline_version=self.pipeline_version).observe(latency)
            QUERY_COUNTER.labels(pipeline_version=self.pipeline_version, status="success").inc()
            logger.debug(f"Chain ended for query: {self.current_query}, latency: {latency:.2f}s")

    def on_chain_error(self, error: Exception, **kwargs) -> None:
        """Record query failure."""
        QUERY_COUNTER.labels(pipeline_version=self.pipeline_version, status="error").inc()
        logger.error(f"Chain error for query: {self.current_query}, error: {str(error)}")

    def on_retriever_end(self, documents: List, **kwargs) -> None:
        """Track reranker scores for retrieved documents."""
        if hasattr(documents, "metadata") and "reranker_scores" in documents.metadata:
            scores = documents.metadata["reranker_scores"]
            avg_score = sum(scores) / len(scores) if scores else 0
            RERANKER_SCORE_GAUGE.labels(pipeline_version=self.pipeline_version).set(avg_score)

class RAGCostCalculator:
    """Calculate and track RAG inference costs."""

    def __init__(self, embedding_cost_per_1k: float = 0.0001, llm_cost_per_1k_tokens: float = 0.001):
        self.embedding_cost_per_1k = embedding_cost_per_1k
        self.llm_cost_per_1k_tokens = llm_cost_per_1k_tokens
        self.total_embedding_tokens = 0
        self.total_llm_tokens = 0

    def track_embedding_tokens(self, num_tokens: int) -> None:
        """Track embedding token usage."""
        self.total_embedding_tokens += num_tokens

    def track_llm_tokens(self, num_tokens: int) -> None:
        """Track LLM token usage."""
        self.total_llm_tokens += num_tokens

    def calculate_cost_per_1k_queries(self, num_queries: int) -> float:
        """Calculate cost per 1000 queries."""
        if num_queries == 0:
            return 0.0

        embedding_cost = (self.total_embedding_tokens / 1000) * self.embedding_cost_per_1k
        llm_cost = (self.total_llm_tokens / 1000) * self.llm_cost_per_1k_tokens
        total_cost = embedding_cost + llm_cost

        return (total_cost / num_queries) * 1000

def start_monitoring_server(port: int = 8000, pipeline_version: str = "langchain-0.3.12-ragatouille-0.7.4") -> None:
    """Start Prometheus metrics server and initialize monitoring."""
    try:
        start_http_server(port)
        logger.info(f"Prometheus metrics server started on port {port}")

        # Initialize callback handler
        callback = RAGMonitoringCallback(pipeline_version=pipeline_version)

        # Initialize cost calculator
        cost_calculator = RAGCostCalculator()

        return callback, cost_calculator
    except Exception as e:
        logger.error(f"Failed to start monitoring server: {str(e)}")
        raise

def update_hallucination_rate(hallucination_rate: float, pipeline_version: str) -> None:
    """Update hallucination rate gauge."""
    HALLUCINATION_GAUGE.labels(pipeline_version=pipeline_version).set(hallucination_rate)
    logger.info(f"Updated hallucination rate to {hallucination_rate:.2f}%")

if __name__ == "__main__":
    # Example usage: start monitoring server
    callback, cost_calc = start_monitoring_server(port=8000)

    # Simulate query processing
    for i in range(10):
        # Simulate latency
        time.sleep(0.8)
        # Simulate cost tracking
        cost_calc.track_embedding_tokens(512)
        cost_calc.track_llm_tokens(256)

        if i == 5:
            # Simulate a hallucinated query
            update_hallucination_rate(17.1, "langchain-0.3.12-ragatouille-0.7.4")

    # Calculate and log cost
    cost_per_1k = cost_calc.calculate_cost_per_1k_queries(10)
    INFERENCE_COST_GAUGE.labels(pipeline_version="langchain-0.3.12-ragatouille-0.7.4").set(cost_per_1k)
    logger.info(f"Cost per 1k queries: ${cost_per_1k:.2f}")

    # Keep server running
    while True:
        time.sleep(60)

# Short snippet to wrap RAGatouille reranker with LangChain retriever
from langchain.schema.runnable import RunnableLambda
from ragatouille import RAGPretrainedModel

reranker = RAGPretrainedModel.from_pretrained("colbert-ir/colbertv2.0")
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 20})

def rerank_retriever(query: str):
    docs = base_retriever.get_relevant_documents(query)
    doc_texts = [doc.page_content for doc in docs]
    reranked = reranker.rerank(query, doc_texts, k=5)
    return [docs[r["result_index"]] for r in reranked]

retriever = RunnableLambda(lambda x: rerank_retriever(x["input"]))

# Short snippet of LCEL Runnable pipeline
from langchain.schema.runnable import RunnablePassthrough
from langchain.chains import create_retrieval_chain

# Compose pipeline with LCEL
chain = (
    {"context": retriever, "input": RunnablePassthrough()}
    | create_stuff_documents_chain(llm, prompt)
    | RunnableLambda(lambda x: {"answer": x, "timestamp": time.time()})
)

# Short snippet to add hallucination score to pipeline output
def query_with_hallucination_score(query: str):
    result = pipeline.query(query)
    is_hallucinated = evaluator._is_hallucinated(
        query, result["answer"], result["answer"], result["source_documents"]
    )
    result["hallucination_score"] = 1.0 if is_hallucinated else 0.0
    return result