diff --git a/src/crewai/agents/agent_builder/base_agent.py b/src/crewai/agents/agent_builder/base_agent.py index 47515d087..4413e0a97 100644 --- a/src/crewai/agents/agent_builder/base_agent.py +++ b/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. diff --git a/src/crewai/task.py b/src/crewai/task.py index 0c063e4f9..10358147c 100644 --- a/src/crewai/task.py +++ b/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 diff --git a/src/crewai/utilities/formatter.py b/src/crewai/utilities/formatter.py index 34da6cc43..19b2a74f9 100644 --- a/src/crewai/utilities/formatter.py +++ b/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] diff --git a/src/crewai/utilities/string_utils.py b/src/crewai/utilities/string_utils.py new file mode 100644 index 000000000..9a1857781 --- /dev/null +++ b/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 diff --git a/tests/task_test.py b/tests/task_test.py index ac25a14f8..67ce99910 100644 --- a/tests/task_test.py +++ b/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 diff --git a/tests/utilities/test_string_utils.py b/tests/utilities/test_string_utils.py new file mode 100644 index 000000000..441aae8c0 --- /dev/null +++ b/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()