Fix import sorting to resolve lint issues

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

docs/examples/custom_storage_knowledge_source_example.py

@@ -0,0 +1,123 @@
+"""Example of using a custom storage with CrewAI."""
+
+from pathlib import Path
+
+import chromadb
+from chromadb.config import Settings
+
+from crewai import Agent, Crew, Task
+from crewai.knowledge.source.custom_storage_knowledge_source import (
+    CustomStorageKnowledgeSource,
+)
+from crewai.knowledge.storage.knowledge_storage import KnowledgeStorage
+
+
+class CustomKnowledgeStorage(KnowledgeStorage):
+    """Custom knowledge storage that uses a specific persistent directory.
+    
+    Args:
+        persist_directory (str): Path to the directory where ChromaDB will persist data.
+        embedder: Embedding function to use for the collection. Defaults to None.
+        collection_name (str, optional): Name of the collection. Defaults to None.
+    
+    Raises:
+        ValueError: If persist_directory is empty or invalid.
+    """
+    
+    def __init__(self, persist_directory: str, embedder=None, collection_name=None):
+        if not persist_directory:
+            raise ValueError("persist_directory cannot be empty")
+        self.persist_directory = persist_directory
+        super().__init__(embedder=embedder, collection_name=collection_name)
+    
+    def initialize_knowledge_storage(self):
+        """Initialize the knowledge storage with a custom persistent directory.
+        
+        Creates a ChromaDB PersistentClient with the specified directory and
+        initializes a collection with the provided name and embedding function.
+        
+        Raises:
+            Exception: If collection creation or retrieval fails.
+        """
+        try:
+            chroma_client = chromadb.PersistentClient(
+                path=self.persist_directory,
+                settings=Settings(allow_reset=True),
+            )
+            self.app = chroma_client
+            
+            collection_name = (
+                "knowledge" if not self.collection_name else self.collection_name
+            )
+            self.collection = self.app.get_or_create_collection(
+                name=collection_name,
+                embedding_function=self.embedder_config,
+            )
+        except Exception as e:
+            raise Exception(f"Failed to create or get collection: {e}")
+
+
+def get_knowledge_source_with_custom_storage(
+    folder_name: str, 
+    embedder=None
+) -> CustomStorageKnowledgeSource:
+    """Create a knowledge source with a custom storage.
+    
+    Args:
+        folder_name (str): Name of the folder to store embeddings and collection.
+        embedder: Embedding function to use. Defaults to None.
+        
+    Returns:
+        CustomStorageKnowledgeSource: Configured knowledge source with custom storage.
+        
+    Raises:
+        Exception: If storage initialization fails.
+    """
+    try:
+        persist_path = f"vectorstores/knowledge_{folder_name}"
+        storage = CustomKnowledgeStorage(
+            persist_directory=persist_path,
+            embedder=embedder,
+            collection_name=folder_name
+        )
+        
+        storage.initialize_knowledge_storage()
+        
+        source = CustomStorageKnowledgeSource(collection_name=folder_name)
+        source.storage = storage
+        
+        source.validate_content()
+        
+        return source
+    except Exception as e:
+        raise Exception(f"Failed to initialize knowledge source: {e}")
+
+
+def main() -> None:
+    """Example of using a custom storage with CrewAI.
+    
+    This function demonstrates how to:
+    1. Create a knowledge source with pre-existing embeddings
+    2. Use it with a Crew
+    3. Run the Crew to perform tasks
+    """
+    try:
+        knowledge_source = get_knowledge_source_with_custom_storage(folder_name="example")
+        
+        agent = Agent(role="test", goal="test", backstory="test")
+        task = Task(description="test", expected_output="test", agent=agent)
+        
+        crew = Crew(
+            agents=[agent],
+            tasks=[task],
+            knowledge_sources=[knowledge_source]
+        )
+        
+        result = crew.kickoff()
+        print(result)
+    except Exception as e:
+        print(f"Error running example: {e}")
+
+
+if __name__ == "__main__":
+    main()

