Implement PR feedback: Add logging, specific exception handling, type hints, and enhanced tests

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

docs/concepts/llms.mdx

@@ -59,7 +59,7 @@ There are three ways to configure LLMs in CrewAI. Choose the method that best fi
         goal: Conduct comprehensive research and analysis
         backstory: A dedicated research professional with years of experience
         verbose: true
-        llm: openai/gpt-4o-mini # your model here 
+        llm: openai/gpt-4o-mini # your model here
         # (see provider configuration examples below for more)
     ```
 
@@ -111,7 +111,7 @@ There are three ways to configure LLMs in CrewAI. Choose the method that best fi
 ## Provider Configuration Examples
 
 
-CrewAI supports a multitude of LLM providers, each offering unique features, authentication methods, and model capabilities. 
+CrewAI supports a multitude of LLM providers, each offering unique features, authentication methods, and model capabilities.
 In this section, you'll find detailed examples that help you select, configure, and optimize the LLM that best fits your project's needs.
 
 <AccordionGroup>
@@ -121,7 +121,7 @@ In this section, you'll find detailed examples that help you select, configure,
     ```toml Code
     # Required
     OPENAI_API_KEY=sk-...
-    
+
     # Optional
     OPENAI_API_BASE=<custom-base-url>
     OPENAI_ORGANIZATION=<your-org-id>
@@ -226,7 +226,7 @@ In this section, you'll find detailed examples that help you select, configure,
     AZURE_API_KEY=<your-api-key>
     AZURE_API_BASE=<your-resource-url>
     AZURE_API_VERSION=<api-version>
-    
+
     # Optional
     AZURE_AD_TOKEN=<your-azure-ad-token>
     AZURE_API_TYPE=<your-azure-api-type>
@@ -289,7 +289,7 @@ In this section, you'll find detailed examples that help you select, configure,
     | Mistral 8x7B Instruct   | Up to 32k tokens     | An MOE LLM that follows instructions, completes requests, and generates creative text. |
 
   </Accordion>
-  
+
   <Accordion title="Amazon SageMaker">
     ```toml Code
     AWS_ACCESS_KEY_ID=<your-access-key>
@@ -474,7 +474,7 @@ In this section, you'll find detailed examples that help you select, configure,
     WATSONX_URL=<your-url>
     WATSONX_APIKEY=<your-apikey>
     WATSONX_PROJECT_ID=<your-project-id>
-    
+
     # Optional
     WATSONX_TOKEN=<your-token>
     WATSONX_DEPLOYMENT_SPACE_ID=<your-space-id>
@@ -491,7 +491,7 @@ In this section, you'll find detailed examples that help you select, configure,
 
   <Accordion title="Ollama (Local LLMs)">
     1. Install Ollama: [ollama.ai](https://ollama.ai/)
-    2. Run a model: `ollama run llama2`
+    2. Run a model: `ollama run llama3`
     3. Configure:
 
     ```python Code
@@ -600,7 +600,7 @@ In this section, you'll find detailed examples that help you select, configure,
     ```toml Code
     OPENROUTER_API_KEY=<your-api-key>
     ```
