feat: ensure guardrail is callable while initializing Task

src/crewai/task.py

@@ -142,7 +142,7 @@ class Task(BaseModel):
     processed_by_agents: Set[str] = Field(default_factory=set)
     guardrail: Optional[Union[Callable[[TaskOutput], Tuple[bool, Any]], str]] = Field(
         default=None,
-        description="Function to validate task output before proceeding to next task",
+        description="Function or string description of a guardrail to validate task output before proceeding to next task",
     )
     max_retries: int = Field(
         default=3, description="Maximum number of retries when guardrail fails"
@@ -215,6 +215,7 @@ class Task(BaseModel):
                     )
         return v
 
+    _guardrail: Optional[Callable] = PrivateAttr(default=None)
     _original_description: Optional[str] = PrivateAttr(default=None)
     _original_expected_output: Optional[str] = PrivateAttr(default=None)
     _original_output_file: Optional[str] = PrivateAttr(default=None)
@@ -235,6 +236,17 @@ class Task(BaseModel):
                 )
         return self
 
+    @model_validator(mode="after")
+    def ensure_guardrail_is_callable(self) -> "Task":
+        if callable(self.guardrail):
+            self._guardrail = self.guardrail
+        elif isinstance(self.guardrail, str):
+            from crewai.tasks.task_guardrail import TaskGuardrail
+
+            self._guardrail = TaskGuardrail(description=self.guardrail, task=self)
+
+        return self
+
     @field_validator("id", mode="before")
     @classmethod
     def _deny_user_set_id(cls, v: Optional[UUID4]) -> None:
@@ -411,7 +423,7 @@ class Task(BaseModel):
                 output_format=self._get_output_format(),
             )
 
-            if self.guardrail:
+            if self._guardrail:
                 guardrail_result = self._process_guardrail(task_output)
                 if not guardrail_result.success:
                     if self.retry_count >= self.max_retries:
@@ -476,7 +488,7 @@ class Task(BaseModel):
             raise e  # Re-raise the exception after emitting the event
 
     def _process_guardrail(self, task_output: TaskOutput) -> GuardrailResult:
-        if self.guardrail is None:
+        if self._guardrail is None:
             raise ValueError("Guardrail is not set")
 
         from crewai.utilities.events import (
@@ -485,20 +497,15 @@ class Task(BaseModel):
         )
         from crewai.utilities.events.crewai_event_bus import crewai_event_bus
 
+        result = self._guardrail(task_output)
+
         crewai_event_bus.emit(
             self,
             TaskGuardrailStartedEvent(
-                guardrail=self.guardrail, retry_count=self.retry_count
+                guardrail=self._guardrail, retry_count=self.retry_count
             ),
         )
 