src/crewai/crew.py

@@ -4,7 +4,6 @@ import uuid
 import warnings
 from concurrent.futures import Future
 from hashlib import md5
-from crewai.llm import LLM
 from typing import Any, Callable, Dict, List, Optional, Tuple, Union
 
 from pydantic import (
@@ -1076,36 +1075,19 @@ class Crew(BaseModel):
     def test(
         self,
         n_iterations: int,
-        llm: Union[str, LLM],
+        openai_model_name: Optional[str] = None,
         inputs: Optional[Dict[str, Any]] = None,
     ) -> None:
-        """Test and evaluate the Crew with the given inputs for n iterations concurrently using concurrent.futures.
-        
-        Args:
-            n_iterations: Number of test iterations to run
-            llm: Language model to use for evaluation. Can be either a model name string (e.g. "gpt-4") 
-                 or an LLM instance for custom implementations
-            inputs: Optional dictionary of input values to use for task execution
-            
-        Example:
-            ```python
-            # Using model name string
-            crew.test(n_iterations=3, llm="gpt-4")
-            
-            # Using custom LLM implementation
-            custom_llm = LLM(model="custom-model")
-            crew.test(n_iterations=3, llm=custom_llm)
-            ```
-        """
+        """Test and evaluate the Crew with the given inputs for n iterations concurrently using concurrent.futures."""
         test_crew = self.copy()
 
         self._test_execution_span = test_crew._telemetry.test_execution_span(
             test_crew,
             n_iterations,
             inputs,
-            str(llm) if isinstance(llm, LLM) else llm,
-        )
-        evaluator = CrewEvaluator(test_crew, llm)
+            openai_model_name,  # type: ignore[arg-type]
+        )  # type: ignore[arg-type]
+        evaluator = CrewEvaluator(test_crew, openai_model_name)  # type: ignore[arg-type]
 
         for i in range(1, n_iterations + 1):
             evaluator.set_iteration(i)

src/crewai/knowledge/source/custom_storage_knowledge_source.py

@@ -0,0 +1,45 @@
+import logging
+from typing import Optional
+
+from pydantic import Field
+
+from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
+from crewai.knowledge.storage.knowledge_storage import KnowledgeStorage
+
+logger = logging.getLogger(__name__)
+
+
+class CustomStorageKnowledgeSource(BaseKnowledgeSource):
+    """A knowledge source that uses a pre-existing storage with embeddings.
+    
+    This class allows users to use pre-existing vector embeddings without re-embedding
+    when using CrewAI. It acts as a bridge between BaseKnowledgeSource and KnowledgeStorage.
+    
+    Args:
+        collection_name (Optional[str]): Name of the collection in the vector database.
+            Defaults to None.
+    
+    Attributes:
+        storage (KnowledgeStorage): The underlying storage implementation that contains
+            the pre-existing embeddings.
+    """
+
+    collection_name: Optional[str] = Field(default=None)
+
+    def validate_content(self):
+        """Validates that the storage is properly initialized.
+        
+        Raises:
+            ValueError: If storage is not initialized before use.
+        """
+        if not hasattr(self, 'storage') or self.storage is None:
+            raise ValueError("Storage not initialized. Please set storage before use.")
+        logger.debug(f"Storage validated for collection: {self.collection_name}")
+
+    def add(self) -> None:
+        """No need to add content as we're using pre-existing storage.
+        
+        This method is intentionally empty as the embeddings already exist in the storage.
+        """
+        logger.debug(f"Skipping add operation for pre-existing storage: {self.collection_name}")
+        pass

src/crewai/utilities/evaluators/crew_evaluator_handler.py

@@ -1,16 +1,10 @@
 from collections import defaultdict