-    
+
     Example usage in your CrewAI project:
     ```python Code
     llm = LLM(
@@ -723,7 +723,7 @@ Learn how to get the most out of your LLM configuration:
         - Small tasks (up to 4K tokens): Standard models
         - Medium tasks (between 4K-32K): Enhanced models
         - Large tasks (over 32K): Large context models
-        
+
         ```python
         # Configure model with appropriate settings
         llm = LLM(
@@ -760,11 +760,11 @@ Learn how to get the most out of your LLM configuration:
     <Warning>
       Most authentication issues can be resolved by checking API key format and environment variable names.
     </Warning>
-    
+
     ```bash
     # OpenAI
     OPENAI_API_KEY=sk-...
-    
+
     # Anthropic
     ANTHROPIC_API_KEY=sk-ant-...
     ```
@@ -773,11 +773,11 @@ Learn how to get the most out of your LLM configuration:
     <Check>
       Always include the provider prefix in model names
     </Check>
-    
+
     ```python
     # Correct
     llm = LLM(model="openai/gpt-4")
-    
+
     # Incorrect
     llm = LLM(model="gpt-4")
     ```
@@ -786,5 +786,10 @@ Learn how to get the most out of your LLM configuration:
     <Tip>
       Use larger context models for extensive tasks
     </Tip>
+    
+    ```python
+    # Large context model
+    llm = LLM(model="openai/gpt-4o")  # 128K tokens
+    ```
   </Tab>
 </Tabs>

src/crewai/agents/agent_builder/base_agent.py

@@ -25,6 +25,7 @@ from crewai.tools.base_tool import BaseTool, Tool
 from crewai.utilities import I18N, Logger, RPMController
 from crewai.utilities.config import process_config
 from crewai.utilities.converter import Converter
+from crewai.utilities.string_utils import interpolate_only
 
 T = TypeVar("T", bound="BaseAgent")
 
@@ -333,9 +334,15 @@ class BaseAgent(ABC, BaseModel):
             self._original_backstory = self.backstory
 
         if inputs:
-            self.role = self._original_role.format(**inputs)
-            self.goal = self._original_goal.format(**inputs)
-            self.backstory = self._original_backstory.format(**inputs)
+            self.role = interpolate_only(
+                input_string=self._original_role, inputs=inputs
+            )
+            self.goal = interpolate_only(
+                input_string=self._original_goal, inputs=inputs
+            )
+            self.backstory = interpolate_only(
+                input_string=self._original_backstory, inputs=inputs
+            )
 
     def set_cache_handler(self, cache_handler: CacheHandler) -> None:
         """Set the cache handler for the agent.

src/crewai/llm.py

@@ -114,6 +114,60 @@ LLM_CONTEXT_WINDOW_SIZES = {
     "Llama-3.2-11B-Vision-Instruct": 16384,
     "Meta-Llama-3.2-3B-Instruct": 4096,
     "Meta-Llama-3.2-1B-Instruct": 16384,
+    # bedrock
+    "us.amazon.nova-pro-v1:0": 300000,
+    "us.amazon.nova-micro-v1:0": 128000,
+    "us.amazon.nova-lite-v1:0": 300000,
+    "us.anthropic.claude-3-5-sonnet-20240620-v1:0": 200000,
+    "us.anthropic.claude-3-5-haiku-20241022-v1:0": 200000,
+    "us.anthropic.claude-3-5-sonnet-20241022-v2:0": 200000,
+    "us.anthropic.claude-3-7-sonnet-20250219-v1:0": 200000,
+    "us.anthropic.claude-3-sonnet-20240229-v1:0": 200000,
+    "us.anthropic.claude-3-opus-20240229-v1:0": 200000,
+    "us.anthropic.claude-3-haiku-20240307-v1:0": 200000,
+    "us.meta.llama3-2-11b-instruct-v1:0": 128000,
+    "us.meta.llama3-2-3b-instruct-v1:0": 131000,
+    "us.meta.llama3-2-90b-instruct-v1:0": 128000,
+    "us.meta.llama3-2-1b-instruct-v1:0": 131000,
+    "us.meta.llama3-1-8b-instruct-v1:0": 128000,
+    "us.meta.llama3-1-70b-instruct-v1:0": 128000,
+    "us.meta.llama3-3-70b-instruct-v1:0": 128000,
+    "us.meta.llama3-1-405b-instruct-v1:0": 128000,
+    "eu.anthropic.claude-3-5-sonnet-20240620-v1:0": 200000,
+    "eu.anthropic.claude-3-sonnet-20240229-v1:0": 200000,
+    "eu.anthropic.claude-3-haiku-20240307-v1:0": 200000,
+    "eu.meta.llama3-2-3b-instruct-v1:0": 131000,
+    "eu.meta.llama3-2-1b-instruct-v1:0": 131000,
+    "apac.anthropic.claude-3-5-sonnet-20240620-v1:0": 200000,
+    "apac.anthropic.claude-3-5-sonnet-20241022-v2:0": 200000,
+    "apac.anthropic.claude-3-sonnet-20240229-v1:0": 200000,
+    "apac.anthropic.claude-3-haiku-20240307-v1:0": 200000,
+    "amazon.nova-pro-v1:0": 300000,
+    "amazon.nova-micro-v1:0": 128000,
+    "amazon.nova-lite-v1:0": 300000,
+    "anthropic.claude-3-5-sonnet-20240620-v1:0": 200000,
+    "anthropic.claude-3-5-haiku-20241022-v1:0": 200000,
+    "anthropic.claude-3-5-sonnet-20241022-v2:0": 200000,
+    "anthropic.claude-3-7-sonnet-20250219-v1:0": 200000,
+    "anthropic.claude-3-sonnet-20240229-v1:0": 200000,
+    "anthropic.claude-3-opus-20240229-v1:0": 200000,
+    "anthropic.claude-3-haiku-20240307-v1:0": 200000,
+    "anthropic.claude-v2:1": 200000,
+    "anthropic.claude-v2": 100000,
+    "anthropic.claude-instant-v1": 100000,
+    "meta.llama3-1-405b-instruct-v1:0": 128000,
+    "meta.llama3-1-70b-instruct-v1:0": 128000,
+    "meta.llama3-1-8b-instruct-v1:0": 128000,
+    "meta.llama3-70b-instruct-v1:0": 8000,
+    "meta.llama3-8b-instruct-v1:0": 8000,
+    "amazon.titan-text-lite-v1": 4000,
+    "amazon.titan-text-express-v1": 8000,
+    "cohere.command-text-v14": 4000,
+    "ai21.j2-mid-v1": 8191,
+    "ai21.j2-ultra-v1": 8191,
+    "ai21.jamba-instruct-v1:0": 256000,
+    "mistral.mistral-7b-instruct-v0:2": 32000,
+    "mistral.mixtral-8x7b-instruct-v0:1": 32000,
     # mistral
     "mistral-tiny": 32768,
     "mistral-small-latest": 32768,

src/crewai/task.py

@@ -2,6 +2,7 @@ import datetime
 import inspect
 import json
 import logging
+import re
 import threading
 import uuid
 from concurrent.futures import Future
@@ -49,6 +50,7 @@ from crewai.utilities.events import (
 from crewai.utilities.events.crewai_event_bus import crewai_event_bus
 from crewai.utilities.i18n import I18N
 from crewai.utilities.printer import Printer
+from crewai.utilities.string_utils import interpolate_only
 
 
 class Task(BaseModel):
@@ -507,7 +509,9 @@ class Task(BaseModel):
             return
 
         try:
-            self.description = self._original_description.format(**inputs)
+            self.description = interpolate_only(
+                input_string=self._original_description, inputs=inputs
+            )
         except KeyError as e:
             raise ValueError(
                 f"Missing required template variable '{e.args[0]}' in description"
@@ -516,7 +520,7 @@ class Task(BaseModel):
             raise ValueError(f"Error interpolating description: {str(e)}") from e
 
         try:
-            self.expected_output = self.interpolate_only(
+            self.expected_output = interpolate_only(
                 input_string=self._original_expected_output, inputs=inputs
             )
         except (KeyError, ValueError) as e:
@@ -524,7 +528,7 @@ class Task(BaseModel):
 
         if self.output_file is not None:
             try:
-                self.output_file = self.interpolate_only(
+                self.output_file = interpolate_only(
                     input_string=self._original_output_file, inputs=inputs
                 )
             except (KeyError, ValueError) as e:
@@ -555,72 +559,6 @@ class Task(BaseModel):
                 f"\n\n{conversation_instruction}\n\n{conversation_history}"
             )
 
-    def interpolate_only(
-        self,
-        input_string: Optional[str],
-        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]],
-    ) -> str:
-        """Interpolate placeholders (e.g., {key}) in a string while leaving JSON untouched.
-
-        Args:
-            input_string: The string containing template variables to interpolate.
-                         Can be None or empty, in which case an empty string is returned.
-            inputs: Dictionary mapping template variables to their values.
-                   Supported value types are strings, integers, floats, and dicts/lists
-                   containing only these types and other nested dicts/lists.
-
-        Returns:
-            The interpolated string with all template variables replaced with their values.
-            Empty string if input_string is None or empty.
-
-        Raises:
-            ValueError: If a value contains unsupported types
-        """
-
-        # Validation function for recursive type checking
-        def validate_type(value: Any) -> None:
-            if value is None:
-                return
-            if isinstance(value, (str, int, float, bool)):
-                return
-            if isinstance(value, (dict, list)):
-                for item in value.values() if isinstance(value, dict) else value:
-                    validate_type(item)
-                return
-            raise ValueError(
-                f"Unsupported type {type(value).__name__} in inputs. "
-                "Only str, int, float, bool, dict, and list are allowed."
-            )
-
-        # Validate all input values
-        for key, value in inputs.items():
-            try:
-                validate_type(value)
-            except ValueError as e:
-                raise ValueError(f"Invalid value for key '{key}': {str(e)}") from e
-
-        if input_string is None or not input_string:
-            return ""
-        if "{" not in input_string and "}" not in input_string:
-            return input_string
-        if not inputs:
-            raise ValueError(
-                "Inputs dictionary cannot be empty when interpolating variables"
-            )
-        try:
-            escaped_string = input_string.replace("{", "{{").replace("}", "}}")
-
-            for key in inputs.keys():
-                escaped_string = escaped_string.replace(f"{{{{{key}}}}}", f"{{{key}}}")
-
-            return escaped_string.format(**inputs)
-        except KeyError as e:
-            raise KeyError(
-                f"Template variable '{e.args[0]}' not found in inputs dictionary"
-            ) from e
-        except ValueError as e:
-            raise ValueError(f"Error during string interpolation: {str(e)}") from e
-
     def increment_tools_errors(self) -> None:
         """Increment the tools errors counter."""
         self.tools_errors += 1

src/crewai/telemetry/telemetry.py

@@ -2,12 +2,13 @@ from __future__ import annotations
 
 import asyncio
 import json
+import logging
 import os
 import platform
 import warnings
 from contextlib import contextmanager
 from importlib.metadata import version
-from typing import TYPE_CHECKING, Any, Optional
+from typing import TYPE_CHECKING, Any, Optional, Sequence
 
 
 @contextmanager
@@ -22,7 +23,7 @@ from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
     OTLPSpanExporter,  # noqa: E402
 )
 from opentelemetry.sdk.resources import SERVICE_NAME, Resource  # noqa: E402
-from opentelemetry.sdk.trace import TracerProvider  # noqa: E402
+from opentelemetry.sdk.trace import ReadableSpan, TracerProvider  # noqa: E402
 from opentelemetry.sdk.trace.export import BatchSpanProcessor  # noqa: E402
 from opentelemetry.trace import Span, Status, StatusCode  # noqa: E402
 
@@ -31,6 +32,62 @@ if TYPE_CHECKING:
     from crewai.task import Task
 
 
+# A custom BatchSpanProcessor that catches and suppresses all exceptions
+logger = logging.getLogger(__name__)
+
+class SafeBatchSpanProcessor(BatchSpanProcessor):
+    """A wrapper around BatchSpanProcessor that suppresses all exceptions.
+    
+    This processor ensures that telemetry operations do not disrupt user code 
+    by catching and suppressing connection and timeout errors that might occur
+    during span export operations.
+    
+    It logs suppressed errors at the debug level for diagnostic purposes without
+    propagating them to calling code.
+    """
+    
+    def force_flush(self, timeout_millis: Optional[int] = None) -> bool:
+        """Override force_flush to catch and suppress all exceptions.
+        
+        Args:
+            timeout_millis: The maximum amount of time to wait for spans to be exported.
+            
+        Returns:
+            bool: True if the flush was successful, False otherwise.
+        """
+        try:
+            return super().force_flush(timeout_millis)
+        except ConnectionError as e:
+            logger.debug(f"Suppressed telemetry force_flush connection error: {str(e)}")
+            return False
+        except TimeoutError as e:
+            logger.debug(f"Suppressed telemetry force_flush timeout: {str(e)}")
+            return False
+        except Exception as e:
+            logger.debug(f"Unexpected telemetry force_flush error: {str(e)}")
+            return False
+            
+    def export(self, spans: Sequence[ReadableSpan]) -> None:
+        """Override export to catch and suppress all exceptions.
+        
+        Args:
+            spans: The spans to export.
+        """
+        try:
+            if hasattr(super(), 'export'):
+                super().export(spans)
+            else:
+                # Call the exporter directly if super().export doesn't exist
+                self._span_exporter.export(spans)
+        except ConnectionError as e:
+            logger.debug(f"Suppressed telemetry export connection error: {str(e)}")
+        except TimeoutError as e:
+            logger.debug(f"Suppressed telemetry export timeout: {str(e)}")
+        except Exception as e:
+            logger.debug(f"Unexpected telemetry export error: {str(e)}")
+
+
+
 class Telemetry:
     """A class to handle anonymous telemetry for the crewai package.
 
@@ -59,7 +116,7 @@ class Telemetry:
             with suppress_warnings():
                 self.provider = TracerProvider(resource=self.resource)
 
-            processor = BatchSpanProcessor(
+            processor = SafeBatchSpanProcessor(
                 OTLPSpanExporter(
                     endpoint=f"{telemetry_endpoint}/v1/traces",
                     timeout=30,

src/crewai/tools/tool_usage.py

@@ -455,7 +455,7 @@ class ToolUsage:
 
         # Attempt 4: Repair JSON
         try:
-            repaired_input = repair_json(tool_input)
+            repaired_input = repair_json(tool_input, skip_json_loads=True)
             self._printer.print(
                 content=f"Repaired JSON: {repaired_input}", color="blue"
             )

src/crewai/utilities/formatter.py

@@ -1,10 +1,12 @@
-from typing import List
+import re
+from typing import TYPE_CHECKING, List
 
-from crewai.task import Task
-from crewai.tasks.task_output import TaskOutput
+if TYPE_CHECKING:
+    from crewai.task import Task
+    from crewai.tasks.task_output import TaskOutput
 
 
-def aggregate_raw_outputs_from_task_outputs(task_outputs: List[TaskOutput]) -> str:
+def aggregate_raw_outputs_from_task_outputs(task_outputs: List["TaskOutput"]) -> str:
     """Generate string context from the task outputs."""
     dividers = "\n\n----------\n\n"
 
@@ -13,7 +15,7 @@ def aggregate_raw_outputs_from_task_outputs(task_outputs: List[TaskOutput]) -> s
     return context
 
 
-def aggregate_raw_outputs_from_tasks(tasks: List[Task]) -> str:
+def aggregate_raw_outputs_from_tasks(tasks: List["Task"]) -> str:
     """Generate string context from the tasks."""
     task_outputs = [task.output for task in tasks if task.output is not None]
 

src/crewai/utilities/string_utils.py

@@ -0,0 +1,82 @@
+import re
+from typing import Any, Dict, List, Optional, Union
+
+
+def interpolate_only(
+    input_string: Optional[str],
+    inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]],
+) -> str:
+    """Interpolate placeholders (e.g., {key}) in a string while leaving JSON untouched.
+    Only interpolates placeholders that follow the pattern {variable_name} where
+    variable_name starts with a letter/underscore and contains only letters, numbers, and underscores.
+
+    Args:
+        input_string: The string containing template variables to interpolate.
+                     Can be None or empty, in which case an empty string is returned.
+        inputs: Dictionary mapping template variables to their values.
+               Supported value types are strings, integers, floats, and dicts/lists
+               containing only these types and other nested dicts/lists.
+
+    Returns:
+        The interpolated string with all template variables replaced with their values.
+        Empty string if input_string is None or empty.
+
+    Raises:
+        ValueError: If a value contains unsupported types or a template variable is missing
+    """
+
+    # Validation function for recursive type checking
+    def validate_type(value: Any) -> None:
+        if value is None:
+            return
+        if isinstance(value, (str, int, float, bool)):
+            return
+        if isinstance(value, (dict, list)):
+            for item in value.values() if isinstance(value, dict) else value:
+                validate_type(item)
+            return
+        raise ValueError(
+            f"Unsupported type {type(value).__name__} in inputs. "
+            "Only str, int, float, bool, dict, and list are allowed."
+        )
+
+    # Validate all input values
+    for key, value in inputs.items():
+        try:
+            validate_type(value)
+        except ValueError as e:
+            raise ValueError(f"Invalid value for key '{key}': {str(e)}") from e
+
+    if input_string is None or not input_string:
+        return ""
+    if "{" not in input_string and "}" not in input_string:
+        return input_string
+    if not inputs:
+        raise ValueError(
+            "Inputs dictionary cannot be empty when interpolating variables"
+        )
+
+    # The regex pattern to find valid variable placeholders
+    # Matches {variable_name} where variable_name starts with a letter/underscore
+    # and contains only letters, numbers, and underscores
+    pattern = r"\{([A-Za-z_][A-Za-z0-9_]*)\}"
+
+    # Find all matching variables in the input string
+    variables = re.findall(pattern, input_string)
+    result = input_string
+
+    # Check if all variables exist in inputs
+    missing_vars = [var for var in variables if var not in inputs]
+    if missing_vars:
+        raise KeyError(
+            f"Template variable '{missing_vars[0]}' not found in inputs dictionary"
+        )
+
+    # Replace each variable with its value
+    for var in variables:
+        if var in inputs:
+            placeholder = "{" + var + "}"
+            value = str(inputs[var])
+            result = result.replace(placeholder, value)
+
+    return result

tests/task_test.py

@@ -15,6 +15,7 @@ from crewai import Agent, Crew, Process, Task
 from crewai.tasks.conditional_task import ConditionalTask
 from crewai.tasks.task_output import TaskOutput
 from crewai.utilities.converter import Converter
+from crewai.utilities.string_utils import interpolate_only
 
 
 def test_task_tool_reflect_agent_tools():
@@ -822,7 +823,7 @@ def test_interpolate_only():
 
     # Test JSON structure preservation
     json_string = '{"info": "Look at {placeholder}", "nested": {"val": "{nestedVal}"}}'
-    result = task.interpolate_only(
+    result = interpolate_only(
         input_string=json_string,
         inputs={"placeholder": "the data", "nestedVal": "something else"},
     )
@@ -833,20 +834,18 @@ def test_interpolate_only():
 
     # Test normal string interpolation
     normal_string = "Hello {name}, welcome to {place}!"
-    result = task.interpolate_only(
+    result = interpolate_only(
         input_string=normal_string, inputs={"name": "John", "place": "CrewAI"}
     )
     assert result == "Hello John, welcome to CrewAI!"
 
     # Test empty string
-    result = task.interpolate_only(input_string="", inputs={"unused": "value"})
+    result = interpolate_only(input_string="", inputs={"unused": "value"})
     assert result == ""
 
     # Test string with no placeholders
     no_placeholders = "Hello, this is a test"
-    result = task.interpolate_only(
-        input_string=no_placeholders, inputs={"unused": "value"}
-    )
+    result = interpolate_only(input_string=no_placeholders, inputs={"unused": "value"})
     assert result == no_placeholders
 
 
@@ -858,7 +857,7 @@ def test_interpolate_only_with_dict_inside_expected_output():
     )
 
     json_string = '{"questions": {"main_question": "What is the user\'s name?", "secondary_question": "What is the user\'s age?"}}'
-    result = task.interpolate_only(
+    result = interpolate_only(
         input_string=json_string,
         inputs={
             "questions": {
@@ -872,18 +871,16 @@ def test_interpolate_only_with_dict_inside_expected_output():
     assert result == json_string
 
     normal_string = "Hello {name}, welcome to {place}!"
-    result = task.interpolate_only(
+    result = interpolate_only(
         input_string=normal_string, inputs={"name": "John", "place": "CrewAI"}
     )
     assert result == "Hello John, welcome to CrewAI!"
 
-    result = task.interpolate_only(input_string="", inputs={"unused": "value"})
+    result = interpolate_only(input_string="", inputs={"unused": "value"})
     assert result == ""
 
     no_placeholders = "Hello, this is a test"
-    result = task.interpolate_only(
-        input_string=no_placeholders, inputs={"unused": "value"}
-    )
+    result = interpolate_only(input_string=no_placeholders, inputs={"unused": "value"})
     assert result == no_placeholders
 
 
@@ -1085,12 +1082,12 @@ def test_interpolate_with_list_of_strings():
     # Test simple list of strings
     input_str = "Available items: {items}"
     inputs = {"items": ["apple", "banana", "cherry"]}
-    result = task.interpolate_only(input_str, inputs)
+    result = interpolate_only(input_str, inputs)
     assert result == f"Available items: {inputs['items']}"
 
     # Test empty list
     empty_list_input = {"items": []}
-    result = task.interpolate_only(input_str, empty_list_input)
+    result = interpolate_only(input_str, empty_list_input)
     assert result == "Available items: []"
 
 
@@ -1106,7 +1103,7 @@ def test_interpolate_with_list_of_dicts():
             {"name": "Bob", "age": 25, "skills": ["Java", "Cloud"]},
         ]
     }
-    result = task.interpolate_only("{people}", input_data)
+    result = interpolate_only("{people}", input_data)
 
     parsed_result = eval(result)
     assert isinstance(parsed_result, list)
@@ -1138,7 +1135,7 @@ def test_interpolate_with_nested_structures():
             ],
         }
     }
-    result = task.interpolate_only("{company}", input_data)
+    result = interpolate_only("{company}", input_data)
     parsed = eval(result)
 
     assert parsed["name"] == "TechCorp"
@@ -1161,7 +1158,7 @@ def test_interpolate_with_special_characters():
             "empty": "",
         }
     }
-    result = task.interpolate_only("{special_data}", input_data)
+    result = interpolate_only("{special_data}", input_data)
     parsed = eval(result)
 
     assert parsed["quotes"] == """This has "double" and 'single' quotes"""
@@ -1188,7 +1185,7 @@ def test_interpolate_mixed_types():
             },
         }
     }
