feat: async knowledge support (#4023)

* feat: add async support for tools, add async tool tests

* chore: improve tool decorator typing

* fix: ensure _run backward compat

* chore: update docs

* chore: make docstrings a little more readable

* feat: add async execution support to agent executor

* chore: add tests

* feat: add aiosqlite dep; regenerate lockfile

* feat: add async ops to memory feat; create tests

* feat: async knowledge support; add tests

* chore: regenerate lockfile

Author: greysonlalonde

lib/crewai/src/crewai/knowledge/knowledge.py

@@ -32,8 +32,8 @@ def __init__(
         sources: list[BaseKnowledgeSource],
         embedder: EmbedderConfig | None = None,
         storage: KnowledgeStorage | None = None,
-        **data,
-    ):
+        **data: object,
+    ) -> None:
         super().__init__(**data)
         if storage:
             self.storage = storage
@@ -75,3 +75,44 @@ def reset(self) -> None:
             self.storage.reset()
         else:
             raise ValueError("Storage is not initialized.")
+
+    async def aquery(
+        self, query: list[str], results_limit: int = 5, score_threshold: float = 0.6
+    ) -> list[SearchResult]:
+        """Query across all knowledge sources asynchronously.
+
+        Args:
+            query: List of query strings.
+            results_limit: Maximum number of results to return.
+            score_threshold: Minimum similarity score for results.
+
+        Returns:
+            The top results matching the query.
+
+        Raises:
+            ValueError: If storage is not initialized.
+        """
+        if self.storage is None:
+            raise ValueError("Storage is not initialized.")
+
+        return await self.storage.asearch(
+            query,
+            limit=results_limit,
+            score_threshold=score_threshold,
+        )
+
+    async def aadd_sources(self) -> None:
+        """Add all knowledge sources to storage asynchronously."""
+        try:
+            for source in self.sources:
+                source.storage = self.storage
+                await source.aadd()
+        except Exception as e:
+            raise e
+
+    async def areset(self) -> None:
+        """Reset the knowledge base asynchronously."""
+        if self.storage:
+            await self.storage.areset()
+        else:
+            raise ValueError("Storage is not initialized.")

lib/crewai/src/crewai/knowledge/source/base_file_knowledge_source.py

@@ -1,5 +1,6 @@
 from abc import ABC, abstractmethod
 from pathlib import Path
+from typing import Any
 
 from pydantic import Field, field_validator
 
@@ -25,7 +26,10 @@ class BaseFileKnowledgeSource(BaseKnowledgeSource, ABC):
     safe_file_paths: list[Path] = Field(default_factory=list)
 
     @field_validator("file_path", "file_paths", mode="before")
-    def validate_file_path(cls, v, info):  # noqa: N805
+    @classmethod
+    def validate_file_path(
+        cls, v: Path | list[Path] | str | list[str] | None, info: Any
+    ) -> Path | list[Path] | str | list[str] | None:
         """Validate that at least one of file_path or file_paths is provided."""
         # Single check if both are None, O(1) instead of nested conditions
         if (
@@ -38,7 +42,7 @@ def validate_file_path(cls, v, info):  # noqa: N805
             raise ValueError("Either file_path or file_paths must be provided")
         return v
 
-    def model_post_init(self, _):
+    def model_post_init(self, _: Any) -> None:
         """Post-initialization method to load content."""
         self.safe_file_paths = self._process_file_paths()
         self.validate_content()
@@ -48,7 +52,7 @@ def model_post_init(self, _):
     def load_content(self) -> dict[Path, str]:
         """Load and preprocess file content. Should be overridden by subclasses. Assume that the file path is relative to the project root in the knowledge directory."""
 
-    def validate_content(self):
+    def validate_content(self) -> None:
         """Validate the paths."""
         for path in self.safe_file_paths:
             if not path.exists():
@@ -65,13 +69,20 @@ def validate_content(self):
                     color="red",
                 )
 
-    def _save_documents(self):
+    def _save_documents(self) -> None:
         """Save the documents to the storage."""
         if self.storage:
             self.storage.save(self.chunks)
         else:
             raise ValueError("No storage found to save documents.")
 
+    async def _asave_documents(self) -> None:
+        """Save the documents to the storage asynchronously."""
+        if self.storage:
+            await self.storage.asave(self.chunks)
+        else:
+            raise ValueError("No storage found to save documents.")
+
     def convert_to_path(self, path: Path | str) -> Path:
         """Convert a path to a Path object."""
         return Path(KNOWLEDGE_DIRECTORY + "/" + path) if isinstance(path, str) else path

lib/crewai/src/crewai/knowledge/source/base_knowledge_source.py

@@ -39,12 +39,32 @@ def _chunk_text(self, text: str) -> list[str]:
             for i in range(0, len(text), self.chunk_size - self.chunk_overlap)
         ]
 
-    def _save_documents(self):
-        """
-        Save the documents to the storage.
+    def _save_documents(self) -> None:
+        """Save the documents to the storage.
+
         This method should be called after the chunks and embeddings are generated.