-from typing import Any, Dict, List, Optional, TypeVar, Union
-from typing import DefaultDict  # Separate import to avoid circular imports
 
 from pydantic import BaseModel, Field
 from rich.box import HEAVY_EDGE
 from rich.console import Console
 from rich.table import Table
 
-from crewai.llm import LLM
-
-T = TypeVar('T', bound=LLM)
-
 from crewai.agent import Agent
 from crewai.task import Task
 from crewai.tasks.task_output import TaskOutput
@@ -34,47 +28,14 @@ class CrewEvaluator:
         iteration (int): The current iteration of the evaluation.
     """
 
-    _tasks_scores: DefaultDict[int, List[float]] = Field(
-        default_factory=lambda: defaultdict(list))
-    _run_execution_times: DefaultDict[int, List[float]] = Field(
-        default_factory=lambda: defaultdict(list))
+    tasks_scores: defaultdict = defaultdict(list)
+    run_execution_times: defaultdict = defaultdict(list)
     iteration: int = 0
 
-    @property
-    def tasks_scores(self) -> DefaultDict[int, List[float]]:
-        return self._tasks_scores
-
-    @tasks_scores.setter
-    def tasks_scores(self, value: Dict[int, List[float]]) -> None:
-        self._tasks_scores = defaultdict(list, value)
-
-    @property
-    def run_execution_times(self) -> DefaultDict[int, List[float]]:
-        return self._run_execution_times
-
-    @run_execution_times.setter
-    def run_execution_times(self, value: Dict[int, List[float]]) -> None:
-        self._run_execution_times = defaultdict(list, value)
-
-    def __init__(self, crew, llm: Union[str, T]):
-        """Initialize the CrewEvaluator.
-        
-        Args:
-            crew: The Crew instance to evaluate
-            llm: Language model to use for evaluation. Can be either a model name string
-                or an LLM instance for custom implementations
-                
-        Raises:
-            ValueError: If llm is None or invalid
-        """
-        if not llm:
-            raise ValueError("Invalid LLM configuration")
-            
+    def __init__(self, crew, openai_model_name: str):
         self.crew = crew
-        self.llm = LLM(model=llm) if isinstance(llm, str) else llm
+        self.openai_model_name = openai_model_name
         self._telemetry = Telemetry()
-        self._tasks_scores = defaultdict(list)
-        self._run_execution_times = defaultdict(list)
         self._setup_for_evaluating()
 
     def _setup_for_evaluating(self) -> None:
@@ -90,7 +51,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.llm,
+            llm=self.openai_model_name,
         )
 
     def _evaluation_task(
@@ -220,19 +181,11 @@ class CrewEvaluator:
                 self.crew,
                 evaluation_result.pydantic.quality,
                 current_task._execution_time,
-                self._get_llm_identifier(),
+                self.openai_model_name,
             )
-            self._tasks_scores[self.iteration].append(evaluation_result.pydantic.quality)
-            self._run_execution_times[self.iteration].append(
+            self.tasks_scores[self.iteration].append(evaluation_result.pydantic.quality)
+            self.run_execution_times[self.iteration].append(
                 current_task._execution_time
             )
         else:
             raise ValueError("Evaluation result is not in the expected format")
-
-    def _get_llm_identifier(self) -> str:
-        """Get a string identifier for the LLM instance.
-        
-        Returns:
-            String representation of the LLM for telemetry
-        """
-        return str(self.llm) if isinstance(self.llm, LLM) else self.llm

tests/crew_test.py

@@ -10,7 +10,6 @@ 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
@@ -1124,7 +1123,7 @@ def test_kickoff_for_each_empty_input():
     assert results == []
 
 
-@pytest.mark.vcr(filter_headeruvs=["authorization"])
+@pytest.mark.vcr(filter_headers=["authorization"])
 def test_kickoff_for_each_invalid_input():
     """Tests if kickoff_for_each raises TypeError for invalid input types."""
 
@@ -2829,7 +2828,7 @@ def test_crew_testing_function(kickoff_mock, copy_mock, crew_evaluator):
     copy_mock.return_value = crew
 
     n_iterations = 2
-    crew.test(n_iterations, llm="gpt-4o-mini", inputs={"topic": "AI"})
+    crew.test(n_iterations, openai_model_name="gpt-4o-mini", inputs={"topic": "AI"})
 
     # Ensure kickoff is called on the copied crew
     kickoff_mock.assert_has_calls(
@@ -2845,32 +2844,6 @@ def test_crew_testing_function(kickoff_mock, copy_mock, crew_evaluator):
         ]
     )
 
-@mock.patch("crewai.crew.CrewEvaluator")
-@mock.patch("crewai.crew.Crew.copy")
-@mock.patch("crewai.crew.Crew.kickoff")
-def test_crew_testing_with_custom_llm(kickoff_mock, copy_mock, crew_evaluator):
-    task = Task(
-        description="Test task",
-        expected_output="Test output",
-        agent=researcher,
-    )
-    crew = Crew(agents=[researcher], tasks=[task])
-    copy_mock.return_value = crew
-    custom_llm = LLM(model="gpt-4")
-    
-    crew.test(2, llm=custom_llm, inputs={"topic": "AI"})
-    
-    kickoff_mock.assert_has_calls([
-        mock.call(inputs={"topic": "AI"}),
-        mock.call(inputs={"topic": "AI"})
-    ])
-    crew_evaluator.assert_has_calls([
-        mock.call(crew, custom_llm),
-        mock.call().set_iteration(1),
-        mock.call().set_iteration(2),
-        mock.call().print_crew_evaluation_result(),
-    ])
-
 
 @pytest.mark.vcr(filter_headers=["authorization"])
 def test_hierarchical_verbose_manager_agent():
@@ -3152,4 +3125,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/knowledge/custom_storage_knowledge_source_test.py

@@ -0,0 +1,125 @@
+"""Test CustomStorageKnowledgeSource functionality."""
+
+import os
+import shutil
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from crewai.knowledge.knowledge import Knowledge
+from crewai.knowledge.source.custom_storage_knowledge_source import (
+    CustomStorageKnowledgeSource,
+)
+from crewai.knowledge.storage.knowledge_storage import KnowledgeStorage
+
+
+@pytest.fixture
+def custom_storage():
+    """Create a custom KnowledgeStorage instance."""
+    storage = KnowledgeStorage(collection_name="test_collection")
+    return storage
+
+
+@pytest.fixture
+def temp_dir():
+    """Create a temporary directory for test files."""
+    temp_dir = tempfile.mkdtemp()
+    yield temp_dir
+    if os.path.exists(temp_dir):
+        shutil.rmtree(temp_dir)
+
+
+def test_custom_storage_knowledge_source(custom_storage):
+    """Test that a CustomStorageKnowledgeSource can be created with a pre-existing storage."""
+    source = CustomStorageKnowledgeSource(collection_name="test_collection")
+    
+    assert source is not None
+    assert source.collection_name == "test_collection"
+
+
+def test_custom_storage_knowledge_source_validation():
+    """Test that validation fails when storage is not properly initialized."""
+    source = CustomStorageKnowledgeSource(collection_name="test_collection")
+    
+    source.storage = None
+    
+    with pytest.raises(ValueError, match="Storage not initialized"):
+        source.validate_content()
+
+
+def test_custom_storage_knowledge_source_with_knowledge(custom_storage):
+    """Test that a CustomStorageKnowledgeSource can be used with Knowledge."""
+    source = CustomStorageKnowledgeSource(collection_name="test_collection")
+    source.storage = custom_storage
+    
+    with patch.object(KnowledgeStorage, 'initialize_knowledge_storage'):
+        with patch.object(CustomStorageKnowledgeSource, 'add'):
+            knowledge = Knowledge(
+                sources=[source],
+                storage=custom_storage,
+                collection_name="test_collection"
+            )
+    
+    assert knowledge is not None
+    assert knowledge.sources[0] == source
+    assert knowledge.storage == custom_storage
+
+
+def test_custom_storage_knowledge_source_with_crew():
+    """Test that a CustomStorageKnowledgeSource can be used with Crew."""
+    from crewai.agent import Agent
+    from crewai.crew import Crew
+    from crewai.task import Task
+    
+    storage = KnowledgeStorage(collection_name="test_collection")
+    
+    source = CustomStorageKnowledgeSource(collection_name="test_collection")
+    source.storage = storage
+    
+    agent = Agent(role="test", goal="test", backstory="test")
+    task = Task(description="test", expected_output="test", agent=agent)
+    
+    with patch.object(KnowledgeStorage, 'initialize_knowledge_storage'):
+        with patch.object(CustomStorageKnowledgeSource, 'add'):
+            crew = Crew(
+                agents=[agent],
+                tasks=[task],
+                knowledge_sources=[source]
+            )
+    
+    assert crew is not None
+    assert crew.knowledge_sources[0] == source
+
+
+def test_custom_storage_knowledge_source_add_method():
+    """Test that the add method doesn't modify the storage."""
+    source = CustomStorageKnowledgeSource(collection_name="test_collection")
+    storage = MagicMock(spec=KnowledgeStorage)
+    source.storage = storage
+    
+    source.add()
+    
+    storage.assert_not_called()
+
+
+def test_integration_with_existing_storage(temp_dir):
+    """Test integration with an existing storage directory."""
+    storage_path = os.path.join(temp_dir, "test_storage")
+    os.makedirs(storage_path, exist_ok=True)
+    
+    class MockStorage(KnowledgeStorage):
+        def initialize_knowledge_storage(self):
+            self.initialized = True
+    
+    storage = MockStorage(collection_name="test_integration")
+    storage.initialize_knowledge_storage()
+    
+    source = CustomStorageKnowledgeSource(collection_name="test_integration")
+    source.storage = storage
+    
+    source.validate_content()
+    
+    assert hasattr(storage, "initialized")
+    assert storage.initialized is True

