Adding guardrail to a task error (crew defined with decorators)

General
1

In docs I have found only 1 way to add guardrail to a task:

blog_task = Task(
    description="Write a blog post about AI",
    expected_output="A blog post under 200 words",
    agent=blog_agent,
    guardrail=validate_blog_content  # Add the guardrail function
)

But I`m using decorators syntax to define my whole crew:

 @CrewBase
class MyCrew:
    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"

    @agent
....

@task
....

So all works well but this task definition with guardrail:

@task
    def my_task(self) -> Task:
        return Task(
            config=self.tasks_config["my_task"],
            guardrail=validate_text_output,
        )

Is it valid: to pass guardrail function this way using decorators approach?
If yes - why Im getting this error?

[Flow._execute_single_listener] Error in method create_content: 1 validation error for Task
guardrail
  Value error, If return type is annotated, it must be Tuple[bool, Any] [type=value_error, input_value=<function validate_text_o...t at 0x000002AC24F9FEC0>, input_type=function]

Yes, I’m passing the function itself, exactly as stated in documentation, but Tuple is expected.
Why?
Is there a way to correctly pass a guardrail function if using Crew Class with Decorators?

1 Like
2

Hello @Alex_Koniushenko can you share your validate_blog_content() function please. let’s debug that !

3

Yes, sure
Here it is:

def validate_text_output(result: str) → Tuple[bool, Union[Dict[str, Any], str]]:
“”“Validate that the content is valid text and not binary data.”“”

try:
if not result.strip():
return (
False,
{
“error”: “Content is empty”,
“code”: “EMPTY_CONTENT”,
“context”: {“length”: len(result)},
},
)

    binary_signatures = ["%PDF", "\x00"]
    if any(result.startswith(sig) for sig in binary_signatures):
        return (
            False,
            {
                "error": "Content appears to be binary data",
                "code": "BINARY_CONTENT",
                "context": {
                    "content_start": result[:10]
                }, 
            },
        )

    return True, result.strip()

except Exception as e:
    return (
        False,
        {
            "error": f"Unexpected error during validation: {str(e)}",
            "code": "SYSTEM_ERROR",
        },
    )
4

Sorry for late reply :
here one exemple of function to extract Json structure of output :

def validate_json_output2(result: str) -> Tuple[bool, Any]:
	json_pattern = r"\{.*\}"
	match = re.search(json_pattern, result.raw, re.DOTALL)
	if match:
		json_data = match.group()
		print(json_data)
		return (True, json_data)
	else:

		return (False, "Output must be JSoN format")

class WriteDescription(BaseModel):
	Description: str = Field(description="la description détaillée du canapé")


	@task
	def DescriBot_task(self) -> Task:
		return Task(
			config=self.tasks_config['DescriBot_task'],
			output_pydantic = WriteDescription,
			guardrail = validate_json_output2
		)

I had the same error which was fix by changing to “Tuple[bool, Any]”

You also need to change result by result.raw which is equal to the string output

1 Like
5

Modified the existing method to below which works

def validate_blog_content(result: str) -> Tuple[bool, Any]:
"""Validate blog content meets requirements."""
try:
    print("Validating blog content...")
    print("The variable, result is of type:", type(result))
    print("The variable, result description is of type:", type(result.description))
    print("The variable, result raw is of type:", type(result.raw))
    print("The variable, result agent is of type:", type(result.agent))
    
    blog= result.raw

    # Check word count
    word_count = len(blog.split())
    print("The variable, word_count is of type:", type(word_count), "and its value is:", word_count)
    if word_count > 200:
        return (False, "Blog content exceeds 200 words")

    # Additional validation logic here
    return (True, blog.strip())
except Exception as e:
    return (False, "Unexpected error during validation")
1 Like
6

Thank you for your suggestion, it really worked!
One little remark: I think in function definition:

def validate_json_output2(result: str)

should be: 
def validate_json_output2(result: TaskOutput)

as tasks output class is TaskOutput.

And then we can use

result.raw

exactly as in your code.