+
+        Raises:
+            ValueError: If no storage is configured.
         """
         if self.storage:
             self.storage.save(self.chunks)
         else:
             raise ValueError("No storage found to save documents.")
+
+    @abstractmethod
+    async def aadd(self) -> None:
+        """Process content, chunk it, compute embeddings, and save them asynchronously."""
+
+    async def _asave_documents(self) -> None:
+        """Save the documents to the storage asynchronously.
+
+        This method should be called after the chunks and embeddings are generated.
+
+        Raises:
+            ValueError: If no storage is configured.
+        """
+        if self.storage:
+            await self.storage.asave(self.chunks)
+        else:
+            raise ValueError("No storage found to save documents.")

lib/crewai/src/crewai/knowledge/source/crew_docling_source.py

@@ -2,27 +2,24 @@
 
 from collections.abc import Iterator
 from pathlib import Path
+from typing import TYPE_CHECKING, Any
 from urllib.parse import urlparse
 
 
 try:
-    from docling.datamodel.base_models import (  # type: ignore[import-not-found]
-        InputFormat,
-    )
-    from docling.document_converter import (  # type: ignore[import-not-found]
-        DocumentConverter,
-    )
-    from docling.exceptions import ConversionError  # type: ignore[import-not-found]
-    from docling_core.transforms.chunker.hierarchical_chunker import (  # type: ignore[import-not-found]
-        HierarchicalChunker,
-    )
-    from docling_core.types.doc.document import (  # type: ignore[import-not-found]
-        DoclingDocument,
-    )
+    from docling.datamodel.base_models import InputFormat
+    from docling.document_converter import DocumentConverter
+    from docling.exceptions import ConversionError
+    from docling_core.transforms.chunker.hierarchical_chunker import HierarchicalChunker
+    from docling_core.types.doc.document import DoclingDocument
 
     DOCLING_AVAILABLE = True
 except ImportError:
     DOCLING_AVAILABLE = False
+    # Provide type stubs for when docling is not available
+    if TYPE_CHECKING:
+        from docling.document_converter import DocumentConverter
+        from docling_core.types.doc.document import DoclingDocument
 
 from pydantic import Field
 
@@ -32,11 +29,13 @@
 
 
 class CrewDoclingSource(BaseKnowledgeSource):
-    """Default Source class for converting documents to markdown or json
-    This will auto support PDF, DOCX, and TXT, XLSX, Images, and HTML files without any additional dependencies and follows the docling package as the source of truth.
+    """Default Source class for converting documents to markdown or json.
+
+    This will auto support PDF, DOCX, and TXT, XLSX, Images, and HTML files without
+    any additional dependencies and follows the docling package as the source of truth.
     """
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         if not DOCLING_AVAILABLE:
             raise ImportError(
                 "The docling package is required to use CrewDoclingSource. "
@@ -66,7 +65,7 @@ def __init__(self, *args, **kwargs):
         )
     )
 
