Por Qué el 83% de Herramientas de Detección de Alucinaciones RAG Fallan en Producción (Y Cómo Solucionarlo)

"""
Pre-Generation Validation
Primera línea de defensa: prevenir alucinaciones antes de generation
"""

from typing import Dict, List
import logging

logger = logging.getLogger(__name__)


class PreGenerationValidator:
    """
    Valida que tenemos suficiente contexto de calidad ANTES de generar.
    Previene alucinaciones rechazando queries sin retrieval confiable.
    """

    def __init__(
        self,
        min_retrieval_score: float = 0.7,
        min_context_length: int = 100,
        max_context_length: int = 4000
    ):
        self.min_retrieval_score = min_retrieval_score
        self.min_context_length = min_context_length
        self.max_context_length = max_context_length

    def validate(
        self,
        query: str,
        retrieved_docs: List[Dict],
        retrieval_scores: List[float]
    ) -> Dict:
        """
        Valida si podemos generar respuesta confiable con contexto actual.

        Returns:
            {
                'should_generate': bool,
                'reason': str,
                'context': str | None,
                'confidence_estimate': float
            }
        """
        # Check #1: ¿Tenemos docs recuperados?
        if not retrieved_docs:
            return {
                'should_generate': False,
                'reason': 'No se encontraron documentos relevantes',
                'context': None,
                'confidence_estimate': 0.0
            }

        # Check #2: ¿Retrieval score es suficiente?
        best_score = max(retrieval_scores) if retrieval_scores else 0.0
        if best_score < self.min_retrieval_score:
            return {
                'should_generate': False,
                'reason': f'Retrieval score demasiado bajo ({best_score:.2f} < {self.min_retrieval_score})',
                'context': None,
                'confidence_estimate': best_score * 0.5  # Penalizar
            }

        # Check #3: ¿Contexto combinado tiene longitud adecuada?
        combined_context = "\n\n".join([doc['content'] for doc in retrieved_docs])

        if len(combined_context) < self.min_context_length:
            return {
                'should_generate': False,
                'reason': f'Contexto demasiado corto ({len(combined_context)} chars)',
                'context': None,
                'confidence_estimate': 0.3
            }

        if len(combined_context) > self.max_context_length:
            # Truncate pero warn
            logger.warning(
                f"Contexto truncado de {len(combined_context)} a {self.max_context_length} chars"
            )
            combined_context = combined_context[:self.max_context_length]

        # Check #4: Query classification - ¿es answerable?
        is_answerable = self._classify_query_answerability(query, combined_context)
        if not is_answerable:
            return {
                'should_generate': False,
                'reason': 'Query no es answerable con contexto disponible',
                'context': combined_context,
                'confidence_estimate': 0.4
            }

        # All checks passed
        confidence = min(best_score, 0.95)  # Cap en 0.95 porque siempre hay incertidumbre

        return {
            'should_generate': True,
            'reason': 'Validación exitosa',
            'context': combined_context,
            'confidence_estimate': confidence
        }

    def _classify_query_answerability(self, query: str, context: str) -> bool:
        """
        Clasifica si query es answerable con context disponible.
        Heurística simple (para producción, usar clasificador ML).
        """
        query_lower = query.lower()

        # Red flags: queries que típicamente requieren conocimiento externo
        external_knowledge_keywords = [
            'última', 'newest', 'latest', 'hoy', 'ayer',
            'esta semana', 'precio actual', 'current price',
            'now', 'real-time'
        ]

        requires_external = any(kw in query_lower for kw in external_knowledge_keywords)

        if requires_external:
            # Check si context menciona timestamps recientes
            has_recent_info = any(year in context for year in ['2024', '2025'])
            return has_recent_info

        # Por defecto, asumir answerable si pasó retrieval
        return True


# Ejemplo de uso en pipeline
def rag_pipeline_with_validation(query: str):
    """RAG pipeline con pre-generation validation."""

    # 1. Retrieval
    retrieved_docs, scores = vector_store.similarity_search_with_score(query, k=5)

    # 2. Pre-Generation Validation
    validator = PreGenerationValidator(
        min_retrieval_score=0.7,
        min_context_length=100
    )

    validation_result = validator.validate(
        query=query,
        retrieved_docs=[{'content': doc.page_content} for doc in retrieved_docs],
        retrieval_scores=scores
    )

    # 3. Decision: generar o rechazar
    if not validation_result['should_generate']:
        return {
            'status': 'REJECTED_PRE_GENERATION',
            'response': f"No puedo responder esta pregunta con confianza. Razón: {validation_result['reason']}",
            'confidence': validation_result['confidence_estimate']
        }

    # 4. Continuar con generation (solo si validación pasa)
    context = validation_result['context']

    # ... resto del pipeline

# Prompt con guardrails anti-hallucination
system_prompt = """Eres un asistente de IA confiable que responde SOLO basándote en el contexto proporcionado.

REGLAS CRÍTICAS:
1. Si la información NO está en el contexto, di "No tengo esa información"
2. NO inventes datos, nombres, fechas o cifras
3. Si hay múltiples interpretaciones, menciona la ambigüedad
4. Cita específicamente qué parte del contexto soporta tu respuesta
5. Prefiere respuestas cortas y factuales sobre respuestas largas y elaboradas

Tu respuesta debe ser VERIFICABLE contra el contexto proporcionado."""

