import hashlib
import redis
import json
import time
from langchain.cache import RedisSemanticCache
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI
from typing import Optional


class IntelligentCachingLayer:
    """
    Capa de caching semántico que reduce requests redundantes en 20-40%.
    """

    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        similarity_threshold: float = 0.95,
        ttl_seconds: int = 86400  # 24 horas
    ):
        self.redis_client = redis.from_url(redis_url)
        self.similarity_threshold = similarity_threshold
        self.ttl = ttl_seconds

        # Embeddings para semantic caching
        self.embeddings = OpenAIEmbeddings()

        # LLM con caching habilitado
        self.llm = ChatOpenAI(
            model_name="gpt-4",
            temperature=0,
            cache=RedisSemanticCache(
                redis_url=redis_url,
                embedding=self.embeddings,
                score_threshold=similarity_threshold
            )
        )

    def _generate_cache_key(self, prompt: str, model: str, temperature: float) -> str:
        """Genera cache key único basado en parámetros."""
        content = f"{prompt}{model}{temperature}"
        return hashlib.sha256(content.encode()).hexdigest()

    def get_cached_response(self, prompt: str) -> Optional[str]:
        """
        Busca respuesta en cache.
        Usa semantic similarity para queries similares.

        Returns:
            Cached response si existe y similarity > threshold, None otherwise
        """
        # El RedisSemanticCache de LangChain maneja esto automáticamente
        # cuando usas el LLM con cache enabled
        pass

    def call_with_caching(self, prompt: str) -> dict:
        """
        Ejecuta LLM call con caching inteligente.

        Returns:
            Dict con response + metadata (cache hit/miss)
        """
        start_time = time.time()

        # LangChain automáticamente checkea cache antes de llamar API
        response = self.llm.predict(prompt)

        execution_time = (time.time() - start_time) * 1000

        # Determinar si fue cache hit (tiempos < 100ms típicamente indican cache hit)
        cache_hit = execution_time < 100

        return {
            'response': response,
            'cache_hit': cache_hit,
            'execution_time_ms': execution_time,
            'prompt': prompt
        }

    def get_cache_stats(self) -> dict:
        """Obtiene estadísticas de uso de cache."""
        # Contar total de keys en cache
        total_keys = self.redis_client.dbsize()

        return {
            'total_cached_responses': total_keys,
            'cache_size_mb': self.redis_client.info()['used_memory'] / (1024 * 1024)
        }


# Uso
cache_layer = IntelligentCachingLayer(
    redis_url="redis://localhost:6379",
    similarity_threshold=0.95
)

# Primera llamada (cache miss)
result1 = cache_layer.call_with_caching("¿Cuál es la capital de Francia?")
print(f"Call 1: {result1['execution_time_ms']:.2f}ms (cache hit: {result1['cache_hit']})")

# Segunda llamada idéntica (cache hit)
result2 = cache_layer.call_with_caching("¿Cuál es la capital de Francia?")
print(f"Call 2: {result2['execution_time_ms']:.2f}ms (cache hit: {result2['cache_hit']})")

# Tercera llamada similar semánticamente (cache hit si similarity > 0.95)
result3 = cache_layer.call_with_caching("¿Cuál es la ciudad capital de Francia?")
print(f"Call 3: {result3['execution_time_ms']:.2f}ms (cache hit: {result3['cache_hit']})")

# Stats
stats = cache_layer.get_cache_stats()
print(f"Cache stats: {stats['total_cached_responses']} responses, {stats['cache_size_mb']:.2f} MB")

import os
import random
from typing import Dict, Any
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.vectorstores import Pinecone