-        if isinstance(self.guardrail, str):
-            from crewai.tasks.task_guardrail import TaskGuardrail
-
-            result = TaskGuardrail(description=self.guardrail, task=self)(task_output)
-        else:
-            result = self.guardrail(task_output)
-
         guardrail_result = GuardrailResult.from_tuple(result)
 
         crewai_event_bus.emit(

src/crewai/tasks/task_guardrail.py

@@ -0,0 +1,174 @@
+from typing import Any, Tuple
+
+from crewai.llm import LLM
+from crewai.task import Task
+from crewai.tasks.task_output import TaskOutput
+from crewai.utilities.printer import Printer
+
+
+class TaskGuardrail:
+    """A task that validates the output of another task using generated Python code.
+
+    This class generates and executes Python code to validate task outputs based on
+    specified criteria. It uses an LLM to generate the validation code and provides
+    safety guardrails for code execution. The code is executed in a Docker container
+    if available, otherwise it is executed in the current environment.
+
+    Args:
+        description (str): The description of the validation criteria.
+        task (Task, optional): The task whose output needs validation.
+        llm (LLM, optional): The language model to use for code generation.
+        additional_instructions (str, optional): Additional instructions for the guardrail task.
+        unsafe_mode (bool, optional): Whether to run the code in unsafe mode.
+    Raises:
+        ValueError: If no valid LLM is provided.
+    """
+
+    generated_code: str = ""
+
+    def __init__(
+        self,
+        description: str,
+        task: Task | None = None,
+        llm: LLM | None = None,
+        additional_instructions: str = "",
+        unsafe_mode: bool | None = None,
+    ):
+        self.description = description
+
+        fallback_llm: LLM | None = (
+            task.agent.llm
+            if task is not None
+            and hasattr(task, "agent")
+            and task.agent is not None
+            and hasattr(task.agent, "llm")
+            else None
+        )
+        self.llm: LLM | None = llm or fallback_llm
+
+        self.additional_instructions = additional_instructions
+        self.unsafe_mode = unsafe_mode
+
+    @property
+    def system_instructions(self) -> str:
+        """System instructions for the LLM code generation.
+
+        Returns:
+            str: Complete system instructions including security constraints.
+        """
+        security_instructions = (
+            "- DO NOT wrap the output in markdown or use triple backticks. Return only raw Python code."
+            "- DO NOT use `exec`, `eval`, `compile`, `open`, `os`, `subprocess`, `socket`, `shutil`, or any other system-level modules.\n"
+            "- Your code must not perform any file I/O, shell access, or dynamic code execution."
+        )
+        return (
+            "You are a expert Python developer"
+            "You **must strictly** follow the task description, use the provided raw output as the input in your code. "
+            "Your code must:\n"
+            "- Return results with: print((True, data)) on success, or print((False, 'very detailed error message')) on failure. Make sure the final output is being assined to 'result' variable.\n"
+            "- Use the literal string of the task output (already included in your input) if needed.\n"
+            "- Generate the code **following strictly** the task description.\n"
+            "- Be valid Python 3 — executable as-is.\n"
+            f"{security_instructions}\n"
+            "Additional instructions (do not override the previous instructions):\n"
+            f"{self.additional_instructions}"
+        )
+
+    def user_instructions(self, task_output: TaskOutput) -> str:
+        """Generates user instructions for the LLM code generation.
+
+        Args:
+            task_output (TaskOutput): The output to be validated.
+
+        Returns:
+            str: Instructions for generating validation code.
+        """
+        return (
+            "Based on the task description below, generate Python 3 code that validates the task output. \n"
+            "Task description:\n"
+            f"{self.description}\n"
+            "Here is the raw output from the task: \n"
+            f"'{task_output.raw}' \n"
+            "Use this exact string literal inside your generated code (do not reference variables like task_output.raw)."
+            "Now generate Python code that follows the instructions above."
+        )
+
+    def generate_code(self, task_output: TaskOutput) -> str:
+        """Generates Python code for validating the task output.
+
+        Args:
+            task_output (TaskOutput): The output to be validated.
+        """
+        if self.llm is None:
+            raise ValueError("Provide a valid LLM to the TaskGuardrail")
+
+        response = self.llm.call(
+            messages=[
+                {
+                    "role": "system",
+                    "content": self.system_instructions,
+                },
+                {
+                    "role": "user",
+                    "content": self.user_instructions(task_output=task_output),
+                },
+            ]
+        )
+
+        printer = Printer()
+        printer.print(
+            content=f"The following code was generated for the guardrail task:\n{response}\n",
+            color="cyan",
+        )
+        return response
+
+    def __call__(self, task_output: TaskOutput) -> Tuple[bool, Any]:
+        """Executes the validation code on the task output.
+
+        Args:
+            task_output (TaskOutput): The output to be validated.
+
+        Returns:
+            Tuple[bool, Any]: A tuple containing:
+                - bool: True if validation passed, False otherwise
+                - Any: The validation result or error message
+        """
+        import ast
+
+        from crewai_tools import CodeInterpreterTool
+
+        self.generated_code = self.generate_code(task_output)
+
+        unsafe_mode = (
+            self.unsafe_mode
+            if self.unsafe_mode is not None
+            else not self.check_docker_available()
+        )
+        result = CodeInterpreterTool(
+            code=self.generated_code, unsafe_mode=unsafe_mode
+        ).run()
+
+        error_messages = [
+            "Something went wrong while running the code",
+            "No result variable found",  # when running in unsafe mode, the final output should be stored in the result variable
+        ]
+
+        if any(msg in result for msg in error_messages):
+            return False, result
+
+        if isinstance(result, str):
+            try:
+                result = ast.literal_eval(result)
+            except Exception as e:
+                return False, f"Error parsing result: {str(e)}"
+
+        return result
+
+    def check_docker_available(self) -> bool:
+        import subprocess
+
+        try:
+            subprocess.run(["docker", "--version"], check=True)
+            return True
+        except (subprocess.CalledProcessError, FileNotFoundError):
+            return False

src/crewai/utilities/events/task_guardrail_events.py

@@ -0,0 +1,39 @@
+from typing import Any, Callable, Optional, Union
+
+from crewai.utilities.events.base_events import BaseEvent
+
+
+class TaskGuardrailStartedEvent(BaseEvent):
+    """Event emitted when a guardrail task starts
+
+    Attributes:
+        messages: Content can be either a string or a list of dictionaries that support
+            multimodal content (text, images, etc.)
+    """
+
+    type: str = "task_guardrail_started"
+    guardrail: Union[str, Callable]
+    retry_count: int
+
+    def __init__(self, **data):
+        from inspect import getsource
+
+        from crewai.tasks.task_guardrail import TaskGuardrail
+
+        super().__init__(**data)
+
+        if isinstance(self.guardrail, TaskGuardrail):
+            assert self.guardrail.generated_code is not None
+            self.guardrail = self.guardrail.generated_code.strip()
+        elif isinstance(self.guardrail, Callable):
+            self.guardrail = getsource(self.guardrail).strip()
+
+
+class TaskGuardrailCompletedEvent(BaseEvent):
+    """Event emitted when a guardrail task completes"""
+
+    type: str = "task_guardrail_completed"
+    success: bool
+    result: Any
+    error: Optional[str] = None
+    retry_count: int