diff --git a/examples/guardrail_example_fixed.py b/examples/guardrail_example_fixed.py new file mode 100644 index 000000000..9fc0c9b73 --- /dev/null +++ b/examples/guardrail_example_fixed.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +""" +Fixed example demonstrating proper guardrail usage with PraisonAI Agents. +This addresses the issues reported in issue #875. +""" + +from praisonaiagents import Agent, Task, GuardrailResult, PraisonAIAgents, TaskOutput +from typing import Tuple, Any +import trafilatura + +# Example 1: Using GuardrailResult return type (now supported!) +def validate_length_guardrailresult(output: TaskOutput) -> GuardrailResult: + """Ensure output is between 100-500 characters using GuardrailResult""" + # Extract the raw text from the TaskOutput object + text = output.raw if hasattr(output, 'raw') else str(output) + length = len(text) + + if 100 <= length <= 500: + return GuardrailResult( + success=True, + result=output, # Pass through the original output + error="" + ) + else: + return GuardrailResult( + success=False, + result=None, + error=f"Output must be 100-500 chars, got {length}" + ) + +# Example 2: Using Tuple[bool, Any] return type (original method) +def validate_length_tuple(output: TaskOutput) -> Tuple[bool, Any]: + """Ensure output is between 100-500 characters using tuple""" + text = output.raw if hasattr(output, 'raw') else str(output) + length = len(text) + + if 100 <= length <= 500: + return True, output + else: + return False, f"Output must be 100-500 chars, got {length}" + +# Tool function +def get_url_context(url): + """Fetch and extract content from a URL""" + downloaded = trafilatura.fetch_url(url) + if not downloaded: + return "Sorry, I couldn't fetch the content from that URL." + + extracted = trafilatura.extract( + downloaded, + include_comments=False, + include_links=True, + output_format='json', + with_metadata=True, + url=url + ) + + if not extracted: + return "Sorry, I couldn't extract readable content from that page." + + return extracted # returns JSON string + +# Create agent with FIXED tools parameter (must be a list!) +agent = Agent( + name="Content Summarizer", + role="Content Analysis Expert", + goal="Summarize web content concisely", + instructions="You are a helpful assistant that summarizes web content", + llm="gemini/gemini-2.5-flash-lite-preview-06-17", + self_reflect=False, + verbose=True, + tools=[get_url_context] # FIX: tools must be a list, not a single function +) + +# Create task with GuardrailResult guardrail +task_with_guardrailresult = Task( + name="summarise article with GuardrailResult", + description="get the context of this url: https://blog.google/technology/ai/dolphingemma/ and produce a summary below 500 characters", + agent=agent, + guardrail=validate_length_guardrailresult, # Using GuardrailResult + expected_output="summary of the article below 500 characters", + max_retries=3 # Will retry up to 3 times if guardrail fails +) + +# Alternative: Create task with tuple guardrail +task_with_tuple = Task( + name="summarise article with tuple", + description="get the context of this url: https://blog.google/technology/ai/dolphingemma/ and produce a summary below 500 characters", + agent=agent, + guardrail=validate_length_tuple, # Using Tuple[bool, Any] + expected_output="summary of the article below 500 characters", + max_retries=3 +) + +# Example with string-based LLM guardrail +task_with_llm_guardrail = Task( + name="summarise with LLM guardrail", + description="get the context of this url: https://blog.google/technology/ai/dolphingemma/ and produce a summary", + agent=agent, + guardrail="Ensure the summary is professional, factual, and between 100-500 characters", + expected_output="professional summary of the article" +) + +# Run with GuardrailResult example +print("=== Running with GuardrailResult guardrail ===") +agents_gr = PraisonAIAgents( + agents=[agent], + tasks=[task_with_guardrailresult] +) + +# Uncomment to run: +# result_gr = agents_gr.start() + +# Run with Tuple example +print("\n=== Running with Tuple[bool, Any] guardrail ===") +agents_tuple = PraisonAIAgents( + agents=[agent], + tasks=[task_with_tuple] +) + +# Uncomment to run: +# result_tuple = agents_tuple.start() + +# Run with LLM guardrail example +print("\n=== Running with LLM-based guardrail ===") +agents_llm = PraisonAIAgents( + agents=[agent], + tasks=[task_with_llm_guardrail] +) + +# Uncomment to run: +# result_llm = agents_llm.start() + +print(""" +Key fixes applied: +1. GuardrailResult is now accepted as a valid return type annotation +2. tools parameter must be a list: tools=[get_url_context] not tools=get_url_context +3. Both GuardrailResult and Tuple[bool, Any] return types are supported +4. String-based LLM guardrails are also supported + +The guardrail will automatically retry (up to max_retries times) if validation fails. +""") \ No newline at end of file diff --git a/src/praisonai-agents/CLAUDE.md b/src/praisonai-agents/CLAUDE.md index e48f7237b..8ce9c4a70 100644 --- a/src/praisonai-agents/CLAUDE.md +++ b/src/praisonai-agents/CLAUDE.md @@ -139,10 +139,32 @@ task = Task( #### Task-Level Guardrails ```python from typing import Tuple, Any +from praisonaiagents import GuardrailResult -# Function-based guardrail -def validate_output(task_output: TaskOutput) -> Tuple[bool, Any]: - """Custom validation function.""" +# Function-based guardrail (Option 1: Using GuardrailResult) +def validate_output(task_output: TaskOutput) -> GuardrailResult: + """Custom validation function returning GuardrailResult.""" + if "error" in task_output.raw.lower(): + return GuardrailResult( + success=False, + result=None, + error="Output contains errors" + ) + if len(task_output.raw) < 10: + return GuardrailResult( + success=False, + result=None, + error="Output is too short" + ) + return GuardrailResult( + success=True, + result=task_output, + error="" + ) + +# Function-based guardrail (Option 2: Using Tuple[bool, Any]) +def validate_output_tuple(task_output: TaskOutput) -> Tuple[bool, Any]: + """Custom validation function returning tuple.""" if "error" in task_output.raw.lower(): return False, "Output contains errors" if len(task_output.raw) < 10: @@ -170,7 +192,27 @@ task = Task( #### Agent-Level Guardrails ```python # Agent guardrails apply to ALL outputs from that agent -def validate_professional_tone(task_output: TaskOutput) -> Tuple[bool, Any]: + +# Option 1: Using GuardrailResult +def validate_professional_tone(task_output: TaskOutput) -> GuardrailResult: + """Ensure professional tone in all agent responses.""" + content = task_output.raw.lower() + casual_words = ['yo', 'dude', 'awesome', 'cool'] + for word in casual_words: + if word in content: + return GuardrailResult( + success=False, + result=None, + error=f"Unprofessional language detected: {word}" + ) + return GuardrailResult( + success=True, + result=task_output, + error="" + ) + +# Option 2: Using Tuple[bool, Any] +def validate_professional_tone_tuple(task_output: TaskOutput) -> Tuple[bool, Any]: """Ensure professional tone in all agent responses.""" content = task_output.raw.lower() casual_words = ['yo', 'dude', 'awesome', 'cool'] diff --git a/src/praisonai-agents/praisonaiagents/task/task.py b/src/praisonai-agents/praisonaiagents/task/task.py index db0a79274..38904663a 100644 --- a/src/praisonai-agents/praisonaiagents/task/task.py +++ b/src/praisonai-agents/praisonaiagents/task/task.py @@ -172,8 +172,15 @@ def _setup_guardrail(self): # Check return annotation if present return_annotation = sig.return_annotation if return_annotation != inspect.Signature.empty: + # Import GuardrailResult for checking + from ..guardrails import GuardrailResult + + # Check if it's a GuardrailResult type + is_guardrail_result = return_annotation is GuardrailResult + + # Check for tuple return type return_annotation_args = get_args(return_annotation) - if not ( + is_tuple = ( get_origin(return_annotation) is tuple and len(return_annotation_args) == 2 and return_annotation_args[0] is bool @@ -183,9 +190,11 @@ def _setup_guardrail(self): or return_annotation_args[1] is TaskOutput or return_annotation_args[1] == Union[str, TaskOutput] ) - ): + ) + + if not (is_guardrail_result or is_tuple): raise ValueError( - "If return type is annotated, it must be Tuple[bool, Any]" + "If return type is annotated, it must be GuardrailResult or Tuple[bool, Any]" ) self._guardrail_fn = self.guardrail @@ -447,7 +456,11 @@ def _process_guardrail(self, task_output: TaskOutput): # Call the guardrail function result = self._guardrail_fn(task_output) - # Convert the result to a GuardrailResult + # Check if result is already a GuardrailResult + if isinstance(result, GuardrailResult): + return result + + # Otherwise, convert the tuple result to a GuardrailResult return GuardrailResult.from_tuple(result) except Exception as e: diff --git a/src/praisonai-agents/test_guardrail_return_types.py b/src/praisonai-agents/test_guardrail_return_types.py new file mode 100644 index 000000000..094f6d289 --- /dev/null +++ b/src/praisonai-agents/test_guardrail_return_types.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +""" +Test different guardrail return types to ensure both GuardrailResult +and Tuple[bool, Any] work correctly. This addresses issue #875. +""" + +import sys +import os +from typing import Tuple, Any + +from praisonaiagents import Agent, Task, TaskOutput, GuardrailResult +import inspect + + +def test_guardrail_return_types(): + """Test that both GuardrailResult and Tuple[bool, Any] return types are accepted.""" + print("Testing guardrail return type validation...") + + # Test 1: GuardrailResult return type + def validate_with_guardrailresult(task_output: TaskOutput) -> GuardrailResult: + """Validation function returning GuardrailResult.""" + if len(task_output.raw) < 10: + return GuardrailResult( + success=False, + result=None, + error="Output is too short" + ) + return GuardrailResult( + success=True, + result=task_output, + error="" + ) + + # Test 2: Tuple[bool, Any] return type + def validate_with_tuple(task_output: TaskOutput) -> Tuple[bool, Any]: + """Validation function returning tuple.""" + if len(task_output.raw) < 10: + return False, "Output is too short" + return True, task_output + + # Test 3: No return type annotation + def validate_no_annotation(task_output: TaskOutput): + """Validation function without return type annotation.""" + if len(task_output.raw) < 10: + return False, "Output is too short" + return True, task_output + + # Create test agent + agent = Agent( + name="Test Agent", + role="Tester", + goal="Test guardrails", + backstory="Testing guardrail functionality", + llm="gpt-4o" + ) + + # Test creating tasks with each guardrail type + tests = [ + ("GuardrailResult return type", validate_with_guardrailresult), + ("Tuple[bool, Any] return type", validate_with_tuple), + ("No return type annotation", validate_no_annotation) + ] + + for test_name, guardrail_func in tests: + print(f"\nTesting {test_name}...") + try: + task = Task( + description="Test task", + expected_output="Test output", + agent=agent, + guardrail=guardrail_func + ) + print(f"✓ Task created successfully with {test_name}") + + # Verify guardrail was set up + assert task._guardrail_fn is not None, "Guardrail function not set" + + # Test guardrail processing + test_output = TaskOutput( + description="Test", + raw="This is a test output that is long enough", + agent="Test Agent" + ) + + result = task._process_guardrail(test_output) + assert isinstance(result, GuardrailResult), f"Expected GuardrailResult, got {type(result)}" + assert result.success, "Guardrail validation should pass" + print(f"✓ Guardrail processing works with {test_name}") + + except Exception as e: + print(f"✗ Failed with {test_name}: {e}") + return False + + print("\n✓ All guardrail return type tests passed!") + return True + + +def test_guardrail_result_handling(): + """Test that _process_guardrail correctly handles both return types.""" + print("\nTesting guardrail result handling...") + + # Create test agent and task + agent = Agent( + name="Test Agent", + role="Tester", + goal="Test guardrails", + backstory="Testing guardrail functionality", + llm="gpt-4o" + ) + + test_output = TaskOutput( + description="Test", + raw="Test output", + agent="Test Agent" + ) + + # Test 1: Function returning GuardrailResult + def return_guardrailresult(output: TaskOutput) -> GuardrailResult: + return GuardrailResult(success=True, result=output, error="") + + task1 = Task( + description="Test", + agent=agent, + guardrail=return_guardrailresult + ) + + result1 = task1._process_guardrail(test_output) + assert isinstance(result1, GuardrailResult), "Should return GuardrailResult" + assert result1.success, "Should be successful" + print("✓ GuardrailResult return handled correctly") + + # Test 2: Function returning tuple + def return_tuple(output: TaskOutput) -> Tuple[bool, Any]: + return True, output + + task2 = Task( + description="Test", + agent=agent, + guardrail=return_tuple + ) + + result2 = task2._process_guardrail(test_output) + assert isinstance(result2, GuardrailResult), "Should convert tuple to GuardrailResult" + assert result2.success, "Should be successful" + print("✓ Tuple return converted correctly") + + print("✓ Guardrail result handling tests passed!") + return True + + +if __name__ == "__main__": + print("Running guardrail return type tests for issue #875...") + + success = True + success &= test_guardrail_return_types() + success &= test_guardrail_result_handling() + + if success: + print("\n✅ All tests passed! The fix for issue #875 is working correctly.") + sys.exit(0) + else: + print("\n❌ Some tests failed!") + sys.exit(1) \ No newline at end of file