Merge branch 'main' into patch-1

src/crewai/__init__.py

@@ -14,5 +14,5 @@ warnings.filterwarnings(
     category=UserWarning,
     module="pydantic.main",
 )
-__version__ = "0.70.1"
+__version__ = "0.74.0"
 __all__ = ["Agent", "Crew", "Process", "Task", "Pipeline", "Router", "LLM", "Flow"]

src/crewai/agents/agent_builder/base_agent_executor_mixin.py

@@ -17,7 +17,7 @@ if TYPE_CHECKING:
 
 class CrewAgentExecutorMixin:
     crew: Optional["Crew"]
-    crew_agent: Optional["BaseAgent"]
+    agent: Optional["BaseAgent"]
     task: Optional["Task"]
     iterations: int
     have_forced_answer: bool
@@ -33,9 +33,9 @@ class CrewAgentExecutorMixin:
         """Create and save a short-term memory item if conditions are met."""
         if (
             self.crew
-            and self.crew_agent
+            and self.agent
             and self.task
-            and "Action: Delegate work to coworker" not in output.log
+            and "Action: Delegate work to coworker" not in output.text
         ):
             try:
                 if (
@@ -43,11 +43,11 @@ class CrewAgentExecutorMixin:
                     and self.crew._short_term_memory
                 ):
                     self.crew._short_term_memory.save(
-                        value=output.log,
+                        value=output.text,
                         metadata={
                             "observation": self.task.description,
                         },
-                        agent=self.crew_agent.role,
+                        agent=self.agent.role,
                     )
             except Exception as e:
                 print(f"Failed to add to short term memory: {e}")
@@ -61,18 +61,18 @@ class CrewAgentExecutorMixin:
             and self.crew._long_term_memory
             and self.crew._entity_memory
             and self.task
-            and self.crew_agent
+            and self.agent
         ):
             try:
-                ltm_agent = TaskEvaluator(self.crew_agent)
-                evaluation = ltm_agent.evaluate(self.task, output.log)
+                ltm_agent = TaskEvaluator(self.agent)
+                evaluation = ltm_agent.evaluate(self.task, output.text)
 
                 if isinstance(evaluation, ConverterError):
                     return
 
                 long_term_memory = LongTermMemoryItem(
                     task=self.task.description,
-                    agent=self.crew_agent.role,
+                    agent=self.agent.role,
                     quality=evaluation.quality,
                     datetime=str(time.time()),
                     expected_output=self.task.expected_output,

src/crewai/agents/crew_agent_executor.py

@@ -19,6 +19,7 @@ from crewai.utilities.exceptions.context_window_exceeding_exception import (
 )
 from crewai.utilities.logger import Logger
 from crewai.utilities.training_handler import CrewTrainingHandler
+from crewai.agents.agent_builder.base_agent import BaseAgent
 
 
 class CrewAgentExecutor(CrewAgentExecutorMixin):
@@ -29,7 +30,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
         llm: Any,
         task: Any,
         crew: Any,
-        agent: Any,
+        agent: BaseAgent,
         prompt: dict[str, str],
         max_iter: int,
         tools: List[Any],
@@ -103,7 +104,8 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
 
             if self.crew and self.crew._train:
                 self._handle_crew_training_output(formatted_answer)
-
+        self._create_short_term_memory(formatted_answer)
+        self._create_long_term_memory(formatted_answer)
         return {"output": formatted_answer.output}
 
     def _invoke_loop(self, formatted_answer=None):
@@ -176,6 +178,8 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
         return formatted_answer
 
     def _show_start_logs(self):
+        if self.agent is None:
+            raise ValueError("Agent cannot be None")
         if self.agent.verbose or (
             hasattr(self, "crew") and getattr(self.crew, "verbose", False)
         ):
@@ -188,6 +192,8 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
             )
 
     def _show_logs(self, formatted_answer: Union[AgentAction, AgentFinish]):
+        if self.agent is None:
+            raise ValueError("Agent cannot be None")
         if self.agent.verbose or (
             hasattr(self, "crew") and getattr(self.crew, "verbose", False)
         ):
@@ -306,7 +312,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
         self, result: AgentFinish, human_feedback: str | None = None
     ) -> None:
         """Function to handle the process of the training data."""
-        agent_id = str(self.agent.id)
+        agent_id = str(self.agent.id)  # type: ignore
 
         # Load training data
         training_handler = CrewTrainingHandler(TRAINING_DATA_FILE)
@@ -339,7 +345,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
                 "initial_output": result.output,
                 "human_feedback": human_feedback,
                 "agent": agent_id,
-                "agent_role": self.agent.role,
+                "agent_role": self.agent.role,  # type: ignore
             }
             if self.crew is not None and hasattr(self.crew, "_train_iteration"):
                 train_iteration = self.crew._train_iteration

src/crewai/cli/constants.py

@@ -0,0 +1,19 @@
+ENV_VARS = {
+    'openai': ['OPENAI_API_KEY'],
+    'anthropic': ['ANTHROPIC_API_KEY'],
+    'gemini': ['GEMINI_API_KEY'],
+    'groq': ['GROQ_API_KEY'],
+    'ollama': ['FAKE_KEY'],
+}
+
+PROVIDERS = ['openai', 'anthropic', 'gemini', 'groq', 'ollama']
+
+MODELS = {
+    'openai': ['gpt-4', 'gpt-4o', 'gpt-4o-mini', 'o1-mini', 'o1-preview'],
+    'anthropic': ['claude-3-5-sonnet-20240620', 'claude-3-sonnet-20240229', 'claude-3-opus-20240229', 'claude-3-haiku-20240307'],
+    'gemini': ['gemini-1.5-flash', 'gemini-1.5-pro', 'gemini-gemma-2-9b-it', 'gemini-gemma-2-27b-it'],
+    'groq': ['llama-3.1-8b-instant', 'llama-3.1-70b-versatile', 'llama-3.1-405b-reasoning', 'gemma2-9b-it', 'gemma-7b-it'],
+    'ollama': ['llama3.1', 'mixtral'],
+}
+
+JSON_URL = "https://raw.githubusercontent.com/BerriAI/litellm/main/model_prices_and_context_window.json"

src/crewai/cli/create_crew.py

@@ -1,12 +1,11 @@
 from pathlib import Path
-
 import click
-
-from crewai.cli.utils import copy_template
+from crewai.cli.utils import copy_template, load_env_vars, write_env_file
+from crewai.cli.provider import get_provider_data, select_provider, PROVIDERS
+from crewai.cli.constants import ENV_VARS
 
 
-def create_crew(name, parent_folder=None):
-    """Create a new crew."""
+def create_folder_structure(name, parent_folder=None):
     folder_name = name.replace(" ", "_").replace("-", "_").lower()
     class_name = name.replace("_", " ").replace("-", " ").title().replace(" ", "")
 
@@ -28,19 +27,84 @@ def create_crew(name, parent_folder=None):
             (folder_path / "src" / folder_name).mkdir(parents=True)
             (folder_path / "src" / folder_name / "tools").mkdir(parents=True)
             (folder_path / "src" / folder_name / "config").mkdir(parents=True)
-            with open(folder_path / ".env", "w") as file:
-                file.write("OPENAI_API_KEY=YOUR_API_KEY")
     else:
         click.secho(
-            f"\tFolder {folder_name} already exists. Please choose a different name.",
-            fg="red",
+            f"\tFolder {folder_name} already exists.",
+            fg="yellow",
         )
