Function timeout

src/smolagents/local_python_executor.py:285–320

Decorator to limit the execution time of a function using threading. This implementation is cross-platform (works on Windows) and thread-safe (works when called from any thread, not just the main thread), unlike signal-based approaches. Args: timeout_seconds (`int`): Maxim

(timeout_seconds: int)

def timeout(timeout_seconds: int):
    """
    Decorator to limit the execution time of a function using threading.

    This implementation is cross-platform (works on Windows) and thread-safe (works when
    called from any thread, not just the main thread), unlike signal-based approaches.

    Args:
        timeout_seconds (`int`): Maximum time in seconds allowed for function execution.

    Raises:
        ExecutionTimeoutError: If the function execution exceeds the timeout period.

    Note:
        If a timeout occurs, the thread running the function cannot be forcefully killed
        in Python, so it will continue running in the background until completion. However,
        the caller will receive a TimeoutError and can continue execution.
    """

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Create a new ThreadPoolExecutor for each call to avoid threading issues
            with ThreadPoolExecutor(max_workers=1) as executor:
                future = executor.submit(func, *args, **kwargs)
                try:
                    result = future.result(timeout=timeout_seconds)
                    return result
                except FuturesTimeoutError:
                    raise ExecutionTimeoutError(
                        f"Code execution exceeded the maximum execution time of {timeout_seconds} seconds"
                    )

        return wrapper

    return decorator


def get_iterable(obj):