class LangfuseObservabilityIntegration:
    """
    Integración completa de Langfuse para observabilidad production-grade.
    """

    def __init__(
        self,
        public_key: str,
        secret_key: str,
        host: str = "https://cloud.langfuse.com"
    ):
        self.langfuse = Langfuse(
            public_key=public_key,
            secret_key=secret_key,
            host=host
        )

        # LLM con observabilidad integrada
        self.llm = ChatOpenAI(
            model_name="gpt-4",
            temperature=0,
            callbacks=[self.langfuse.callback_handler]
        )

    @observe(name="rag_query", capture_input=True, capture_output=True)
    def query_rag_system(
        self,
        query: str,
        vectorstore: Pinecone,
        user_id: str = None
    ) -> Dict[str, Any]:
        """
        Ejecuta query RAG con observabilidad completa.

        El decorador @observe automáticamente:
        - Crea trace en Langfuse
        - Captura input/output
        - Trackea latency
        - Asocia con user_id
        """
        # Configurar metadata del trace
        if user_id:
            langfuse_context.update_current_trace(
                user_id=user_id,
                tags=["rag", "production"]
            )

        # 1. Retrieval step (span separado)
        with langfuse_context.observation(name="vector_search"):
            retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
            docs = retriever.get_relevant_documents(query)

            # Loggear metadata de retrieval
            langfuse_context.update_current_observation(
                metadata={
                    'num_docs_retrieved': len(docs),
                    'similarity_scores': [doc.metadata.get('score', 0) for doc in docs]
                }
            )

        # 2. Generation step (span separado)
        with langfuse_context.observation(name="llm_generation"):
            qa_chain = RetrievalQA.from_chain_type(
                llm=self.llm,
                retriever=retriever,
                return_source_documents=True
            )
            result = qa_chain({"query": query})

            # Loggear metadata de generación
            langfuse_context.update_current_observation(
                metadata={
                    'model': 'gpt-4',
                    'temperature': 0,
                    'num_source_docs': len(result['source_documents'])
                }
            )

        # 3. Quality assessment (span separado)
        with langfuse_context.observation(name="quality_check"):
            # Ejecutar hallucination detection (simplificado)
            hallucination_score = self._assess_hallucination(
                context=" ".join([doc.page_content for doc in docs]),
                response=result['result']
            )

            langfuse_context.update_current_observation(
                metadata={
                    'hallucination_score': hallucination_score,
                    'quality_gate_passed': hallucination_score < 0.3
                }
            )

            # Loggear score como métrica
            langfuse_context.score_current_trace(
                name="hallucination_risk",
                value=hallucination_score,
                comment="Lower is better"
            )

        return {
            'answer': result['result'],
            'sources': result['source_documents'],
            'hallucination_score': hallucination_score,
            'trace_url': langfuse_context.get_current_trace_url()
        }

    def _assess_hallucination(self, context: str, response: str) -> float:
        """
        Assess hallucination risk (simplificado para ejemplo).
        En producción, usar LLM-as-judge o semantic entropy.
        """
        # Placeholder: en realidad ejecutarías hallucination detection aquí
        return random.random()  # 0.0-1.0 score

    @observe(name="batch_evaluation")
    def run_batch_evaluation(
        self,
        test_cases: list[Dict[str, str]],
        vectorstore: Pinecone
    ) -> Dict[str, Any]:
        """
        Ejecuta batch evaluation con tracking completo.
        """
        results = []

        for i, test_case in enumerate(test_cases):
            # Cada test case es un sub-trace
            with langfuse_context.observation(name=f"test_case_{i}"):
                result = self.query_rag_system(
                    query=test_case['query'],
                    vectorstore=vectorstore
                )

                # Comparar con expected output
                is_correct = self._compare_outputs(
                    expected=test_case['expected_answer'],
                    actual=result['answer']
                )

                # Loggear score
                langfuse_context.score_current_observation(
                    name="correctness",
                    value=1.0 if is_correct else 0.0
                )

                results.append({
                    'test_case': test_case,
                    'result': result,
                    'correct': is_correct
                })

        # Calcular métricas agregadas
        total_correct = sum(1 for r in results if r['correct'])
        accuracy = total_correct / len(test_cases)

        # Loggear accuracy como métrica del trace principal
        langfuse_context.score_current_trace(
            name="batch_accuracy",
            value=accuracy
        )

        return {
            'accuracy': accuracy,
            'total_cases': len(test_cases),
            'correct': total_correct,
            'results': results
        }

    def _compare_outputs(self, expected: str, actual: str) -> bool:
        """Compara outputs (simplificado)."""
        # En producción: semantic similarity, exact match, o LLM-as-judge
        return expected.lower() in actual.lower()