user_prompt = f"""Contexto:
{context}

Pregunta: {query}

Respuesta (basada SOLO en el contexto):"""

response = openai.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt}
    ],
    temperature=0.0,      # Determinístico
    max_tokens=300,       # Limita verbosidad
    presence_penalty=0.0,
    frequency_penalty=0.0
)

# Stage 3: Hallucination Detection
detector = HybridHallucinationDetector(...)  # O TLMDetector, BERTDetector

detection_result = detector.detect_hallucination(
    query=query,
    context=context,
    response=generated_response
)

if detection_result['is_hallucination']:
    # Log para análisis
    logger.warning(f"Hallucination detected: {detection_result['reasoning']}")

    # Retornar rechazo al usuario
    return {
        'status': 'HALLUCINATION_DETECTED',
        'response': "Lo siento, no puedo proporcionar una respuesta confiable para esta pregunta.",
        'detection_details': detection_result  # Para debugging
    } 

def get_dynamic_threshold(query: str, user_context: Dict) -> float:
    """Determina threshold basado en query type y user context."""

    # Clasificar query
    is_financial = any(
        kw in query.lower()
        for kw in ['precio', 'cost', 'revenue', 'financ']
    )

    is_legal = any(
        kw in query.lower()
        for kw in ['legal', 'contrato', 'policy', 'compliance']
    )

    is_medical = any(
        kw in query.lower()
        for kw in ['diagnóstico', 'tratamiento', 'médico', 'salud']
    )

    # Threshold por tipo
    if is_legal or is_medical:
        return 0.9  # Muy estricto
    elif is_financial:
        return 0.7  # Moderado
    else:
        return 0.5  # General


# Uso
threshold = get_dynamic_threshold(query, user_metadata)

if detection_result['confidence_score'] < threshold:
    # Rechazar
    pass
else:
    # Mostrar respuesta
    pass

# Logging completo con structured data
logger.info(
    "RAG detection result",
    extra={
        'query_id': query_id,
        'user_id': user_id,
        'query_text': query[:100],  # Truncate para privacy
        'retrieval_score': best_retrieval_score,
        'detection_method': 'hybrid',
        'is_hallucination': detection_result['is_hallucination'],
        'confidence_score': detection_result['confidence_score'],
        'threshold_used': threshold,
        'latency_ms': detection_result['latency_ms'],
        'cost_usd': estimated_cost,
        'timestamp': datetime.utcnow().isoformat()
    }
)

# Dashboard metrics (Prometheus, Datadog, etc.)
metrics.increment('rag.queries_total')

if detection_result['is_hallucination']:
    metrics.increment('rag.hallucinations_detected')

metrics.histogram('rag.detection_latency_ms', detection_result['latency_ms']) 

"""
Legal RAG Pipeline con Hybrid Detection + Human Review
"""

from enum import Enum


class QueryType(Enum):
    SIMPLE_LOOKUP = "simple"      # "¿Qué dice artículo X?"
    CASE_CITATION = "citation"    # "Encuentra casos similares"
    MULTI_STEP = "reasoning"       # "¿Cómo aplica precedente?"


def classify_legal_query(query: str) -> QueryType:
    """Clasifica tipo de query legal."""
    q = query.lower()

    if any(kw in q for kw in ['qué dice', 'artículo', 'sección', 'definición']):
        return QueryType.SIMPLE_LOOKUP

    if any(kw in q for kw in ['casos similares', 'precedentes', 'jurisprudencia']):
        return QueryType.CASE_CITATION

    # Default: reasoning (más conservador)
    return QueryType.MULTI_STEP


def legal_rag_pipeline(query: str, user_id: str):
    """Pipeline completo legal RAG."""

    # 1. Classify query type
    query_type = classify_legal_query(query)

    # 2. Get dynamic threshold
    thresholds = {
        QueryType.SIMPLE_LOOKUP: 0.5,
        QueryType.CASE_CITATION: 0.7,
        QueryType.MULTI_STEP: 0.9
    }
    threshold = thresholds[query_type]

    # 3. Retrieval con reranking
    retrieved_docs = vector_store.similarity_search(query, k=10)
    reranked_docs = cross_encoder_reranker.rerank(query, retrieved_docs, top_k=5)
    context = "\n\n".join([doc.page_content for doc in reranked_docs])

    # 4. Generation
    response = llm.generate(query, context)

    # 5. Hybrid Detection
    detection = hybrid_detector.detect_hallucination(query, context, response)

    # 6. Decision Logic
    confidence = detection['confidence_score']

    if confidence >= 0.9:
        # AUTO-APPROVED
        return {
            'status': 'APPROVED',
            'response': response,
            'confidence': confidence
        }

    elif confidence >= threshold:
        # PENDING REVIEW (human-in-loop)
        ticket_id = human_review_queue.create_ticket({
            'query': query,
            'response': response,
            'confidence': confidence,
            'query_type': query_type.value,
            'user_id': user_id,
            'priority': 'HIGH' if query_type == QueryType.MULTI_STEP else 'MEDIUM'
        })

        return {
            'status': 'PENDING_REVIEW',
            'response': f"Tu consulta está siendo revisada por nuestro equipo legal. "
                        f"Recibirás respuesta en 2-4 horas. Ticket: {ticket_id}",
            'ticket_id': ticket_id
        }

    else:
        # REJECTED
        return {
            'status': 'REJECTED',
            'response': "Para consultas de esta complejidad, te recomendamos contactar "
                        "directamente con un abogado senior de nuestro equipo."
        }

