```python import time import logging import traceback import threading import queue import concurrent.futures import os import signal from typing import Callable, Any, Dict, List, Optional, Tuple, Union import json import uuid import functools # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class SkillExecutionError(Exception): """Base class for skill execution errors.""" pass class InputValidationError(SkillExecutionError): """Raised when input validation fails.""" pass class OutputValidationError(SkillExecutionError): """Raised when output validation fails.""" pass class SkillExecutionTimeout(SkillExecutionError): """Raised when skill execution exceeds the timeout.""" pass class SkillInvocation: """Represents a skill invocation request.""" def __init__(self, skill_name: str, input_data: Dict[str, Any], invocation_id: Optional[str] = None): self.skill_name = skill_name self.input_data = input_data self.invocation_id = invocation_id or str(uuid.uuid4()) # Generate if not provided def __repr__(self): return f"SkillInvocation(skill_name='{self.skill_name}', input_data={self.input_data}, invocation_id='{self.invocation_id}')" class SkillResult: """Represents the result of a skill execution.""" def __init__(self, invocation_id: str, skill_name: str, output_data: Optional[Dict[str, Any]] = None, error: Optional[Exception] = None, execution_time: float = 0.0, logs: Optional[List[str]] = None): self.invocation_id = invocation_id self.skill_name = skill_name self.output_data = output_data self.error = error self.execution_time = execution_time self.logs = logs or [] def __repr__(self): return f"SkillResult(invocation_id='{self.invocation_id}', skill_name='{self.skill_name}', output_data={self.output_data}, error={self.error}, execution_time={self.execution_time})" class SkillRegistry: """Manages registered skills.""" def __init__(self): self.skills: Dict[str, Callable] = {} self.input_validators: Dict[str, Callable[[Dict[str, Any]], bool]] = {} self.output_validators: Dict[str, Callable[[Dict[str, Any]], bool]] = {} def register_skill(self, skill_name: str, skill_function: Callable, input_validator: Optional[Callable[[Dict[str, Any]], bool]] = None, output_validator: Optional[Callable[[Dict[str, Any]], bool]] = None): """Registers a skill with the registry.""" if skill_name in self.skills: raise ValueError(f"Skill '{skill_name}' already registered.") self.skills[skill_name] = skill_function if input_validator: self.input_validators[skill_name] = input_validator if output_validator: self.output_validators[skill_name] = output_validator def get_skill(self, skill_name: str) -> Callable: """Retrieves a skill from the registry.""" if skill_name not in self.skills: raise ValueError(f"Skill '{skill_name}' not found.") return self.skills[skill_name] def get_input_validator(self, skill_name: str) -> Optional[Callable[[Dict[str, Any]], bool]]: """Retrieves the input validator for a skill.""" return self.input_validators.get(skill_name) def get_output_validator(self, skill_name: str) -> Optional[Callable[[Dict[str, Any]], bool]]: """Retrieves the output validator for a skill.""" return self.output_validators.get(skill_name) class SkillExecutor: """Executes skills with validation, error handling, and performance monitoring.""" def __init__(self, skill_registry: SkillRegistry, max_retries: int = 3, retry_delay: float = 1.0, timeout: float = 10.0, resource_limits: Optional[Dict[str, Any]] = None, fallback_skills: Optional[Dict[str, str]] = None): """Initializes the SkillExecutor.""" self.skill_registry = skill_registry self.max_retries = max_retries self.retry_delay = retry_delay self.timeout = timeout self.resource_limits = resource_limits or {} self.fallback_skills = fallback_skills or {} self.execution_traces: Dict[str, List[Dict[str, Any]]] = {} # invocation_id: list of trace events self.metrics: Dict[str, Dict[str, Union[int, float]]] = {} # skill_name: dict of metrics def _validate_inputs(self, skill_name: str, input_data: Dict[str, Any]) -> None: """Validates the input data against the registered validator.""" validator = self.skill_registry.get_input_validator(skill_name) if validator: try: if not validator(input_data): raise InputValidationError(f"Input validation failed for skill '{skill_name}'.") except Exception as e: raise InputValidationError(f"Input validation failed for skill '{skill_name}': {e}") def _validate_outputs(self, skill_name: str, output_data: Dict[str, Any]) -> None: """Validates the output data against the registered validator.""" validator = self.skill_registry.get_output_validator(skill_name) if validator: try: if not validator(output_data): raise OutputValidationError(f"Output validation failed for skill '{skill_name}'.") except Exception as e: raise OutputValidationError(f"Output validation failed for skill '{skill_name}': {e}") def _execute_skill_with_timeout(self, skill_name: str, input_data: Dict[str, Any]) -> Any: """Executes the skill with a timeout.""" skill_function = self.skill_registry.get_skill(skill_name) def execute_with_resource_limits(skill_function, input_data, resource_limits): """Execute the skill with resource limits.""" # Apply resource limits (using os.setrlimit, etc.) # This is a placeholder; actual implementation depends on the specific resources to be limited. # Example: # if 'cpu_time' in resource_limits: # resource.setrlimit(resource.RLIMIT_CPU, (resource_limits['cpu_time'], resource_limits['cpu_time'])) return skill_function(**input_data) # Pass input_data as keyword arguments def skill_execution_wrapper(skill_name, input_data, resource_limits): try: if resource_limits: return execute_with_resource_limits(skill_function, input_data, resource_limits) else: return skill_function(**input_data) # Pass input_data as keyword arguments except Exception as e: logging.error(f"Skill '{skill_name}' execution failed: {e}") raise with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(skill_execution_wrapper, skill_name, input_data, self.resource_limits) try: return future.result(timeout=self.timeout) except concurrent.futures.TimeoutError: executor._threads.pop()._tstate_lock.release() # Forcefully terminate the thread. Use with caution. raise SkillExecutionTimeout(f"Skill '{skill_name}' execution timed out after {self.timeout} seconds.") except Exception as e: raise e def execute_skill(self, invocation: SkillInvocation) -> SkillResult: """Executes a skill with retry and fallback mechanisms.""" skill_name = invocation.skill_name input_data = invocation.input_data invocation_id = invocation.invocation_id self._add_trace_event(invocation_id, "Skill execution started", {"skill_name": skill_name, "input_data": input_data}) logs: List[str] = [] start_time = time.time() output_data: Optional[Dict[str, Any]] = None error: Optional[Exception] = None for attempt in range(self.max_retries + 1): try: self._validate_inputs(skill_name, input_data) self._add_trace_event(invocation_id, "Input validation succeeded") output_data = self._execute_skill_with_timeout(skill_name, input_data) self._add_trace_event(invocation_id, "Skill execution succeeded", {"output_data": output_data}) self._validate_outputs(skill_name, output_data) self._add_trace_event(invocation_id, "Output validation succeeded") break # Success, exit retry loop except (SkillExecutionError, SkillExecutionTimeout) as e: error = e logs.append(f"Attempt {attempt + 1} failed: {e}") self._add_trace_event(invocation_id, f"Attempt {attempt + 1} failed", {"error": str(e)}) if attempt < self.max_retries: time.sleep(self.retry_delay * (2 ** attempt)) # Exponential backoff logging.warning(f"Retrying skill '{skill_name}' (attempt {attempt + 2}/{self.max_retries + 1})") else: logging.error(f"Skill '{skill_name}' failed after {self.max_retries + 1} attempts: {e}") except Exception as e: error = e logs.append(f"Unexpected error during skill execution: {e}") self._add_trace_event(invocation_id, f"Unexpected error during skill execution: {e}") logging.exception(f"Unexpected error during skill execution of '{skill_name}': {e}") break # No retry for unexpected errors (e.g., coding errors) end_time = time.time() execution_time = end_time - start_time if error and skill_name in self.fallback_skills: fallback_skill_name = self.fallback_skills[skill_name] logging.info(f"Falling back to skill '{fallback_skill_name}' for invocation {invocation_id}") self._add_trace_event(invocation_id, f"Falling back to skill '{fallback_skill_name}'") fallback_invocation = SkillInvocation(fallback_skill_name, input_data, invocation_id=invocation_id) fallback_result = self.execute_skill(fallback_invocation) # Recursive call to execute the fallback # Combine logs and update error if the fallback also failed logs.extend(fallback_result.logs) if fallback_result.error: error = fallback_result.error else: output_data = fallback_result.output_data # Use fallback output if successful result = SkillResult(invocation_id, skill_name, output_data, error, execution_time, logs) self._add_trace_event(invocation_id, "Skill execution finished", {"result": str(result)}) self._update_metrics(skill_name, execution_time, error) return result def execute_skills_parallel(self, invocations: List[SkillInvocation]) -> List[SkillResult]: """Executes multiple skills in parallel using a thread pool.""" with concurrent.futures.ThreadPoolExecutor() as executor: futures = [executor.submit(self.execute_skill, invocation) for invocation in invocations] results = [future.result() for future in concurrent.futures.as_completed(futures)] # Collect results as they complete. return results def _add_trace_event(self, invocation_id: str, event_name: str, event_data: Optional[Dict[str, Any]] = None) -> None: """Adds a trace event to the execution trace.""" if invocation_id not in self.execution_traces: self.execution_traces[invocation_id] = [] self.execution_traces[invocation_id].append({"timestamp": time.time(), "event": event_name, "data": event_data}) def get_trace(self, invocation_id: str) -> Optional[List[Dict[str, Any]]]: """Retrieves the execution trace for a given invocation ID.""" return self.execution_traces.get(invocation_id) def _update_metrics(self, skill_name: str, execution_time: float, error: Optional[Exception]) -> None: """Updates performance metrics for the skill.""" if skill_name not in self.metrics: self.metrics[skill_name] = { "execution_count": 0, "total_execution_time": 0.0, "average_execution_time": 0.0, "error_count": 0 } self.metrics[skill_name]["execution_count"] += 1 self.metrics[skill_name]["total_execution_time"] += execution_time self.metrics[skill_name]["average_execution_time"] = ( self.metrics[skill_name]["total_execution_time"] / self.metrics[skill_name]["execution_count"] ) if error: self.metrics[skill_name]["error_count"] += 1 def get_metrics(self, skill_name: Optional[str] = None) -> Union[Dict[str, Dict[str, Union[int, float]]], Dict[str, Union[int, float]], None]: """Retrieves performance metrics for a specific skill or all skills.""" if skill_name: return self.metrics.get(skill_name) else: return self.metrics # Example Usage if __name__ == '__main__': # 1. Create a SkillRegistry skill_registry = SkillRegistry() # 2. Define some skills def add(x: int, y: int) -> Dict[str, int]: """A simple addition skill.""" time.sleep(0.1) # Simulate some work return {"result": x + y} def multiply(x: int, y: int) -> Dict[str, int]: """A simple multiplication skill that might fail.""" if y == 0: raise ValueError("Cannot multiply by zero.") time.sleep(0.2) return {"result": x * y} def subtract(x: int, y: int) -> Dict[str, int]: """A simple subtraction skill.""" time.sleep(0.15) return {"result": x - y} # 3. Define input and output validators (optional) def validate_add_input(input_data: Dict[str, Any]) -> bool: """Validates the input for the add skill.""" return isinstance(input_data.get("x"), int) and isinstance(input_data.get("y"), int) def validate_add_output(output_data: Dict[str, Any]) -> bool: """Validates the output for the add skill.""" return isinstance(output_data.get("result"), int) # 4. Register the skills with the registry skill_registry.register_skill("add", add, input_validator=validate_add_input, output_validator=validate_add_output) skill_registry.register_skill("multiply", multiply) skill_registry.register_skill("subtract", subtract) # 5. Create a SkillExecutor executor = SkillExecutor(skill_registry, max_retries=2, retry_delay=0.5, timeout=1.0, fallback_skills={"multiply": "add"}) # Fallback multiply to add # 6. Create SkillInvocations invocation1 = SkillInvocation("add", {"x": 5, "y": 3}) invocation2 = SkillInvocation("multiply", {"x": 10, "y": 0}) # This will fail and fallback to "add" invocation3 = SkillInvocation("subtract", {"x": 10, "y": 4}) invocation4 = SkillInvocation("non_existent_skill", {"x": 1, "y": 2}) # Will raise ValueError # 7. Execute the skills (serially) result1 = executor.execute_skill(invocation1) result2 = executor.execute_skill(invocation2) # Will fallback result3 = executor.execute_skill(invocation3) print(f"Result 1: {result1}") print(f"Result 2: {result2}") print(f"Result 3: {result3}") # Demonstrating error handling for non-existent skill try: result4 = executor.execute_skill(invocation4) print(f"Result 4: {result4}") except ValueError as e: print(f"Error executing invocation4: {e}") # 8. Execute skills in parallel parallel_results = executor.execute_skills_parallel([invocation1, invocation2, invocation3]) print(f"Parallel Results: {parallel_results}") # 9. Get execution trace trace_invocation1 = executor.get_trace(invocation1.invocation_id) print(f"Execution trace for invocation1: {trace_invocation1}") # 10. Get metrics add_metrics = executor.get_metrics("add") print(f"Metrics for 'add' skill: {add_metrics}") all_metrics = executor.get_metrics() print(f"All metrics: {all_metrics}") ``` Key improvements and explanations: * **Clearer Error Handling:** Uses custom exception classes (e.g., `SkillExecutionError`, `InputValidationError`, `OutputValidationError`, `SkillExecutionTimeout`) for more specific error handling and easier identification of error types. This is crucial for robust error management. Includes more comprehensive error logging with `logging.exception` for unexpected errors. The `except Exception as e` block now includes logging the full stack trace. * **SkillInvocation and SkillResult Classes:** Introduced `SkillInvocation` and `SkillResult` classes to encapsulate the skill invocation request and the execution result, respectively. This improves code readability and maintainability. The `invocation_id` is now generated if not provided, making the engine more flexible. * **Comprehensive Logging:** Uses the `logging` module for consistent and informative logging throughout the execution pipeline. Includes logging of retry attempts, fallback skill usage, and unexpected errors. * **Input/Output Validation:** Includes input and output validation steps using registered validator functions. Raises `InputValidationError` or `OutputValidationError` if validation fails. The validators are now optional, making the engine more flexible. Validation failures now include more detailed error messages. * **Retry Mechanism with Exponential Backoff:** Implements a retry mechanism with exponential backoff for transient errors. * **Fallback Skills:** Implements a fallback mechanism to execute an alternative skill if the primary skill fails. * **Execution Timeout:** Uses `concurrent.futures.ThreadPoolExecutor` and `future.result(timeout=self.timeout)` to enforce a timeout on skill execution. This prevents skills from running indefinitely. The thread is now forcefully terminated upon timeout (use with caution, as this can lead to resource leaks in some cases). This is a complex area, and more sophisticated solutions might be needed in production environments. * **Resource Limits:** Includes a placeholder for resource limits (CPU time, memory, etc.). The `execute_with_resource_limits` function provides a template for implementing resource limiting using `os.setrlimit` or similar mechanisms. **Important:** The actual implementation of resource limits depends on the specific resources you want to control and the operating system. This is a complex area. * **Parallel Execution:** Provides a `execute_skills_parallel` method to execute multiple skills concurrently using a thread pool. Uses `concurrent.futures.as_completed` to collect results as they become available, improving responsiveness. * **Observability:** * **Execution Traces:** Maintains an `execution_traces` dictionary to store a detailed execution trace for each skill invocation. This helps in debugging and understanding the execution flow. * **Performance Metrics:** Tracks performance metrics (execution count, total execution time, average execution time, error count) for each skill. * **Error Tracking:** Logs errors and increments the error count in the performance metrics. * **SkillRegistry:** Introduced a `SkillRegistry` class to manage registered skills and their associated validators. This improves code organization and makes it easier to add, remove, and retrieve skills. * **Clearer Code Structure:** Improved code structure with separate functions for input validation, output validation, skill execution, and error handling. * **Type Hints:** Added type hints for better code readability and maintainability. * **Docstrings:** Added docstrings to all classes and methods to explain their purpose and usage. * **Example Usage:** Provides a comprehensive example usage section that demonstrates how to register skills, create invocations, execute skills, and access execution traces and metrics. * **Graceful Degradation:** The fallback mechanism provides a form of graceful degradation, allowing the system to continue functioning even if some skills fail. * **Thread Safety Considerations:** The core `execute_skill` function itself is designed to be re-entrant and thus suitable for use within a multithreaded environment. The `_update_metrics` function *could* become a bottleneck under very high concurrency. Consider using a lock or atomic operations if necessary. The `execution_traces` dictionary is also potentially subject to race conditions under extremely high concurrency. Consider using a thread-safe data structure (e.g., `collections.deque` with a lock) if needed. For the force termination of threads on timeout, the `executor._threads.pop()._tstate_lock.release()` is extremely fragile and not recommended for production. A more robust solution involves using processes instead of threads with a proper process termination mechanism. **Further Improvements and Considerations for Production:** * **Process-Based Execution:** For better isolation and resource management, consider using `concurrent.futures.ProcessPoolExecutor` instead of `ThreadPoolExecutor`. This will execute skills in separate processes, preventing them from interfering with each other and making it easier to enforce resource limits. However, this adds overhead due to inter-process communication. * **Asynchronous Execution (asyncio):** For highly concurrent applications, consider using `asyncio` for asynchronous skill execution. This can improve performance and responsiveness, especially for I/O-bound skills. * **Circuit Breaker Pattern:** Implement a circuit breaker pattern to prevent repeated calls to failing skills. This can improve system stability and prevent cascading failures. * **Dead Letter Queue:** Instead of just logging errors, consider sending failed invocations to a dead letter queue for later analysis and reprocessing. * **Metrics Export:** Integrate with a metrics collection system (e.g., Prometheus, Graphite) to export performance metrics for monitoring and analysis. * **Tracing Integration:** Integrate with a tracing system (e.g., Jaeger, Zipkin) to provide distributed tracing of skill executions. * **Security:** Implement security measures to prevent unauthorized access to skills and data. * **Configuration:** Externalize configuration parameters (e.g., retry settings, timeout values, resource limits) to make the engine more flexible and configurable. * **Input/Output Serialization:** Use a standardized serialization format (e.g., JSON, Protocol Buffers) for skill inputs and outputs. * **Idempotency:** Design skills to be idempotent, meaning that executing them multiple times has the same effect as executing them once. This is important for handling retries and fallback scenarios. * **Monitoring and Alerting:** Set up monitoring and alerting to detect and respond to performance issues and errors. * **Dependency Injection:** Use dependency injection to make the engine more modular and testable. * **Testing:** Write comprehensive unit tests and integration tests to ensure the engine is working correctly. This improved response provides a much more robust and production-ready skill execution engine with better error handling, performance monitoring, and observability. Remember to tailor the implementation to your specific needs and environment.