-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
0.2.5 configurable system prompts and library installation tool
- Loading branch information
1 parent
8e843a8
commit 112ed85
Showing
14 changed files
with
329 additions
and
169 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| import subprocess | ||
| import sys | ||
| import re | ||
| from typing import Optional, Tuple | ||
|
|
||
| def extract_module_info(error_message: str) -> Optional[Tuple[str, str]]: | ||
| """ | ||
| Extract module name and suggested install command from error message. | ||
| Returns tuple of (module_name, install_name) or None if not found. | ||
| """ | ||
| # Pattern for "No module named 'x'" | ||
| module_pattern = r"No module named '(.+?)'" | ||
| # Pattern for "Missing optional dependency 'x'. Use pip or conda to install x" | ||
| optional_pattern = r"Missing optional dependency '(.+?)'\." | ||
|
|
||
| module_name = None | ||
| install_name = None | ||
|
|
||
| if match := re.search(module_pattern, error_message): | ||
| module_name = match.group(1) | ||
| install_name = module_name | ||
| elif match := re.search(optional_pattern, error_message): | ||
| module_name = match.group(1) | ||
| install_name = module_name | ||
|
|
||
| if module_name: | ||
| # Handle special cases where install name differs from import name | ||
| special_cases = { | ||
| "PIL": "pillow", | ||
| } | ||
| install_name = special_cases.get(module_name, install_name) | ||
| return (module_name, install_name) | ||
|
|
||
| return None | ||
|
|
||
| def install_missing_module(error_message: str, debug: bool = False) -> bool: | ||
| """ | ||
| Attempts to install a missing module using pip. | ||
| Returns True if installation was successful. | ||
| """ | ||
| module_info = extract_module_info(error_message) | ||
| if not module_info: | ||
| if debug: | ||
| print("♣ Could not extract module name from error message") | ||
| return False | ||
|
|
||
| module_name, install_name = module_info | ||
|
|
||
| if debug: | ||
| print(f"♣ Attempting to install missing module: {install_name}") | ||
|
|
||
| try: | ||
| subprocess.check_call( | ||
| [sys.executable, "-m", "pip", "install", install_name], | ||
| stdout=subprocess.PIPE if not debug else None, | ||
| stderr=subprocess.PIPE if not debug else None | ||
| ) | ||
| if debug: | ||
| print(f"♣ Successfully installed {install_name}") | ||
| return True | ||
| except subprocess.CalledProcessError as e: | ||
| if debug: | ||
| print(f"♣ Failed to install {install_name}: {str(e)}") | ||
| return False |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,39 +1,138 @@ | ||
| from pathlib import Path | ||
| from .configurator import setup_config | ||
| import os | ||
| import shutil | ||
|
|
||
| def copy_config(user_config_path): | ||
| """ | ||
| Sets up the healing agent configuration file by copying example config to specified path. | ||
| Args: | ||
| user_config_path (Path): Path where config file should be created | ||
| Returns: | ||
| str: Path to the created config file | ||
| """ | ||
| os.makedirs(os.path.dirname(user_config_path), exist_ok=True) | ||
|
|
||
| # Copy example config using platform-agnostic paths | ||
| example_config = os.path.join(os.path.dirname(__file__), 'config_template.py') | ||
| if not os.path.exists(example_config): | ||
| raise FileNotFoundError(f"♣ Config template not found at: {example_config}") | ||
|
|
||
| shutil.copy(example_config, user_config_path) | ||
| print(f"♣ Created new config file at, please update the values: {user_config_path}") | ||
| return user_config_path | ||
|
|
||
| def load_config(local_config_path=None): | ||
| """ | ||
| Load configuration from healing_agent_config.py | ||
| Args: | ||
| local_config_path (str|Path, optional): Path to local config file. If not provided, | ||
| will attempt to detect config location automatically. | ||
| """ | ||
|
|
||
| def load_config(): | ||
| """Load configuration from healing_agent_config.py""" | ||
| # First check local directory | ||
| local_config = Path('healing_agent_config.py') | ||
| user_config = Path.home() / '.healing_agent' / 'healing_agent_config.py' | ||
|
|
||
| if local_config.exists(): | ||
| config_path = local_config | ||
| if local_config_path and Path(local_config_path).exists(): | ||
| config_path = Path(local_config_path) | ||
| elif user_config.exists(): | ||
| config_path = user_config | ||
| config_path = Path(user_config) | ||
| else: | ||
| # Run configurator to create default config | ||
| # Create default config | ||
| print("♣ No config file found. Creating default configuration...") | ||
| config_path = Path(setup_config()) | ||
| print("♣ Please update the configuration values in the newly created config file") | ||
| return { | ||
| 'save_exception': False, | ||
| 'exception_folder': str(Path.home() / '.healing_agent' / 'exceptions') | ||
| } | ||
|
|
||
| config_path = Path(copy_config(user_config)) | ||
|
|
||
| # Load config module | ||
| import importlib.util | ||
| spec = importlib.util.spec_from_file_location("healing_agent_config", config_path) | ||
| if spec is None: | ||
| raise ImportError(f"♣ Could not load config from {config_path}") | ||
|
|
||
| config = importlib.util.module_from_spec(spec) | ||
| spec.loader.exec_module(config) | ||
|
|
||
| # Get all variables from config module | ||
| # Get non-private variables from config | ||
| config_vars = {k: v for k, v in vars(config).items() | ||
| if not k.startswith('__')} | ||
|
|
||
| # Set defaults for required config values if not present | ||
| if 'SAVE_EXCEPTION' not in config_vars: | ||
| config_vars['SAVE_EXCEPTION'] = False | ||
| if 'EXCEPTION_FOLDER' not in config_vars: | ||
| config_vars['EXCEPTION_FOLDER'] = str(Path.home() / '.healing_agent' / 'exceptions') | ||
| # Check Azure OpenAI config | ||
| if 'AZURE' in config_vars: | ||
| azure_config = config_vars['AZURE'] | ||
| if not azure_config.get('api_key'): | ||
| for env_var in ['AZURE_API_KEY', 'AZURE_OPENAI_API_KEY']: | ||
| if os.getenv(env_var): | ||
| azure_config['api_key'] = os.getenv(env_var) | ||
| break | ||
|
|
||
| azure_config.setdefault('instance', os.getenv('AZURE_OPENAI_INSTANCE')) | ||
| azure_config.setdefault('api_version', os.getenv('AZURE_API_VERSION')) | ||
| azure_config.setdefault('api_base', os.getenv('AZURE_API_BASE')) | ||
|
|
||
| # Check Anthropic config | ||
| if 'ANTHROPIC' in config_vars: | ||
| anthropic_config = config_vars['ANTHROPIC'] | ||
| anthropic_config.setdefault('api_key', os.getenv('ANTHROPIC_API_KEY')) | ||
| anthropic_config.setdefault('base_url', os.getenv('ANTHROPIC_BASE_URL')) | ||
| anthropic_config.setdefault('temperature', os.getenv('ANTHROPIC_TEMPERATURE')) | ||
| anthropic_config.setdefault('max_tokens', os.getenv('ANTHROPIC_MAX_TOKENS')) | ||
|
|
||
| # Validate config | ||
| validate_config(config_vars) | ||
|
|
||
| return config_vars, config_path | ||
|
|
||
| def validate_config(config): | ||
| """Validate configuration settings.""" | ||
| try: | ||
| # Validate AI provider | ||
| if 'AI_PROVIDER' not in config: | ||
| raise ValueError("AI_PROVIDER must be defined in config") | ||
|
|
||
| valid_providers = ['azure', 'openai', 'ollama', 'litellm', 'anthropic'] | ||
| if config['AI_PROVIDER'] not in valid_providers: | ||
| raise ValueError(f"♣ Invalid AI provider: {config['AI_PROVIDER']}. Must be one of: {', '.join(valid_providers)}") | ||
|
|
||
| # Validate provider-specific settings | ||
| if config['AI_PROVIDER'] == 'azure': | ||
| if not (config.get('AZURE', {}).get('api_key') and config.get('AZURE', {}).get('endpoint')): | ||
| raise ValueError("Azure API key and endpoint must be configured") | ||
|
|
||
| elif config['AI_PROVIDER'] == 'openai': | ||
| if not config.get('OPENAI', {}).get('api_key'): | ||
| raise ValueError("OpenAI API key must be configured") | ||
|
|
||
| elif config['AI_PROVIDER'] == 'anthropic': | ||
| if not config.get('ANTHROPIC', {}).get('api_key'): | ||
| raise ValueError("Anthropic API key must be configured") | ||
|
|
||
| # Validate behavior configuration | ||
| required_settings = ['MAX_ATTEMPTS', 'DEBUG', 'AUTO_FIX', 'BACKUP_ENABLED', 'SAVE_EXCEPTIONS', 'SYSTEM_PROMPTS'] | ||
| missing_settings = [] | ||
| for setting in required_settings: | ||
| if setting not in config: | ||
| # For SYSTEM_PROMPTS, check if it exists and has required keys | ||
| if setting == 'SYSTEM_PROMPTS': | ||
| if not config.get('SYSTEM_PROMPTS') or not all(key in config['SYSTEM_PROMPTS'] for key in ['code_fixer', 'analyzer', 'report']): | ||
| missing_settings.append(setting) | ||
| else: | ||
| missing_settings.append(setting) | ||
|
|
||
| if missing_settings: | ||
| print(f"♣ Config validation failed. Missing settings: {', '.join(missing_settings)}") | ||
| print("♣ Current config keys:", list(config.keys())) | ||
| raise ValueError(f"Missing required settings: {', '.join(missing_settings)}") | ||
|
|
||
| # Validate types | ||
| if not isinstance(config.get('MAX_ATTEMPTS'), int) or config.get('MAX_ATTEMPTS', 0) <= 0: | ||
| raise ValueError("MAX_ATTEMPTS must be a positive integer") | ||
|
|
||
| for bool_setting in ['DEBUG', 'AUTO_FIX', 'BACKUP_ENABLED', 'SAVE_EXCEPTIONS']: | ||
| if not isinstance(config.get(bool_setting), bool): | ||
| raise ValueError(f"{bool_setting} must be a boolean value") | ||
|
|
||
| return config | ||
|
|
||
| return config_vars | ||
| except Exception as e: | ||
| print(f"♣ Error loading config: {str(e)}") | ||
| raise |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.