Supporting no-code Guardrail creation (#2636)

* feat: support to define a guardrail task no-code

* feat: add auto-discovery for Guardrail code execution mode

* feat: handle malformed or invalid response from CodeInterpreterTool

* feat: allow to set unsafe_mode from Guardrail task

* feat: renaming GuardrailTask to TaskGuardrail

* feat: ensure guardrail is callable while initializing Task

* feat: remove Docker availability check from TaskGuardrail

The CodeInterpreterTool already ensures compliance with this requirement.

* refactor: replace if/raise with assert

For this use case `assert` is more appropriate choice

* test: remove useless or duplicated test

* fix: attempt to fix type-checker

* feat: support to define a task guardrail using YAML config

* refactor: simplify TaskGuardrail to use LLM for validation, no code generation

* docs: update TaskGuardrail doc strings

* refactor: drop task paramenter from TaskGuardrail

This parameter was used to get the model from the `task.agent` which is a quite bit redudant since we could propagate the llm directly
This commit is contained in:
Lucas Gomide
2025-04-30 11:47:58 -03:00
committed by GitHub
parent 94b1a6cfb8
commit 015e1a41b2
18 changed files with 4935 additions and 1162 deletions

View File

@@ -0,0 +1,92 @@
from typing import Any, Optional, Tuple
from pydantic import BaseModel, Field
from crewai.agent import Agent, LiteAgentOutput
from crewai.llm import LLM
from crewai.task import Task
from crewai.tasks.task_output import TaskOutput
class TaskGuardrailResult(BaseModel):
valid: bool = Field(
description="Whether the task output complies with the guardrail"
)
feedback: str | None = Field(
description="A feedback about the task output if it is not valid",
default=None,
)
class TaskGuardrail:
"""It validates the output of another task using an LLM.
This class is used to validate the output from a Task based on specified criteria.
It uses an LLM to validate the output and provides a feedback if the output is not valid.
Args:
description (str): The description of the validation criteria.
llm (LLM, optional): The language model to use for code generation.
"""
def __init__(
self,
description: str,
llm: LLM,
):
self.description = description
self.llm: LLM = llm
def _validate_output(self, task_output: TaskOutput) -> LiteAgentOutput:
agent = Agent(
role="Guardrail Agent",
goal="Validate the output of the task",
backstory="You are a expert at validating the output of a task. By providing effective feedback if the output is not valid.",
llm=self.llm,
)
query = f"""
Ensure the following task result complies with the given guardrail.
Task result:
{task_output.raw}
Guardrail:
{self.description}
Your task:
- Confirm if the Task result complies with the guardrail.
- If not, provide clear feedback explaining what is wrong (e.g., by how much it violates the rule, or what specific part fails).
- Focus only on identifying issues — do not propose corrections.
- If the Task result complies with the guardrail, saying that is valid
"""
result = agent.kickoff(query, response_format=TaskGuardrailResult)
return result
def __call__(self, task_output: TaskOutput) -> Tuple[bool, Any]:
"""Validates the output of a task based on specified criteria.
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
"""
try:
result = self._validate_output(task_output)
assert isinstance(
result.pydantic, TaskGuardrailResult
), "The guardrail result is not a valid pydantic model"
if result.pydantic.valid:
return True, task_output.raw
else:
return False, result.pydantic.feedback
except Exception as e:
return False, f"Error while validating the task output: {str(e)}"