+
+    return folder_path, folder_name, class_name
+
+
+def copy_template_files(folder_path, name, class_name, parent_folder):
+    package_dir = Path(__file__).parent
+    templates_dir = package_dir / "templates" / "crew"
+
+    root_template_files = (
+        [".gitignore", "pyproject.toml", "README.md"] if not parent_folder else []
+    )
+    tools_template_files = ["tools/custom_tool.py", "tools/__init__.py"]
+    config_template_files = ["config/agents.yaml", "config/tasks.yaml"]
+    src_template_files = (
+        ["__init__.py", "main.py", "crew.py"] if not parent_folder else ["crew.py"]
+    )
+
+    for file_name in root_template_files:
+        src_file = templates_dir / file_name
+        dst_file = folder_path / file_name
+        copy_template(src_file, dst_file, name, class_name, folder_path.name)
+
+    src_folder = (
+        folder_path / "src" / folder_path.name if not parent_folder else folder_path
+    )
+
+    for file_name in src_template_files:
+        src_file = templates_dir / file_name
+        dst_file = src_folder / file_name
+        copy_template(src_file, dst_file, name, class_name, folder_path.name)
+
+    if not parent_folder:
+        for file_name in tools_template_files + config_template_files:
+            src_file = templates_dir / file_name
+            dst_file = src_folder / file_name
+            copy_template(src_file, dst_file, name, class_name, folder_path.name)
+
+
+def create_crew(name, parent_folder=None):
+    folder_path, folder_name, class_name = create_folder_structure(name, parent_folder)
+    env_vars = load_env_vars(folder_path)
+
+    provider_models = get_provider_data()
+    if not provider_models:
         return
 
+    selected_provider = select_provider(provider_models)
+    if not selected_provider:
+        return
+    provider = selected_provider
+
+    # selected_model = select_model(provider, provider_models)
+    # if not selected_model:
+    #     return
+    # model = selected_model
+
+    if provider in PROVIDERS:
+        api_key_var = ENV_VARS[provider][0]
+    else:
+        api_key_var = click.prompt(
+            f"Enter the environment variable name for your {provider.capitalize()} API key",
+            type=str,
+        )
+
+    env_vars = {api_key_var: "YOUR_API_KEY_HERE"}
+    write_env_file(folder_path, env_vars)
+
+    # env_vars['MODEL'] = model
+    # click.secho(f"Selected model: {model}", fg="green")
+
     package_dir = Path(__file__).parent
     templates_dir = package_dir / "templates" / "crew"
 
-    # List of template files to copy
     root_template_files = (
         [".gitignore", "pyproject.toml", "README.md"] if not parent_folder else []
     )

src/crewai/cli/provider.py

@@ -0,0 +1,186 @@
+import json
+import time
+import requests
+from collections import defaultdict
+import click
+from pathlib import Path 
+from crewai.cli.constants import PROVIDERS, MODELS, JSON_URL
+
+def select_choice(prompt_message, choices):
+    """
+    Presents a list of choices to the user and prompts them to select one.
+    
+    Args:
+    - prompt_message (str): The message to display to the user before presenting the choices.
+    - choices (list): A list of options to present to the user.
+    
+    Returns:
+    - str: The selected choice from the list, or None if the operation is aborted or an invalid selection is made.
+    """
+    click.secho(prompt_message, fg="cyan")
+    for idx, choice in enumerate(choices, start=1):
+        click.secho(f"{idx}. {choice}", fg="cyan")
+    try:
+        selected_index = click.prompt("Enter the number of your choice", type=int) - 1
+    except click.exceptions.Abort:
+        click.secho("Operation aborted by the user.", fg="red")
+        return None
+    if not (0 <= selected_index < len(choices)):
+        click.secho("Invalid selection.", fg="red")
+        return None
+    return choices[selected_index]
+
+def select_provider(provider_models):
+    """
+    Presents a list of providers to the user and prompts them to select one.
+    
+    Args:
+    - provider_models (dict): A dictionary of provider models.
+    
+    Returns:
+    - str: The selected provider, or None if the operation is aborted or an invalid selection is made.
+    """
+    predefined_providers = [p.lower() for p in PROVIDERS]
+    all_providers = sorted(set(predefined_providers + list(provider_models.keys())))
+
+    provider = select_choice("Select a provider to set up:", predefined_providers + ['other'])
+    if not provider:
+        return None
+    provider = provider.lower()
+
+    if provider == 'other':
+        provider = select_choice("Select a provider from the full list:", all_providers)
+        if not provider:
+            return None
+    return provider
+
+def select_model(provider, provider_models):
+    """
+    Presents a list of models for a given provider to the user and prompts them to select one.
+    
+    Args:
+    - provider (str): The provider for which to select a model.
+    - provider_models (dict): A dictionary of provider models.
+    
+    Returns:
+    - str: The selected model, or None if the operation is aborted or an invalid selection is made.
+    """
+    predefined_providers = [p.lower() for p in PROVIDERS]
+
+    if provider in predefined_providers:
+        available_models = MODELS.get(provider, [])
+    else:
+        available_models = provider_models.get(provider, [])
+
+    if not available_models:
+        click.secho(f"No models available for provider '{provider}'.", fg="red")
+        return None
+
+    selected_model = select_choice(f"Select a model to use for {provider.capitalize()}:", available_models)
+    return selected_model
+
+def load_provider_data(cache_file, cache_expiry):
+    """
+    Loads provider data from a cache file if it exists and is not expired. If the cache is expired or corrupted, it fetches the data from the web.
+    
+    Args:
+    - cache_file (Path): The path to the cache file.
+    - cache_expiry (int): The cache expiry time in seconds.
+    
+    Returns:
+    - dict or None: The loaded provider data or None if the operation fails.
+    """
+    current_time = time.time()
+    if cache_file.exists() and (current_time - cache_file.stat().st_mtime) < cache_expiry:
+        data = read_cache_file(cache_file)
+        if data:
+            return data
+        click.secho("Cache is corrupted. Fetching provider data from the web...", fg="yellow")
+    else:
+        click.secho("Cache expired or not found. Fetching provider data from the web...", fg="cyan")
+    return fetch_provider_data(cache_file)
+
+def read_cache_file(cache_file):
+    """
+    Reads and returns the JSON content from a cache file. Returns None if the file contains invalid JSON.
+    
+    Args:
+    - cache_file (Path): The path to the cache file.
+    
+    Returns:
+    - dict or None: The JSON content of the cache file or None if the JSON is invalid.
+    """
+    try:
+        with open(cache_file, "r") as f:
+            return json.load(f)
+    except json.JSONDecodeError:
+        return None
+
+def fetch_provider_data(cache_file):
+    """
+    Fetches provider data from a specified URL and caches it to a file.
+    
+    Args:
+    - cache_file (Path): The path to the cache file.
+    
+    Returns:
+    - dict or None: The fetched provider data or None if the operation fails.
+    """
+    try:
+        response = requests.get(JSON_URL, stream=True, timeout=10)
+        response.raise_for_status()
+        data = download_data(response)
+        with open(cache_file, "w") as f:
+            json.dump(data, f)
+        return data
+    except requests.RequestException as e:
+        click.secho(f"Error fetching provider data: {e}", fg="red")
+    except json.JSONDecodeError:
+        click.secho("Error parsing provider data. Invalid JSON format.", fg="red")
+    return None
+
+def download_data(response):
+    """
+    Downloads data from a given HTTP response and returns the JSON content.
+    
+    Args:
+    - response (requests.Response): The HTTP response object.
+    
+    Returns:
+    - dict: The JSON content of the response.
+    """
+    total_size = int(response.headers.get('content-length', 0))
+    block_size = 8192
+    data_chunks = []
+    with click.progressbar(length=total_size, label='Downloading', show_pos=True) as progress_bar:
+        for chunk in response.iter_content(block_size):
+            if chunk:
+                data_chunks.append(chunk)
+                progress_bar.update(len(chunk))
+    data_content = b''.join(data_chunks)
+    return json.loads(data_content.decode('utf-8'))
+
+def get_provider_data():
+    """
+    Retrieves provider data from a cache file, filters out models based on provider criteria, and returns a dictionary of providers mapped to their models.
+    
+    Returns:
+    - dict or None: A dictionary of providers mapped to their models or None if the operation fails.
+    """
+    cache_dir = Path.home() / '.crewai'
+    cache_dir.mkdir(exist_ok=True)
+    cache_file = cache_dir / 'provider_cache.json'
+    cache_expiry = 24 * 3600 
+
+    data = load_provider_data(cache_file, cache_expiry)
+    if not data:
+        return None
+
+    provider_models = defaultdict(list)
+    for model_name, properties in data.items():
+        provider = properties.get("litellm_provider", "").strip().lower()
+        if 'http' in provider or provider == 'other':
+            continue
+        if provider:
+            provider_models[provider].append(model_name)
+    return provider_models

