Three-Layer Hallucination Prevention in RAG Systems: A Practical Guide

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

class RetrievalFilter:
    def __init__(self, vector_store, llm):
        self.vector_store = vector_store
        self.llm = llm
    
    def filter_documents(self, query: str, threshold: float = 0.7):
        """Filter retrieved documents by relevance score."""
        # Get initial results
        results = self.vector_store.similarity_search_with_score(query, k=10)
        
        # Filter by score threshold
        filtered = [
            (doc, score) 
            for doc, score in results 
            if score >= threshold
        ]
        
        # Rerank using LLM
        reranked = self.rerank_with_llm(query, filtered)
        
        return reranked
    
    def rerank_with_llm(self, query: str, documents):
        """Use LLM to rerank documents by relevance."""
        prompt = f"""Rank these documents by relevance to the query: {query}
        
Documents:
{[doc.page_content for doc, _ in documents]}

Return the most relevant document indices (comma-separated):"""
        
        response = self.llm.predict(prompt)
        # Parse and return top documents
        # Implementation details...

HALLUCINATION_PREVENTION_PROMPT = """You are a helpful assistant. Answer the question based ONLY on the provided context.

Rules:
1. If the context doesn't contain enough information, say "I don't have enough information to answer this question."
2. Never make up information not present in the context.
3. Always cite the source document when possible.
4. If multiple sources conflict, present both perspectives.
5. For numerical data, quote exact figures from the context.

Context:
{context}

Question: {input}

Answer:"""

ECOMMERC_PROMPT = """You are a product expert. Answer questions about products using ONLY the provided product information.

Rules:
1. Never invent prices, specifications, or availability
2. If product info is incomplete, say "I don't have complete information about this product"
3. Always mention the product name and key features from the data
4. For comparisons, stick strictly to the provided data

Product Information:
{context}

Question: {input}
"""

AGRICULTURAL_PROMPT = """You are an agricultural expert. Answer farming questions using ONLY the provided knowledge base.

Rules:
1. Never invent pesticide dosages or treatment methods
2. If the knowledge base doesn't cover the specific crop/disease, recommend consulting a local expert
3. Always cite the source manual when possible
4. For safety-critical advice, include appropriate warnings

Knowledge Base:
{context}

Question: {input}
"""

import re
from typing import Dict, List, Tuple

class OutputValidator:
    def __init__(self, llm):
        self.llm = llm
    
    def validate_response(self, response: str, context: str) -> Dict:
        """Validate LLM response against source context."""
        validation_results = {
            "is_valid": True,
            "issues": [],
            "corrected_response": response
        }
        
        # Check 1: Numerical claims
        numerical_issues = self.check_numerical_claims(response, context)
        if numerical_issues:
            validation_results["issues"].extend(numerical_issues)
        
        # Check 2: Entity verification
        entity_issues = self.verify_entities(response, context)
        if entity_issues:
            validation_results["issues"].extend(entity_issues)
        
        # Check 3: Source citation
        citation_issues = self.check_citations(response, context)
        if citation_issues:
            validation_results["issues"].extend(citation_issues)
        
        # If issues found, generate corrected response
        if validation_results["issues"]:
            validation_results["corrected_response"] = self.generate_corrected_response(
                response, validation_results["issues"], context
            )
            validation_results["is_valid"] = False
        
        return validation_results
    
    def check_numerical_claims(self, response: str, context: str) -> List[str]:
        """Verify numerical values in response exist in context."""
        issues = []
        
        # Extract numbers from response
        numbers = re.findall(r'\b\d+(?:\.\d+)?(?:%|元|kg|ml)?\b', response)
        
        for number in numbers:
            if number not in context:
                issues.append(f"Numerical value '{number}' not found in source context")
        
        return issues
    
    def verify_entities(self, response: str, context: str) -> List[str]:
        """Verify product/entity names exist in context."""
        issues = []
        
        # Simple entity extraction (can be enhanced with NER)
        entities = re.findall(r'[A-Z][a-z]+(?:\s[A-Z][a-z]+)*', response)
        
        for entity in entities:
            if entity.lower() not in context.lower():
                issues.append(f"Entity '{entity}' not found in source context")
        
        return issues

from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate

class HallucinationPreventionSystem:
    def __init__(self, vector_store, llm):
        self.vector_store = vector_store
        self.llm = llm
        self.retrieval_filter = RetrievalFilter(vector_store, llm)
        self.output_validator = OutputValidator(llm)
    
    def query(self, user_input: str) -> Dict:
        """Process query with three-layer hallucination prevention."""
        
        # Layer 1: Filter retrieval
        filtered_docs = self.retrieval_filter.filter_documents(user_input)
        context = "\n\n".join([doc.page_content for doc in filtered_docs])
        
        # Layer 2: Generate response with constrained prompt
        prompt = ChatPromptTemplate.from_template(HALLUCINATION_PREVENTION_PROMPT)
        chain = LLMChain(llm=self.llm, prompt=prompt)
        response = chain.invoke({"context": context, "input": user_input})
        
        # Layer 3: Validate output
        validation = self.output_validator.validate_response(
            response["text"], context
        )
        
        return {
            "response": validation["corrected_response"],
            "is_valid": validation["is_valid"],
            "issues": validation["issues"],
            "sources": [doc.metadata.get("source") for doc in filtered_docs]
        }

class HallucinationMetrics:
    def __init__(self):
        self.metrics = {
            "total_queries": 0,
            "hallucinations_detected": 0,
            "false_positives": 0,
            "retrieval_accuracy": 0
        }
    
    def evaluate_response(self, response: str, context: str, ground_truth: str):
        """Evaluate response quality and hallucination presence."""
        self.metrics["total_queries"] += 1
        
        # Check for hallucinations
        if self.detect_hallucination(response, context):
            self.metrics["hallucinations_detected"] += 1
        
        # Compare with ground truth
        accuracy = self.calculate_accuracy(response, ground_truth)
        self.metrics["retrieval_accuracy"] += accuracy
    
    def detect_hallucination(self, response: str, context: str) -> bool:
        """Simple hallucination detection."""
        # Check if response contains information not in context
        response_words = set(response.lower().split())
        context_words = set(context.lower().split())
        
        # Words in response but not in context
        novel_words = response_words - context_words
        
        # If too many novel words, likely hallucination
        return len(novel_words) > len(response_words) * 0.3