| """ |
| Package Manager for CodeAct Agent. |
| Handles Python package management and imports. |
| """ |
|
|
| from core.constants import LIBRARY_CONTENT_DICT |
|
|
|
|
| class PackageManager: |
| """Manages Python packages available to the agent.""" |
|
|
| def __init__(self, default_packages: dict[str, str] = None): |
| self.default_packages = default_packages or LIBRARY_CONTENT_DICT.copy() |
| self.custom_packages = {} |
|
|
| def add_packages(self, packages: dict[str, str]) -> bool: |
| """Add new packages to the available packages.""" |
| try: |
| if not isinstance(packages, dict): |
| raise ValueError( |
| "Packages must be a dictionary with package name as key and description as value" |
| ) |
|
|
| for package_name, description in packages.items(): |
| if not isinstance(package_name, str) or not isinstance(description, str): |
| print( |
| "Warning: Skipping invalid package entry - package_name and description must be strings" |
| ) |
| continue |
|
|
| self.custom_packages[package_name] = description |
| print(f"Added package '{package_name}': {description}") |
|
|
| print(f"Successfully added {len(packages)} package(s) to the library") |
| return True |
|
|
| except Exception as e: |
| print(f"Error adding packages: {e}") |
| return False |
|
|
| def get_all_packages(self) -> dict[str, str]: |
| """Get all available packages (default + custom).""" |
| all_packages = self.default_packages.copy() |
| all_packages.update(self.custom_packages) |
| return all_packages |
|
|
| def import_packages(self, python_executor) -> tuple: |
| """Import all packages into the executor environment.""" |
| all_packages = self.get_all_packages() |
| imported_packages = [] |
| failed_packages = [] |
|
|
| for package_name in all_packages.keys(): |
| try: |
| import_code = f"import {package_name}" |
| python_executor(import_code) |
| imported_packages.append(package_name) |
| except Exception as e: |
| failed_packages.append((package_name, str(e))) |
|
|
| return imported_packages, failed_packages |
|
|