# Breakdown detallado Hybrid Detection (100k queries/mes)

# Stage 1: Token Similarity
stage1_queries = 100_000
stage1_cost_per_query = 0.00000  # Negligible (CPU local)
stage1_total = 0  # €

# Stage 2: LLM-as-Judge (solo ~30% queries complejas)
stage2_queries = 30_000  # 30% llegan a stage 2
stage2_cost_per_query = 0.001  # GPT-4 API call
stage2_total = 30  # €

# Total mensual
total_cost = stage1_total + stage2_total  # 30€/mes

# Ahorro vs RAGAS
ragas_cost = 200  # € - 100k queries * 0.002€
savings = ragas_cost - total_cost  # 170€/mes (85% reducción)

print(f"Hybrid Detection: {total_cost}€/mes")
print(f"Ahorro vs RAGAS: {savings}€/mes ({savings/ragas_cost:.0%})")

"""
TLM Hallucination Detector - Production Implementation
Integración con LangChain para sistemas RAG
"""

from cleanlab_studio import Studio
from langchain.callbacks.base import BaseCallbackHandler
import os


class TLMHallucinationDetector:
    """
    Detector de alucinaciones usando TLM de Cleanlab.
    95% AUROC en benchmarks multi-dataset.
    """

    def __init__(self, api_key: str = None):
        """
        Inicializa TLM detector.
        
        Args:
            api_key: Cleanlab Studio API key (o usa CLEANLAB_API_KEY env var)
        """
        self.api_key = api_key or os.getenv("CLEANLAB_API_KEY")
        if not self.api_key:
            raise ValueError("Cleanlab API key requerida")

        self.studio = Studio(self.api_key)
        self.tlm = self.studio.TLM()

    def detect_hallucination(
        self, 
        query: str, 
        context: str, 
        response: str,
        threshold: float = 0.7
    ) -> dict:
        """
        Detecta si response contiene alucinaciones dado context.
        
        Args:
            query: Pregunta del usuario
            context: Contexto recuperado por RAG
            response: Respuesta generada por LLM
            threshold: Score mínimo para considerarlo confiable (0.0-1.0)
        
        Returns:
            {
                'is_hallucination': bool,
                'trustworthiness_score': float,  # 0.0-1.0
                'should_show_response': bool,
                'confidence': float
            }
        """
        # TLM evalúa trustworthiness de response dado context
        result = self.tlm.get_trustworthiness_score(
            prompt=query,
            response=response,
            context=context
        )

        trustworthiness = result['trustworthiness_score']

        return {
            'is_hallucination': trustworthiness < threshold,
            'trustworthiness_score': trustworthiness,
            'should_show_response': trustworthiness >= threshold,
            'confidence': result.get('confidence', 0.0)
        }


# Ejemplo de uso en pipeline RAG
def rag_pipeline_with_tlm(query: str):
    """Pipeline RAG con TLM hallucination detection."""

    # 1. Retrieval (ejemplo con vector store)
    retrieved_docs = vector_store.similarity_search(query, k=5)
    context = "\n\n".join([doc.page_content for doc in retrieved_docs])

    # 2. Generation
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    response = llm.predict(
        f"Context: {context}\n\n"
        f"Question: {query}\n\n"
        f"Answer:"
    )

    # 3. Hallucination Detection con TLM
    detector = TLMHallucinationDetector()
    detection_result = detector.detect_hallucination(
        query=query,
        context=context,
        response=response,
        threshold=0.7  # 70% confianza mínima
    )

    # 4. Decision Logic
    if detection_result['is_hallucination']:
        return {
            'status': 'HALLUCINATION_DETECTED',
            'response': "Lo siento, no tengo suficiente información confiable para responder esta pregunta.",
            'trustworthiness_score': detection_result['trustworthiness_score'],
            'original_response': response  # Log para análisis
        }
    else:
        return {
            'status': 'OK',
            'response': response,
            'trustworthiness_score': detection_result['trustworthiness_score']
        }

"""
Hybrid Hallucination Detector - Production Implementation
Combina Token Similarity (rápido) + LLM-as-Judge (preciso)
72% mejora vs single-method según benchmarks
"""

from typing import Dict, List, Set
import re
from collections import Counter
from openai import OpenAI
import logging

logger = logging.getLogger(__name__)