src/crewai/cli/run_crew.py

@@ -2,6 +2,9 @@ import subprocess
 
 import click
 import tomllib
+from packaging import version
+
+from crewai.cli.utils import get_crewai_version
 
 
 def run_crew() -> None:
@@ -9,6 +12,22 @@ def run_crew() -> None:
     Run the crew by running a command in the UV environment.
     """
     command = ["uv", "run", "run_crew"]
+    crewai_version = get_crewai_version()
+    min_required_version = "0.71.0"
+
+    with open("pyproject.toml", "rb") as f:
+        data = tomllib.load(f)
+
+    if data.get("tool", {}).get("poetry") and (
+        version.parse(crewai_version) < version.parse(min_required_version)
+    ):
+        click.secho(
+            f"You are running an older version of crewAI ({crewai_version}) that uses poetry pyproject.toml. "
+            f"Please run `crewai update` to update your pyproject.toml to use uv.",
+            fg="red",
+        )
+        print()
+
     try:
         subprocess.run(command, capture_output=False, text=True, check=True)
 

src/crewai/cli/templates/crew/pyproject.toml

@@ -5,7 +5,7 @@ description = "{{name}} using crewAI"
 authors = [{ name = "Your Name", email = "you@example.com" }]
 requires-python = ">=3.10,<=3.13"
 dependencies = [
-    "crewai[tools]>=0.67.1,<1.0.0"
+    "crewai[tools]>=0.74.0,<1.0.0"
 ]
 
 [project.scripts]

src/crewai/cli/templates/flow/pyproject.toml

@@ -5,7 +5,7 @@ description = "{{name}} using crewAI"
 authors = [{ name = "Your Name", email = "you@example.com" }]
 requires-python = ">=3.10,<=3.13"
 dependencies = [
-    "crewai[tools]>=0.67.1,<1.0.0",
+    "crewai[tools]>=0.74.0,<1.0.0",
     "asyncio"
 ]
 

src/crewai/cli/templates/pipeline/pyproject.toml

@@ -6,7 +6,7 @@ authors = ["Your Name <you@example.com>"]
 
 [tool.poetry.dependencies]
 python = ">=3.10,<=3.13"
-crewai = { extras = ["tools"], version = ">=0.70.1,<1.0.0" }
+crewai = { extras = ["tools"], version = ">=0.74.0,<1.0.0" }
 asyncio = "*"
 
 [tool.poetry.scripts]

src/crewai/cli/templates/pipeline_router/pyproject.toml

@@ -5,7 +5,7 @@ description = "{{name}} using crewAI"
 authors = ["Your Name <you@example.com>"]
 requires-python = ">=3.10,<=3.13"
 dependencies = [
-    "crewai[tools]>=0.67.1,<1.0.0"
+    "crewai[tools]>=0.74.0,<1.0.0"
 ]
 
 [project.scripts]

src/crewai/cli/templates/tool/pyproject.toml

@@ -5,6 +5,6 @@ description = "Power up your crews with {{folder_name}}"
 readme = "README.md"
 requires-python = ">=3.10,<=3.13"
 dependencies = [
-    "crewai[tools]>=0.70.1"
+    "crewai[tools]>=0.74.0"
 ]
 

src/crewai/cli/utils.py

@@ -9,6 +9,7 @@ import click
 from rich.console import Console
 
 from crewai.cli.authentication.utils import TokenManager
+from crewai.cli.constants import ENV_VARS
 
 if sys.version_info >= (3, 11):
     import tomllib
@@ -200,3 +201,76 @@ def tree_find_and_replace(directory, find, replace):
                 new_dirpath = os.path.join(path, new_dirname)
                 old_dirpath = os.path.join(path, dirname)
                 os.rename(old_dirpath, new_dirpath)
+
+
+def load_env_vars(folder_path):
+    """
+    Loads environment variables from a .env file in the specified folder path.
+
+    Args:
+    - folder_path (Path): The path to the folder containing the .env file.
+
+    Returns:
+    - dict: A dictionary of environment variables.
+    """
+    env_file_path = folder_path / ".env"
+    env_vars = {}
+    if env_file_path.exists():
+        with open(env_file_path, "r") as file:
+            for line in file:
+                key, _, value = line.strip().partition("=")
+                if key and value:
+                    env_vars[key] = value
+    return env_vars
+
+
+def update_env_vars(env_vars, provider, model):
+    """
+    Updates environment variables with the API key for the selected provider and model.
+
+    Args:
+    - env_vars (dict): Environment variables dictionary.
+    - provider (str): Selected provider.
+    - model (str): Selected model.
+
+    Returns:
+    - None
+    """
+    api_key_var = ENV_VARS.get(
+        provider,
+        [
+            click.prompt(
+                f"Enter the environment variable name for your {provider.capitalize()} API key",
+                type=str,
+            )
+        ],
+    )[0]
+
+    if api_key_var not in env_vars:
+        try:
+            env_vars[api_key_var] = click.prompt(
+                f"Enter your {provider.capitalize()} API key", type=str, hide_input=True
+            )
+        except click.exceptions.Abort:
+            click.secho("Operation aborted by the user.", fg="red")
+            return None
+    else:
+        click.secho(f"API key already exists for {provider.capitalize()}.", fg="yellow")
+
+    env_vars["MODEL"] = model
+    click.secho(f"Selected model: {model}", fg="green")
+    return env_vars
+
+
+def write_env_file(folder_path, env_vars):
+    """
+    Writes environment variables to a .env file in the specified folder.
+
+    Args:
+    - folder_path (Path): The path to the folder where the .env file will be written.
+    - env_vars (dict): A dictionary of environment variables to write.
+    """
+    env_file_path = folder_path / ".env"
+    with open(env_file_path, "w") as file:
+        for key, value in env_vars.items():
+            file.write(f"{key}={value}\n")

src/crewai/crew.py

@@ -126,8 +126,8 @@ class Crew(BaseModel):
         default=None,
         description="An Instance of the EntityMemory to be used by the Crew",
     )
-    embedder: Optional[dict] = Field(
-        default={"provider": "openai"},
+    embedder: Optional[Any] = Field(
+        default=None,
         description="Configuration for the embedder to be used for the crew.",
     )
     usage_metrics: Optional[UsageMetrics] = Field(
@@ -774,7 +774,9 @@ class Crew(BaseModel):
 
     def _log_task_start(self, task: Task, role: str = "None"):
         if self.output_log_file:
-            self._file_handler.log(task_name=task.name, task=task.description, agent=role, status="started")
+            self._file_handler.log(
+                task_name=task.name, task=task.description, agent=role, status="started"
+            )
 
     def _update_manager_tools(self, task: Task):
         if self.manager_agent:
@@ -796,7 +798,13 @@ class Crew(BaseModel):
     def _process_task_result(self, task: Task, output: TaskOutput) -> None:
         role = task.agent.role if task.agent is not None else "None"
         if self.output_log_file:
-            self._file_handler.log(task_name=task.name, task=task.description, agent=role, status="completed", output=output.raw)
+            self._file_handler.log(
+                task_name=task.name,
+                task=task.description,
+                agent=role,
+                status="completed",
+                output=output.raw,
+            )
 
     def _create_crew_output(self, task_outputs: List[TaskOutput]) -> CrewOutput:
         if len(task_outputs) != 1:

src/crewai/memory/contextual/contextual_memory.py

@@ -31,7 +31,9 @@ class ContextualMemory:
         formatted as bullet points.
         """
         stm_results = self.stm.search(query)
