feat: add console logging for LLM guardrail events (#3105)

* feat: add console logging for memory events

* fix: emit guardrail events in correct order and handle exceptions

* fix: remove unreachable elif conditions in guardrail event listener

* fix: resolve mypy type errors in guardrail event handler

src/crewai/task.py

@@ -97,7 +97,7 @@ class Task(BaseModel):
     )
     context: Union[List["Task"], None, _NotSpecified] = Field(
         description="Other tasks that will have their output used as context for this task.",
-        default=NOT_SPECIFIED
+        default=NOT_SPECIFIED,
     )
     async_execution: Optional[bool] = Field(
         description="Whether the task should be executed asynchronously or not.",
@@ -158,9 +158,7 @@ class Task(BaseModel):
     end_time: Optional[datetime.datetime] = Field(
         default=None, description="End time of the task execution"
     )
-    model_config = {
-        "arbitrary_types_allowed": True
-    }
+    model_config = {"arbitrary_types_allowed": True}
 
     @field_validator("guardrail")
     @classmethod
@@ -204,7 +202,6 @@ class Task(BaseModel):
             # Check return annotation if present, but don't require it
             return_annotation = sig.return_annotation
             if return_annotation != inspect.Signature.empty:
-
                 return_annotation_args = get_args(return_annotation)
                 if not (
                     get_origin(return_annotation) is tuple
@@ -437,7 +434,7 @@ class Task(BaseModel):
                 guardrail_result = process_guardrail(
                     output=task_output,
                     guardrail=self._guardrail,
-                    retry_count=self.retry_count
+                    retry_count=self.retry_count,
                 )
                 if not guardrail_result.success:
                     if self.retry_count >= self.max_retries:
@@ -510,8 +507,6 @@ class Task(BaseModel):
         )
         from crewai.utilities.events.crewai_event_bus import crewai_event_bus
 
-        result = self._guardrail(task_output)
-
         crewai_event_bus.emit(
             self,
             LLMGuardrailStartedEvent(
@@ -519,7 +514,13 @@ class Task(BaseModel):
             ),
         )
 
-        guardrail_result = GuardrailResult.from_tuple(result)
+        try:
+            result = self._guardrail(task_output)
+            guardrail_result = GuardrailResult.from_tuple(result)
+        except Exception as e:
+            guardrail_result = GuardrailResult(
+                success=False, result=None, error=f"Guardrail execution error: {str(e)}"
+            )
 
         crewai_event_bus.emit(
             self,

src/crewai/utilities/events/event_listener.py

@@ -22,6 +22,10 @@ from crewai.utilities.events.llm_events import (
     LLMCallStartedEvent,
     LLMStreamChunkEvent,
 )
+from crewai.utilities.events.llm_guardrail_events import (
+    LLMGuardrailStartedEvent,
+    LLMGuardrailCompletedEvent,
+)
 from crewai.utilities.events.utils.console_formatter import ConsoleFormatter
 
 from .agent_events import (
@@ -370,6 +374,23 @@ class EventListener(BaseEventListener):
             print(content, end="", flush=True)
             self.next_chunk = self.text_stream.tell()
 
+        # ----------- LLM GUARDRAIL EVENTS -----------
+
+        @crewai_event_bus.on(LLMGuardrailStartedEvent)
+        def on_llm_guardrail_started(source, event: LLMGuardrailStartedEvent):
+            guardrail_str = str(event.guardrail)
+            guardrail_name = (
+                guardrail_str[:50] + "..." if len(guardrail_str) > 50 else guardrail_str
+            )
+
+            self.formatter.handle_guardrail_started(guardrail_name, event.retry_count)
+
+        @crewai_event_bus.on(LLMGuardrailCompletedEvent)
+        def on_llm_guardrail_completed(source, event: LLMGuardrailCompletedEvent):
+            self.formatter.handle_guardrail_completed(
+                event.success, event.error, event.retry_count
+            )
+
         @crewai_event_bus.on(CrewTestStartedEvent)
         def on_crew_test_started(source, event: CrewTestStartedEvent):
             cloned_crew = source.copy()

src/crewai/utilities/events/utils/console_formatter.py

@@ -1473,9 +1473,7 @@ class ConsoleFormatter:
                 return None
 
         memory_branch = branch_to_use.add("")
-        self.update_tree_label(
-            memory_branch, "🧠", "Memory Retrieval Started", "blue"
-        )
+        self.update_tree_label(memory_branch, "🧠", "Memory Retrieval Started", "blue")
 
         self.print(tree_to_use)
         self.print()
@@ -1549,7 +1547,6 @@ class ConsoleFormatter:
         if memory_content:
             add_panel()
 
-
     def handle_memory_query_completed(
         self,
         agent_branch: Optional[Tree],
@@ -1616,11 +1613,8 @@ class ConsoleFormatter:
                     sources_branch.add(f"❌ {memory_type} - Error: {error}")
                     break
 
-
     def handle_memory_save_started(
-        self,
-        agent_branch: Optional[Tree],
-        crew_tree: Optional[Tree]
+        self, agent_branch: Optional[Tree], crew_tree: Optional[Tree]
     ) -> None:
         if not self.verbose:
             return None
@@ -1633,7 +1627,7 @@ class ConsoleFormatter:
 
         for child in tree_to_use.children:
             if "Memory Update" in str(child.label):
-               break
+                break
         else:
             memory_branch = tree_to_use.add("")
             self.update_tree_label(
@@ -1700,4 +1694,62 @@ class ConsoleFormatter:
             memory_branch.add(content)
 
         self.print(tree_to_use)
-        self.print()
+        self.print()
+
+    def handle_guardrail_started(
+        self,
+        guardrail_name: str,
+        retry_count: int,
+    ) -> None:
+        """Display guardrail evaluation started status.
+
+        Args:
+            guardrail_name: Name/description of the guardrail being evaluated.
+            retry_count: Zero-based retry count (0 = first attempt).
+        """
+        if not self.verbose:
+            return
+
+        content = self.create_status_content(
+            "Guardrail Evaluation Started",
+            guardrail_name,
+            "yellow",
+            Status="🔄 Evaluating",
+            Attempt=f"{retry_count + 1}",
+        )
+        self.print_panel(content, "🛡️ Guardrail Check", "yellow")
+
+    def handle_guardrail_completed(
+        self,
+        success: bool,
+        error: Optional[str],
+        retry_count: int,
+    ) -> None:
+        """Display guardrail evaluation result.
+
+        Args:
+            success: Whether validation passed.
+            error: Error message if validation failed.
+            retry_count: Zero-based retry count.
+        """
+        if not self.verbose:
+            return
+
+        if success:
+            content = self.create_status_content(
+                "Guardrail Passed",
+                "Validation Successful",
+                "green",
+                Status="✅ Validated",
+                Attempts=f"{retry_count + 1}",
+            )
+            self.print_panel(content, "🛡️ Guardrail Success", "green")
+        else:
+            content = self.create_status_content(
+                "Guardrail Failed",
+                "Validation Error",
+                "red",
+                Error=str(error) if error else "Unknown error",
+                Attempts=f"{retry_count + 1}",
+            )
+            self.print_panel(content, "🛡️ Guardrail Failed", "red")