Overview
The RecallrAI SDK implements a comprehensive exception hierarchy to help you handle different error scenarios gracefully. All SDK exceptions inherit from the base RecallrAIError class.
Exception Hierarchy
Importing Exceptions
Import exceptions from the recallrai.exceptions module:
# Import specific exceptions
from recallrai.exceptions import (
UserNotFoundError,
SessionNotFoundError,
InvalidCategoriesError,
MergeConflictNotFoundError,
)
# Import all exceptions
from recallrai.exceptions import (
RecallrAIError,
AuthenticationError,
TimeoutError,
ConnectionError,
InternalServerError,
RateLimitError,
UserNotFoundError,
UserAlreadyExistsError,
InvalidCategoriesError,
SessionNotFoundError,
InvalidSessionStateError,
ValidationError,
MergeConflictError,
MergeConflictNotFoundError,
MergeConflictAlreadyResolvedError,
MergeConflictInvalidQuestionsError,
MergeConflictMissingAnswersError,
MergeConflictInvalidAnswerError,
)
Base Exception
RecallrAIError
The base exception class for all SDK-specific errors.
from recallrai.exceptions import RecallrAIError
try:
# SDK operation
user = client.get_user("user123")
except RecallrAIError as e:
print(f"RecallrAI error occurred: {e}")
Catch RecallrAIError to handle all SDK-specific exceptions in one place.
Authentication Errors
AuthenticationError
Raised when there’s an issue with your API key or project ID authentication.
from recallrai.exceptions import AuthenticationError
try:
client = RecallrAI(api_key="invalid_key", project_id="project-uuid")
user = client.get_user("user123")
except AuthenticationError as e:
print(f"Authentication failed: {e}")
# Prompt user to check their API credentials
- Invalid or expired API key
- Incorrect project ID
- Missing authentication headers
Network Errors
TimeoutError
Occurs when a request takes too long to complete.
from recallrai.exceptions import TimeoutError
try:
user = client.get_user("user123")
except TimeoutError as e:
print(f"Request timed out: {e}")
# Implement retry logic or increase timeout
ConnectionError
Happens when the SDK cannot establish a connection to the RecallrAI API.
from recallrai.exceptions import ConnectionError
try:
user = client.get_user("user123")
except ConnectionError as e:
print(f"Connection failed: {e}")
# Check network connectivity
Server Errors
InternalServerError
Raised when the RecallrAI API returns a 5xx error code.
from recallrai.exceptions import InternalServerError
try:
user = client.create_user("user123")
except InternalServerError as e:
print(f"Server error: {e}")
# Retry after a delay or contact support
RateLimitError
Raised when the API rate limit has been exceeded (HTTP 429).
from recallrai.exceptions import RateLimitError
try:
users = client.list_users(limit=100)
except RateLimitError as e:
print(f"Rate limit exceeded: {e}")
if hasattr(e, 'retry_after'):
print(f"Retry after {e.retry_after} seconds")
# Wait and retry
When available, the retry_after value indicates how long to wait before retrying the request.
User Errors
UserNotFoundError
Raised when attempting to access a user that doesn’t exist.
from recallrai.exceptions import UserNotFoundError
try:
user = client.get_user("nonexistent_user")
except UserNotFoundError as e:
print(f"User not found: {e}")
# Create the user or handle gracefully
UserAlreadyExistsError
Occurs when creating a user with an ID that already exists.
from recallrai.exceptions import UserAlreadyExistsError
try:
user = client.create_user("user123")
except UserAlreadyExistsError as e:
print(f"User already exists: {e}")
# Get the existing user instead
user = client.get_user("user123")
InvalidCategoriesError
Raised when filtering user memories by categories that don’t exist in the project.
from recallrai.exceptions import InvalidCategoriesError
try:
user = client.get_user("user123")
memories = user.list_memories(categories=["invalid_category"])
except InvalidCategoriesError as e:
print(f"Invalid categories: {e.invalid_categories}")
# Use valid categories from your project
The exception contains the list of invalid categories in the invalid_categories attribute.
Session Errors
SessionNotFoundError
Raised when attempting to access a non-existent session.
from recallrai.exceptions import SessionNotFoundError
try:
user = client.get_user("user123")
session = user.get_session("nonexistent_session")
except SessionNotFoundError as e:
print(f"Session not found: {e}")
# Create a new session
InvalidSessionStateError
Occurs when performing an operation that’s not valid for the current session state.
from recallrai.exceptions import InvalidSessionStateError
from recallrai.models import MessageRole
try:
user = client.get_user("user123")
session = user.get_session("session-uuid")
# Trying to add a message to a processed session
session.add_message(role=MessageRole.USER, content="Hello")
except InvalidSessionStateError as e:
print(f"Invalid session state: {e}")
# Create a new session for new messages
You cannot add messages to a session that has already been processed.
Merge Conflict Errors
MergeConflictNotFoundError
Raised when attempting to access a merge conflict that doesn’t exist.
from recallrai.exceptions import MergeConflictNotFoundError
try:
user = client.get_user("user123")
conflict = user.get_merge_conflict("nonexistent_conflict")
except MergeConflictNotFoundError as e:
print(f"Merge conflict not found: {e}")
MergeConflictAlreadyResolvedError
Occurs when trying to resolve a merge conflict that has already been processed.
from recallrai.exceptions import MergeConflictAlreadyResolvedError
try:
conflict.resolve(answers)
except MergeConflictAlreadyResolvedError as e:
print(f"Conflict already resolved: {e}")
# Refresh to get latest status
conflict.refresh()
MergeConflictInvalidQuestionsError
Raised when the provided questions don’t match the original clarifying questions.
from recallrai.exceptions import MergeConflictInvalidQuestionsError
try:
conflict.resolve(answers)
except MergeConflictInvalidQuestionsError as e:
print(f"Invalid questions: {e}")
if e.invalid_questions:
print(f"These questions are invalid: {e.invalid_questions}")
MergeConflictMissingAnswersError
Occurs when not all required clarifying questions have been answered.
from recallrai.exceptions import MergeConflictMissingAnswersError
try:
conflict.resolve(answers)
except MergeConflictMissingAnswersError as e:
print(f"Missing answers: {e}")
if e.missing_questions:
print(f"Missing answers for: {e.missing_questions}")
MergeConflictInvalidAnswerError
Raised when an answer is not one of the valid options for a question.
from recallrai.exceptions import MergeConflictInvalidAnswerError
try:
conflict.resolve(answers)
except MergeConflictInvalidAnswerError as e:
print(f"Invalid answer: {e}")
if e.question and e.valid_options:
print(f"Question: {e.question}")
print(f"Valid options: {e.valid_options}")
Validation Errors
ValidationError
Raised when provided data doesn’t meet the required format or constraints.
from recallrai.exceptions import ValidationError
try:
# Invalid parameter value
context = session.get_context(min_top_k=200) # Out of range
except ValidationError as e:
print(f"Validation error: {e}")
Best Practices
1. Handle Specific Exceptions First
Catch more specific exceptions before general ones:
from recallrai.exceptions import UserNotFoundError, RecallrAIError
try:
user = client.get_user("user123")
except UserNotFoundError as e:
# Handle specific case
print(f"Creating new user...")
user = client.create_user("user123")
except RecallrAIError as e:
# General fallback
print(f"SDK error: {e}")
2. Implement Retry Logic for Transient Errors
Network and timeout errors might be temporary:
import time
from recallrai.exceptions import TimeoutError, ConnectionError, RateLimitError
def get_user_with_retry(client, user_id, max_retries=3):
for attempt in range(max_retries):
try:
return client.get_user(user_id)
except (TimeoutError, ConnectionError) as e:
if attempt < max_retries - 1:
wait_time = 2 ** attempt # Exponential backoff
print(f"Retry in {wait_time}s...")
time.sleep(wait_time)
else:
raise
except RateLimitError as e:
if hasattr(e, 'retry_after'):
time.sleep(e.retry_after)
else:
time.sleep(60)
import logging
from recallrai.exceptions import RecallrAIError
logger = logging.getLogger(__name__)
try:
user = client.get_user("user123")
except RecallrAIError as e:
logger.error(f"RecallrAI error: {type(e).__name__} - {e}")
# Re-raise or handle appropriately
4. Handle Common User Flows
Check if resources exist before operations:
from recallrai.exceptions import UserNotFoundError
def get_or_create_user(client, user_id, metadata=None):
try:
return client.get_user(user_id)
except UserNotFoundError:
return client.create_user(user_id, metadata=metadata)
5. Graceful Degradation
Provide fallback behavior when errors occur:
from recallrai.exceptions import SessionNotFoundError
def get_conversation_context(user, session_id=None):
if session_id:
try:
session = user.get_session(session_id)
return session.get_context()
except SessionNotFoundError:
# Fallback: create new session
pass
# Default behavior
session = user.create_session()
return session.get_context()
Complete Example
Here’s a comprehensive example showing proper exception handling:
from recallrai import RecallrAI
from recallrai.exceptions import (
AuthenticationError,
UserNotFoundError,
UserAlreadyExistsError,
SessionNotFoundError,
InvalidSessionStateError,
TimeoutError,
RecallrAIError,
)
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def safe_chat_operation(api_key, project_id, user_id, session_id=None):
try:
# Initialize client
client = RecallrAI(api_key=api_key, project_id=project_id)
# Get or create user
try:
user = client.get_user(user_id)
logger.info(f"Retrieved user: {user_id}")
except UserNotFoundError:
user = client.create_user(user_id)
logger.info(f"Created new user: {user_id}")
# Get or create session
if session_id:
try:
session = user.get_session(session_id)
logger.info(f"Retrieved session: {session_id}")
except SessionNotFoundError:
session = user.create_session()
logger.info(f"Created new session: {session.session_id}")
else:
session = user.create_session()
logger.info(f"Created new session: {session.session_id}")
# Add message
try:
session.add_message(role=MessageRole.USER, content="Hello")
logger.info("Added message to session")
except InvalidSessionStateError:
logger.warning("Session already processed, creating new one")
session = user.create_session()
session.add_message(role=MessageRole.USER, content="Hello")
# Get context
context = session.get_context()
return session, context
except AuthenticationError as e:
logger.error(f"Authentication failed: {e}")
raise
except TimeoutError as e:
logger.error(f"Request timed out: {e}")
# Could implement retry logic here
raise
except RecallrAIError as e:
logger.error(f"RecallrAI error: {type(e).__name__} - {e}")
raise
except Exception as e:
logger.error(f"Unexpected error: {e}")
raise