-    result = task.interpolate_only("{data}", input_data)
+    result = interpolate_only("{data}", input_data)
     parsed = eval(result)
 
     assert parsed["name"] == "Test Dataset"
@@ -1216,7 +1213,7 @@ def test_interpolate_complex_combination():
             },
         ]
     }
-    result = task.interpolate_only("{report}", input_data)
+    result = interpolate_only("{report}", input_data)
     parsed = eval(result)
 
     assert len(parsed) == 2
@@ -1233,7 +1230,7 @@ def test_interpolate_invalid_type_validation():
 
     # Test with invalid top-level type
     with pytest.raises(ValueError) as excinfo:
-        task.interpolate_only("{data}", {"data": set()})  # type: ignore we are purposely testing this failure
+        interpolate_only("{data}", {"data": set()})  # type: ignore we are purposely testing this failure
 
     assert "Unsupported type set" in str(excinfo.value)
 
@@ -1246,7 +1243,7 @@ def test_interpolate_invalid_type_validation():
         }
     }
     with pytest.raises(ValueError) as excinfo:
-        task.interpolate_only("{data}", {"data": invalid_nested})
+        interpolate_only("{data}", {"data": invalid_nested})
     assert "Unsupported type set" in str(excinfo.value)
 
 
@@ -1265,24 +1262,22 @@ def test_interpolate_custom_object_validation():
 
     # Test with custom object at top level
     with pytest.raises(ValueError) as excinfo:
