File size: 5,345 Bytes
c922f8b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 |
"""
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 |