feat: Add interpolate_only method and improve error handling (#1791)

* Fixed output_file not respecting system path

* Fixed yaml config is not escaped properly for output requirements

* feat: Add interpolate_only method and improve error handling

- Add interpolate_only method for string interpolation while preserving JSON structure
- Add comprehensive test coverage for interpolate_only
- Add proper type annotation for logger using ClassVar
- Improve error handling and documentation for _save_file method

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

* fix: Sort imports to fix lint issues

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

* fix: Reorganize imports using ruff --fix

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

* fix: Consolidate imports and fix formatting

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

* fix: Apply ruff automatic import sorting

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

* fix: Sort imports using ruff --fix

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

---------

Co-authored-by: Frieda (Jingying) Huang <jingyingfhuang@gmail.com>
Co-authored-by: Brandon Hancock (bhancock_ai) <109994880+bhancockio@users.noreply.github.com>
Co-authored-by: Frieda Huang <124417784+frieda-huang@users.noreply.github.com>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Joe Moura <joao@crewai.com>

src/crewai/task.py

@@ -1,13 +1,25 @@
 import datetime
 import inspect
 import json
+import logging
 import threading
 import uuid
 from concurrent.futures import Future
 from copy import copy
 from hashlib import md5
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Type, Union
+from typing import (
+    Any,
+    Callable,
+    ClassVar,
+    Dict,
+    List,
+    Optional,
+    Set,
+    Tuple,
+    Type,
+    Union,
+)
 
 from opentelemetry.trace import Span
 from pydantic import (
@@ -51,6 +63,7 @@ class Task(BaseModel):
     """
 
     __hash__ = object.__hash__  # type: ignore
+    logger: ClassVar[logging.Logger] = logging.getLogger(__name__)
     used_tools: int = 0
     tools_errors: int = 0
     delegations: int = 0
@@ -389,7 +402,18 @@ class Task(BaseModel):
 
         if inputs:
             self.description = self._original_description.format(**inputs)
-            self.expected_output = self._original_expected_output.format(**inputs)
+            self.expected_output = self.interpolate_only(
+                input_string=self._original_expected_output, inputs=inputs
+            )
+
+    def interpolate_only(self, input_string: str, inputs: Dict[str, Any]) -> str:
+        """Interpolate placeholders (e.g., {key}) in a string while leaving JSON untouched."""
+        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)
 
     def increment_tools_errors(self) -> None:
         """Increment the tools errors counter."""
@@ -471,22 +495,33 @@ class Task(BaseModel):
         return OutputFormat.RAW
 
     def _save_file(self, result: Any) -> None:
+        """Save task output to a file.
+        
+        Args:
+            result: The result to save to the file. Can be a dict or any stringifiable object.
+            
+        Raises:
+            ValueError: If output_file is not set
+            RuntimeError: If there is an error writing to the file
+        """
         if self.output_file is None:
             raise ValueError("output_file is not set.")
 
-        resolved_path = Path(self.output_file).expanduser().resolve()
-        directory = resolved_path.parent
+        try:
+            resolved_path = Path(self.output_file).expanduser().resolve()
+            directory = resolved_path.parent
 
-        if not directory.exists():
-            directory.mkdir(parents=True, exist_ok=True)
+            if not directory.exists():
+                directory.mkdir(parents=True, exist_ok=True)
 
-        with resolved_path.open("w", encoding="utf-8") as file:
-            if isinstance(result, dict):
-                import json
-
-                json.dump(result, file, ensure_ascii=False, indent=2)
-            else:
-                file.write(str(result))
+            with resolved_path.open("w", encoding="utf-8") as file:
+                if isinstance(result, dict):
+                    import json
+                    json.dump(result, file, ensure_ascii=False, indent=2)
+                else:
+                    file.write(str(result))
+        except (OSError, IOError) as e:
+            raise RuntimeError(f"Failed to save output file: {e}")
         return None
 
     def __repr__(self):

tests/task_test.py

@@ -736,6 +736,48 @@ def test_interpolate_inputs():
     assert task.expected_output == "Bullet point list of 5 interesting ideas about ML."
 
 
+def test_interpolate_only():
+    """Test the interpolate_only method for various scenarios including JSON structure preservation."""
+    task = Task(
+        description="Unused in this test",
+        expected_output="Unused in this test"
+    )
+    
+    # Test JSON structure preservation
+    json_string = '{"info": "Look at {placeholder}", "nested": {"val": "{nestedVal}"}}'
+    result = task.interpolate_only(
+        input_string=json_string,
+        inputs={"placeholder": "the data", "nestedVal": "something else"}
+    )
+    assert '"info": "Look at the data"' in result
+    assert '"val": "something else"' in result
+    assert "{placeholder}" not in result
+    assert "{nestedVal}" not in result
+    
+    # Test normal string interpolation
+    normal_string = "Hello {name}, welcome to {place}!"
+    result = task.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"}
+    )
+    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"}
+    )
+    assert result == no_placeholders
+
+
 def test_task_output_str_with_pydantic():
     from crewai.tasks.output_format import OutputFormat