feat: support to define a task guardrail using YAML config

This commit is contained in:
Lucas Gomide
2025-04-24 12:11:38 -03:00
parent e5d0cc8ac1
commit 50b603d3d2
8 changed files with 2350 additions and 4 deletions

View File

@@ -1,4 +1,5 @@
from typing import List
from unittest.mock import patch
import pytest
@@ -142,8 +143,31 @@ def test_agent_function_calling_llm():
), "agent's function_calling_llm is incorrect"
# VCR could not record the request to localhost from Docker to get it version, so we need to mock the tool.
# TODO: We can remove this mock after some issue such as https://github.com/kevin1024/vcrpy/issues/519 been addressed
@pytest.fixture
def code_interpreter_tool_mock():
with patch(
"crewai_tools.tools.code_interpreter_tool.code_interpreter_tool.CodeInterpreterTool._run",
return_value="(True, 'good result')",
):
yield
def test_task_guardrail(code_interpreter_tool_mock):
crew = InternalCrew()
research_task = crew.research_task()
assert (
research_task.guardrail
== "make sure each bullet contains a minimum of 100 words"
)
reporting_task = crew.reporting_task()
assert reporting_task.guardrail is None
@pytest.mark.vcr(filter_headers=["authorization"])
def test_before_kickoff_modification():
def test_before_kickoff_modification(code_interpreter_tool_mock):
crew = InternalCrew()
inputs = {"topic": "LLMs"}
result = crew.crew().kickoff(inputs=inputs)
@@ -151,7 +175,7 @@ def test_before_kickoff_modification():
@pytest.mark.vcr(filter_headers=["authorization"])
def test_after_kickoff_modification():
def test_after_kickoff_modification(code_interpreter_tool_mock):
crew = InternalCrew()
# Assuming the crew execution returns a dict
result = crew.crew().kickoff({"topic": "LLMs"})
@@ -162,14 +186,14 @@ def test_after_kickoff_modification():
@pytest.mark.vcr(filter_headers=["authorization"])
def test_before_kickoff_with_none_input():
def test_before_kickoff_with_none_input(code_interpreter_tool_mock):
crew = InternalCrew()
crew.crew().kickoff(None)
# Test should pass without raising exceptions
@pytest.mark.vcr(filter_headers=["authorization"])
def test_multiple_before_after_kickoff():
def test_multiple_before_after_kickoff(code_interpreter_tool_mock):
@CrewBase
class MultipleHooksCrew:
agents: List[BaseAgent]