Agent Errors

from agentica import agentic

# Define custom exceptions
class InsufficientDataError(Exception):
    """Raised when the input data is incomplete or insufficient for analysis."""
    pass

class DataQualityError(Exception):
    """Raised when data quality is too poor for reliable results."""
    pass

class UnsupportedFormatError(Exception):
    """Raised when the data format is not supported."""
    pass

@agentic(InsufficientDataError, DataQualityError, UnsupportedFormatError)
async def analyze_dataset(data: str) -> dict:
    """
    Analyze the dataset and return insights.
    
    Raises:
        InsufficientDataError: If the dataset has fewer than 10 rows
        DataQualityError: If more than 50% of values are missing or invalid
        UnsupportedFormatError: If the data format is not CSV or JSON
        ValueError: If the data cannot be parsed
    
    Returns a dictionary with analysis results.
    """
    ...

# Use with try/except
try:
    results = await analyze_dataset(raw_data)
    print(f"Analysis complete: {results}")
except InsufficientDataError as e:
    logger.warning(f"Not enough data: {e}")
    results = {"status": "insufficient_data", "message": str(e)}
except DataQualityError as e:
    logger.warning(f"Poor data quality: {e}")
    results = perform_basic_analysis(raw_data)  # Fallback
except UnsupportedFormatError as e:
    logger.error(f"Format not supported: {e}")
    results = {"status": "error", "message": "Please provide CSV or JSON"}
except ValueError as e:
    logger.error(f"Parsing failed: {e}")
    raise

from dataclasses import dataclass
from agentica import agentic

@dataclass
class RefactorResult:
    file_path: str
    success: bool
    updated_code: str | None
    error: str | None

@agentic()
async def refactor_file(code: str, instruction: str) -> str:
    """
    Refactor the given code according to the instruction.
    Preserve functionality while improving code quality.
    """
    ...

async def refactor_codebase(files: dict[str, str], instruction: str) -> list[RefactorResult]:
    """Refactor multiple files, continuing even if some fail."""
    results = []

    for file_path, code in files.items():
        try:
            updated = await refactor_file(code, instruction)
            results.append(RefactorResult(
                file_path=file_path,
                success=True,
                updated_code=updated,
                error=None
            ))
            logger.info(f"Successfully refactored {file_path}")
        except Exception as e:
            results.append(RefactorResult(
                file_path=file_path,
                success=False,
                updated_code=None,
                error=str(e)
            ))
            logger.warning(f"Failed to refactor {file_path}: {e}")

    return results

async def apply_refactoring(files: dict[str, str], instruction: str) -> dict:
    """Apply refactoring and report on partial success."""
    results = await refactor_codebase(files, instruction)

    successful = [r for r in results if r.success]
    failed = [r for r in results if not r.success]

    # Write successful refactorings
    for result in successful:
        with open(result.file_path, 'w') as f:
            f.write(result.updated_code)

    # Log summary
    if len(successful) == len(results):
        logger.info(f"All {len(results)} files refactored successfully")
    elif len(successful) > 0:
        logger.warning(
            f"Partial success: {len(successful)}/{len(results)} files refactored. "
            f"Failed: {[r.file_path for r in failed]}"
        )
    else:
        logger.error("All refactoring attempts failed")

    return {
        "total": len(results),
        "successful": len(successful),
        "failed": len(failed),
        "failed_files": [r.file_path for r in failed]
    }

from dataclasses import dataclass, field
from enum import Enum
from time import time

from agentica import agentic

class TaskCategory(Enum):
    BUSINESS = "business"
    PERSONAL = "personal"
    FREELANCE = "freelance"

@dataclass
class Task:
    user: str
    category: TaskCategory
    description: str
    time_created: float = field(default_factory=time)

class TaskTooComplicatedError(Exception):
    """Raised when a task is too complex to complete automatically."""
    pass

class InsufficientPermissionsError(Exception):
    """Raised when the user lacks permissions for the requested task."""
    pass

@agentic(TaskTooComplicatedError, InsufficientPermissionsError)
async def perform_task(task: Task) -> str:
    """
    Perform the task and return the result.

    Raises:
        TaskTooComplicatedError: If the task requires human intervention
        InsufficientPermissionsError: If the user lacks necessary permissions
        ValueError: If the task description is empty or invalid

    Returns:
        A description of the completed task.
    """
    ...

# Usage with error handling
try:
    result = await perform_task(task)
    print(f"Task completed: {result}")
except TaskTooComplicatedError as e:
    print(f"Manual intervention required: {e}")
    # Escalate to human
    assign_to_human(task)
except InsufficientPermissionsError as e:
    print(f"Permission denied: {e}")
    # Request additional permissions
    request_permissions(task.user, task.category)
except ValueError as e:
    print(f"Invalid task: {e}")