# Uso
obs = LangfuseObservabilityIntegration(
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    secret_key=os.getenv("LANGFUSE_SECRET_KEY")
)

# Query individual con full tracing
result = obs.query_rag_system(
    query="¿Cuál es la política de devoluciones?",
    vectorstore=my_pinecone_vectorstore,
    user_id="user_12345"
)

print(f"Respuesta: {result['answer']}")
print(f"Hallucination score: {result['hallucination_score']:.2f}")
print(f"Ver trace: {result['trace_url']}")

# Batch evaluation
test_cases = [
    {"query": "¿Política de devoluciones?", "expected_answer": "30 días"},
    {"query": "¿Cómo contactar soporte?", "expected_answer": "soporte@empresa.com"},
    # ... más casos
]

eval_results = obs.run_batch_evaluation(
    test_cases=test_cases,
    vectorstore=my_pinecone_vectorstore
)

print(f"Accuracy: {eval_results['accuracy']:.2%}") 

import json
from typing import Dict, List
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.chains import RetrievalQA
from langchain.vectorstores import Pinecone


class HallucinationDetector:
    """
    Detector de alucinaciones usando patrón LLM-as-a-Judge.
    Precisión: 70-85% según benchmarks de Datadog.
    """

    JUDGE_PROMPT_TEMPLATE = """Eres un verificador de facts especializado en detectar alucinaciones en respuestas de IA.

Tu tarea es analizar si la RESPUESTA contiene información que NO está respaldada por el CONTEXTO proporcionado.

CONTEXTO (fuente de verdad):
{context}

RESPUESTA A VERIFICAR:
{response}

Analiza la respuesta paso a paso:
1. Identifica cada claim factual en la respuesta
2. Para cada claim, verifica si está directamente respaldado por el contexto
3. Marca como alucinación cualquier información que:
    - No aparece en el contexto
    - Contradice el contexto
    - Es una extrapolación no justificada

Responde en formato JSON:
{{
    "is_hallucination": true/false,
    "confidence": 0.0-1.0,
    "hallucinated_claims": ["claim 1", "claim 2", ...],
    "reasoning": "explicación detallada"
}}"""

    def __init__(self, model_name: str = "gpt-4", temperature: float = 0.0):
        self.llm = ChatOpenAI(model_name=model_name, temperature=temperature)
        self.prompt = ChatPromptTemplate.from_template(self.JUDGE_PROMPT_TEMPLATE)

    def detect(self, context: str, response: str) -> Dict:
        """
        Detecta si la respuesta contiene alucinaciones respecto al contexto.

        Args:
            context: Información de referencia (ej: chunks recuperados por RAG)
            response: Respuesta generada por el LLM a verificar

        Returns:
            Diccionario con resultado de detección
        """
        # Generar prompt para el judge
        messages = self.prompt.format_messages(
            context=context,
            response=response
        )

        # Llamar al LLM judge
        result = self.llm(messages)

        try:
            # Parsear respuesta JSON
            detection_result = json.loads(result.content)

            # Validar estructura
            required_keys = ['is_hallucination', 'confidence', 'hallucinated_claims', 'reasoning']
            if not all(key in detection_result for key in required_keys):
                raise ValueError("Respuesta del judge con formato inválido")

            return detection_result

        except json.JSONDecodeError:
            # Fallback: intentar parseo básico
            is_hallucination = "true" in result.content.lower()
            return {
                'is_hallucination': is_hallucination,
                'confidence': 0.5,
                'hallucinated_claims': [],
                'reasoning': result.content
            }


