MemGPT/letta/errors.py

import json
from enum import Enum
from typing import TYPE_CHECKING, Dict, List, Optional, Union

# Avoid circular imports
if TYPE_CHECKING:
    from letta.schemas.message import Message


class ErrorCode(Enum):
    """Enum for error codes used by client."""

    NOT_FOUND = "NOT_FOUND"
    UNAUTHENTICATED = "UNAUTHENTICATED"
    PERMISSION_DENIED = "PERMISSION_DENIED"
    INVALID_ARGUMENT = "INVALID_ARGUMENT"
    INTERNAL_SERVER_ERROR = "INTERNAL_SERVER_ERROR"
    CONTEXT_WINDOW_EXCEEDED = "CONTEXT_WINDOW_EXCEEDED"
    RATE_LIMIT_EXCEEDED = "RATE_LIMIT_EXCEEDED"


class LettaError(Exception):
    """Base class for all Letta related errors."""

    def __init__(self, message: str, code: Optional[ErrorCode] = None, details: Optional[Union[Dict, str, object]] = None):
        if details is None:
            details = {}
        self.message = message
        self.code = code
        self.details = details
        super().__init__(message)

    def __str__(self) -> str:
        if self.code:
            return f"{self.code.value}: {self.message}"
        return self.message

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(message='{self.message}', code='{self.code}', details={self.details})"


class LettaToolCreateError(LettaError):
    """Error raised when a tool cannot be created."""

    default_error_message = "Error creating tool."

    def __init__(self, message=None):
        super().__init__(message=message or self.default_error_message)


class LettaConfigurationError(LettaError):
    """Error raised when there are configuration-related issues."""

    def __init__(self, message: str, missing_fields: Optional[List[str]] = None):
        self.missing_fields = missing_fields or []
        super().__init__(message=message, details={"missing_fields": self.missing_fields})


class LettaAgentNotFoundError(LettaError):
    """Error raised when an agent is not found."""


class LettaUserNotFoundError(LettaError):
    """Error raised when a user is not found."""


class LLMError(LettaError):
    pass


class LLMConnectionError(LLMError):
    """Error when unable to connect to LLM service"""


class LLMRateLimitError(LLMError):
    """Error when rate limited by LLM service"""


class LLMBadRequestError(LLMError):
    """Error when LLM service cannot process request"""


class LLMAuthenticationError(LLMError):
    """Error when authentication fails with LLM service"""


class LLMPermissionDeniedError(LLMError):
    """Error when permission is denied by LLM service"""


class LLMNotFoundError(LLMError):
    """Error when requested resource is not found"""


class LLMUnprocessableEntityError(LLMError):
    """Error when request is well-formed but semantically invalid"""


class LLMServerError(LLMError):
    """Error indicating an internal server error occurred within the LLM service itself
    while processing the request."""


class BedrockPermissionError(LettaError):
    """Exception raised for errors in the Bedrock permission process."""

    def __init__(self, message="User does not have access to the Bedrock model with the specified ID."):
        super().__init__(message=message)


class BedrockError(LettaError):
    """Exception raised for errors in the Bedrock process."""

    def __init__(self, message="Error with Bedrock model."):
        super().__init__(message=message)


class LLMJSONParsingError(LettaError):
    """Exception raised for errors in the JSON parsing process."""

    def __init__(self, message="Error parsing JSON generated by LLM"):
        super().__init__(message=message)


class LocalLLMError(LettaError):
    """Generic catch-all error for local LLM problems"""

    def __init__(self, message="Encountered an error while running local LLM"):
        super().__init__(message=message)


class LocalLLMConnectionError(LettaError):
    """Error for when local LLM cannot be reached with provided IP/port"""

    def __init__(self, message="Could not connect to local LLM"):
        super().__init__(message=message)


class ContextWindowExceededError(LettaError):
    """Error raised when the context window is exceeded but further summarization fails."""

    def __init__(self, message: str, details: dict = {}):
        error_message = f"{message} ({details})"
        super().__init__(
            message=error_message,
            code=ErrorCode.CONTEXT_WINDOW_EXCEEDED,
            details=details,
        )


class RateLimitExceededError(LettaError):
    """Error raised when the llm rate limiter throttles api requests."""

    def __init__(self, message: str, max_retries: int):
        error_message = f"{message} ({max_retries})"
        super().__init__(
            message=error_message,
            code=ErrorCode.RATE_LIMIT_EXCEEDED,
            details={"max_retries": max_retries},
        )


class LettaMessageError(LettaError):
    """Base error class for handling message-related errors."""

    messages: List[Union["Message", "LettaMessage"]]
    default_error_message: str = "An error occurred with the message."

    def __init__(self, *, messages: List[Union["Message", "LettaMessage"]], explanation: Optional[str] = None) -> None:
        error_msg = self.construct_error_message(messages, self.default_error_message, explanation)
        super().__init__(error_msg)
        self.messages = messages

    @staticmethod
    def construct_error_message(messages: List[Union["Message", "LettaMessage"]], error_msg: str, explanation: Optional[str] = None) -> str:
        """Helper method to construct a clean and formatted error message."""
        if explanation:
            error_msg += f" (Explanation: {explanation})"

        # Pretty print out message JSON
        message_json = json.dumps([message.model_dump() for message in messages], indent=4)
        return f"{error_msg}\n\n{message_json}"


class MissingToolCallError(LettaMessageError):
    """Error raised when a message is missing a tool call."""

    default_error_message = "The message is missing a tool call."


class InvalidToolCallError(LettaMessageError):
    """Error raised when a message uses an invalid tool call."""

    default_error_message = "The message uses an invalid tool call or has improper usage of a tool call."


class MissingInnerMonologueError(LettaMessageError):
    """Error raised when a message is missing an inner monologue."""

    default_error_message = "The message is missing an inner monologue."


class InvalidInnerMonologueError(LettaMessageError):
    """Error raised when a message has a malformed inner monologue."""

    default_error_message = "The message has a malformed inner monologue."


class HandleNotFoundError(LettaError):
    """Error raised when a handle is not found."""

    def __init__(self, handle: str, available_handles: List[str]):
        super().__init__(
            message=f"Handle {handle} not found, must be one of {available_handles}",
            code=ErrorCode.NOT_FOUND,
        )