update interpolation to work with example response types in yaml docs

This commit is contained in:
Brandon Hancock
2025-03-20 10:58:18 -04:00
parent c62e6def62
commit 0e4e24d7d1
3 changed files with 99 additions and 73 deletions

View File

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

View File

@@ -2,6 +2,7 @@ import datetime
import inspect import inspect
import json import json
import logging import logging
import re
import threading import threading
import uuid import uuid
from concurrent.futures import Future from concurrent.futures import Future
@@ -47,6 +48,7 @@ from crewai.utilities.events import (
TaskStartedEvent, TaskStartedEvent,
) )
from crewai.utilities.events.crewai_event_bus import crewai_event_bus 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.i18n import I18N
from crewai.utilities.printer import Printer from crewai.utilities.printer import Printer
@@ -507,7 +509,9 @@ class Task(BaseModel):
return return
try: try:
self.description = self._original_description.format(**inputs) self.description = interpolate_only(
input_string=self._original_description, inputs=inputs
)
except KeyError as e: except KeyError as e:
raise ValueError( raise ValueError(
f"Missing required template variable '{e.args[0]}' in description" 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 raise ValueError(f"Error interpolating description: {str(e)}") from e
try: try:
self.expected_output = self.interpolate_only( self.expected_output = interpolate_only(
input_string=self._original_expected_output, inputs=inputs input_string=self._original_expected_output, inputs=inputs
) )
except (KeyError, ValueError) as e: except (KeyError, ValueError) as e:
@@ -524,7 +528,7 @@ class Task(BaseModel):
if self.output_file is not None: if self.output_file is not None:
try: try:
self.output_file = self.interpolate_only( self.output_file = interpolate_only(
input_string=self._original_output_file, inputs=inputs input_string=self._original_output_file, inputs=inputs
) )
except (KeyError, ValueError) as e: except (KeyError, ValueError) as e:
@@ -555,72 +559,6 @@ class Task(BaseModel):
f"\n\n{conversation_instruction}\n\n{conversation_history}" 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: def increment_tools_errors(self) -> None:
"""Increment the tools errors counter.""" """Increment the tools errors counter."""
self.tools_errors += 1 self.tools_errors += 1

View File

@@ -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.task import Task
from crewai.tasks.task_output import TaskOutput 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] task_outputs = [task.output for task in tasks if task.output is not None]
return aggregate_raw_outputs_from_task_outputs(task_outputs) 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