# Uso en pipeline RAG
def rag_with_hallucination_detection(query: str, vectorstore: Pinecone):
    """
    Pipeline RAG con detección de alucinaciones integrada.
    """
    # 1. Búsqueda de contexto
    retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
    docs = retriever.get_relevant_documents(query)
    context = " ".join([doc.page_content for doc in docs])

    # 2. Generar respuesta
    qa_chain = RetrievalQA.from_chain_type(
        llm=ChatOpenAI(model_name="gpt-4", temperature=0),
        retriever=retriever
    )
    response = qa_chain.run(query)

    # 3. Detectar alucinaciones
    detector = HallucinationDetector()
    detection = detector.detect(context=context, response=response)

    # 4. Actuar según resultado
    if detection['is_hallucination'] and detection['confidence'] > 0.7:
        # Rechazar respuesta y usar fallback
        return {
            'response': "Lo siento, no encontré información suficiente para responder con confianza.",
            'hallucination_detected': True,
            'original_response': response,
            'detection_details': detection
        }
    else:
        return {
            'response': response,
            'hallucination_detected': False,
            'detection_details': detection
        }


# Ejemplo de uso
result = rag_with_hallucination_detection(
    query="¿Cuál es la política de devoluciones?",
    vectorstore=my_vectorstore
)

print(f"Respuesta: {result['response']}")
print(f"Alucinación detectada: {result['hallucination_detected']}")

import hashlib
import json
import pandas as pd
from datetime import datetime
from typing import Dict, Any, Optional
from langchain.chat_models import ChatOpenAI


class DeterministicLLMWrapper:
    """
    Wrapper que enforces seed + loggea system fingerprint para troubleshooting.
    """

    def __init__(
        self,
        model_name: str = "gpt-4",
        default_seed: int = 42,
        log_file: str = "llm_executions.jsonl"
    ):
        self.model_name = model_name
        self.default_seed = default_seed
        self.log_file = log_file

    def _generate_request_id(self, prompt: str, seed: int) -> str:
        """Genera ID único para request basado en prompt + seed."""
        content = f"{prompt}{seed}{datetime.utcnow().isoformat()}"
        return hashlib.sha256(content.encode()).hexdigest()[:16]

    def _log_execution(self, execution_data: Dict[str, Any]):
        """Loggea ejecución completa a archivo JSONL para troubleshooting."""
        with open(self.log_file, 'a') as f:
            f.write(json.dumps(execution_data) + '\n')

    def call(
        self,
        prompt: str,
        temperature: float = 0.0,
        seed: Optional[int] = None,
        **kwargs
    ) -> Dict[str, Any]:
        """
        Ejecuta LLM call con seed enforcement y logging completo.

        Returns:
            Dict con response + metadata completa para reproducibilidad
        """
        # Usar seed por defecto si no se especifica
        if seed is None:
            seed = self.default_seed

        # Generar request ID
        request_id = self._generate_request_id(prompt, seed)

        # Crear LLM con parámetros deterministas
        llm = ChatOpenAI(
            model_name=self.model_name,
            temperature=temperature,
            model_kwargs={"seed": seed}  # Enforces seed en API call
        )

        # Timestamp de inicio
        start_time = datetime.utcnow()

        # Ejecutar LLM call
        try:
            response = llm.predict(prompt)

            # Extraer system_fingerprint si disponible (OpenAI)
            system_fingerprint = getattr(llm, 'system_fingerprint', None)

            execution_data = {
                'request_id': request_id,
                'timestamp': start_time.isoformat(),
                'model': self.model_name,
                'temperature': temperature,
                'seed': seed,
                'system_fingerprint': system_fingerprint,
                'prompt': prompt,
                'response': response,
                'status': 'success',
                'latency_ms': (datetime.utcnow() - start_time).total_seconds() * 1000
            }

        except Exception as e:
            execution_data = {
                'request_id': request_id,
                'timestamp': start_time.isoformat(),
                'model': self.model_name,
                'temperature': temperature,
                'seed': seed,
                'prompt': prompt,
                'status': 'error',
                'error': str(e),
                'latency_ms': (datetime.utcnow() - start_time).total_seconds() * 1000
            }
            raise

        finally:
            # Siempre loggear ejecución
            self._log_execution(execution_data)

        return execution_data


