|
""" |
|
Common utility functions for GAIA implementation. |
|
|
|
This module contains shared utility functions used across different components |
|
of the GAIA system, including error handling, logging, and configuration helpers. |
|
""" |
|
|
|
import re |
|
import json |
|
import logging |
|
import traceback |
|
from typing import Dict, Any, List, Optional, Union, Tuple, Set, Callable |
|
|
|
logger = logging.getLogger("gaia_agent.utils.common") |
|
|
|
def setup_logging(level: str = "INFO", log_file: Optional[str] = None, |
|
format_str: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"): |
|
""" |
|
Set up logging with a consistent format across the application. |
|
|
|
Args: |
|
level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) |
|
log_file: Optional file path for logging to a file |
|
format_str: String format for log messages |
|
""" |
|
log_level = getattr(logging, level.upper(), logging.INFO) |
|
|
|
handlers = [] |
|
if log_file: |
|
handlers.append(logging.FileHandler(log_file)) |
|
handlers.append(logging.StreamHandler()) |
|
|
|
logging.basicConfig( |
|
level=log_level, |
|
format=format_str, |
|
handlers=handlers |
|
) |
|
|
|
def safe_execute(func: Callable, *args, **kwargs) -> Tuple[Any, Optional[Exception]]: |
|
""" |
|
Safely execute a function and catch any exceptions. |
|
|
|
Args: |
|
func: Function to execute |
|
*args: Positional arguments to pass to the function |
|
**kwargs: Keyword arguments to pass to the function |
|
|
|
Returns: |
|
Tuple of (result, exception) where exception is None if no exception occurred |
|
""" |
|
try: |
|
result = func(*args, **kwargs) |
|
return result, None |
|
except Exception as e: |
|
logger.error(f"Error executing {func.__name__}: {str(e)}") |
|
logger.error(traceback.format_exc()) |
|
return None, e |
|
|
|
def get_nested_value(data: Dict[str, Any], key_path: str, default: Any = None) -> Any: |
|
""" |
|
Get a value from a nested dictionary using dot notation. |
|
|
|
Args: |
|
data: Dictionary to search |
|
key_path: Path to the value using dot notation (e.g., 'api.keys.openai') |
|
default: Default value to return if the key is not found |
|
|
|
Returns: |
|
The value if found, otherwise the default |
|
""" |
|
if not key_path: |
|
return default |
|
|
|
parts = key_path.split('.') |
|
current = data |
|
|
|
try: |
|
for part in parts: |
|
if not isinstance(current, dict): |
|
return default |
|
if part not in current: |
|
return default |
|
current = current.get(part) |
|
return current |
|
except (KeyError, TypeError, AttributeError): |
|
return default |
|
|
|
def set_nested_value(data: Dict[str, Any], key_path: str, value: Any) -> None: |
|
""" |
|
Set a value in a nested dictionary using dot notation. |
|
|
|
Args: |
|
data: Dictionary to modify |
|
key_path: Path to the value using dot notation (e.g., 'api.keys.openai') |
|
value: Value to set |
|
""" |
|
if not key_path: |
|
return |
|
|
|
parts = key_path.split('.') |
|
current = data |
|
|
|
for part in parts[:-1]: |
|
if part not in current or not isinstance(current[part], dict): |
|
current[part] = {} |
|
current = current[part] |
|
|
|
current[parts[-1]] = value |
|
|
|
def load_file_content(file_path: str) -> Optional[str]: |
|
""" |
|
Load the content of a file. |
|
|
|
Args: |
|
file_path: Path to the file |
|
|
|
Returns: |
|
The content of the file, or None if the file could not be loaded |
|
""" |
|
try: |
|
with open(file_path, 'r', encoding='utf-8') as f: |
|
return f.read() |
|
except Exception as e: |
|
logger.error(f"Error loading file {file_path}: {str(e)}") |
|
return None |
|
|
|
def load_json_file(file_path: str) -> Optional[Dict[str, Any]]: |
|
""" |
|
Load a JSON file. |
|
|
|
Args: |
|
file_path: Path to the JSON file |
|
|
|
Returns: |
|
The parsed JSON data, or None if the file could not be loaded |
|
""" |
|
content = load_file_content(file_path) |
|
if content is None: |
|
return None |
|
|
|
try: |
|
return json.loads(content) |
|
except json.JSONDecodeError as e: |
|
logger.error(f"Error parsing JSON file {file_path}: {str(e)}") |
|
return None |
|
|
|
def validate_api_key(api_key: Optional[str]) -> bool: |
|
""" |
|
Validate that an API key is not None or empty. |
|
|
|
Args: |
|
api_key: API key to validate |
|
|
|
Returns: |
|
True if the API key is valid, False otherwise |
|
""" |
|
return api_key is not None and api_key.strip() != "" |
|
|
|
def format_error_message(error: Union[Exception, str], include_traceback: bool = True) -> str: |
|
""" |
|
Format an error message with optional traceback. |
|
|
|
Args: |
|
error: Exception or error message |
|
include_traceback: Whether to include the traceback |
|
|
|
Returns: |
|
Formatted error message |
|
""" |
|
if isinstance(error, Exception): |
|
error_message = str(error) |
|
stack_trace = traceback.format_exc() if include_traceback else "" |
|
else: |
|
error_message = error |
|
stack_trace = "".join(traceback.format_stack()[:-1]) if include_traceback else "" |
|
|
|
if stack_trace: |
|
return f"{error_message}\n\nTraceback:\n{stack_trace}" |
|
else: |
|
return error_message |