-        task.interpolate_only("{obj}", {"obj": CustomObject(5)})  # type: ignore we are purposely testing this failure
+        interpolate_only("{obj}", {"obj": CustomObject(5)})  # type: ignore we are purposely testing this failure
     assert "Unsupported type CustomObject" in str(excinfo.value)
 
     # Test with nested custom object in dictionary
     with pytest.raises(ValueError) as excinfo:
-        task.interpolate_only(
-            "{data}", {"data": {"valid": 1, "invalid": CustomObject(5)}}
-        )
+        interpolate_only("{data}", {"data": {"valid": 1, "invalid": CustomObject(5)}})
     assert "Unsupported type CustomObject" in str(excinfo.value)
 
     # Test with nested custom object in list
     with pytest.raises(ValueError) as excinfo:
-        task.interpolate_only("{data}", {"data": [1, "valid", CustomObject(5)]})
+        interpolate_only("{data}", {"data": [1, "valid", CustomObject(5)]})
     assert "Unsupported type CustomObject" in str(excinfo.value)
 
     # Test with deeply nested custom object
     with pytest.raises(ValueError) as excinfo:
-        task.interpolate_only(
+        interpolate_only(
             "{data}", {"data": {"level1": {"level2": [{"level3": CustomObject(5)}]}}}
         )
     assert "Unsupported type CustomObject" in str(excinfo.value)
@@ -1306,7 +1301,7 @@ def test_interpolate_valid_complex_types():
     }
 
     # Should not raise any errors