-        formatted_results = "\n".join([f"- {result}" for result in stm_results])
+        formatted_results = "\n".join(
+            [f"- {result['context']}" for result in stm_results]
+        )
         return f"Recent Insights:\n{formatted_results}" if stm_results else ""
 
     def _fetch_ltm_context(self, task) -> Optional[str]:

src/crewai/memory/long_term/long_term_memory.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 from crewai.memory.long_term.long_term_memory_item import LongTermMemoryItem
 from crewai.memory.memory import Memory
@@ -28,7 +28,7 @@ class LongTermMemory(Memory):
             datetime=item.datetime,
         )
 
-    def search(self, task: str, latest_n: int = 3) -> Dict[str, Any]:
+    def search(self, task: str, latest_n: int = 3) -> List[Dict[str, Any]]:  # type: ignore # signature of "search" incompatible with supertype "Memory"
         return self.storage.load(task, latest_n)  # type: ignore # BUG?: "Storage" has no attribute "load"
 
     def reset(self) -> None:

src/crewai/memory/memory.py

@@ -1,6 +1,6 @@
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, List
 
-from crewai.memory.storage.interface import Storage
+from crewai.memory.storage.rag_storage import RAGStorage
 
 
 class Memory:
@@ -8,7 +8,7 @@ class Memory:
     Base class for memory, now supporting agent tags and generic metadata.
     """
 
-    def __init__(self, storage: Storage):
+    def __init__(self, storage: RAGStorage):
         self.storage = storage
 
     def save(
@@ -23,5 +23,5 @@ class Memory:
 
         self.storage.save(value, metadata)
 
-    def search(self, query: str) -> Dict[str, Any]:
+    def search(self, query: str) -> List[Dict[str, Any]]:
         return self.storage.search(query)

src/crewai/memory/storage/base_rag_storage.py

@@ -0,0 +1,76 @@
+from abc import ABC, abstractmethod
+from typing import Any, Dict, List, Optional
+
+
+class BaseRAGStorage(ABC):
+    """
+    Base class for RAG-based Storage implementations.
+    """
+
+    app: Any | None = None
+
+    def __init__(
+        self,
+        type: str,
+        allow_reset: bool = True,
+        embedder_config: Optional[Any] = None,
+        crew: Any = None,
+    ):
+        self.type = type
+        self.allow_reset = allow_reset
+        self.embedder_config = embedder_config
+        self.crew = crew
+        self.agents = self._initialize_agents()
+
+    def _initialize_agents(self) -> str:
+        if self.crew:
+            return "_".join(
+                [self._sanitize_role(agent.role) for agent in self.crew.agents]
+            )
+        return ""
+
+    @abstractmethod
+    def _sanitize_role(self, role: str) -> str:
+        """Sanitizes agent roles to ensure valid directory names."""
+        pass
+
+    @abstractmethod
+    def save(self, value: Any, metadata: Dict[str, Any]) -> None:
+        """Save a value with metadata to the storage."""
+        pass
+
+    @abstractmethod
+    def search(
+        self,
+        query: str,
+        limit: int = 3,
+        filter: Optional[dict] = None,
+        score_threshold: float = 0.35,
+    ) -> List[Any]:
+        """Search for entries in the storage."""
+        pass
+
+    @abstractmethod
+    def reset(self) -> None:
+        """Reset the storage."""
+        pass
+
+    @abstractmethod
+    def _generate_embedding(
+        self, text: str, metadata: Optional[Dict[str, Any]] = None
+    ) -> Any:
+        """Generate an embedding for the given text and metadata."""
+        pass
+
+    @abstractmethod
+    def _initialize_app(self):
+        """Initialize the vector db."""
+        pass
+
+    def setup_config(self, config: Dict[str, Any]):
+        """Setup the config of the storage."""
+        pass
+
+    def initialize_client(self):
+        """Initialize the client of the storage. This should setup the app and the db collection"""
+        pass

src/crewai/memory/storage/interface.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 
 class Storage:
@@ -7,7 +7,7 @@ class Storage:
     def save(self, value: Any, metadata: Dict[str, Any]) -> None:
         pass
 
-    def search(self, key: str) -> Dict[str, Any]:  # type: ignore
+    def search(self, key: str) -> List[Dict[str, Any]]:  # type: ignore
         pass
 
     def reset(self) -> None:

src/crewai/memory/storage/rag_storage.py

@@ -3,10 +3,11 @@ import io
 import logging
 import os
 import shutil
+import uuid
 from typing import Any, Dict, List, Optional
-
-from crewai.memory.storage.interface import Storage
+from crewai.memory.storage.base_rag_storage import BaseRAGStorage
 from crewai.utilities.paths import db_storage_path
+from chromadb.api import ClientAPI
 
 
 @contextlib.contextmanager
@@ -24,61 +25,42 @@ def suppress_logging(
     logger.setLevel(original_level)
 
 
-class RAGStorage(Storage):
+class RAGStorage(BaseRAGStorage):
     """
     Extends Storage to handle embeddings for memory entries, improving
     search efficiency.
     """
 
-    def __init__(self, type, allow_reset=True, embedder_config=None, crew=None):
-        super().__init__()
-        if (
-            not os.getenv("OPENAI_API_KEY")
-            and not os.getenv("OPENAI_BASE_URL") == "https://api.openai.com/v1"
-        ):
-            os.environ["OPENAI_API_KEY"] = "fake"
+    app: ClientAPI | None = None
 
+    def __init__(self, type, allow_reset=True, embedder_config=None, crew=None):
+        super().__init__(type, allow_reset, embedder_config, crew)
         agents = crew.agents if crew else []
         agents = [self._sanitize_role(agent.role) for agent in agents]
         agents = "_".join(agents)
+        self.agents = agents
 
-        config = {
-            "app": {
-                "config": {"name": type, "collect_metrics": False, "log_level": "ERROR"}
-            },
-            "chunker": {
-                "chunk_size": 5000,
-                "chunk_overlap": 100,
-                "length_function": "len",
-                "min_chunk_size": 150,
-            },
-            "vectordb": {
-                "provider": "chroma",
-                "config": {
-                    "collection_name": type,
-                    "dir": f"{db_storage_path()}/{type}/{agents}",
-                    "allow_reset": allow_reset,
-                },
-            },
-        }
-
-        if embedder_config:
-            config["embedder"] = embedder_config
         self.type = type
-        self.config = config
+        self.embedder_config = embedder_config or self._create_embedding_function()
         self.allow_reset = allow_reset
+        self._initialize_app()
 
     def _initialize_app(self):
-        from embedchain import App
-        from embedchain.llm.base import BaseLlm
+        import chromadb
 
-        class FakeLLM(BaseLlm):
-            pass
+        chroma_client = chromadb.PersistentClient(
+            path=f"{db_storage_path()}/{self.type}/{self.agents}"
+        )
+        self.app = chroma_client
 
-        self.app = App.from_config(config=self.config)
-        self.app.llm = FakeLLM()
-        if self.allow_reset:
-            self.app.reset()
+        try:
+            self.collection = self.app.get_collection(
+                name=self.type, embedding_function=self.embedder_config
+            )
+        except Exception:
+            self.collection = self.app.create_collection(
+                name=self.type, embedding_function=self.embedder_config
+            )
 
     def _sanitize_role(self, role: str) -> str:
         """
