mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-05-02 07:42:40 +00:00
feat: Add guardrail validation improvements
- Add result/error exclusivity validation in GuardrailResult - Make return type annotations optional in Task guardrail validator - Improve error messages for validation failures Co-Authored-By: Joe Moura <joao@crewai.com>
This commit is contained in:
@@ -1,4 +1,5 @@
|
|||||||
import datetime
|
import datetime
|
||||||
|
import inspect
|
||||||
import json
|
import json
|
||||||
import threading
|
import threading
|
||||||
import uuid
|
import uuid
|
||||||
@@ -124,6 +125,22 @@ class Task(BaseModel):
|
|||||||
description="Current number of retries"
|
description="Current number of retries"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@field_validator("guardrail")
|
||||||
|
@classmethod
|
||||||
|
def validate_guardrail_function(cls, v: Optional[Callable]) -> Optional[Callable]:
|
||||||
|
"""Validate that the guardrail function has the correct signature."""
|
||||||
|
if v is not None:
|
||||||
|
sig = inspect.signature(v)
|
||||||
|
if len(sig.parameters) != 1:
|
||||||
|
raise ValueError("Guardrail function must accept exactly one parameter")
|
||||||
|
|
||||||
|
# Check return annotation if present, but don't require it
|
||||||
|
return_annotation = sig.return_annotation
|
||||||
|
if return_annotation != inspect.Signature.empty:
|
||||||
|
if not (return_annotation == Tuple[bool, Any] or str(return_annotation) == 'Tuple[bool, Any]'):
|
||||||
|
raise ValueError("If return type is annotated, it must be Tuple[bool, Any]")
|
||||||
|
return v
|
||||||
|
|
||||||
_telemetry: Telemetry = PrivateAttr(default_factory=Telemetry)
|
_telemetry: Telemetry = PrivateAttr(default_factory=Telemetry)
|
||||||
_execution_span: Optional[Span] = PrivateAttr(default=None)
|
_execution_span: Optional[Span] = PrivateAttr(default=None)
|
||||||
_original_description: Optional[str] = PrivateAttr(default=None)
|
_original_description: Optional[str] = PrivateAttr(default=None)
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ the way task guardrails return their validation results.
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
from typing import Any, Optional, Tuple, Union
|
from typing import Any, Optional, Tuple, Union
|
||||||
from pydantic import BaseModel
|
from pydantic import BaseModel, field_validator
|
||||||
|
|
||||||
|
|
||||||
class GuardrailResult(BaseModel):
|
class GuardrailResult(BaseModel):
|
||||||
@@ -25,6 +25,17 @@ class GuardrailResult(BaseModel):
|
|||||||
result: Optional[Any] = None
|
result: Optional[Any] = None
|
||||||
error: Optional[str] = None
|
error: Optional[str] = None
|
||||||
|
|
||||||
|
@field_validator("result", "error")
|
||||||
|
@classmethod
|
||||||
|
def validate_result_error_exclusivity(cls, v: Any, info) -> Any:
|
||||||
|
values = info.data
|
||||||
|
if "success" in values:
|
||||||
|
if values["success"] and v and "error" in values and values["error"]:
|
||||||
|
raise ValueError("Cannot have both result and error when success is True")
|
||||||
|
if not values["success"] and v and "result" in values and values["result"]:
|
||||||
|
raise ValueError("Cannot have both result and error when success is False")
|
||||||
|
return v
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_tuple(cls, result: Tuple[bool, Union[Any, str]]) -> "GuardrailResult":
|
def from_tuple(cls, result: Tuple[bool, Union[Any, str]]) -> "GuardrailResult":
|
||||||
"""Create a GuardrailResult from a validation tuple.
|
"""Create a GuardrailResult from a validation tuple.
|
||||||
|
|||||||
Reference in New Issue
Block a user