-    def model_post_init(self, _) -> None:
+    def model_post_init(self, _: Any) -> None:
         if self.file_path:
             self._logger.log(
                 "warning",
@@ -99,6 +98,15 @@ def add(self) -> None:
             self.chunks.extend(list(new_chunks_iterable))
         self._save_documents()
 
+    async def aadd(self) -> None:
+        """Add docling content asynchronously."""
+        if self.content is None:
+            return
+        for doc in self.content:
+            new_chunks_iterable = self._chunk_doc(doc)
+            self.chunks.extend(list(new_chunks_iterable))
+        await self._asave_documents()
+
     def _convert_source_to_docling_documents(self) -> list[DoclingDocument]:
         conv_results_iter = self.document_converter.convert_all(self.safe_file_paths)
         return [result.document for result in conv_results_iter]

lib/crewai/src/crewai/knowledge/source/csv_knowledge_source.py

@@ -31,6 +31,15 @@ def add(self) -> None:
         self.chunks.extend(new_chunks)
         self._save_documents()
 
+    async def aadd(self) -> None:
+        """Add CSV file content asynchronously."""
+        content_str = (
+            str(self.content) if isinstance(self.content, dict) else self.content
+        )
+        new_chunks = self._chunk_text(content_str)
+        self.chunks.extend(new_chunks)
+        await self._asave_documents()
+
     def _chunk_text(self, text: str) -> list[str]:
         """Utility method to split text into chunks."""
         return [

lib/crewai/src/crewai/knowledge/source/excel_knowledge_source.py

@@ -1,4 +1,6 @@
 from pathlib import Path
+from types import ModuleType
+from typing import Any
 
 from pydantic import Field, field_validator
 
@@ -26,7 +28,10 @@ class ExcelKnowledgeSource(BaseKnowledgeSource):
     safe_file_paths: list[Path] = Field(default_factory=list)
 
     @field_validator("file_path", "file_paths", mode="before")
-    def validate_file_path(cls, v, info):  # noqa: N805
+    @classmethod
+    def validate_file_path(
+        cls, v: Path | list[Path] | str | list[str] | None, info: Any
+    ) -> Path | list[Path] | str | list[str] | None:
         """Validate that at least one of file_path or file_paths is provided."""
         # Single check if both are None, O(1) instead of nested conditions
         if (
@@ -69,7 +74,7 @@ def _process_file_paths(self) -> list[Path]:
 
         return [self.convert_to_path(path) for path in path_list]
 
-    def validate_content(self):
+    def validate_content(self) -> None:
         """Validate the paths."""
         for path in self.safe_file_paths:
             if not path.exists():
@@ -86,7 +91,7 @@ def validate_content(self):
                     color="red",
                 )
 
-    def model_post_init(self, _) -> None:
+    def model_post_init(self, _: Any) -> None:
         if self.file_path:
             self._logger.log(
                 "warning",
@@ -128,12 +133,12 @@ def convert_to_path(self, path: Path | str) -> Path:
         """Convert a path to a Path object."""
         return Path(KNOWLEDGE_DIRECTORY + "/" + path) if isinstance(path, str) else path
 
-    def _import_dependencies(self):
+    def _import_dependencies(self) -> ModuleType:
         """Dynamically import dependencies."""
         try:
-            import pandas as pd  # type: ignore[import-untyped,import-not-found]
+            import pandas as pd  # type: ignore[import-untyped]
 
-            return pd
+            return pd  # type: ignore[no-any-return]
         except ImportError as e:
             missing_package = str(e).split()[-1]
             raise ImportError(
@@ -159,6 +164,20 @@ def add(self) -> None:
         self.chunks.extend(new_chunks)
         self._save_documents()
 
+    async def aadd(self) -> None:
+        """Add Excel file content asynchronously."""
+        content_str = ""
+        for value in self.content.values():
+            if isinstance(value, dict):
+                for sheet_value in value.values():
+                    content_str += str(sheet_value) + "\n"
+            else:
+                content_str += str(value) + "\n"
+
+        new_chunks = self._chunk_text(content_str)
+        self.chunks.extend(new_chunks)
+        await self._asave_documents()
+
     def _chunk_text(self, text: str) -> list[str]:
         """Utility method to split text into chunks."""
         return [

lib/crewai/src/crewai/knowledge/source/json_knowledge_source.py

@@ -44,6 +44,15 @@ def add(self) -> None:
         self.chunks.extend(new_chunks)
         self._save_documents()
 
+    async def aadd(self) -> None:
+        """Add JSON file content asynchronously."""
+        content_str = (
+            str(self.content) if isinstance(self.content, dict) else self.content
+        )
+        new_chunks = self._chunk_text(content_str)
+        self.chunks.extend(new_chunks)
+        await self._asave_documents()
+
     def _chunk_text(self, text: str) -> list[str]:
         """Utility method to split text into chunks."""
         return [

lib/crewai/src/crewai/knowledge/source/pdf_knowledge_source.py

@@ -1,4 +1,5 @@
 from pathlib import Path
+from types import ModuleType
 
 from crewai.knowledge.source.base_file_knowledge_source import BaseFileKnowledgeSource
 
@@ -23,7 +24,7 @@ def load_content(self) -> dict[Path, str]:
             content[path] = text
         return content
 
-    def _import_pdfplumber(self):
+    def _import_pdfplumber(self) -> ModuleType:
         """Dynamically import pdfplumber."""
         try:
             import pdfplumber
@@ -44,6 +45,13 @@ def add(self) -> None:
             self.chunks.extend(new_chunks)
         self._save_documents()
 
+    async def aadd(self) -> None:
+        """Add PDF file content asynchronously."""
+        for text in self.content.values():
+            new_chunks = self._chunk_text(text)
+            self.chunks.extend(new_chunks)
+        await self._asave_documents()
+
     def _chunk_text(self, text: str) -> list[str]:
         """Utility method to split text into chunks."""
         return [

lib/crewai/src/crewai/knowledge/source/string_knowledge_source.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from pydantic import Field
 
 from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
@@ -9,11 +11,11 @@ class StringKnowledgeSource(BaseKnowledgeSource):
     content: str = Field(...)
     collection_name: str | None = Field(default=None)
 
-    def model_post_init(self, _):
+    def model_post_init(self, _: Any) -> None:
         """Post-initialization method to validate content."""
         self.validate_content()
 
-    def validate_content(self):
+    def validate_content(self) -> None:
         """Validate string content."""
         if not isinstance(self.content, str):
             raise ValueError("StringKnowledgeSource only accepts string content")
@@ -24,6 +26,12 @@ def add(self) -> None:
         self.chunks.extend(new_chunks)
         self._save_documents()
 
+    async def aadd(self) -> None:
+        """Add string content asynchronously."""
+        new_chunks = self._chunk_text(self.content)
+        self.chunks.extend(new_chunks)
+        await self._asave_documents()
+
     def _chunk_text(self, text: str) -> list[str]:
         """Utility method to split text into chunks."""
         return [