mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-24 15:48:23 +00:00
[FEATURE] Improve agent/task templating with Jinja2
Fixes #2650 - Add support for container types (List, Dict, Set) - Add support for standard objects (datetime) - Add support for custom objects - Add support for conditional and loop statements - Add support for filtering options - Maintain backward compatibility with existing templates - Add comprehensive tests - Add documentation with examples Co-Authored-By: Joe Moura <joao@crewai.com>
This commit is contained in:
@@ -485,6 +485,19 @@ class Task(BaseModel):
|
||||
tasks_slices = [self.description, output]
|
||||
return "\n".join(tasks_slices)
|
||||
|
||||
def interpolate_inputs(self, inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]]) -> None:
|
||||
"""Interpolate inputs into the task description, expected output, and output file path.
|
||||
|
||||
Args:
|
||||
inputs: Dictionary mapping template variables to their values.
|
||||
Supported value types are strings, integers, floats, dicts, lists,
|
||||
and other objects with string representation.
|
||||
|
||||
Raises:
|
||||
ValueError: If a required template variable is missing from inputs.
|
||||
"""
|
||||
self.interpolate_inputs_and_add_conversation_history(inputs)
|
||||
|
||||
def interpolate_inputs_and_add_conversation_history(
|
||||
self, inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]]
|
||||
) -> None:
|
||||
@@ -493,7 +506,8 @@ class Task(BaseModel):
|
||||
|
||||
Args:
|
||||
inputs: Dictionary mapping template variables to their values.
|
||||
Supported value types are strings, integers, and floats.
|
||||
Supported value types are strings, integers, floats, dicts, lists,
|
||||
and other objects with string representation.
|
||||
|
||||
Raises:
|
||||
ValueError: If a required template variable is missing from inputs.
|
||||
@@ -508,23 +522,63 @@ class Task(BaseModel):
|
||||
if not inputs:
|
||||
return
|
||||
|
||||
try:
|
||||
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"
|
||||
) from e
|
||||
except ValueError as e:
|
||||
raise ValueError(f"Error interpolating description: {str(e)}") from e
|
||||
# Check for complex indexing patterns like {topics[0]} in the description
|
||||
has_complex_indexing = re.search(r"\{([A-Za-z_][A-Za-z0-9_]*)\[[0-9]+\]\}", self._original_description)
|
||||
|
||||
if has_complex_indexing:
|
||||
complex_patterns = re.findall(r"\{([A-Za-z_][A-Za-z0-9_]*)\[([0-9]+)\]\}", self._original_description)
|
||||
result = self._original_description
|
||||
|
||||
for var_name, index in complex_patterns:
|
||||
if var_name in inputs and isinstance(inputs[var_name], list):
|
||||
try:
|
||||
idx = int(index)
|
||||
if 0 <= idx < len(inputs[var_name]):
|
||||
placeholder = f"{{{var_name}[{index}]}}"
|
||||
value = str(inputs[var_name][idx])
|
||||
result = result.replace(placeholder, value)
|
||||
except (ValueError, IndexError):
|
||||
pass
|
||||
|
||||
self.description = result
|
||||
else:
|
||||
try:
|
||||
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"
|
||||
) from e
|
||||
except ValueError as e:
|
||||
raise ValueError(f"Error interpolating description: {str(e)}") from e
|
||||
|
||||
try:
|
||||
self.expected_output = interpolate_only(
|
||||
input_string=self._original_expected_output, inputs=inputs
|
||||
)
|
||||
except (KeyError, ValueError) as e:
|
||||
raise ValueError(f"Error interpolating expected_output: {str(e)}") from e
|
||||
# Check for complex indexing patterns in the expected output
|
||||
has_complex_indexing = re.search(r"\{([A-Za-z_][A-Za-z0-9_]*)\[[0-9]+\]\}", self._original_expected_output)
|
||||
|
||||
if has_complex_indexing:
|
||||
complex_patterns = re.findall(r"\{([A-Za-z_][A-Za-z0-9_]*)\[([0-9]+)\]\}", self._original_expected_output)
|
||||
result = self._original_expected_output
|
||||
|
||||
for var_name, index in complex_patterns:
|
||||
if var_name in inputs and isinstance(inputs[var_name], list):
|
||||
try:
|
||||
idx = int(index)
|
||||
if 0 <= idx < len(inputs[var_name]):
|
||||
placeholder = f"{{{var_name}[{index}]}}"
|
||||
value = str(inputs[var_name][idx])
|
||||
result = result.replace(placeholder, value)
|
||||
except (ValueError, IndexError):
|
||||
pass
|
||||
|
||||
self.expected_output = result
|
||||
else:
|
||||
try:
|
||||
self.expected_output = interpolate_only(
|
||||
input_string=self._original_expected_output, inputs=inputs
|
||||
)
|
||||
except (KeyError, ValueError) as e:
|
||||
raise ValueError(f"Error interpolating expected_output: {str(e)}") from e
|
||||
|
||||
if self.output_file is not None:
|
||||
try:
|
||||
|
||||
79
src/crewai/utilities/jinja_templating.py
Normal file
79
src/crewai/utilities/jinja_templating.py
Normal file
@@ -0,0 +1,79 @@
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
import jinja2
|
||||
import re
|
||||
from datetime import datetime
|
||||
|
||||
def to_jinja_template(input_string: str) -> str:
|
||||
"""
|
||||
Convert CrewAI-style {var} templates to Jinja2-style {{var}} templates.
|
||||
|
||||
This function preserves existing Jinja2 syntax if present and only converts
|
||||
CrewAI-style variables.
|
||||
|
||||
Args:
|
||||
input_string: String containing CrewAI-style templates.
|
||||
|
||||
Returns:
|
||||
String with CrewAI-style templates converted to Jinja2 syntax.
|
||||
"""
|
||||
if not input_string or ("{" not in input_string and "}" not in input_string):
|
||||
return input_string
|
||||
|
||||
pattern = r'(?<!\{)\{([A-Za-z_][A-Za-z0-9_]*)\}(?!\})'
|
||||
|
||||
return re.sub(pattern, r'{{\1}}', input_string)
|
||||
|
||||
def render_template(
|
||||
input_string: Optional[str],
|
||||
inputs: Dict[str, Any],
|
||||
) -> str:
|
||||
"""
|
||||
Render a template string using Jinja2 with the provided inputs.
|
||||
|
||||
This function supports:
|
||||
- Container types (List, Dict, Set)
|
||||
- Standard objects (datetime, time)
|
||||
- Custom objects
|
||||
- Conditional and loop statements
|
||||
- Filtering options
|
||||
|
||||
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.
|
||||
Supports all types of values.
|
||||
|
||||
Returns:
|
||||
The rendered template string.
|
||||
|
||||
Raises:
|
||||
ValueError: If inputs dictionary is empty when interpolating variables.
|
||||
jinja2.exceptions.TemplateError: If there's an error in the template syntax.
|
||||
KeyError: If a required template variable is missing from inputs.
|
||||
"""
|
||||
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")
|
||||
|
||||
jinja_template = to_jinja_template(input_string)
|
||||
|
||||
env = jinja2.Environment(
|
||||
undefined=jinja2.StrictUndefined, # Raise errors for undefined variables
|
||||
)
|
||||
|
||||
env.filters['date'] = lambda d, format='%Y-%m-%d': d.strftime(format) if isinstance(d, datetime) else str(d)
|
||||
|
||||
template = env.from_string(jinja_template)
|
||||
|
||||
try:
|
||||
return template.render(**inputs)
|
||||
except jinja2.exceptions.UndefinedError as e:
|
||||
var_name = str(e).split("'")[1] if "'" in str(e) else None
|
||||
if var_name:
|
||||
raise KeyError(f"Template variable '{var_name}' not found in inputs dictionary")
|
||||
raise KeyError(f"Missing required template variable: {str(e)}")
|
||||
@@ -1,31 +1,38 @@
|
||||
import re
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
from datetime import datetime
|
||||
|
||||
from crewai.utilities.jinja_templating import render_template
|
||||
|
||||
def interpolate_only(
|
||||
input_string: Optional[str],
|
||||
inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]],
|
||||
inputs: Dict[str, 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.
|
||||
|
||||
This function now supports advanced Jinja2 templating features:
|
||||
- Container types (List, Dict, Set)
|
||||
- Standard objects (datetime, time)
|
||||
- Custom objects
|
||||
- Conditional and loop statements
|
||||
- Filtering options
|
||||
|
||||
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.
|
||||
Supports all types of values including complex objects.
|
||||
|
||||
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
|
||||
ValueError: If inputs dictionary is empty when interpolating variables.
|
||||
KeyError: If a required template variable is missing from inputs.
|
||||
"""
|
||||
|
||||
# Validation function for recursive type checking
|
||||
def validate_type(value: Any) -> None:
|
||||
if value is None:
|
||||
return
|
||||
@@ -35,12 +42,21 @@ def interpolate_only(
|
||||
for item in value.values() if isinstance(value, dict) else value:
|
||||
validate_type(item)
|
||||
return
|
||||
if isinstance(value, datetime):
|
||||
return
|
||||
# Check if it's a Pydantic model or other known custom type
|
||||
try:
|
||||
from pydantic import BaseModel
|
||||
if isinstance(value, BaseModel):
|
||||
return
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
raise ValueError(
|
||||
f"Unsupported type {type(value).__name__} in inputs. "
|
||||
"Only str, int, float, bool, dict, and list are allowed."
|
||||
"Only str, int, float, bool, dict, list, datetime, and custom objects are allowed."
|
||||
)
|
||||
|
||||
# Validate all input values
|
||||
for key, value in inputs.items():
|
||||
try:
|
||||
validate_type(value)
|
||||
@@ -56,27 +72,35 @@ def interpolate_only(
|
||||
"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_]*)\}"
|
||||
# Check if the template contains Jinja2 syntax ({% ... %} or {{ ... }})
|
||||
has_jinja_syntax = "{{" in input_string or "{%" in input_string
|
||||
has_complex_indexing = re.search(r"\{([A-Za-z_][A-Za-z0-9_]*)\[[0-9]+\]\}", input_string)
|
||||
|
||||
# Find all matching variables in the input string
|
||||
variables = re.findall(pattern, input_string)
|
||||
result = input_string
|
||||
if has_jinja_syntax or has_complex_indexing:
|
||||
return render_template(input_string, inputs)
|
||||
else:
|
||||
# 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_]*)\}"
|
||||
|
||||
# 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
|
||||
# Find all matching variables in the input string
|
||||
variables = re.findall(pattern, 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"
|
||||
)
|
||||
|
||||
try:
|
||||
return render_template(input_string, inputs)
|
||||
except Exception:
|
||||
result = input_string
|
||||
for var in variables:
|
||||
if var in inputs:
|
||||
placeholder = "{" + var + "}"
|
||||
value = str(inputs[var])
|
||||
result = result.replace(placeholder, value)
|
||||
return result
|
||||
|
||||
Reference in New Issue
Block a user