class HybridHallucinationDetector:
    """
    Detector híbrido de dos etapas para alucinaciones RAG.
    Stage 1: Token similarity (fast, cheap)
    Stage 2: LLM-as-judge (slow, expensive pero accurate)
    """

    def __init__(
        self,
        openai_api_key: str,
        token_similarity_threshold: float = 0.5,
        llm_confidence_threshold: float = 0.7,
        stage1_only_for_simple: bool = True
    ):
        """
        Args:
            token_similarity_threshold: Score mínimo stage 1 (0.0-1.0)
            llm_confidence_threshold: Score mínimo stage 2 (0.0-1.0)
            stage1_only_for_simple: Si True, queries simples solo usan stage 1
        """
        self.openai_client = OpenAI(api_key=openai_api_key)
        self.token_threshold = token_similarity_threshold
        self.llm_threshold = llm_confidence_threshold
        self.stage1_only_simple = stage1_only_for_simple

    def detect_hallucination(
        self,
        query: str,
        context: str,
        response: str
    ) -> Dict:
        """
        Detecta alucinaciones usando enfoque híbrido de dos etapas.
        
        Returns:
            {
                'is_hallucination': bool,
                'stage_used': str,  # 'stage1_only' o 'stage2_deep'
                'token_similarity_score': float,
                'llm_confidence_score': float | None,
                'reasoning': str,
                'latency_ms': int
            }
        """
        import time
        start_time = time.time()

        # STAGE 1: Token Similarity Detection (rápido)
        token_score = self._calculate_token_similarity(context, response)

        # Si token similarity es MUY bajo, es alucinación obvia
        if token_score < 0.3:
            return {
                'is_hallucination': True,
                'stage_used': 'stage1_only',
                'token_similarity_score': token_score,
                'llm_confidence_score': None,
                'reasoning': f"Alucinación obvia: {token_score:.1%} tokens en response aparecen en context",
                'latency_ms': int((time.time() - start_time) * 1000)
            }

        # Si token similarity es ALTO y query es simple, skip stage 2
        if token_score >= 0.8 and self._is_simple_query(query):
            return {
                'is_hallucination': False,
                'stage_used': 'stage1_only',
                'token_similarity_score': token_score,
                'llm_confidence_score': None,
                'reasoning': f"Alta similitud tokens ({token_score:.1%}) + query simple",
                'latency_ms': int((time.time() - start_time) * 1000)
            }

        # STAGE 2: LLM-as-Judge (análisis profundo)
        llm_result = self._llm_as_judge(query, context, response)
        is_hallucination = llm_result['confidence_score'] < self.llm_threshold

        return {
            'is_hallucination': is_hallucination,
            'stage_used': 'stage2_deep',
            'token_similarity_score': token_score,
            'llm_confidence_score': llm_result['confidence_score'],
            'reasoning': llm_result['reasoning'],
            'latency_ms': int((time.time() - start_time) * 1000)
        }

    def _calculate_token_similarity(self, context: str, response: str) -> float:
        """
        Calcula porcentaje de tokens en response que aparecen en context.
        Método rápido para detectar información completamente inventada.
        """
        # Tokenización simple (para producción, usa tokenizer real)
        context_tokens = set(self._tokenize(context.lower()))
        response_tokens = self._tokenize(response.lower())

        if not response_tokens:
            return 0.0

        # Cuántos tokens de response aparecen en context
        matching_tokens = sum(
            1 for token in response_tokens
            if token in context_tokens
        )

        similarity = matching_tokens / len(response_tokens)
        return similarity

    def _tokenize(self, text: str) -> List[str]:
        """Tokenización simple. Para producción, usar tiktoken o similar."""
        # Remove punctuation, split on whitespace
        text = re.sub(r'[^\w\s]', ' ', text)
        tokens = text.split()

        # Filter stopwords comunes (versión simplificada)
        stopwords = {
            'el', 'la', 'de', 'en', 'y', 'a', 'que', 'es', 'por', 'un', 'una',
            'the', 'is', 'and', 'to', 'of'
        }

        return [t for t in tokens if t not in stopwords and len(t) > 2]

    def _is_simple_query(self, query: str) -> bool:
        """
        Clasifica query como simple vs compleja.
        Simple: facts directos, lookups
        Compleja: razonamiento, multi-step, comparaciones
        """
        # Heurísticas simples (para producción, usar clasificador ML)
        complexity_keywords = [
            'por qué', 'cómo', 'compara', 'diferencia', 'explica',
            'why', 'how', 'compare', 'explain', 'difference'
        ]

        query_lower = query.lower()
        is_complex = any(kw in query_lower for kw in complexity_keywords)

        return not is_complex and len(query.split()) < 15

    def _llm_as_judge(self, query: str, context: str, response: str) -> Dict:
        """
        Usa GPT-4 como judge para evaluar si response está soportado por context.
        """
        prompt = f"""Eres un evaluador experto de sistemas RAG.
Tu tarea: determinar si la RESPUESTA está completamente soportada por el CONTEXTO proporcionado.

CONSULTA USUARIO:
{query}

CONTEXTO RECUPERADO:
{context}

RESPUESTA GENERADA:
{response}

INSTRUCCIONES:
1. Identifica cada claim factual en la respuesta
2. Verifica si cada claim está explícitamente soportado por el contexto
3. Asigna un confidence score de 0.0 a 1.0:
    - 1.0 = Todos los claims completamente soportados
    - 0.5-0.9 = Mayoría soportados, algunos inferidos
    - 0.0-0.4 = Claims contradicen contexto o están inventados

Responde en formato JSON:
{{
    "confidence_score": 0.0-1.0,
    "reasoning": "Explicación de tu evaluación",
    "unsupported_claims": ["claim 1", "claim 2"]
}}"""

        try:
            completion = self.openai_client.chat.completions.create(
                model="gpt-4",
                messages=[
                    {
                        "role": "system",
                        "content": "Eres un evaluador experto de alucinaciones en sistemas RAG."
                    },
                    {
                        "role": "user",
                        "content": prompt
                    }
                ],
                temperature=0,
                response_format={"type": "json_object"}
            )

            import json
            result = json.loads(completion.choices[0].message.content)

            return {
                'confidence_score': float(result.get('confidence_score', 0.5)),
                'reasoning': result.get('reasoning', ''),
                'unsupported_claims': result.get('unsupported_claims', [])
            }

        except Exception as e:
            logger.error(f"LLM-as-judge failed: {e}")
            # Fallback conservador: asumir posible alucinación
            return {
                'confidence_score': 0.5,
                'reasoning': f"Error en evaluación LLM: {str(e)}",
                'unsupported_claims': []
            }