@@ -87,11 +69,14 @@ class RAGStorage(Storage):
         return role.replace("\n", "").replace(" ", "_").replace("/", "_")
 
     def save(self, value: Any, metadata: Dict[str, Any]) -> None:
-        if not hasattr(self, "app"):
+        if not hasattr(self, "app") or not hasattr(self, "collection"):
             self._initialize_app()
-        self._generate_embedding(value, metadata)
+        try:
+            self._generate_embedding(value, metadata)
+        except Exception as e:
+            logging.error(f"Error during {self.type} save: {str(e)}")
 
-    def search(  # type: ignore # BUG?: Signature of "search" incompatible with supertype "Storage"
+    def search(
         self,
         query: str,
         limit: int = 3,
@@ -100,31 +85,50 @@ class RAGStorage(Storage):
     ) -> List[Any]:
         if not hasattr(self, "app"):
             self._initialize_app()
-        from embedchain.vectordb.chroma import InvalidDimensionException
 
-        with suppress_logging():
-            try:
-                results = (
-                    self.app.search(query, limit, where=filter)
-                    if filter
-                    else self.app.search(query, limit)
-                )
-            except InvalidDimensionException:
-                self.app.reset()
-                return []
-        return [r for r in results if r["metadata"]["score"] >= score_threshold]
+        try:
+            with suppress_logging():
+                response = self.collection.query(query_texts=query, n_results=limit)
 
-    def _generate_embedding(self, text: str, metadata: Dict[str, Any]) -> Any:
-        if not hasattr(self, "app"):
+            results = []
+            for i in range(len(response["ids"][0])):
+                result = {
+                    "id": response["ids"][0][i],
+                    "metadata": response["metadatas"][0][i],
+                    "context": response["documents"][0][i],
+                    "score": response["distances"][0][i],
+                }
+                if result["score"] >= score_threshold:
+                    results.append(result)
+
+            return results
+        except Exception as e:
+            logging.error(f"Error during {self.type} search: {str(e)}")
+            return []
+
+    def _generate_embedding(self, text: str, metadata: Dict[str, Any]) -> None:  # type: ignore
+        if not hasattr(self, "app") or not hasattr(self, "collection"):
             self._initialize_app()
-        from embedchain.models.data_type import DataType
 
-        self.app.add(text, data_type=DataType.TEXT, metadata=metadata)
+        self.collection.add(
+            documents=[text],
+            metadatas=[metadata or {}],
+            ids=[str(uuid.uuid4())],
+        )
 
     def reset(self) -> None:
         try:
             shutil.rmtree(f"{db_storage_path()}/{self.type}")
+            if self.app:
+                self.app.reset()
         except Exception as e:
             raise Exception(
                 f"An error occurred while resetting the {self.type} memory: {e}"
             )
+
+    def _create_embedding_function(self):
+        import chromadb.utils.embedding_functions as embedding_functions
+
+        return embedding_functions.OpenAIEmbeddingFunction(
+            api_key=os.getenv("OPENAI_API_KEY"), model_name="text-embedding-3-small"
+        )

src/crewai/project/annotations.py

@@ -76,27 +76,13 @@ def crew(func) -> Callable[..., Crew]:
         instantiated_agents = []
         agent_roles = set()
 
-        # Collect methods from crew in order
-        all_functions = [
-            (name, getattr(self, name))
-            for name, attr in self.__class__.__dict__.items()
-            if callable(attr)
-        ]
-        tasks = [
-            (name, method)
-            for name, method in all_functions
-            if hasattr(method, "is_task")
-        ]
-
-        agents = [
-            (name, method)
-            for name, method in all_functions
-            if hasattr(method, "is_agent")
-        ]
+        # Use the preserved task and agent information
+        tasks = self._original_tasks.items()
+        agents = self._original_agents.items()
 
         # Instantiate tasks in order
         for task_name, task_method in tasks:
-            task_instance = task_method()
+            task_instance = task_method(self)
             instantiated_tasks.append(task_instance)
             agent_instance = getattr(task_instance, "agent", None)
             if agent_instance and agent_instance.role not in agent_roles:
@@ -105,7 +91,7 @@ def crew(func) -> Callable[..., Crew]:
 
         # Instantiate agents not included by tasks
         for agent_name, agent_method in agents:
-            agent_instance = agent_method()
+            agent_instance = agent_method(self)
             if agent_instance.role not in agent_roles:
                 instantiated_agents.append(agent_instance)
                 agent_roles.add(agent_instance.role)

src/crewai/project/crew_base.py

@@ -34,6 +34,18 @@ def CrewBase(cls: T) -> T:
             self.map_all_agent_variables()
             self.map_all_task_variables()
 
+            # Preserve task and agent information
+            self._original_tasks = {
+                name: method
+                for name, method in cls.__dict__.items()
+                if hasattr(method, "is_task") and method.is_task
+            }
+            self._original_agents = {
+                name: method
+                for name, method in cls.__dict__.items()
+                if hasattr(method, "is_agent") and method.is_agent
+            }
+
         @staticmethod
         def load_yaml(config_path: Path):
             try:

src/crewai/telemetry/telemetry.py

@@ -65,7 +65,7 @@ class Telemetry:
 
             self.provider.add_span_processor(processor)
             self.ready = True
-        except BaseException as e:
+        except Exception as e:
             if isinstance(
                 e,
                 (SystemExit, KeyboardInterrupt, GeneratorExit, asyncio.CancelledError),
@@ -83,404 +83,33 @@ class Telemetry:
                 self.ready = False
                 self.trace_set = False
 
+    def _safe_telemetry_operation(self, operation):
+        if not self.ready:
+            return
+        try:
+            operation()
+        except Exception:
+            pass
+
     def crew_creation(self, crew: Crew, inputs: dict[str, Any] | None):
         """Records the creation of a crew."""
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Crew Created")
-                self._add_attribute(
-                    span,
-                    "crewai_version",
-                    pkg_resources.get_distribution("crewai").version,
-                )
-                self._add_attribute(span, "python_version", platform.python_version())
-                self._add_attribute(span, "crew_key", crew.key)
-                self._add_attribute(span, "crew_id", str(crew.id))
-                self._add_attribute(span, "crew_process", crew.process)
-                self._add_attribute(span, "crew_memory", crew.memory)
-                self._add_attribute(span, "crew_number_of_tasks", len(crew.tasks))
-                self._add_attribute(span, "crew_number_of_agents", len(crew.agents))
-                if crew.share_crew:
-                    self._add_attribute(
-                        span,
-                        "crew_agents",
-                        json.dumps(
-                            [
-                                {
-                                    "key": agent.key,
-                                    "id": str(agent.id),
-                                    "role": agent.role,
-                                    "goal": agent.goal,
-                                    "backstory": agent.backstory,
-                                    "verbose?": agent.verbose,
-                                    "max_iter": agent.max_iter,
-                                    "max_rpm": agent.max_rpm,
-                                    "i18n": agent.i18n.prompt_file,
-                                    "function_calling_llm": (
-                                        agent.function_calling_llm.model
-                                        if agent.function_calling_llm
-                                        else ""
-                                    ),
-                                    "llm": agent.llm.model,
-                                    "delegation_enabled?": agent.allow_delegation,
-                                    "allow_code_execution?": agent.allow_code_execution,
-                                    "max_retry_limit": agent.max_retry_limit,
-                                    "tools_names": [
-                                        tool.name.casefold()
-                                        for tool in agent.tools or []
-                                    ],
-                                }
-                                for agent in crew.agents
-                            ]
-                        ),
-                    )
-                    self._add_attribute(
-                        span,
-                        "crew_tasks",
-                        json.dumps(
-                            [
-                                {
-                                    "key": task.key,
-                                    "id": str(task.id),
-                                    "description": task.description,
-                                    "expected_output": task.expected_output,
-                                    "async_execution?": task.async_execution,
-                                    "human_input?": task.human_input,
-                                    "agent_role": (
-                                        task.agent.role if task.agent else "None"
-                                    ),
-                                    "agent_key": task.agent.key if task.agent else None,
-                                    "context": (
-                                        [task.description for task in task.context]
-                                        if task.context
-                                        else None
-                                    ),
-                                    "tools_names": [
-                                        tool.name.casefold()
-                                        for tool in task.tools or []
-                                    ],
-                                }
-                                for task in crew.tasks
-                            ]
-                        ),
-                    )
-                    self._add_attribute(span, "platform", platform.platform())
-                    self._add_attribute(span, "platform_release", platform.release())
-                    self._add_attribute(span, "platform_system", platform.system())
-                    self._add_attribute(span, "platform_version", platform.version())
-                    self._add_attribute(span, "cpus", os.cpu_count())
-                    self._add_attribute(
-                        span, "crew_inputs", json.dumps(inputs) if inputs else None
-                    )
-                else:
-                    self._add_attribute(
-                        span,
-                        "crew_agents",
-                        json.dumps(
-                            [
-                                {
-                                    "key": agent.key,
-                                    "id": str(agent.id),
-                                    "role": agent.role,
-                                    "verbose?": agent.verbose,
-                                    "max_iter": agent.max_iter,
-                                    "max_rpm": agent.max_rpm,
-                                    "function_calling_llm": (
-                                        agent.function_calling_llm.model
-                                        if agent.function_calling_llm
-                                        else ""
-                                    ),
-                                    "llm": agent.llm.model,
-                                    "delegation_enabled?": agent.allow_delegation,
-                                    "allow_code_execution?": agent.allow_code_execution,
-                                    "max_retry_limit": agent.max_retry_limit,
-                                    "tools_names": [
-                                        tool.name.casefold()
-                                        for tool in agent.tools or []
-                                    ],
-                                }
-                                for agent in crew.agents
-                            ]
-                        ),
-                    )
-                    self._add_attribute(
-                        span,
-                        "crew_tasks",
-                        json.dumps(
-                            [
-                                {
-                                    "key": task.key,
-                                    "id": str(task.id),
-                                    "async_execution?": task.async_execution,
-                                    "human_input?": task.human_input,
-                                    "agent_role": (
-                                        task.agent.role if task.agent else "None"
-                                    ),
-                                    "agent_key": task.agent.key if task.agent else None,
-                                    "tools_names": [
-                                        tool.name.casefold()
-                                        for tool in task.tools or []
-                                    ],
-                                }
-                                for task in crew.tasks
-                            ]
-                        ),
-                    )
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
 
-    def task_started(self, crew: Crew, task: Task) -> Span | None:
-        """Records task started in a crew."""
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-
-                created_span = tracer.start_span("Task Created")
-
-                self._add_attribute(created_span, "crew_key", crew.key)
-                self._add_attribute(created_span, "crew_id", str(crew.id))
-                self._add_attribute(created_span, "task_key", task.key)
-                self._add_attribute(created_span, "task_id", str(task.id))
-
-                if crew.share_crew:
-                    self._add_attribute(
-                        created_span, "formatted_description", task.description
-                    )
-                    self._add_attribute(
-                        created_span, "formatted_expected_output", task.expected_output
-                    )
-
-                created_span.set_status(Status(StatusCode.OK))
-                created_span.end()
-
-                span = tracer.start_span("Task Execution")
-
-                self._add_attribute(span, "crew_key", crew.key)
-                self._add_attribute(span, "crew_id", str(crew.id))
-                self._add_attribute(span, "task_key", task.key)
-                self._add_attribute(span, "task_id", str(task.id))
-
-                if crew.share_crew:
-                    self._add_attribute(span, "formatted_description", task.description)
-                    self._add_attribute(
-                        span, "formatted_expected_output", task.expected_output
-                    )
-
-                return span
-            except Exception:
-                pass
-
-        return None
-
-    def task_ended(self, span: Span, task: Task, crew: Crew):
-        """Records task execution in a crew."""
-        if self.ready:
-            try:
-                if crew.share_crew:
-                    self._add_attribute(
-                        span,
-                        "task_output",
-                        task.output.raw if task.output else "",
-                    )
-
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
-
-    def tool_repeated_usage(self, llm: Any, tool_name: str, attempts: int):
-        """Records the repeated usage 'error' of a tool by an agent."""
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Tool Repeated Usage")
-                self._add_attribute(
-                    span,
-                    "crewai_version",
-                    pkg_resources.get_distribution("crewai").version,
-                )
-                self._add_attribute(span, "tool_name", tool_name)
-                self._add_attribute(span, "attempts", attempts)
-                if llm:
-                    self._add_attribute(span, "llm", llm.model)
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
-
-    def tool_usage(self, llm: Any, tool_name: str, attempts: int):
-        """Records the usage of a tool by an agent."""
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Tool Usage")
-                self._add_attribute(
-                    span,
-                    "crewai_version",
-                    pkg_resources.get_distribution("crewai").version,
-                )
-                self._add_attribute(span, "tool_name", tool_name)
-                self._add_attribute(span, "attempts", attempts)
-                if llm:
-                    self._add_attribute(span, "llm", llm.model)
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
-
-    def tool_usage_error(self, llm: Any):
-        """Records the usage of a tool by an agent."""
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Tool Usage Error")
-                self._add_attribute(
-                    span,
-                    "crewai_version",
-                    pkg_resources.get_distribution("crewai").version,
-                )
-                if llm:
-                    self._add_attribute(span, "llm", llm.model)
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
-
-    def individual_test_result_span(
-        self, crew: Crew, quality: float, exec_time: int, model_name: str
-    ):
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Crew Individual Test Result")
-
-                self._add_attribute(
-                    span,
-                    "crewai_version",
-                    pkg_resources.get_distribution("crewai").version,
-                )
-                self._add_attribute(span, "crew_key", crew.key)
-                self._add_attribute(span, "crew_id", str(crew.id))
-                self._add_attribute(span, "quality", str(quality))
-                self._add_attribute(span, "exec_time", str(exec_time))
-                self._add_attribute(span, "model_name", model_name)
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
-
-    def test_execution_span(
-        self,
-        crew: Crew,
-        iterations: int,
-        inputs: dict[str, Any] | None,
-        model_name: str,
-    ):
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Crew Test Execution")
-
-                self._add_attribute(
-                    span,
-                    "crewai_version",
-                    pkg_resources.get_distribution("crewai").version,
-                )
-                self._add_attribute(span, "crew_key", crew.key)
-                self._add_attribute(span, "crew_id", str(crew.id))
-                self._add_attribute(span, "iterations", str(iterations))
-                self._add_attribute(span, "model_name", model_name)
-
-                if crew.share_crew:
-                    self._add_attribute(
-                        span, "inputs", json.dumps(inputs) if inputs else None
-                    )
-
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
-
-    def deploy_signup_error_span(self):
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Deploy Signup Error")
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
-
-    def start_deployment_span(self, uuid: Optional[str] = None):
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Start Deployment")
-                if uuid:
-                    self._add_attribute(span, "uuid", uuid)
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
-
-    def create_crew_deployment_span(self):
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Create Crew Deployment")
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
-
-    def get_crew_logs_span(self, uuid: Optional[str], log_type: str = "deployment"):
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Get Crew Logs")
-                self._add_attribute(span, "log_type", log_type)
-                if uuid:
-                    self._add_attribute(span, "uuid", uuid)
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
-
-    def remove_crew_span(self, uuid: Optional[str] = None):
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Remove Crew")
-                if uuid:
-                    self._add_attribute(span, "uuid", uuid)
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
-
-    def crew_execution_span(self, crew: Crew, inputs: dict[str, Any] | None):
-        """Records the complete execution of a crew.
-        This is only collected if the user has opted-in to share the crew.
-        """
-        self.crew_creation(crew, inputs)
-
-        if (self.ready) and (crew.share_crew):
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Crew Execution")
-                self._add_attribute(
-                    span,
-                    "crewai_version",
-                    pkg_resources.get_distribution("crewai").version,
-                )
-                self._add_attribute(span, "crew_key", crew.key)
-                self._add_attribute(span, "crew_id", str(crew.id))
-                self._add_attribute(
-                    span, "crew_inputs", json.dumps(inputs) if inputs else None
-                )
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Crew Created")
+            self._add_attribute(
+                span,
+                "crewai_version",
+                pkg_resources.get_distribution("crewai").version,
+            )
+            self._add_attribute(span, "python_version", platform.python_version())
+            self._add_attribute(span, "crew_key", crew.key)
+            self._add_attribute(span, "crew_id", str(crew.id))
+            self._add_attribute(span, "crew_process", crew.process)
+            self._add_attribute(span, "crew_memory", crew.memory)
+            self._add_attribute(span, "crew_number_of_tasks", len(crew.tasks))
+            self._add_attribute(span, "crew_number_of_agents", len(crew.agents))
+            if crew.share_crew:
                 self._add_attribute(
                     span,
                     "crew_agents",
@@ -496,8 +125,15 @@ class Telemetry:
                                 "max_iter": agent.max_iter,
                                 "max_rpm": agent.max_rpm,
                                 "i18n": agent.i18n.prompt_file,
+                                "function_calling_llm": (
+                                    agent.function_calling_llm.model
+                                    if agent.function_calling_llm
+                                    else ""
+                                ),
                                 "llm": agent.llm.model,
                                 "delegation_enabled?": agent.allow_delegation,
+                                "allow_code_execution?": agent.allow_code_execution,
+                                "max_retry_limit": agent.max_retry_limit,
                                 "tools_names": [
                                     tool.name.casefold() for tool in agent.tools or []
                                 ],
@@ -512,12 +148,15 @@ class Telemetry:
                     json.dumps(
                         [
                             {
+                                "key": task.key,
                                 "id": str(task.id),
                                 "description": task.description,
                                 "expected_output": task.expected_output,
                                 "async_execution?": task.async_execution,
                                 "human_input?": task.human_input,
-                                "agent_role": task.agent.role if task.agent else "None",
+                                "agent_role": (
+                                    task.agent.role if task.agent else "None"
+                                ),
                                 "agent_key": task.agent.key if task.agent else None,
                                 "context": (
                                     [task.description for task in task.context]
@@ -532,78 +171,433 @@ class Telemetry:
                         ]
                     ),
                 )
-                return span
-            except Exception:
-                pass
-
-    def end_crew(self, crew, final_string_output):
-        if (self.ready) and (crew.share_crew):
-            try:
+                self._add_attribute(span, "platform", platform.platform())
+                self._add_attribute(span, "platform_release", platform.release())
+                self._add_attribute(span, "platform_system", platform.system())
+                self._add_attribute(span, "platform_version", platform.version())
+                self._add_attribute(span, "cpus", os.cpu_count())
                 self._add_attribute(
-                    crew._execution_span,
-                    "crewai_version",
-                    pkg_resources.get_distribution("crewai").version,
+                    span, "crew_inputs", json.dumps(inputs) if inputs else None
                 )
+            else:
                 self._add_attribute(
-                    crew._execution_span, "crew_output", final_string_output
-                )
-                self._add_attribute(
-                    crew._execution_span,
-                    "crew_tasks_output",
+                    span,
+                    "crew_agents",
                     json.dumps(
                         [
                             {
+                                "key": agent.key,
+                                "id": str(agent.id),
+                                "role": agent.role,
+                                "verbose?": agent.verbose,
+                                "max_iter": agent.max_iter,
+                                "max_rpm": agent.max_rpm,
+                                "function_calling_llm": (
+                                    agent.function_calling_llm.model
+                                    if agent.function_calling_llm
+                                    else ""
+                                ),
+                                "llm": agent.llm.model,
+                                "delegation_enabled?": agent.allow_delegation,
+                                "allow_code_execution?": agent.allow_code_execution,
+                                "max_retry_limit": agent.max_retry_limit,
+                                "tools_names": [
+                                    tool.name.casefold() for tool in agent.tools or []
+                                ],
+                            }
+                            for agent in crew.agents
+                        ]
+                    ),
+                )
+                self._add_attribute(
+                    span,
+                    "crew_tasks",
+                    json.dumps(
+                        [
+                            {
+                                "key": task.key,
                                 "id": str(task.id),
-                                "description": task.description,
-                                "output": task.output.raw_output,
+                                "async_execution?": task.async_execution,
+                                "human_input?": task.human_input,
+                                "agent_role": (
+                                    task.agent.role if task.agent else "None"
+                                ),
+                                "agent_key": task.agent.key if task.agent else None,
+                                "tools_names": [
+                                    tool.name.casefold() for tool in task.tools or []
+                                ],
                             }
                             for task in crew.tasks
                         ]
                     ),
                 )
-                crew._execution_span.set_status(Status(StatusCode.OK))
-                crew._execution_span.end()
-            except Exception:
-                pass
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
+
+    def task_started(self, crew: Crew, task: Task) -> Span | None:
+        """Records task started in a crew."""
+
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+
+            created_span = tracer.start_span("Task Created")
+
+            self._add_attribute(created_span, "crew_key", crew.key)
+            self._add_attribute(created_span, "crew_id", str(crew.id))
+            self._add_attribute(created_span, "task_key", task.key)
+            self._add_attribute(created_span, "task_id", str(task.id))
+
+            if crew.share_crew:
+                self._add_attribute(
+                    created_span, "formatted_description", task.description
+                )
+                self._add_attribute(
+                    created_span, "formatted_expected_output", task.expected_output
+                )
+
+            created_span.set_status(Status(StatusCode.OK))
+            created_span.end()
+
+            span = tracer.start_span("Task Execution")
+
+            self._add_attribute(span, "crew_key", crew.key)
+            self._add_attribute(span, "crew_id", str(crew.id))
+            self._add_attribute(span, "task_key", task.key)
+            self._add_attribute(span, "task_id", str(task.id))
+
+            if crew.share_crew:
+                self._add_attribute(span, "formatted_description", task.description)
+                self._add_attribute(
+                    span, "formatted_expected_output", task.expected_output
+                )
+
+            return span
+
+        return self._safe_telemetry_operation(operation)
+
+    def task_ended(self, span: Span, task: Task, crew: Crew):
+        """Records task execution in a crew."""
+
+        def operation():
+            if crew.share_crew:
+                self._add_attribute(
+                    span,
+                    "task_output",
+                    task.output.raw if task.output else "",
+                )
+
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
+
+    def tool_repeated_usage(self, llm: Any, tool_name: str, attempts: int):
+        """Records the repeated usage 'error' of a tool by an agent."""
+
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Tool Repeated Usage")
+            self._add_attribute(
+                span,
+                "crewai_version",
+                pkg_resources.get_distribution("crewai").version,
+            )
+            self._add_attribute(span, "tool_name", tool_name)
+            self._add_attribute(span, "attempts", attempts)
+            if llm:
+                self._add_attribute(span, "llm", llm.model)
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
+
+    def tool_usage(self, llm: Any, tool_name: str, attempts: int):
+        """Records the usage of a tool by an agent."""
+
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Tool Usage")
+            self._add_attribute(
+                span,
+                "crewai_version",
+                pkg_resources.get_distribution("crewai").version,
+            )
+            self._add_attribute(span, "tool_name", tool_name)
+            self._add_attribute(span, "attempts", attempts)
+            if llm:
+                self._add_attribute(span, "llm", llm.model)
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
+
+    def tool_usage_error(self, llm: Any):
+        """Records the usage of a tool by an agent."""
+
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Tool Usage Error")
+            self._add_attribute(
+                span,
+                "crewai_version",
+                pkg_resources.get_distribution("crewai").version,
+            )
+            if llm:
+                self._add_attribute(span, "llm", llm.model)
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
+
+    def individual_test_result_span(
+        self, crew: Crew, quality: float, exec_time: int, model_name: str
+    ):
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Crew Individual Test Result")
+
+            self._add_attribute(
+                span,
+                "crewai_version",
+                pkg_resources.get_distribution("crewai").version,
+            )
+            self._add_attribute(span, "crew_key", crew.key)
+            self._add_attribute(span, "crew_id", str(crew.id))
+            self._add_attribute(span, "quality", str(quality))
+            self._add_attribute(span, "exec_time", str(exec_time))
+            self._add_attribute(span, "model_name", model_name)
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
+
+    def test_execution_span(
+        self,
+        crew: Crew,
+        iterations: int,
+        inputs: dict[str, Any] | None,
+        model_name: str,
+    ):
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Crew Test Execution")
+
+            self._add_attribute(
+                span,
+                "crewai_version",
+                pkg_resources.get_distribution("crewai").version,
+            )
+            self._add_attribute(span, "crew_key", crew.key)
+            self._add_attribute(span, "crew_id", str(crew.id))
+            self._add_attribute(span, "iterations", str(iterations))
+            self._add_attribute(span, "model_name", model_name)
+
+            if crew.share_crew:
+                self._add_attribute(
+                    span, "inputs", json.dumps(inputs) if inputs else None
+                )
+
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
+
+    def deploy_signup_error_span(self):
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Deploy Signup Error")
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
+
+    def start_deployment_span(self, uuid: Optional[str] = None):
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Start Deployment")
+            if uuid:
+                self._add_attribute(span, "uuid", uuid)
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
+
+    def create_crew_deployment_span(self):
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Create Crew Deployment")
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
+
+    def get_crew_logs_span(self, uuid: Optional[str], log_type: str = "deployment"):
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Get Crew Logs")
+            self._add_attribute(span, "log_type", log_type)
+            if uuid:
+                self._add_attribute(span, "uuid", uuid)
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
+
+    def remove_crew_span(self, uuid: Optional[str] = None):
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Remove Crew")
+            if uuid:
+                self._add_attribute(span, "uuid", uuid)
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
+
+    def crew_execution_span(self, crew: Crew, inputs: dict[str, Any] | None):
+        """Records the complete execution of a crew.
+        This is only collected if the user has opted-in to share the crew.
+        """
+        self.crew_creation(crew, inputs)
+
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Crew Execution")
+            self._add_attribute(
+                span,
+                "crewai_version",
+                pkg_resources.get_distribution("crewai").version,
+            )
+            self._add_attribute(span, "crew_key", crew.key)
+            self._add_attribute(span, "crew_id", str(crew.id))
+            self._add_attribute(
+                span, "crew_inputs", json.dumps(inputs) if inputs else None
+            )
+            self._add_attribute(
+                span,
+                "crew_agents",
+                json.dumps(
+                    [
+                        {
+                            "key": agent.key,
+                            "id": str(agent.id),
+                            "role": agent.role,
+                            "goal": agent.goal,
+                            "backstory": agent.backstory,
+                            "verbose?": agent.verbose,
+                            "max_iter": agent.max_iter,
+                            "max_rpm": agent.max_rpm,
+                            "i18n": agent.i18n.prompt_file,
+                            "llm": agent.llm.model,
+                            "delegation_enabled?": agent.allow_delegation,
+                            "tools_names": [
+                                tool.name.casefold() for tool in agent.tools or []
+                            ],
+                        }
+                        for agent in crew.agents
+                    ]
+                ),
+            )
+            self._add_attribute(
+                span,
+                "crew_tasks",
+                json.dumps(
+                    [
+                        {
+                            "id": str(task.id),
+                            "description": task.description,
+                            "expected_output": task.expected_output,
+                            "async_execution?": task.async_execution,
+                            "human_input?": task.human_input,
+                            "agent_role": task.agent.role if task.agent else "None",
+                            "agent_key": task.agent.key if task.agent else None,
+                            "context": (
+                                [task.description for task in task.context]
+                                if task.context
+                                else None
+                            ),
+                            "tools_names": [
+                                tool.name.casefold() for tool in task.tools or []
+                            ],
+                        }
+                        for task in crew.tasks
+                    ]
+                ),
+            )
+            return span
+
+        if crew.share_crew:
+            return self._safe_telemetry_operation(operation)
+        return None
+
+    def end_crew(self, crew, final_string_output):
+        def operation():
+            self._add_attribute(
+                crew._execution_span,
+                "crewai_version",
+                pkg_resources.get_distribution("crewai").version,
+            )
+            self._add_attribute(
+                crew._execution_span, "crew_output", final_string_output
+            )
+            self._add_attribute(
+                crew._execution_span,
+                "crew_tasks_output",
+                json.dumps(
+                    [
+                        {
+                            "id": str(task.id),
+                            "description": task.description,
+                            "output": task.output.raw_output,
+                        }
+                        for task in crew.tasks
+                    ]
+                ),
+            )
+            crew._execution_span.set_status(Status(StatusCode.OK))
+            crew._execution_span.end()
+
+        if crew.share_crew:
+            self._safe_telemetry_operation(operation)
 
     def _add_attribute(self, span, key, value):
         """Add an attribute to a span."""
-        try:
+
+        def operation():
             return span.set_attribute(key, value)
-        except Exception:
-            pass
+
+        self._safe_telemetry_operation(operation)
 
     def flow_creation_span(self, flow_name: str):
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Flow Creation")
-                self._add_attribute(span, "flow_name", flow_name)
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Flow Creation")
+            self._add_attribute(span, "flow_name", flow_name)
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
 
     def flow_plotting_span(self, flow_name: str, node_names: list[str]):
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Flow Plotting")
-                self._add_attribute(span, "flow_name", flow_name)
-                self._add_attribute(span, "node_names", json.dumps(node_names))
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Flow Plotting")
+            self._add_attribute(span, "flow_name", flow_name)
+            self._add_attribute(span, "node_names", json.dumps(node_names))
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)
 
     def flow_execution_span(self, flow_name: str, node_names: list[str]):
-        if self.ready:
-            try:
-                tracer = trace.get_tracer("crewai.telemetry")
-                span = tracer.start_span("Flow Execution")
-                self._add_attribute(span, "flow_name", flow_name)
-                self._add_attribute(span, "node_names", json.dumps(node_names))
-                span.set_status(Status(StatusCode.OK))
-                span.end()
-            except Exception:
-                pass
+        def operation():
+            tracer = trace.get_tracer("crewai.telemetry")
+            span = tracer.start_span("Flow Execution")
+            self._add_attribute(span, "flow_name", flow_name)
+            self._add_attribute(span, "node_names", json.dumps(node_names))
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
+        self._safe_telemetry_operation(operation)