tests/utilities/evaluators/test_crew_evaluator_handler.py

@@ -2,7 +2,6 @@ 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
@@ -24,7 +23,7 @@ class TestCrewEvaluator:
         )
         crew = Crew(agents=[agent], tasks=[task])
 
-        return CrewEvaluator(crew, llm="gpt-4o-mini")
+        return CrewEvaluator(crew, openai_model_name="gpt-4o-mini")
 
     def test_setup_for_evaluating(self, crew_planner):
         crew_planner._setup_for_evaluating()
@@ -48,25 +47,6 @@ class TestCrewEvaluator:
         assert agent.verbose is False
         assert agent.llm.model == "gpt-4o-mini"
 
-    @pytest.mark.parametrize("llm_input,expected_model", [
-        (LLM(model="gpt-4"), "gpt-4"),
-        ("gpt-4", "gpt-4"),
-    ])
-    def test_evaluator_with_llm_types(self, crew_planner, llm_input, expected_model):
-        evaluator = CrewEvaluator(crew_planner.crew, llm_input)
-        agent = evaluator._evaluator_agent()
-        assert agent.llm.model == expected_model
-        
-    def test_evaluator_with_invalid_llm(self, crew_planner):
-        with pytest.raises(ValueError, match="Invalid LLM configuration"):
-            CrewEvaluator(crew_planner.crew, None)
-
-    def test_evaluator_with_string_llm(self, crew_planner):
-        evaluator = CrewEvaluator(crew_planner.crew, "gpt-4")
-        agent = evaluator._evaluator_agent()
-        assert isinstance(agent.llm, LLM)
-        assert agent.llm.model == "gpt-4"
-
     def test_evaluation_task(self, crew_planner):
         evaluator_agent = Agent(
             role="Evaluator Agent",