# Ejemplo de uso
if __name__ == "__main__":
    detector = HybridHallucinationDetector(
        openai_api_key="tu-api-key",
        token_similarity_threshold=0.5,
        llm_confidence_threshold=0.7
    )

    # Caso de prueba: alucinación numérica
    result = detector.detect_hallucination(
        query="¿Cuál fue el revenue de Apple Q4 2023?",
        context="Apple Inc. reportó ingresos de $89.5 mil millones en Q4 2023",
        response="Apple generó $120 mil millones en Q4 2023"  # ALUCINACIÓN
    )

    print(f"Hallucination: {result['is_hallucination']}")
    print(f"Stage: {result['stage_used']}")
    print(f"Token similarity: {result['token_similarity_score']:.2%}")
    print(f"LLM confidence: {result['llm_confidence_score']}")
    print(f"Reasoning: {result['reasoning']}")
    print(f"Latency: {result['latency_ms']}ms")

"""
BERT Stochastic Hallucination Detector
Basado en SelfCheckGPT approach: generación múltiple + consistency check
"""

from transformers import AutoTokenizer, AutoModel, AutoModelForSequenceClassification
import torch
from typing import List, Dict
import numpy as np


class BERTStochasticDetector:
    """
    Detector basado en consistency check de múltiples generaciones.
    Highest recall pero mayor latencia (~2-3s).
    """

    def __init__(
        self,
        bert_model: str = "sentence-transformers/all-MiniLM-L6-v2",
        cross_encoder_model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2",
        num_generations: int = 5,
        consistency_threshold: float = 0.8
    ):
        """
        Args:
            bert_model: BERT para similarity scoring
            cross_encoder_model: Cross-encoder para reranking
            num_generations: Cuántas respuestas generar para consistency check
            consistency_threshold: Score mínimo para considerar consistentes
        """
        self.num_generations = num_generations
        self.consistency_threshold = consistency_threshold

        # Load BERT para embeddings
        self.tokenizer = AutoTokenizer.from_pretrained(bert_model)
        self.bert_model = AutoModel.from_pretrained(bert_model)

        # Load cross-encoder para reranking
        self.cross_encoder_tokenizer = AutoTokenizer.from_pretrained(cross_encoder_model)
        self.cross_encoder = AutoModelForSequenceClassification.from_pretrained(cross_encoder_model)

    def detect_hallucination(
        self,
        query: str,
        context: str,
        llm_generation_func: callable
    ) -> Dict:
        """
        Detecta alucinaciones generando múltiples respuestas y verificando consistency.
        
        Args:
            query: Consulta usuario
            context: Contexto RAG
            llm_generation_func: Función que genera respuesta dado query + context
                                Debe aceptar temperature parameter
        
        Returns:
            {
                'is_hallucination': bool,
                'best_response': str,
                'consistency_score': float,
                'all_responses': List[str],
                'variance_analysis': str
            }
        """
        # Paso 1: Genera múltiples respuestas con sampling
        responses = []
        for i in range(self.num_generations):
            response = llm_generation_func(
                query=query,
                context=context,
                temperature=0.7 + (i * 0.1)  # Varía temperature para diversity
            )
            responses.append(response)

        # Paso 2: Calcula consistency score con BERT
        consistency_score = self._calculate_consistency(responses)

        # Paso 3: Si consistency es baja, probablemente alucinación
        is_hallucination = consistency_score < self.consistency_threshold

        # Paso 4: Rerank responses con cross-encoder, selecciona best
        best_response = self._rerank_responses(query, context, responses)[0]

        # Paso 5: Analiza variance para reasoning
        variance_analysis = self._analyze_variance(responses, consistency_score)

        return {
            'is_hallucination': is_hallucination,
            'best_response': best_response,
            'consistency_score': consistency_score,
            'all_responses': responses,
            'variance_analysis': variance_analysis
        }

    def _calculate_consistency(self, responses: List[str]) -> float:
        """
        Calcula consistency score entre múltiples responses usando BERT similarity.
        """
        # Get BERT embeddings para cada response
        embeddings = []
        for response in responses:
            inputs = self.tokenizer(
                response,
                return_tensors="pt",
                truncation=True,
                max_length=512,
                padding=True
            )

            with torch.no_grad():
                outputs = self.bert_model(**inputs)
                # Mean pooling
                embedding = outputs.last_hidden_state.mean(dim=1)
                embeddings.append(embedding)

        # Calcula pairwise cosine similarity
        similarities = []
        for i in range(len(embeddings)):
            for j in range(i + 1, len(embeddings)):
                sim = torch.nn.functional.cosine_similarity(
                    embeddings[i],
                    embeddings[j]
                )
                similarities.append(sim.item())

        # Consistency = promedio de similarities
        consistency = np.mean(similarities) if similarities else 0.0

        return consistency

    def _rerank_responses(
        self,
        query: str,
        context: str,
        responses: List[str]
    ) -> List[str]:
        """
        Rerank responses usando cross-encoder basado en relevancia a context.
        """
        scores = []

        for response in responses:
            # Cross-encoder score: qué tan bien response coincide con context
            inputs = self.cross_encoder_tokenizer(
                context,
                response,
                return_tensors="pt",
                truncation=True,
                max_length=512,
                padding=True
            )

            with torch.no_grad():
                logits = self.cross_encoder(**inputs).logits
                score = logits[0][0].item()  # Relevance score
                scores.append(score)

        # Sort responses por score descendente
        ranked = sorted(
            zip(responses, scores),
            key=lambda x: x[1],
            reverse=True
        )

        return [resp for resp, score in ranked]

    def _analyze_variance(self, responses: List[str], consistency: float) -> str:
        """Genera análisis explicativo de variance en responses."""
        if consistency >= 0.9:
            return "Todas las respuestas son altamente consistentes. Alta confianza."
        elif consistency >= 0.7:
            return "Respuestas mayormente consistentes con variaciones menores en wording."
        elif consistency >= 0.5:
            return "Variación moderada entre respuestas. Posible incertidumbre del modelo."
        else:
            return "Alta variación entre respuestas. Indica posibles alucinaciones o conocimiento insuficiente."