-    result = task.interpolate_only("{data}", {"data": valid_data})
+    result = interpolate_only("{data}", {"data": valid_data})
     parsed = eval(result)
     assert parsed["name"] == "Valid Dataset"
     assert parsed["stats"]["nested"]["deeper"]["b"] == 2.5
@@ -1319,16 +1314,16 @@ def test_interpolate_edge_cases():
     )
 
     # Test empty dict and list
-    assert task.interpolate_only("{}", {"data": {}}) == "{}"
-    assert task.interpolate_only("[]", {"data": []}) == "[]"
+    assert interpolate_only("{}", {"data": {}}) == "{}"
+    assert interpolate_only("[]", {"data": []}) == "[]"
 
     # Test numeric types
-    assert task.interpolate_only("{num}", {"num": 42}) == "42"
-    assert task.interpolate_only("{num}", {"num": 3.14}) == "3.14"
+    assert interpolate_only("{num}", {"num": 42}) == "42"
+    assert interpolate_only("{num}", {"num": 3.14}) == "3.14"
 
     # Test boolean values (valid JSON types)
-    assert task.interpolate_only("{flag}", {"flag": True}) == "True"
-    assert task.interpolate_only("{flag}", {"flag": False}) == "False"
+    assert interpolate_only("{flag}", {"flag": True}) == "True"
+    assert interpolate_only("{flag}", {"flag": False}) == "False"
 
 
 def test_interpolate_valid_types():
