From 0e4e24d7d12c5921f7a066d48f96e59e9de112d0 Mon Sep 17 00:00:00 2001 From: Brandon Hancock Date: Thu, 20 Mar 2025 10:58:18 -0400 Subject: [PATCH] update interpolation to work with example response types in yaml docs --- src/crewai/agents/agent_builder/base_agent.py | 13 ++- src/crewai/task.py | 76 ++--------------- src/crewai/utilities/formatter.py | 83 ++++++++++++++++++- 3 files changed, 99 insertions(+), 73 deletions(-) diff --git a/src/crewai/agents/agent_builder/base_agent.py b/src/crewai/agents/agent_builder/base_agent.py index 47515d087..622b198ab 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.formatter 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..c3b56ec6c 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 @@ -47,6 +48,7 @@ from crewai.utilities.events import ( TaskStartedEvent, ) from crewai.utilities.events.crewai_event_bus import crewai_event_bus +from crewai.utilities.formatter import interpolate_only from crewai.utilities.i18n import I18N from crewai.utilities.printer import Printer @@ -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..c6c14a8c2 100644 --- a/src/crewai/utilities/formatter.py +++ b/src/crewai/utilities/formatter.py @@ -1,4 +1,5 @@ -from typing import List +import re +from typing import Any, Dict, List, Optional, Union from crewai.task import Task from crewai.tasks.task_output import TaskOutput @@ -18,3 +19,83 @@ def aggregate_raw_outputs_from_tasks(tasks: List[Task]) -> str: task_outputs = [task.output for task in tasks if task.output is not None] return aggregate_raw_outputs_from_task_outputs(task_outputs) + + +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