app/exceptions/rag.py

"""Author RAG Chatbot SaaS — RAG Pipeline Exceptions."""

from app.exceptions.base import AppException


class PipelineError(AppException):
    """Generic RAG pipeline failure."""
    def __init__(self, step: str, message: str) -> None:
        super().__init__(f"Pipeline error at step [{step}]: {message}")
        self.step = step


class HallucinationError(AppException):
    """Response failed the faithfulness guardrail check."""
    def __init__(self, score: float) -> None:
        super().__init__(f"Response failed faithfulness check (score={score:.2f})")
        self.score = score


class NoContextError(AppException):
    """No relevant context chunks were retrieved from the vector DB."""
    def __init__(self) -> None:
        super().__init__("No relevant context found for this query")


class BoundaryViolationError(AppException):
    """Query or response violates the content boundary rules."""
    def __init__(self, violation_type: str) -> None:
        super().__init__(f"Boundary violation detected: {violation_type}")
        self.violation_type = violation_type


class OpenAIError(AppException):
    """OpenAI API call failed after all retries."""
    def __init__(self, message: str) -> None:
        super().__init__(f"OpenAI API error: {message}")


class CircuitBreakerOpenError(AppException):
    """Circuit breaker is open due to repeated OpenAI failures."""
    def __init__(self) -> None:
        super().__init__("Service temporarily unavailable. Please try again shortly.")