@@ -1346,7 +1341,7 @@ def test_interpolate_valid_types():
         "nested": {"flag": True, "empty": None},
     }
 
-    result = task.interpolate_only("{data}", {"data": valid_data})
+    result = interpolate_only("{data}", {"data": valid_data})
     parsed = eval(result)
 
     assert parsed["active"] is True

tests/telemetry_test.py

@@ -0,0 +1,86 @@
+import os
+from unittest.mock import Mock, patch
+
+import pytest
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+
+from crewai.telemetry.telemetry import SafeBatchSpanProcessor, Telemetry
+
+
+class TestTelemetry:
+    """Test suite for Telemetry functionality focusing on error handling and span processing."""
+    
+    def test_safe_batch_span_processor(self):
+        """Test that SafeBatchSpanProcessor properly suppresses exceptions."""
+        # Create a mock exporter that will be used by the processor
+        mock_exporter = Mock()
+        
+        # Create a SafeBatchSpanProcessor with the mock exporter
+        processor = SafeBatchSpanProcessor(mock_exporter)
+        
+        # Test force_flush with an exception
+        with patch.object(BatchSpanProcessor, 'force_flush', side_effect=ConnectionError("Test error")):
+            # This should not raise an exception
+            processor.force_flush()
+        
+        # Test that the processor's export method suppresses exceptions
+        with patch.object(mock_exporter, 'export', side_effect=ConnectionError("Test error")):
+            # This should not raise an exception
+            processor.export([])
+    
+    def test_telemetry_with_connection_error(self):
+        """Test that telemetry connection errors are properly handled in real usage."""
+        # Make sure telemetry is enabled for the test
+        os.environ["OTEL_SDK_DISABLED"] = "false"
+        
+        # Create a telemetry instance
+        telemetry = Telemetry()
+        
+        # Verify telemetry is initialized
+        assert telemetry.ready is True
+        
+        # Test a real telemetry operation
+        # This should not raise an exception even if there are connection issues
+        telemetry.flow_creation_span("test_flow")
+        
+        # Reset environment variables
+        os.environ["OTEL_SDK_DISABLED"] = "true"
+        
+    def test_safe_batch_span_processor_with_timeout(self):
+        """Test that SafeBatchSpanProcessor properly handles timeout errors."""
+        # Create a mock exporter that will be used by the processor
+        mock_exporter = Mock()
+        
+        # Create a SafeBatchSpanProcessor with the mock exporter
+        processor = SafeBatchSpanProcessor(mock_exporter)
+        
+        # Test force_flush with a timeout error
+        with patch.object(BatchSpanProcessor, 'force_flush', side_effect=TimeoutError("Test timeout")):
+            # This should not raise an exception
+            processor.force_flush()
+        
+        # Test that the processor's export method suppresses timeout exceptions
+        with patch.object(mock_exporter, 'export', side_effect=TimeoutError("Test timeout")):
+            # This should not raise an exception
+            processor.export([])
+    
+    def test_safe_batch_span_processor_with_valid_data(self):
+        """Test SafeBatchSpanProcessor normal operation with valid data."""
+        # Create a mock exporter that will be used by the processor
+        mock_exporter = Mock()
+        
+        # Create a SafeBatchSpanProcessor with the mock exporter
+        processor = SafeBatchSpanProcessor(mock_exporter)
+        
+        # Test force_flush with no exception
+        with patch.object(BatchSpanProcessor, 'force_flush', return_value=None):
+            # This should complete normally
+            processor.force_flush()
+            
+        # Mock some valid spans
+        mock_spans = [Mock() for _ in range(3)]
+        
+        # Test that the processor's export method works with valid data
+        with patch.object(mock_exporter, 'export', return_value=None):
+            # This should complete normally
+            processor.export(mock_spans)

