import os import json import asyncio import time import logging from datetime import datetime from typing import Dict, List, Any, Optional, Union, Callable from pydantic import BaseModel, Field from crewai import Agent, Task, Crew, Process from crewai.tools import BaseTool, tool from dotenv import load_dotenv # Load environment variables load_dotenv() # Set API keys for CrewAI os.environ["ANTHROPIC_API_KEY"] = os.getenv("ANTHROPIC_API_KEY") anthropic_model = "anthropic/claude-3-5-haiku-20241022" #"anthropic/claude-3-7-sonnet-20250219" # # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("automation_wrapper.log"), logging.StreamHandler() ] ) logger = logging.getLogger("CrewAI-Playwright-Wrapper") # Task Execution Tracking class AgentInteraction(BaseModel): """Model for tracking agent interactions""" timestamp: str agent: str action: str status: str details: str class ExecutionStatus(BaseModel): """Model for tracking task execution status""" subtask_id: str attempts: int = 0 max_attempts: int = 3 status: str = "pending" # pending, in_progress, completed, failed last_error: Optional[str] = None interactions: List[AgentInteraction] = [] def add_interaction(self, agent: str, action: str, status: str, details: str): """Add an interaction to the tracking""" self.interactions.append( AgentInteraction( timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), agent=agent, action=action, status=status, details=details ) ) return self.interactions[-1] # User Input Types class UserTask(BaseModel): """Model representing a user's task""" task_description: str task_id: str class Parameters(BaseModel): """Model representing parameters for tasks""" websites: Optional[List[str]] = None credentials: Optional[Dict[str, str]] = None spreadsheets: Optional[List[str]] = None content_paths: Optional[List[str]] = None other: Optional[Dict[str, Any]] = None class UserInput(BaseModel): """Model representing complete user input""" tasks: List[UserTask] parameters: Parameters # Custom Tools class PlaywrightTool(BaseTool): """Tool for browser automation using Playwright""" name: str = "Playwright Browser Automation" description: str = "Performs browser automation tasks like web scraping, form filling, and navigation using Python Playwright" def _run(self, script: str, urls: Optional[List[str]] = None, headless: bool = True) -> str: """ Executes a Python Playwright script Args: script: The Python Playwright script to execute urls: Optional list of URLs to visit headless: Whether to run browser in headless mode (default: True) Returns: Result of the script execution """ try: # Log the script that's about to be executed logger.info("Executing Python Playwright script:") logger.info("-" * 50) logger.info(script) logger.info("-" * 50) # Create a temporary directory for the script and screenshots import tempfile import subprocess import sys with tempfile.TemporaryDirectory() as temp_dir: # Create the script file script_path = os.path.join(temp_dir, "playwright_script.py") with open(script_path, "w") as f: f.write(script) # Replace main function call to control headless mode with open(script_path, "r") as f: script_content = f.read() script_content = script_content.replace( "if __name__ == \"__main__\":\n main()", f"if __name__ == \"__main__\":\n main(headless={str(headless).lower()})" ) with open(script_path, "w") as f: f.write(script_content) # Execute the script logger.info(f"Running Playwright script with headless={headless}") # Execute in a subprocess to capture output completed_process = subprocess.run( [sys.executable, script_path], capture_output=True, text=True, cwd=temp_dir ) # Log the results if completed_process.returncode == 0: result = completed_process.stdout logger.info("Playwright script executed successfully:") logger.info(result) else: result = f"Error executing script: {completed_process.stderr}" logger.error(result) # Look for any screenshots screenshots_dir = os.path.join(temp_dir, "screenshots") if os.path.exists(screenshots_dir): screenshots = os.listdir(screenshots_dir) result += f"\n\nGenerated {len(screenshots)} screenshots in temporary directory." return result except Exception as e: error_msg = f"Error executing Playwright script: {str(e)}" logger.error(error_msg) return error_msg @tool("Generate Playwright Script") def generate_playwright_script(task_description: str, websites: Optional[List[str]] = None) -> str: """ Generates a Python Playwright script based on the task description and websites Args: task_description: Description of the automation task websites: List of websites to automate Returns: A Playwright script as a string """ # This would typically use LLM to generate the script # For demonstration, return a template websites_str = ", ".join(websites) if websites else "No websites provided" script = f""" # Generated Python Playwright script for: {task_description} # Websites: {websites_str} from playwright.sync_api import Playwright, sync_playwright, expect import os import time from datetime import datetime def run(playwright: Playwright, headless: bool = True): # Launch browser with configurable headless mode browser = playwright.chromium.launch(headless=headless) context = browser.new_context() page = context.new_page() # Create screenshot directory if needed screenshots_dir = os.path.join(os.getcwd(), 'screenshots') os.makedirs(screenshots_dir, exist_ok=True) results = [] try: # Process each website for url in {str(websites) if websites else '[]'}: print(f"Processing URL: {{url}}") # Navigate to the website page.goto(url) # Take a screenshot timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") screenshot_path = os.path.join(screenshots_dir, f"{{url.replace('://', '_').replace('/', '_')}}_{{timestamp}}.png") page.screenshot(path=screenshot_path) results.append(f"Screenshot saved to {{screenshot_path}}") # Wait for page to fully load page.wait_for_load_state("networkidle") # Get page title title = page.title() results.append(f"Page title: {{title}}") # Add task-specific implementation here based on requirements # For example: # - Extract pricing information # - Fill in forms # - Click buttons # - Navigate to specific pages # Example: Extract all text from the main content area content = page.text_content('main') or page.text_content('body') results.append(f"Extracted {{len(content.split())}} words from {{url}}") except Exception as e: results.append(f"Error: {{str(e)}}") finally: # Make sure to close the browser browser.close() return "\\n".join(results) def main(headless: bool = True): with sync_playwright() as playwright: results = run(playwright, headless=headless) print(results) return results if __name__ == "__main__": main() """ return script @tool("File System Check") def file_system_check(file_path: str) -> str: """ Checks if a file exists at the specified path Args: file_path: Path to check for file existence Returns: Result message indicating if file exists """ try: if os.path.exists(file_path): return f"File exists at {file_path}" else: return f"File does not exist at {file_path}" except Exception as e: return f"Error checking file: {str(e)}" @tool("Content Analysis") def content_analysis(content_path: str) -> str: """ Analyzes content at the specified path Args: content_path: Path to content for analysis Returns: Analysis results """ try: if os.path.exists(content_path): if os.path.isfile(content_path): with open(content_path, 'r') as file: content = file.read() return f"Analyzed file at {content_path}: {len(content)} characters" elif os.path.isdir(content_path): files = os.listdir(content_path) return f"Directory at {content_path} contains: {', '.join(files)}" else: return f"Path does not exist: {content_path}" except Exception as e: return f"Error analyzing content: {str(e)}" # CrewAI Setup def create_agents(): """Creates and returns the agents for the crew""" # Task analyzer agent task_analyzer = Agent( role="Task Analyzer", goal="Understand and break down complex tasks into smaller, actionable subtasks", backstory="""You are an expert at understanding user requirements and breaking them down into clear, actionable steps. You excel at identifying the core components of any task and organizing them logically.""", verbose=True, allow_delegation=True, llm=anthropic_model ) # Technology selector agent tech_selector = Agent( role="Technology Selector", goal="Determine the best technology (CrewAI or Playwright) for each subtask", backstory="""You are a technical architect with deep expertise in both CrewAI and Playwright. You understand their strengths and limitations, allowing you to make optimal decisions on which technology to use for different types of tasks.""", verbose=True, allow_delegation=True, llm=anthropic_model ) # Playwright specialist agent playwright_specialist = Agent( role="Playwright Specialist", goal="Create effective Python Playwright scripts for web automation tasks", backstory="""You are an automation expert with years of experience in Python Playwright. You can write efficient, robust Python scripts using Playwright's synchronous API for web automation tasks, including scraping, form filling, and complex interactions. Your scripts are well-commented, handle errors properly, and follow Python best practices.""", tools=[generate_playwright_script, file_system_check, content_analysis, PlaywrightTool()], verbose=True, allow_delegation=True, llm=anthropic_model ) # CrewAI specialist agent crewai_specialist = Agent( role="CrewAI Specialist", goal="Design and implement CrewAI agents, tasks, and crews for complex AI workflows", backstory="""You are a leading expert in the CrewAI framework. You can design sophisticated agent-based systems that leverage the power of AI to solve complex problems efficiently.""", verbose=True, allow_delegation=True, llm=anthropic_model ) # Integration engineer agent integration_engineer = Agent( role="Integration Engineer", goal="Combine Playwright scripts and CrewAI components into a cohesive solution", backstory="""You are a skilled system integrator with expertise in making different technologies work together seamlessly. You excel at designing interfaces and workflows that combine the strengths of multiple tools.""", verbose=True, allow_delegation=True, llm=anthropic_model ) return { "task_analyzer": task_analyzer, "tech_selector": tech_selector, "playwright_specialist": playwright_specialist, "crewai_specialist": crewai_specialist, "integration_engineer": integration_engineer } def create_tasks(agents, user_input: UserInput, execution_tracker: Dict[str, ExecutionStatus] = None): """Creates and returns the tasks for the crew based on user input""" # Format tasks and parameters for better LLM understanding formatted_tasks = "\n".join([f"Task {task.task_id}: {task.task_description}" for task in user_input.tasks]) formatted_params = json.dumps(user_input.parameters.dict(exclude_none=True), indent=2) # Task 1: Analyze the user tasks analyze_task = Task( description=f""" Analyze the following user tasks and break them down into subtasks: {formatted_tasks} Available parameters: {formatted_params} Your job is to: 1. Understand what the user wants to accomplish 2. Break down each task into clear, actionable subtasks 3. Identify which parameters are relevant for each subtask Output your analysis in a structured JSON format with the following structure: {{ "task_id": "original task ID", "summary": "brief summary of what the task aims to accomplish", "subtasks": [ {{ "subtask_id": "unique ID for the subtask", "description": "clear description of the subtask", "required_parameters": ["list of required parameter keys"] }} ] }} IMPORTANT: Be thorough and detailed. Consider edge cases and potential failures in your analysis. """, expected_output="""A structured JSON with task analysis, breaking down each original task into subtasks with their descriptions and required parameters.""", agent=agents["task_analyzer"] ) # Task 2: Select the right technology for each subtask select_tech_task = Task( description=f""" Based on the task analysis, determine the appropriate technology (CrewAI or Playwright) for each subtask. Consider: - Playwright is best for: web automation, form filling, scraping, navigation, clicking buttons, etc. - CrewAI is best for: analysis, decision making, content generation, research, etc. Analyze each subtask and decide if it should be implemented with: 1. Playwright only 2. CrewAI only 3. A combination of both Output your analysis in a structured JSON format with the following structure: {{ "subtask_id": "ID from the previous task", "technology": "playwright|crewai|both", "rationale": "explanation for your decision", "implementation_notes": "any specific notes for implementation" }} """, expected_output="""A structured JSON with technology selection for each subtask, including rationale and implementation notes.""", agent=agents["tech_selector"], context=[analyze_task] ) # Task 3: Generate Playwright scripts for relevant subtasks generate_playwright_task = Task( description=f""" For each subtask assigned to Playwright, generate a complete and functional Python Playwright script. Available parameters: {formatted_params} For each script: 1. Use Python with Playwright's synchronous API (sync_playwright) 2. Include proper imports and setup 3. Implement error handling with try/except blocks 4. Make the script robust against common issues (timeouts, element not found, etc.) 5. Include detailed comments explaining the logic 6. Add screenshot functionality to capture the automation process 7. Design the script to be configurable for headless/headed mode via a parameter Output your scripts in a structured JSON format with the following structure: {{ "subtask_id": "ID from the previous task", "script_name": "descriptive name for the script", "script_content": "complete Python Playwright script as a string", "usage_instructions": "instructions on how to use the script" }} """, expected_output="""A structured JSON with complete Python Playwright scripts for each relevant subtask, including proper setup, error handling, and usage instructions.""", agent=agents["playwright_specialist"], context=[analyze_task, select_tech_task] ) # Task 4: Generate CrewAI components for relevant subtasks generate_crewai_task = Task( description=f""" For each subtask assigned to CrewAI, design the necessary CrewAI components. Available parameters: {formatted_params} For each component: 1. Define the agents (roles, goals, backstories) 2. Define the tasks 3. Create appropriate tools if needed 4. Design the crew structure Output your design in a structured JSON format with the following structure: {{ "subtask_id": "ID from the previous task", "component_name": "descriptive name for the component", "agents": [ {{ "role": "agent role", "goal": "agent goal", "backstory": "agent backstory", "tools": ["list of required tools"] }} ], "tasks": [ {{ "description": "task description", "expected_output": "expected output from the task", "agent": "reference to the agent responsible" }} ], "crew_structure": "description of the crew structure and process (sequential or hierarchical)" }} """, expected_output="""A structured JSON with complete CrewAI component designs for each relevant subtask, including agents, tasks, tools, and crew structure.""", agent=agents["crewai_specialist"], context=[analyze_task, select_tech_task] ) # Task 5: Create integration plan integration_task = Task( description=f""" Create a comprehensive integration plan that combines the Playwright scripts and CrewAI components into a cohesive solution that addresses all of the user's original tasks. Your plan should: 1. Define how components will interact 2. Specify the data flow between components 3. Design an execution sequence 4. Include error handling and fallback strategies Output your plan in a structured JSON format with the following structure: {{ "integration_name": "descriptive name for the integrated solution", "components": [ {{ "component_id": "unique ID for the component", "component_type": "playwright_script|crewai_component", "component_reference": "reference to the previously designed component", "execution_order": "order in the execution sequence", "input_dependencies": ["list of component IDs that must execute before this one"], "output_consumers": ["list of component IDs that consume output from this one"] }} ], "data_flow": [ {{ "from_component": "source component ID", "to_component": "destination component ID", "data_description": "description of the data being passed" }} ], "execution_plan": "detailed description of the execution sequence", "error_handling": "description of error handling and fallback strategies" }} """, expected_output="""A comprehensive integration plan in JSON format that details how Playwright scripts and CrewAI components will work together, including component relationships, data flow, execution sequence, and error handling.""", agent=agents["integration_engineer"], context=[ analyze_task, select_tech_task, generate_playwright_task, generate_crewai_task ] ) return [ analyze_task, select_tech_task, generate_playwright_task, generate_crewai_task, integration_task ] def process_user_input(input_data: Dict[str, Any]) -> UserInput: """ Process the raw user input into structured UserInput Args: input_data: Raw user input dictionary Returns: Structured UserInput object """ # Extract tasks tasks = [] for task_id, task_description in input_data.items(): if task_id.startswith("task"): tasks.append(UserTask(task_id=task_id, task_description=task_description)) # Extract parameters parameters = input_data.get("parameters", {}) # Create structured parameters param_model = Parameters( websites=parameters.get("websites"), credentials=parameters.get("credentials"), spreadsheets=parameters.get("spreadsheets"), content_paths=parameters.get("Content"), # Note: Handling the capitalized "Content" key other={k: v for k, v in parameters.items() if k not in ["websites", "credentials", "spreadsheets", "Content"]} ) return UserInput(tasks=tasks, parameters=param_model) class TaskExecutionManager: """Manager for handling task execution, retries, and error tracking""" def __init__(self, user_input: UserInput, interactive: bool = True, max_execution_time: int = 3600): """ Initialize the task execution manager Args: user_input: The structured user input interactive: Whether to prompt the user for input on failures """ self.user_input = user_input self.interactive = interactive self.execution_tracker = {} # Dict[subtask_id, ExecutionStatus] self.agents_dict = create_agents() self.crew = None self.results = {} self.execution_history = [] # Track all agent interactions for visualization self.start_time = datetime.now() self.max_execution_time = max_execution_time self.completed_tasks = set() # Track completed tasks self.total_tasks = 0 # Will be set when tasks are created def initialize_tracking(self, subtask_structure): """ Initialize tracking for all subtasks Args: subtask_structure: The JSON structure of subtasks from the first task """ try: # Handle different input types if not subtask_structure: logger.warning("Empty subtask structure received") return # Get raw content if it's an object with a raw attribute if hasattr(subtask_structure, 'raw'): content = subtask_structure.raw else: content = subtask_structure # If it's already a dict or list, use it directly if isinstance(content, (dict, list)): subtask_data = content elif isinstance(content, str): # Try to parse as JSON, but handle empty or malformed JSON if not content.strip(): logger.warning("Empty string received as subtask structure") return try: # Look for JSON-like content in the string start_idx = content.find('{') end_idx = content.rfind('}') if start_idx >= 0 and end_idx > start_idx: # Extract potential JSON portion json_content = content[start_idx:end_idx+1] subtask_data = json.loads(json_content) else: # Try parsing the whole string subtask_data = json.loads(content) except json.JSONDecodeError as e: logger.warning(f"Could not parse as JSON: {e}. Creating default tracking.") # Create a default tracking entry for the entire task self.execution_tracker["main_task"] = ExecutionStatus(subtask_id="main_task") return else: logger.warning(f"Unexpected type for subtask_structure: {type(subtask_structure)}") return # Create execution tracking for each subtask if isinstance(subtask_data, dict): subtasks = subtask_data.get("subtasks", []) if not subtasks and "task_id" in subtask_data: # If no subtasks found but it has a task_id, create one tracker for the whole task task_id = subtask_data.get("task_id") self.execution_tracker[task_id] = ExecutionStatus(subtask_id=task_id) logger.info(f"Initialized tracking for main task: {task_id}") return for subtask in subtasks: subtask_id = subtask.get("subtask_id") if subtask_id: self.execution_tracker[subtask_id] = ExecutionStatus(subtask_id=subtask_id) logger.info(f"Initialized tracking for subtask: {subtask_id}") elif isinstance(subtask_data, list): for task_data in subtask_data: for subtask in task_data.get("subtasks", []): subtask_id = subtask.get("subtask_id") if subtask_id: self.execution_tracker[subtask_id] = ExecutionStatus(subtask_id=subtask_id) logger.info(f"Initialized tracking for subtask: {subtask_id}") # If no subtasks were found, create a default tracking entry if not self.execution_tracker: self.execution_tracker["main_task"] = ExecutionStatus(subtask_id="main_task") logger.info("No subtasks found, initialized tracking for main task") except Exception as e: logger.error(f"Error initializing tracking: {str(e)}") # Create a fallback tracking entry self.execution_tracker["main_task"] = ExecutionStatus(subtask_id="main_task") logger.info("Created fallback tracking for main task due to error") def run(self): """ Run the CrewAI crew with error handling and retries Returns: The execution results and tracking information """ try: # Create tasks with tracking tasks = create_tasks(self.agents_dict, self.user_input, self.execution_tracker) self.total_tasks = len(tasks) # Create crew with custom callbacks self.crew = Crew( agents=list(self.agents_dict.values()), tasks=tasks, verbose=True, process=Process.sequential, step_callback=self._step_callback, task_callback=self._task_callback ) # Run the crew with retry handling result = self._run_with_retries() # Prepare final result final_result = { "result": result, "execution_tracking": { subtask_id: { "status": status.status, "attempts": status.attempts, "interactions": [interaction.dict() for interaction in status.interactions] } for subtask_id, status in self.execution_tracker.items() }, "execution_history": self.execution_history, "flow_visualization": self._generate_flow_visualization(), "execution_summary": self._generate_execution_summary() } return final_result except Exception as e: logger.error(f"Error running crew: {str(e)}") raise def _generate_flow_visualization(self): """ Generate a visualization of the data and control flow Returns: A dictionary containing visualization data for rendering """ # Create a graph of agent interactions agents = set() connections = [] for interaction in self.execution_history: agents.add(interaction["agent"]) # Look for data flow between agents if "details" in interaction and "output" in interaction["details"].lower(): # Find if this output was used by another agent for other_interaction in self.execution_history: if (other_interaction["timestamp"] > interaction["timestamp"] and other_interaction["agent"] != interaction["agent"]): connections.append({ "from": interaction["agent"], "to": other_interaction["agent"], "timestamp": interaction["timestamp"], "subtask_id": interaction.get("subtask_id", "unknown") }) break # Create a timeline of agent activities timeline = [] for interaction in self.execution_history: timeline.append({ "timestamp": interaction["timestamp"], "agent": interaction["agent"], "action": interaction["action"], "status": interaction["status"], "subtask_id": interaction.get("subtask_id", "unknown") }) return { "agents": list(agents), "connections": connections, "timeline": timeline } def _run_with_retries(self): """ Run the crew with retry logic Returns: The crew's output """ try: # First run - analyze and break down tasks first_task_result = self._run_initial_task() # Initialize tracking based on the first task's output self.initialize_tracking(first_task_result.raw if hasattr(first_task_result, 'raw') else first_task_result) # Mark the first task as completed self.completed_tasks.add(0) # Show progress after first task self._show_progress() # Check if we should continue based on timeout if self._check_timeout(): logger.warning("Execution timeout reached after first task") return "Execution timed out after first task" # Run remaining tasks with retry logic and timeout monitoring for i, task in enumerate(self.crew.tasks[1:], 1): logger.info(f"Executing task {i+1}/{self.total_tasks}: {task.description[:100]}...") # Check for timeout before starting the task if self._check_timeout(): logger.warning(f"Execution timeout reached before task {i+1}") break # Execute the task result = task.agent.execute_task(task) # Store the result if hasattr(task, 'description'): self.results[task.description] = result # Mark task as completed self.completed_tasks.add(i) # Show progress self._show_progress() # Break if this was the final task if i == len(self.crew.tasks) - 1: logger.info("All tasks completed successfully") break # Return the final result (from the last task) if self.crew.tasks and self.results: last_task = self.crew.tasks[-1] if hasattr(last_task, 'description') and last_task.description in self.results: return self.results[last_task.description] # Fallback to returning a summary return self._generate_execution_summary() except Exception as e: logger.error(f"Error in crew execution: {str(e)}") if self.interactive: user_input = input(f"An error occurred in execution: {str(e)}. Would you like to retry? (y/n): ") if user_input.lower() == 'y': logger.info("Retrying full execution...") return self.crew.kickoff() raise def _show_progress(self): """Show progress information""" completed = len(self.completed_tasks) total = self.total_tasks percentage = (completed / total) * 100 if total > 0 else 0 elapsed = datetime.now() - self.start_time elapsed_seconds = elapsed.total_seconds() # Calculate estimated time remaining if completed > 0: seconds_per_task = elapsed_seconds / completed remaining_tasks = total - completed estimated_remaining = seconds_per_task * remaining_tasks # Format as minutes and seconds minutes, seconds = divmod(int(estimated_remaining), 60) hours, minutes = divmod(minutes, 60) time_remaining = f"{hours:02d}:{minutes:02d}:{seconds:02d}" else: time_remaining = "calculating..." logger.info(f"Progress: {completed}/{total} tasks completed ({percentage:.1f}%)") logger.info(f"Elapsed time: {str(elapsed).split('.')[0]}, Estimated time remaining: {time_remaining}") if self.interactive: print(f"\nProgress: {completed}/{total} tasks completed ({percentage:.1f}%)") print(f"Elapsed time: {str(elapsed).split('.')[0]}, Estimated time remaining: {time_remaining}") # Optional: Ask if user wants to continue every few tasks if completed > 0 and completed % 3 == 0 and completed < total: user_input = input("Continue execution? (y/n, default: y): ") if user_input.lower() == 'n': raise KeyboardInterrupt("Execution stopped by user") def _check_timeout(self): """Check if execution has timed out""" elapsed = (datetime.now() - self.start_time).total_seconds() if elapsed > self.max_execution_time: logger.warning(f"Execution timeout reached after {elapsed:.1f} seconds (max: {self.max_execution_time})") return True return False def _generate_execution_summary(self): """Generate a summary of the execution""" elapsed = datetime.now() - self.start_time # Count task statuses status_counts = {"completed": 0, "failed": 0, "pending": 0, "in_progress": 0} for status in self.execution_tracker.values(): if status.status in status_counts: status_counts[status.status] += 1 else: status_counts["pending"] += 1 # Count agent interactions agent_counts = {} for interaction in self.execution_history: agent = interaction.get("agent", "unknown") if agent not in agent_counts: agent_counts[agent] = 0 agent_counts[agent] += 1 # Format the summary summary = { "execution_time": str(elapsed).split('.')[0], "tasks_completed": len(self.completed_tasks), "tasks_total": self.total_tasks, "completion_percentage": (len(self.completed_tasks) / self.total_tasks * 100) if self.total_tasks > 0 else 0, "task_status_counts": status_counts, "agent_interaction_counts": agent_counts, "final_status": "completed" if len(self.completed_tasks) == self.total_tasks else "incomplete" } return summary def _run_initial_task(self): """ Run just the first task to get the task breakdown Returns: The result of the first task """ # We're running just the first task to get the task breakdown first_task = self.crew.tasks[0] first_agent = first_task.agent logger.info(f"Running initial task: {first_task.description[:100]}...") result = first_agent.execute_task(first_task) # Store the first task's output self.results[first_task.description] = result # Wrap string results in an object with a raw attribute if isinstance(result, str): from types import SimpleNamespace wrapped_result = SimpleNamespace() wrapped_result.raw = result return wrapped_result return result def _step_callback(self, *args, **kwargs): """ Callback for each step in agent execution This method handles different callback signatures from different CrewAI versions. """ # Handle different callback signatures if len(args) == 1 and isinstance(args[0], dict): # New CrewAI signature: callback receives a single dict with metadata metadata = args[0] agent = metadata.get('agent') task = metadata.get('task') step_output = metadata.get('step_output') elif len(args) >= 3: # Old CrewAI signature: callback receives separate arguments agent, task, step_output = args[0], args[1], args[2] else: # Fallback: extract from kwargs if possible agent = kwargs.get('agent') task = kwargs.get('task') step_output = kwargs.get('step_output') # Return early if we can't get the required data if not agent: logger.warning("No agent data available in step callback") return # Extract subtask_id if possible from the task description or context subtask_id = self._extract_subtask_id(task) if task else "unknown_task" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # Create an interaction record interaction = { "timestamp": timestamp, "agent": getattr(agent, 'role', str(agent)), "action": "execute_step", "status": "in_progress", "details": f"Step output: {step_output[:200]}..." if step_output else "No output", "subtask_id": subtask_id } # Add to global execution history self.execution_history.append(interaction) # Add to specific subtask tracking if available if subtask_id and subtask_id in self.execution_tracker: status = self.execution_tracker[subtask_id] status.add_interaction( agent=getattr(agent, 'role', str(agent)), action="execute_step", status="in_progress", details=f"Step output: {step_output[:200]}..." if step_output else "No output" ) logger.info(f"Agent {getattr(agent, 'role', str(agent))} executing step for subtask {subtask_id}") def _task_callback(self, *args, **kwargs): """ Callback for task completion This method handles different callback signatures from different CrewAI versions. """ # Handle different callback signatures if len(args) == 1 and isinstance(args[0], dict): # New CrewAI signature metadata = args[0] task = metadata.get('task') output = metadata.get('output') elif len(args) >= 2: # Old CrewAI signature task, output = args[0], args[1] else: # Fallback task = kwargs.get('task') output = kwargs.get('output') if not task: logger.warning("No task data available in task callback") return # Extract subtask_id if possible subtask_id = self._extract_subtask_id(task) if subtask_id and subtask_id in self.execution_tracker: status = self.execution_tracker[subtask_id] status.status = "completed" # Get output text safely output_text = "" if output: if hasattr(output, 'raw'): output_text = output.raw elif isinstance(output, str): output_text = output else: output_text = str(output) status.add_interaction( agent=task.agent.role if hasattr(task, 'agent') and hasattr(task.agent, 'role') else "unknown", action="complete_task", status="success", details=f"Task completed successfully. Output: {output_text[:200]}..." if output_text else "No output" ) logger.info(f"Task completed for subtask {subtask_id}") # Store the task output for reference if hasattr(task, 'description'): self.results[task.description] = output def _extract_subtask_id(self, task): """ Extract the subtask ID from a task description or context Args: task: The task to extract from Returns: The extracted subtask ID or None """ # This is a simplistic implementation - in a real system, # you'd have a more robust way to associate tasks with subtask IDs try: # Try to find subtask_id in the task description if hasattr(task, 'description') and '"subtask_id":' in task.description: start_pos = task.description.find('"subtask_id":') + len('"subtask_id":') end_pos = task.description.find(',', start_pos) if end_pos == -1: end_pos = task.description.find('}', start_pos) if end_pos != -1: subtask_id_str = task.description[start_pos:end_pos].strip(' "\'') return subtask_id_str # If we have context tasks, check their outputs if hasattr(task, 'context') and task.context: for context_task in task.context: if context_task.output and context_task.output.raw: try: context_data = json.loads(context_task.output.raw) if isinstance(context_data, list): for item in context_data: if 'subtask_id' in item: return item['subtask_id'] elif isinstance(context_data, dict) and 'subtask_id' in context_data: return context_data['subtask_id'] except: pass except Exception as e: logger.error(f"Error extracting subtask ID: {str(e)}") return None def handle_task_failure(self, subtask_id: str, error: str, suggestion_callback: Callable = None): """ Handle a task failure with retry logic Args: subtask_id: The ID of the failed subtask error: The error message suggestion_callback: Optional callback to get suggestions for retrying Returns: Whether to retry the task """ if subtask_id not in self.execution_tracker: logger.warning(f"No tracking information for subtask {subtask_id}") return False status = self.execution_tracker[subtask_id] status.last_error = error status.attempts += 1 status.status = "failed" if status.attempts >= status.max_attempts else "retry" status.add_interaction( agent="system", action="handle_failure", status="failure", details=f"Error: {error}" ) # Log the failure logger.error(f"Task failed for subtask {subtask_id} (Attempt {status.attempts}/{status.max_attempts}): {error}") # If we've reached max attempts, mark as failed if status.attempts >= status.max_attempts: logger.warning(f"Maximum retry attempts ({status.max_attempts}) reached for subtask {subtask_id}") if self.interactive: # Ask user if they want to continue retrying user_input = input(f"Maximum retry attempts reached for subtask {subtask_id}. Would you like to continue retrying? (y/n): ") if user_input.lower() == 'y': # Reset attempts and get user suggestions if needed status.attempts = 0 if suggestion_callback: suggestion = suggestion_callback() if suggestion: status.add_interaction( agent="user", action="provide_suggestion", status="retry", details=f"User suggestion: {suggestion}" ) return True else: return False else: return False else: # We haven't reached max attempts, so retry return True def run_crew(user_input: UserInput, interactive: bool = True): """ Sets up and runs the CrewAI crew to process the user's tasks with error handling Args: user_input: Structured user input interactive: Whether to prompt the user for input on failures Returns: The crew's output and execution tracking information """ # Create and run the task execution manager execution_manager = TaskExecutionManager(user_input, interactive) # Run the crew with error handling and tracking result = execution_manager.run() return result def execute_integration_plan(integration_plan, execution_tracking=None, interactive=True): """ Executes the integration plan generated by the crew with robust error handling Args: integration_plan: The integration plan JSON execution_tracking: Execution tracking information interactive: Whether to prompt the user for input on failures Returns: Result of the execution """ try: # Parse the integration plan plan = json.loads(integration_plan) if isinstance(integration_plan, str) else integration_plan # Get components and their execution order components = plan.get("components", []) components.sort(key=lambda c: c.get("execution_order", 0)) results = {} # Execute components in order for component in components: component_id = component.get("component_id") component_type = component.get("component_type") logger.info(f"Executing component: {component_id} (Type: {component_type})") # Execute based on component type if component_type == "playwright_script": # Extract script content and determine if it should run in headless mode script_content = component.get("script_content", "") # Check if user wants headed mode headless = True if interactive: user_input = input(f"Run Playwright script for component {component_id} in headless mode? (y/n, default: y): ") headless = user_input.lower() != "n" # Execute Playwright script with headless option playwright_tool = PlaywrightTool() component_result = playwright_tool._run( script=script_content, urls=component.get("urls", []), headless=headless ) # Log the script and its result logger.info(f"Executed Playwright script for component {component_id}") logger.info(f"Result: {component_result}") # Check if all dependencies are satisfied dependencies = component.get("input_dependencies", []) missing_deps = [dep for dep in dependencies if dep not in results] if missing_deps: error_msg = f"Cannot execute component {component_id}: Missing dependencies: {', '.join(missing_deps)}" logger.error(error_msg) if interactive: user_input = input(f"{error_msg}\nWould you like to skip this component and continue? (y/n): ") if user_input.lower() != 'y': raise ValueError(error_msg) else: logger.warning(f"Skipping component {component_id} due to missing dependencies") continue else: raise ValueError(error_msg) # Execute with retry logic max_retries = 3 retries = 0 while retries < max_retries: try: # Get inputs from dependencies inputs = {dep: results[dep] for dep in dependencies if dep in results} # Execute based on component type if component_type == "playwright_script": # For demonstration, we'll simulate successful execution component_result = f"Successfully executed Playwright script for component {component_id}" elif component_type == "crewai_component": # For demonstration, we'll simulate successful execution component_result = f"Successfully executed CrewAI component {component_id}" else: component_result = f"Unknown component type: {component_type}" # Store the result results[component_id] = component_result logger.info(f"Component {component_id} executed successfully") break except Exception as e: retries += 1 error_msg = f"Error executing component {component_id} (Attempt {retries}/{max_retries}): {str(e)}" logger.error(error_msg) if retries >= max_retries: if interactive: user_input = input(f"{error_msg}\nWould you like to retry this component? (y/n): ") if user_input.lower() == 'y': suggestion = input("Please provide any suggestions for the retry approach: ") logger.info(f"Retrying component {component_id} with user suggestion: {suggestion}") retries = 0 # Reset retries else: logger.warning(f"Failed to execute component {component_id} after {max_retries} attempts") break else: logger.warning(f"Failed to execute component {component_id} after {max_retries} attempts") break # Wait before retrying time.sleep(2) # Summarize results successful = sum(1 for comp in components if comp.get("component_id") in results) total = len(components) return { "status": "complete", "summary": f"Executed {successful}/{total} components successfully", "component_results": results, "execution_tracking": execution_tracking } except Exception as e: logger.error(f"Error executing integration plan: {str(e)}") return { "status": "failed", "error": str(e), "execution_tracking": execution_tracking } def main(user_input_data: Dict[str, Any], interactive: bool = True, max_execution_time: int = 3600): """ Main function to process user input and execute tasks with robust error handling Args: user_input_data: Raw user input data interactive: Whether to prompt the user for input on failures max_execution_time: Maximum execution time in seconds (default: 1 hour) Returns: Result of the execution """ execution_start = datetime.now() logger.info(f"Starting execution at {execution_start}") try: # Process user input user_input = process_user_input(user_input_data) # Extract headless mode preference headless = user_input.parameters.other.get("headless", True) if user_input.parameters.other else True logger.info(f"Playwright automation will run in {'headless' if headless else 'headed'} mode") logger.info(f"Processed user input: {len(user_input.tasks)} tasks, {len(user_input.parameters.dict(exclude_none=True))} parameters") # Run the crew to analyze and design the solution with error handling and timeout execution_manager = TaskExecutionManager( user_input, interactive=interactive, max_execution_time=max_execution_time ) crew_result = execution_manager.run() logger.info("CrewAI analysis and design completed successfully") # In a real implementation, parse the crew's output to get the integration plan try: # Extract the integration plan from the results integration_plan = None final_results = None execution_tracking = crew_result.get("execution_tracking", {}) if isinstance(crew_result, dict) and "result" in crew_result: raw_result = None # Handle different types of results if hasattr(crew_result["result"], "raw"): raw_result = crew_result["result"].raw elif isinstance(crew_result["result"], str): raw_result = crew_result["result"] elif isinstance(crew_result["result"], dict): raw_result = json.dumps(crew_result["result"]) # If it's already a dictionary, it might be the final output final_results = crew_result["result"] else: raw_result = str(crew_result["result"]) # raw_result = crew_result["result"].raw if hasattr(crew_result["result"], "raw") else str(crew_result["result"]) # Try to find and parse the integration plan try: # This is a simplistic approach - in a real system you would have a more robust way to extract the plan if "integration_plan" in raw_result or "integration plan" in raw_result.lower(): start_index = raw_result.find('{') if start_index != -1: # Try to parse the JSON part of the output integration_plan = json.loads(raw_result[start_index:]) logger.info("Successfully extracted integration plan from crew result") except json.JSONDecodeError: logger.warning("Failed to parse integration plan as JSON") integration_plan = raw_result # Prepare the final output with clear sections execution_time = datetime.now() - execution_start final_output = { "status": "completed", "execution_time": str(execution_time).split('.')[0], "summary": crew_result.get("execution_summary", {}), "final_results": final_results, "analysis_result": crew_result, } # Execute the integration plan if found if integration_plan: logger.info("Executing integration plan") execution_result = execute_integration_plan( integration_plan, execution_tracking=execution_tracking, interactive=interactive ) final_output["execution_result"] = execution_result # Update final status based on execution result if execution_result.get("status") == "failed": final_output["status"] = "partial" # Add a clearly formatted final output section formatted_output = "\n" + "="*50 + "\n" formatted_output += "EXECUTION COMPLETED\n" formatted_output += "="*50 + "\n" formatted_output += f"Status: {final_output['status']}\n" formatted_output += f"Total execution time: {final_output['execution_time']}\n" formatted_output += f"Tasks completed: {final_output['summary'].get('tasks_completed', 0)}/{final_output['summary'].get('tasks_total', 0)}\n" # Add final results if available if final_results: formatted_output += "\n--- FINAL RESULTS ---\n" if isinstance(final_results, dict): for key, value in final_results.items(): formatted_output += f"{key}: {value}\n" else: formatted_output += str(final_results) + "\n" logger.info(formatted_output) if interactive: print(formatted_output) return final_output except Exception as e: logger.error(f"Error executing integration plan: {str(e)}") if interactive: user_input = input(f"Error executing integration plan: {str(e)}\nWould you like to return the analysis result anyway? (y/n): ") if user_input.lower() == 'y': return { "analysis_result": crew_result, "execution_error": str(e), "execution_time": str(datetime.now() - execution_start) } raise except KeyboardInterrupt: logger.info("Execution stopped by user") return { "status": "stopped", "execution_time": str(datetime.now() - execution_start).split('.')[0], "message": "Execution was stopped by the user" } except Exception as e: logger.error(f"Error in execution: {str(e)}") return { "status": "failed", "error": str(e), "execution_time": str(datetime.now() - execution_start) } # Interactive prompt for user suggestions def get_user_suggestion(): """Get a suggestion from the user for retry approach""" print("\n" + "="*50) print("Execution has encountered issues. Please provide suggestions:") print("1. Try a different approach") print("2. Skip this subtask") print("3. Provide additional information") print("4. Abort execution") print("="*50) choice = input("Enter your choice (1-4): ") if choice == "1": return input("Please describe an alternative approach: ") elif choice == "2": return "skip" elif choice == "3": return input("Please provide additional information: ") elif choice == "4": raise KeyboardInterrupt("Execution aborted by user") else: return input("Please provide your suggestion: ") # Command-line interface def cli(): """Command-line interface for the CrewAI-Playwright wrapper""" import argparse import json parser = argparse.ArgumentParser(description="CrewAI-Playwright Automation Wrapper") parser.add_argument("--input", "-i", type=str, help="Path to JSON input file") parser.add_argument("--output", "-o", type=str, help="Path to output file") parser.add_argument("--non-interactive", "-n", action="store_true", help="Run in non-interactive mode") parser.add_argument("--log-level", "-l", type=str, choices=["DEBUG", "INFO", "WARNING", "ERROR"], default="INFO", help="Logging level") parser.add_argument("--show-tracking", "-t", action="store_true", help="Show detailed execution tracking") parser.add_argument("--visualize", "-v", action="store_true", help="Generate flow visualizations") parser.add_argument("--max-retries", "-r", type=int, default=3, help="Maximum retry attempts per task") parser.add_argument("--timeout", "-to", type=int, default=3600, help="Maximum execution time in seconds (default: 1 hour)") parser.add_argument("--headed", action="store_true", help="Run Playwright in headed mode instead of headless") args = parser.parse_args() # Set log level logging.getLogger().setLevel(getattr(logging, args.log_level)) # Pass headless mode preference to the execution headless = not args.headed # Load input data if args.input: try: with open(args.input, 'r') as f: user_input_data = json.load(f) except Exception as e: logger.error(f"Error loading input file: {str(e)}") return else: # Example user input if no file is provided user_input_data = { "task1": """ Research competitors in the AI agent space. Collect pricing information from their websites. Also monitor social media for recent announcements and write a blog for the competitors data. Please refer to these urls to explore the competitors: """, "task2": "Check if there is a file named knowledgebase.pdf in the directories : ", "parameters": { "websites": ["https://crew.ai", "https://minions.com/"], "Content": ["/root/usr/test/material/"] } } # Update user_input_data to include headless option if "parameters" not in user_input_data: user_input_data["parameters"] = {} user_input_data["parameters"]["headless"] = headless # Run the main function with timeout from concurrent.futures import ThreadPoolExecutor, TimeoutError interactive = not args.non_interactive # Set maximum retries for all execution statuses for task_id in user_input_data: if task_id.startswith("task"): if "max_retries" not in user_input_data: user_input_data["max_retries"] = args.max_retries # Run with timeout handling with ThreadPoolExecutor() as executor: future = executor.submit(main, user_input_data, interactive, args.timeout) try: result = future.result(timeout=args.timeout + 60) except TimeoutError: print(f"Execution timed out after {args.timeout} seconds") result = {"status": "timeout", "error": f"Execution timed out after {args.timeout} seconds"} # Generate visualizations if requested if args.visualize and isinstance(result, dict): try: from visualization_utility import visualize_execution vis_paths = visualize_execution(result) result["visualizations"] = vis_paths print(f"Generated visualizations: {', '.join(vis_paths.keys())}") except Exception as e: print(f"Error generating visualizations: {str(e)}") # Show detailed tracking if requested if args.show_tracking and "analysis_result" in result and "execution_tracking" in result["analysis_result"]: print("\n" + "="*50) print("EXECUTION TRACKING:") print("="*50) for subtask_id, status in result["analysis_result"]["execution_tracking"].items(): print(f"\nSubtask: {subtask_id}") print(f"Status: {status['status']}") print(f"Attempts: {status['attempts']}") print("Interactions:") for interaction in status["interactions"]: print(f" {interaction['timestamp']} - {interaction['agent']} - {interaction['action']} - {interaction['status']}") print(f" {interaction['details'][:100]}..." if len(interaction["details"]) > 100 else f" {interaction['details']}") # Save or print output if args.output: output_path = args.output if os.path.isdir(output_path): output_path = os.path.join(output_path, "execution_result.json") try: with open(output_path, 'w') as f: json.dump(result, f, indent=2) print(f"Results saved to {output_path}") except Exception as e: logger.error(f"Error saving output file: {str(e)}") print(json.dumps(result, indent=2)) else: # Print a simplified version of the result if isinstance(result, dict): if "analysis_result" in result and "execution_result" in result: print("\n" + "="*50) print("EXECUTION SUMMARY:") print("="*50) print(f"Execution Time: {result.get('execution_time', 'Unknown')}") print(f"Analysis Status: {'Success' if 'result' in result['analysis_result'] else 'Failed'}") print(f"Execution Status: {result['execution_result'].get('status', 'Unknown')}") print(f"Summary: {result['execution_result'].get('summary', 'No summary available')}") if "component_results" in result['execution_result']: print("\nComponent Results:") for comp_id, comp_result in result['execution_result']['component_results'].items(): print(f" {comp_id}: {comp_result[:100]}..." if len(comp_result) > 100 else f" {comp_id}: {comp_result}") else: print(json.dumps(result, indent=2)) else: print(result) # Example usage if __name__ == "__main__": cli()