Add kickoff_for_each_parallel method using ThreadPoolExecutor

[devin-ai-integration]

Fixes #2406 - Adds a new method to run a crew multiple times in parallel on different inputs using ThreadPoolExecutor. This allows for better performance when running the same crew on many inputs.

Link to Devin run: https://app.devin.ai/sessions/3184eb8f13bf4af58cd59a2dc3133f63
Requested by: Joe Moura (joao@crewai.com)

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add "(aside)" to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment and CI monitoring

[joaomdmoura]

Disclaimer: This review was made by a crew of AI Agents.

Code Review Comment for PR #2407

Overview

This PR introduces a new parallel execution method, kickoff_for_each_parallel, within the Crew class, leveraging ThreadPoolExecutor for improved task processing efficiency. The implementation is accompanied by thorough test coverage, reflecting best practices in both design and testing.

Implementation Findings

Strengths:

• Effective Use of Threads: The use of ThreadPoolExecutor for parallel processing is well-considered, notably improving the class's handling of concurrent tasks.
• Documentation and Type Hints: The documentation accompanying the method is clear and includes appropriate type hints, enhancing code readability and maintainability.
• Consistent Error Handling: The error handling strategy maintains consistency with the existing codebase.

Areas for Improvement:

1. Import Organization:

   • Current:

     import concurrent.futures
     from concurrent.futures import ThreadPoolExecutor

   • Recommendation: Consolidate imports to remove redundancy:

     from concurrent.futures import ThreadPoolExecutor

2. Enhance Type Hints:

   • Current:

     def kickoff_for_each_parallel(self, inputs: List[Dict[str, Any]], max_workers: Optional[int] = None) -> List[CrewOutput]:

   • Recommendation: Use Sequence for more flexibility:

     from typing import Sequence
     def kickoff_for_each_parallel(self, inputs: Sequence[Dict[str, Any]], max_workers: Optional[int] = None) -> List[CrewOutput]:

3. Improve Error Messages:

   • Current:

     raise TypeError("Inputs must be a list of dictionaries.")

   • Recommendation: Provide clearer information:

     raise TypeError("Inputs must be a list of dictionaries. Received {type(inputs).__name__} instead.")

4. Memory Management:

   • Consider adding a cleanup section to assist garbage collection:

     finally:
         crew_copies.clear()

Testing Findings

Strengths:

• Comprehensive Coverage: The tests are thorough, covering a variety of scenarios, including edge cases and error handling.
• Use of Mocking: Appropriate use of mocking effectively isolates the tests from external dependencies.

Areas for Improvement:

1. Introduce Helper Functions: To reduce duplication in the test cases, create a fixture:

   @pytest.fixture
   def sample_crew():
       # Returns a predefined Crew instance

2. Expand Async Test Coverage: Add async tests to cover asynchronous scenarios:

   @pytest.mark.asyncio
   async def test_kickoff_for_each_parallel_with_async_tasks():
       # Test implementation for async tasks

3. Add Edge Case Tests: Such as testing large inputs for performance:

   def test_kickoff_for_each_parallel_with_large_input():
       # Implementation for testing large input sets

Additional Recommendations

• Performance Metrics: Incorporate execution time logging to monitor performance:

  from time import perf_counter
  start_time = perf_counter()
  # Rest of the method
  logger.debug(f"Parallel execution completed in {execution_time:.2f} seconds")

• Progress Callback: Introduce an optional parameter for progress reporting:

  def kickoff_for_each_parallel(self, inputs: List[Dict[str, Any]], progress_callback: Optional[Callable[[int, int], None]] = None):
      # Existing method code

Conclusion

The PR demonstrates a strong understanding of parallel processing and is well-structured. After addressing the suggested improvements, especially regarding import optimization and type hint enhancements, the implementation will be ready to merge. The related testing framework is robust, but implementing the additional recommendations will further enhance the overall quality and usability of the code.