tests/utilities/test_string_utils.py

@@ -0,0 +1,187 @@
+from typing import Any, Dict, List, Union
+
+import pytest
+
+from crewai.utilities.string_utils import interpolate_only
+
+
+class TestInterpolateOnly:
+    """Tests for the interpolate_only function in string_utils.py."""
+
+    def test_basic_variable_interpolation(self):
+        """Test basic variable interpolation works correctly."""
+        template = "Hello, {name}! Welcome to {company}."
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "name": "Alice",
+            "company": "CrewAI",
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert result == "Hello, Alice! Welcome to CrewAI."
+
+    def test_multiple_occurrences_of_same_variable(self):
+        """Test that multiple occurrences of the same variable are replaced."""
+        template = "{name} is using {name}'s account."
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "name": "Bob"
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert result == "Bob is using Bob's account."
+
+    def test_json_structure_preservation(self):
+        """Test that JSON structures are preserved and not interpolated incorrectly."""
+        template = """
+        Instructions for {agent}:
+
+        Please return the following object:
+
+        {"name": "person's name", "age": 25, "skills": ["coding", "testing"]}
+        """
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "agent": "DevAgent"
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert "Instructions for DevAgent:" in result
+        assert (
+            '{"name": "person\'s name", "age": 25, "skills": ["coding", "testing"]}'
+            in result
+        )
+
+    def test_complex_nested_json(self):
+        """Test with complex JSON structures containing curly braces."""
+        template = """
+        {agent} needs to process:
+        {
+          "config": {
+            "nested": {
+              "value": 42
+            },
+            "arrays": [1, 2, {"inner": "value"}]
+          }
+        }
+        """
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "agent": "DataProcessor"
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert "DataProcessor needs to process:" in result
+        assert '"nested": {' in result
+        assert '"value": 42' in result
+        assert '[1, 2, {"inner": "value"}]' in result
+
+    def test_missing_variable(self):
+        """Test that an error is raised when a required variable is missing."""
+        template = "Hello, {name}!"
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "not_name": "Alice"
+        }
+
+        with pytest.raises(KeyError) as excinfo:
+            interpolate_only(template, inputs)
+
+        assert "template variable" in str(excinfo.value).lower()
+        assert "name" in str(excinfo.value)
+
+    def test_invalid_input_types(self):
+        """Test that an error is raised with invalid input types."""
+        template = "Hello, {name}!"
+        # Using Any for this test since we're intentionally testing an invalid type
+        inputs: Dict[str, Any] = {"name": object()}  # Object is not a valid input type
+
+        with pytest.raises(ValueError) as excinfo:
+            interpolate_only(template, inputs)
+
+        assert "unsupported type" in str(excinfo.value).lower()
+
+    def test_empty_input_string(self):
+        """Test handling of empty or None input string."""
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "name": "Alice"
+        }
+
+        assert interpolate_only("", inputs) == ""
+        assert interpolate_only(None, inputs) == ""
+
+    def test_no_variables_in_template(self):
+        """Test a template with no variables to replace."""
+        template = "This is a static string with no variables."
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "name": "Alice"
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert result == template
+
+    def test_variable_name_starting_with_underscore(self):
+        """Test variables starting with underscore are replaced correctly."""
+        template = "Variable: {_special_var}"
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "_special_var": "Special Value"
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert result == "Variable: Special Value"
+
+    def test_preserves_non_matching_braces(self):
+        """Test that non-matching braces patterns are preserved."""
+        template = (
+            "This {123} and {!var} should not be replaced but {valid_var} should."
+        )
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "valid_var": "works"
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert (
+            result == "This {123} and {!var} should not be replaced but works should."
+        )
+
+    def test_complex_mixed_scenario(self):
+        """Test a complex scenario with both valid variables and JSON structures."""
+        template = """
+        {agent_name} is working on task {task_id}.
+        
+        Instructions:
+        1. Process the data
+        2. Return results as:
+        
+        {
+          "taskId": "{task_id}",
+          "results": {
+            "processed_by": "agent_name",
+            "status": "complete",
+            "values": [1, 2, 3]
+          }
+        }
+        """
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "agent_name": "AnalyticsAgent",
+            "task_id": "T-12345",
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert "AnalyticsAgent is working on task T-12345" in result
+        assert '"taskId": "T-12345"' in result
+        assert '"processed_by": "agent_name"' in result  # This shouldn't be replaced
+        assert '"values": [1, 2, 3]' in result
+
+    def test_empty_inputs_dictionary(self):
+        """Test that an error is raised with empty inputs dictionary."""
+        template = "Hello, {name}!"
+        inputs: Dict[str, Any] = {}
+
+        with pytest.raises(ValueError) as excinfo:
+            interpolate_only(template, inputs)
+
+        assert "inputs dictionary cannot be empty" in str(excinfo.value).lower()