# Uso
wrapper = DeterministicLLMWrapper(
    model_name="gpt-4",
    default_seed=42,
    log_file="production_llm_logs.jsonl"
)

# Ejecutar múltiples veces
for i in range(5):
    result = wrapper.call(
        prompt="¿Cuál es la capital de Francia?",
        temperature=0.0,
        seed=42
    )
    print(f"Run {i+1}: {result['response']}")
    print(f"Fingerprint: {result['system_fingerprint']}")
    print(f"Request ID: {result['request_id']}")
    print("---")

# Troubleshooting: buscar todas las ejecuciones con mismo prompt + seed
logs_df = pd.read_json("production_llm_logs.jsonl", lines=True)

# Filtrar por prompt específico
same_prompt_logs = logs_df[logs_df['prompt'] == "¿Cuál es la capital de Francia?"]

# Verificar si system_fingerprint cambió (explicaría variaciones)
unique_fingerprints = same_prompt_logs['system_fingerprint'].nunique()
print(f"Fingerprints únicos: {unique_fingerprints}")

if unique_fingerprints > 1:
    print("⚠️ System fingerprint cambió entre ejecuciones - backend updates de OpenAI")

import json
import random
from datetime import datetime
from typing import TypedDict, Annotated, List, Dict, Any
from langgraph.graph import StateGraph, END
from langchain.chat_models import ChatOpenAI


class AgentState(TypedDict):
    """Estado compartido entre agentes."""
    messages: List[str]
    current_step: str
    intermediate_results: Dict[str, Any]
    errors: List[Dict[str, str]]
    execution_trace: List[Dict[str, Any]]