# Ejemplo de uso
if __name__ == "__main__":
    from openai import OpenAI

    # Setup OpenAI client
    openai_client = OpenAI(api_key="tu-api-key")

    # Define generation function
    def generate_response(query: str, context: str, temperature: float) -> str:
        completion = openai_client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": f"Context: {context}"},
                {"role": "user", "content": query}
            ],
            temperature=temperature
        )
        return completion.choices[0].message.content

    # Initialize detector
    detector = BERTStochasticDetector(
        num_generations=5,
        consistency_threshold=0.8
    )

    # Test
    result = detector.detect_hallucination(
        query="¿Cuál fue el revenue de Apple Q4 2023?",
        context="Apple Inc. reportó ingresos de $89.5 mil millones en Q4 2023",
        llm_generation_func=generate_response
    )

    print(f"Hallucination: {result['is_hallucination']}")
    print(f"Consistency: {result['consistency_score']:.2%}")
    print(f"Best response: {result['best_response']}")
    print(f"Analysis: {result['variance_analysis']}") 

"""
Shadow Mode: Correr RAGAS + Hybrid en paralelo sin afectar producción
"""

async def rag_pipeline_shadow_mode(query: str, context: str, response: str):
    """
    Producción usa RAGAS (current).
    Hybrid Detection corre en background, solo logea resultados.
    """
    # PRODUCCIÓN: RAGAS (como siempre)
    ragas_result = ragas_detector.evaluate(query, context, response)

    # Production decision basada en RAGAS (sin cambios)
    should_show_response = ragas_result['faithfulness_score'] >= 0.7

    # SHADOW: Hybrid Detection (no afecta decisión)
    # Corre async en background
    asyncio.create_task(
        run_hybrid_detector_shadow(query, context, response, ragas_result)
    )

    return {
        'response': response if should_show_response else "No puedo responder...",
        'ragas_score': ragas_result['faithfulness_score']
    }


async def run_hybrid_detector_shadow(query, context, response, ragas_result):
    """Background task: corre Hybrid, compara con RAGAS, logea."""

    hybrid_result = hybrid_detector.detect_hallucination(query, context, response)

    # Comparación
    ragas_says_hallucination = ragas_result['faithfulness_score'] < 0.7
    hybrid_says_hallucination = hybrid_result['is_hallucination']
    agreement = ragas_says_hallucination == hybrid_says_hallucination

    # Log para análisis
    logger.info(
        "Shadow mode comparison",
        extra={
            'query_id': generate_query_id(),
            'ragas_score': ragas_result['faithfulness_score'],
            'hybrid_score': hybrid_result['confidence_score'],
            'ragas_hallucination': ragas_says_hallucination,
            'hybrid_hallucination': hybrid_says_hallucination,
            'agreement': agreement,
            'hybrid_latency_ms': hybrid_result['latency_ms']
        }
    )