class MultiAgentDebugger:
    """
    Sistema de debugging para multi-agent LangGraph con trace completo.
    """

    def __init__(self, trace_file: str = "agent_traces.jsonl"):
        self.trace_file = trace_file
        self.llm = ChatOpenAI(model_name="gpt-4", temperature=0)

    def _log_agent_execution(
        self,
        agent_name: str,
        input_data: Any,
        output_data: Any,
        execution_time_ms: float,
        error: str = None
    ):
        """Loggea ejecución de cada agente individual."""
        trace_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'agent_name': agent_name,
            'input': str(input_data)[:500],  # Truncar para no explotar logs
            'output': str(output_data)[:500],
            'execution_time_ms': execution_time_ms,
            'error': error,
            'status': 'error' if error else 'success'
        }

        with open(self.trace_file, 'a') as f:
            f.write(json.dumps(trace_entry) + '\n')

        return trace_entry

    def search_agent(self, state: AgentState) -> AgentState:
        """Agente 1: Búsqueda de información."""
        start_time = datetime.utcnow()

        try:
            # Simular búsqueda
            query = state['messages'][-1]
            search_result = f"Resultados de búsqueda para: {query}"

            # Simular posible alucinación (5% error rate)
            if random.random() < 0.05:
                search_result += " [ALUCINACIÓN: precio 99€]"  # Error inyectado

            state['intermediate_results']['search'] = search_result
            state['current_step'] = 'search_complete'

            # Log trace
            execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000
            trace = self._log_agent_execution(
                agent_name='search_agent',
                input_data=query,
                output_data=search_result,
                execution_time_ms=execution_time
            )
            state['execution_trace'].append(trace)

        except Exception as e:
            state['errors'].append({
                'agent': 'search_agent',
                'error': str(e)
            })
            execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000
            trace = self._log_agent_execution(
                agent_name='search_agent',
                input_data=state['messages'][-1],
                output_data=None,
                execution_time_ms=execution_time,
                error=str(e)
            )
            state['execution_trace'].append(trace)

        return state

    def pricing_agent(self, state: AgentState) -> AgentState:
        """Agente 2: Cálculo de pricing (propaga error si existe)."""
        start_time = datetime.utcnow()

        try:
            search_data = state['intermediate_results'].get('search', '')

            # Si el agente anterior alucinó, este agente propagará el error
            if "ALUCINACIÓN" in search_data:
                # Error se amplifica (de 5% a 12%)
                price_calculation = "Precio final: 79.2€ [ERROR PROPAGADO]"
            else:
                price_calculation = "Precio final: 71.2€ [CORRECTO]"

            state['intermediate_results']['pricing'] = price_calculation
            state['current_step'] = 'pricing_complete'

            execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000
            trace = self._log_agent_execution(
                agent_name='pricing_agent',
                input_data=search_data,
                output_data=price_calculation,
                execution_time_ms=execution_time
            )
            state['execution_trace'].append(trace)

        except Exception as e:
            state['errors'].append({
                'agent': 'pricing_agent',
                'error': str(e)
            })
            execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000
            trace = self._log_agent_execution(
                agent_name='pricing_agent',
                input_data=state['intermediate_results'].get('search', ''),
                output_data=None,
                execution_time_ms=execution_time,
                error=str(e)
            )
            state['execution_trace'].append(trace)

        return state

    def verification_agent(self, state: AgentState) -> AgentState:
        """Agente 3: Verificación de calidad (detecta errores propagados)."""
        start_time = datetime.utcnow()

        try:
            pricing_data = state['intermediate_results'].get('pricing', '')

            # Verificar si hay error propagado
            if "ERROR PROPAGADO" in pricing_data:
                verification_result = "❌ VERIFICACIÓN FALLIDA: Error detectado en cadena"
                state['errors'].append({
                    'agent': 'verification_agent',
                    'error': 'Error compound detected from upstream agents'
                })
            else:
                verification_result = "✅ VERIFICACIÓN EXITOSA: Todos los datos correctos"

            state['intermediate_results']['verification'] = verification_result
            state['current_step'] = 'verification_complete'

            execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000
            trace = self._log_agent_execution(
                agent_name='verification_agent',
                input_data=pricing_data,
                output_data=verification_result,
                execution_time_ms=execution_time
            )
            state['execution_trace'].append(trace)

        except Exception as e:
            state['errors'].append({
                'agent': 'verification_agent',
                'error': str(e)
            })
            execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000
            trace = self._log_agent_execution(
                agent_name='verification_agent',
                input_data=state['intermediate_results'].get('pricing', ''),
                output_data=None,
                execution_time_ms=execution_time,
                error=str(e)
            )
            state['execution_trace'].append(trace)

        return state

    def build_graph(self) -> StateGraph:
        """Construye el grafo de agentes con debugging integrado."""
        workflow = StateGraph(AgentState)

        # Añadir nodos
        workflow.add_node("search", self.search_agent)
        workflow.add_node("pricing", self.pricing_agent)
        workflow.add_node("verification", self.verification_agent)

        # Definir edges
        workflow.set_entry_point("search")
        workflow.add_edge("search", "pricing")
        workflow.add_edge("pricing", "verification")
        workflow.add_edge("verification", END)

        return workflow.compile()


# Uso
debugger = MultiAgentDebugger(trace_file="production_agent_traces.jsonl")
app = debugger.build_graph()

# Ejecutar workflow
initial_state = {
    'messages': ["Buscar precio de laptop XYZ"],
    'current_step': 'start',
    'intermediate_results': {},
    'errors': [],
    'execution_trace': []
}

final_state = app.invoke(initial_state)

# Analizar trace completo
print("=== EJECUCIÓN COMPLETA ===")
for trace in final_state['execution_trace']:
    status = "✅" if trace['status'] == 'success' else "❌"
    print(f"{status} {trace['agent_name']}: {trace['execution_time_ms']:.2f}ms")

if final_state['errors']:
    print("\n=== ERRORES DETECTADOS ===")
    for error in final_state['errors']:
        print(f"❌ {error['agent']}: {error['error']}")