# Ejemplo: análisis threshold optimal

# Data de human review
human_labels = [1, 1, 0, 1, 0, ...]  # 1=correct, 0=hallucination
hybrid_scores = [0.92, 0.85, 0.45, 0.78, 0.32, ...]

# Probar diferentes thresholds
for threshold in [0.5, 0.6, 0.7, 0.8, 0.9]:
    predictions = [1 if score >= threshold else 0 for score in hybrid_scores]

    precision = calculate_precision(human_labels, predictions)
    recall = calculate_recall(human_labels, predictions)
    f1 = 2 * (precision * recall) / (precision + recall)

    print(f"Threshold {threshold}: P={precision:.2%}, R={recall:.2%}, F1={f1:.2%}")

# Output típico:
# Threshold 0.5: P=78%, R=92%, F1=85% ← Alto recall pero más false positives
# Threshold 0.7: P=88%, R=85%, F1=86% ← BALANCED (usual recomendación)
# Threshold 0.9: P=95%, R=70%, F1=81% ← Muy estricto, pierde hallucinations 

"""
Feature flag based gradual rollout
"""

import random
import hashlib

HYBRID_ROLLOUT_PERCENTAGE = 0.10  # Start 10%, increase gradualmente


def should_use_hybrid_detector(user_id: str) -> bool:
    """
    Determina si este usuario debe usar Hybrid (vs RAGAS legacy).
    Usa consistent hashing para que mismo user siempre tenga mismo detector.
    """
    # Hash user_id para 0-1 range
    hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100 / 100
    return hash_val < HYBRID_ROLLOUT_PERCENTAGE


def rag_pipeline_with_rollout(query: str, user_id: str):
    """Pipeline con gradual rollout."""

    # ... retrieval y generation ...

    # Detection: RAGAS (legacy) vs Hybrid (new)
    if should_use_hybrid_detector(user_id):
        # NEW: Hybrid Detection
        detection_result = hybrid_detector.detect_hallucination(query, context, response)
        is_hallucination = detection_result['is_hallucination']

        # Log que usamos Hybrid
        logger.info("Using Hybrid detector", extra={'user_id': user_id})
    else:
        # LEGACY: RAGAS
        ragas_result = ragas_detector.evaluate(query, context, response)
        is_hallucination = ragas_result['faithfulness_score'] < 0.7

        logger.info("Using RAGAS detector", extra={'user_id': user_id})

    # ... resto del pipeline ...

Usuario: "¿Cuál fue el revenue de Apple en Q4 2023?" 
Contexto recuperado: "Apple Inc. reportó ingresos de $89.5 mil millones en el cuarto trimestre fiscal 2023..." 
Respuesta RAG: "$89.5 mil millones" 
RAGAS Faithfulness: ❌ FALLA GENERAR SCORE (devuelve None o 0.0) 

Usuario: "¿Puedo aplicar tarifas de duelo después de mi viaje?" 
Respuesta ALUCINADA: "Sí, puedes aplicar dentro de 90 días posteriores al viaje" 
Respuesta CORRECTA: "No, debes aplicar ANTES del viaje"
RAGAS Answer Relevancy Score: 0.95 (ALTO - porque habla de tarifas de duelo) 
RAGAS Faithfulness Score: 0.0 (correcto, detecta alucinación) 
Problema: Si solo usas Answer Relevancy, NO DETECTAS la alucinación. 

"No tengo suficiente información en mi base de conocimiento para responder esta pregunta con confianza. Por favor, consulta directamente con el departamento de políticas."

# Contexto problemático para TruLens context = """ 
Producto A: - Precio: $99 - Stock: 50 unidades 
Producto B: - Precio: $149 - Stock: 0 unidades """ 
# TruLens a menudo falla parsear bullet points y líneas múltiples 
# Resultado: score None o error de evaluación 

"""
Contradiction Detector: segunda capa detección para false negatives
"""

from openai import OpenAI


def detect_contradiction(context: str, response: str) -> dict:
    """
    Usa LLM específicamente entrenado para detectar contradicciones.
    Complementa faithfulness detection.
    """
    client = OpenAI()

    prompt = f"""Analiza si la RESPUESTA contradice información en el CONTEXTO.

CONTEXTO:
{context}

RESPUESTA:
{response}

¿Hay alguna contradicción directa? Responde en JSON:
{{
    "has_contradiction": true/false,
    "contradiction_description": "explicación si la hay",
    "confidence": 0.0-1.0
}}"""

    completion = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        response_format={"type": "json_object"}
    )

    import json
    result = json.loads(completion.choices[0].message.content)
    return result


# Uso: después de faithfulness detection
faithfulness_result = hybrid_detector.detect_hallucination(query, context, response)

if faithfulness_result['confidence_score'] >= 0.7:
    # Pasó faithfulness, pero verificar contradicciones
    contradiction_result = detect_contradiction(context, response)

    if contradiction_result['has_contradiction'] and contradiction_result['confidence'] > 0.8:
        # Marca como hallucination a pesar de faithfulness score alto
        return {
            'status': 'CONTRADICTION_DETECTED',
            'response': "Detecté información contradictoria. No puedo dar una respuesta confiable.",
            'details': contradiction_result
        }
 

# Preprocessing para mejorar RAGAS compatibility
def preprocess_response_for_ragas(response: str) -> str:
    """
    Convierte respuestas problemáticas a formato que RAGAS puede procesar.
    """
    # Si es número solo, expandir a sentencia
    if response.strip().replace('.', '').replace(',', '').isdigit():
        response = f"La respuesta es: {response}."

    # Si es lista bullets, convertir a prosa
    if '\n-' in response or '\n•' in response:
        items = [
            line.strip('- •').strip()
            for line in response.split('\n')
            if line.strip()
        ]
        response = "La respuesta incluye los siguientes puntos: " + ", ".join(items) + "."

    # Truncate si muy largo (RAGAS tiene límite)
    if len(response) > 2000:
        response = response[:2000] + "..."

    return response


# Uso
processed_response = preprocess_response_for_ragas(original_response)
ragas_score = ragas_evaluator.evaluate(query, context, processed_response)

"""
Latency optimization strategies
"""

import asyncio
import hashlib
from functools import lru_cache


# Strategy #1: Async detection (no bloquea response)
async def async_detection_workflow(query: str, context: str, response: str):
    """
    Muestra respuesta inmediatamente, detection corre en background.
    Si detector marca hallucination, retractamos.
    """
    # Mostrar respuesta AHORA
    yield {
        'status': 'STREAMING',
        'response': response,
        'confidence': 'pending'
    }

    # Detection async en background
    detection_task = asyncio.create_task(
        detector.detect_hallucination(query, context, response)
    )

    detection_result = await detection_task

    if detection_result['is_hallucination']:
        # Retractar respuesta
        yield {
            'status': 'RETRACTED',
            'response': "Lo siento, la respuesta anterior contenía información no verificada. Fue retractada.",
            'original_response': response
        }
    else:
        # Confirmar
        yield {
            'status': 'CONFIRMED',
            'response': response,
            'confidence': detection_result['confidence_score']
        }


# Strategy #2: Caching agresivo
@lru_cache(maxsize=10000)
def cached_detection(query_hash: str, context_hash: str, response_hash: str):
    """
    Cache detection results. Queries idénticas = resultado instantáneo.
    """
    # En producción: usar Redis con TTL 7 días
    pass


def get_hash(text: str) -> str:
    """Hash para caching."""
    return hashlib.md5(text.encode()).hexdigest()


# Uso
query_hash = get_hash(query)
context_hash = get_hash(context)
response_hash = get_hash(response)

cached_result = redis_client.get(
    f"detection:{query_hash}:{context_hash}:{response_hash}"
)

if cached_result:
    # Cache HIT (típicamente 30-40% queries en chatbots)
    return json.loads(cached_result)
else:
    # Cache MISS, run detection
    result = detector.detect_hallucination(query, context, response)
    redis_client.setex(
        f"detection:{query_hash}:{context_hash}:{response_hash}",
        604800,  # 7 días TTL
        json.dumps(result)
    )
    return result


# Strategy #3: Usar modelo más rápido para LLM-as-judge
# GPT-4: 500-800ms → GPT-3.5-turbo: 200-400ms (60% más rápido, 90% accuracy)
llm_judge_config = {
    'model': 'gpt-3.5-turbo',  # Más rápido que GPT-4
    'temperature': 0,
    'max_tokens': 200  # Limita tokens = reduce latency
}

# Cost safeguards
import time
from collections import defaultdict

# Rate limiter por usuario
user_request_counts = defaultdict(
    lambda: {'count': 0, 'reset_time': time.time() + 3600}
)
MAX_REQUESTS_PER_HOUR = 100


def check_rate_limit(user_id: str) -> bool:
    """Previene cost spikes por abuso."""
    user_data = user_request_counts[user_id]

    # Reset counter si pasó 1 hora
    if time.time() > user_data['reset_time']:
        user_data['count'] = 0
        user_data['reset_time'] = time.time() + 3600

    # Check limit
    if user_data['count'] >= MAX_REQUESTS_PER_HOUR:
        logger.warning(f"Rate limit exceeded: user {user_id}")
        return False

    user_data['count'] += 1
    return True


# Budget alerting
MONTHLY_BUDGET_USD = 1000
current_spend = 0


def track_detection_cost(tokens_used: int, model: str):
    """Track y alert si excedemos budget."""
    global current_spend

    # Cost per token (GPT-4: $0.03/1k tokens)
    cost = (tokens_used / 1000) * 0.03
    current_spend += cost

    if current_spend > MONTHLY_BUDGET_USD * 0.8:
        # 80% budget usado
        alert_ops_team(f"Detection costs at {current_spend:.2f} USD (80% budget)")

    if current_spend > MONTHLY_BUDGET_USD:
        # Budget excedido, switch a modo frugal
        logger.critical("Budget exceeded, switching to token similarity only")
        # Desactivar LLM-as-judge temporalmente
        return 'BUDGET_EXCEEDED'

    return 'OK'